Iurii Monitors: Site Pulse — Open-Source Website & Domain Health Monitor

# website_monitor.py
def check_pages() -> None:
    """Check all websites from the list"""
    timestamp = datetime.now().strftime("%d.%m.%Y %H:%M:%S")
    messages_to_send = []
 
    with requests.Session() as session:
        def check_with_session(url):
            return check_single_page(url, session)
 
        with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
            future_to_url = {
                executor.submit(check_with_session, url): url
                for url in URLS_TO_CHECK
            }
 
            for future in concurrent.futures.as_completed(future_to_url):
                message, is_success, is_warning, is_error = future.result()
 
                if (is_success and NOTIFY_SUCCESS) or \
                   (is_warning and NOTIFY_WARNING) or \
                   (is_error and NOTIFY_ERROR):
                    messages_to_send.append(message)
 
    if messages_to_send:
        message = f"📅 {timestamp}\n\n" + "\n\n".join(messages_to_send)
        send_telegram_message(message)

# website_monitor.py
def check_single_page(url: str, session: requests.Session) -> tuple:
    def try_request(headers, device_type):
        start_time = time.time()
        response = session.get(
            url,
            timeout=(CONNECT_TIMEOUT, READ_TIMEOUT),
            headers=headers
        )
        content_length = len(response.content)
        elapsed_time = time.time() - start_time
 
        if response.status_code == 200 and content_length >= MIN_RESPONSE_SIZE:
            return response, content_length, elapsed_time, None
        else:
            if response.status_code != 200:
                raise Exception(f"Response code: {response.status_code}")
            else:
                raise Exception(f"Response size too small: {content_length} bytes")
 
    # Desktop check with retry
    try:
        try:
            response, content_length, elapsed_time, _ = try_request(
                DESKTOP_HEADERS, "Desktop"
            )
        except Exception:
            time.sleep(5)
            response, content_length, elapsed_time, _ = try_request(
                DESKTOP_HEADERS, "Desktop"
            )

# website_monitor.py
def check_domain(domain: str) -> tuple[str, bool]:
    """Check domain registration expiry and nameservers"""
    w = whois.whois(domain)
 
    if isinstance(w.expiration_date, list):
        expiry_date = min(w.expiration_date)
    else:
        expiry_date = w.expiration_date
 
    days_until_expiry = (expiry_date - datetime.now()).days
 
    domain_ns = [ns.lower() for ns in w.name_servers]
    our_ns = [ns.lower() for ns in OUR_NAMESERVERS]
    using_our_ns = any(ns in domain_ns for ns in our_ns)
 
    messages = []
    has_warning = False
 
    if days_until_expiry <= EXPIRY_WARNING_DAYS:
        messages.append(f"⚠️ Domain will expire in {days_until_expiry} days")
        has_warning = True
    if not using_our_ns:
        messages.append(f"⚠️ Domain is using external nameservers")
        has_warning = True