#!/usr/bin/env python3
"""
JonasSearch Crawler Gamma 0.1

PostgreSQL-based parallel crawler.

Features:
- PostgreSQL queue with row-level locking (FOR UPDATE SKIP LOCKED)
- Multiple crawler workers can safely share one queue
- Stores title, headings, cleaned text and links
- Search vector is maintained by the PostgreSQL trigger
- URL deduplication via UNIQUE(url)
- Priority queue
- Per-host crawl delay
- robots.txt support with in-memory cache
- Graceful shutdown
- Basic crawler statistics

Expected PostgreSQL schema:
    pages
    crawl_queue
    crawler_stats

Connection defaults can be overridden with environment variables:
    JONASSEARCH_DB_HOST
    JONASSEARCH_DB_PORT
    JONASSEARCH_DB_NAME
    JONASSEARCH_DB_USER
    JONASSEARCH_DB_PASSWORD

Example:
    python3 crawler.py
"""

import os
import re
import signal
import time
import logging
import multiprocessing as mp
from collections import defaultdict
from html import unescape
from pathlib import Path
from urllib.parse import urljoin, urlparse, urldefrag
from urllib.robotparser import RobotFileParser
from dotenv import load_dotenv

import psycopg2
from psycopg2.extras import execute_values
from urllib.request import Request, urlopen
from urllib.error import HTTPError, URLError

try:
    from bs4 import BeautifulSoup
except ImportError:
    BeautifulSoup = None

# .env zuerst aus dem Arbeitsverzeichnis laden (Standard), zusätzlich aus
# dem Projektordner – so findet der Prozess die Konfiguration auch dann,
# wenn er von woanders gestartet wird (z. B. via systemd).
load_dotenv()
load_dotenv(Path(__file__).resolve().parent / ".env")


# ============================================================
# CONFIG
# ============================================================

DB_HOST = os.getenv("JONASSEARCH_DB_HOST", "127.0.0.1")
DB_PORT = int(os.getenv("JONASSEARCH_DB_PORT", "25432"))
DB_NAME = os.getenv("JONASSEARCH_DB_NAME", "jonassearch")
DB_USER = os.getenv("JONASSEARCH_DB_USER", "jonassearch")
DB_PASSWORD = os.getenv("JONASSEARCH_DB_PASSWORD", "")

START_URL = os.getenv("JONASSEARCH_START_URL", "https://github.com/")

WORKERS = int(os.getenv("JONASSEARCH_WORKERS", "3"))

# Minimum delay between requests to the same host, per worker.
# robots.txt can specify a larger crawl delay.
DEFAULT_DELAY = float(os.getenv("JONASSEARCH_DELAY", "3.0"))

ROBOTS_CACHE_SECONDS = int(
    os.getenv("JONASSEARCH_ROBOTS_CACHE_SECONDS", str(24 * 60 * 60))
)

REQUEST_TIMEOUT = float(os.getenv("JONASSEARCH_TIMEOUT", "20"))

MAX_CONTENT_LENGTH = int(
    os.getenv("JONASSEARCH_MAX_CONTENT_LENGTH", str(5 * 1024 * 1024))
)

USER_AGENT = os.getenv(
    "JONASSEARCH_USER_AGENT",
    "JonasSearchBot/0.1 (+https://jonassearch.local/bot)"
)

# Only crawl normal HTTP(S) pages.
ALLOWED_SCHEMES = {"http", "https"}

# Ignore obvious non-HTML resources.
BLOCKED_EXTENSIONS = {
    ".7z", ".avi", ".bin", ".bmp", ".bz2", ".class", ".csv", ".doc",
    ".docx", ".gif", ".gz", ".ico", ".iso", ".jpeg", ".jpg", ".m4a",
    ".mkv", ".mov", ".mp3", ".mp4", ".mpeg", ".pdf", ".png", ".ppt",
    ".pptx", ".rar", ".svg", ".tar", ".tgz", ".tif", ".tiff", ".webm",
    ".webp", ".xls", ".xlsx", ".xml", ".zip"
}

# Do not follow these URL schemes.
BLOCKED_PREFIXES = (
    "mailto:",
    "javascript:",
    "tel:",
    "data:",
    "ftp:",
    "file:",
)

# Keep generated pages reasonably useful for search.
MAX_TITLE_LENGTH = 1000
MAX_HEADINGS = 200
MAX_HEADING_LENGTH = 1000
MAX_TEXT_LENGTH = 2_000_000
MAX_LINKS = 10_000


# ============================================================
# LOGGING / SHUTDOWN
# ============================================================

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(processName)s] %(levelname)s %(message)s",
    datefmt="%H:%M:%S",
)

STOP = mp.Event()


def handle_signal(signum, _frame):
    logging.info("Shutdown signal received.")
    STOP.set()


signal.signal(signal.SIGINT, handle_signal)
signal.signal(signal.SIGTERM, handle_signal)


# ============================================================
# DATABASE
# ============================================================

def db_connect():
    return psycopg2.connect(
        host=DB_HOST,
        port=DB_PORT,
        dbname=DB_NAME,
        user=DB_USER,
        password=DB_PASSWORD,
        connect_timeout=10,
        application_name="jonassearch-crawler",
    )


def ensure_stats_table(conn):
    with conn.cursor() as cur:
        cur.execute("""
            CREATE TABLE IF NOT EXISTS crawler_stats (
                id BIGSERIAL PRIMARY KEY,
                recorded_at TIMESTAMPTZ NOT NULL DEFAULT now(),
                crawled INTEGER NOT NULL DEFAULT 0,
                queued INTEGER NOT NULL DEFAULT 0,
                failed INTEGER NOT NULL DEFAULT 0
            )
        """)
    conn.commit()


def write_stats(conn, crawled, queued, failed):
    with conn.cursor() as cur:
        cur.execute("""
            INSERT INTO crawler_stats (crawled, queued, failed)
            VALUES (%s, %s, %s)
        """, (crawled, queued, failed))
    conn.commit()


def get_next_queue_item(conn):
    """
    Atomically claim one pending URL.

    SKIP LOCKED is the important part: several crawler processes can
    consume the same PostgreSQL queue without waiting for each other.
    """
    with conn.cursor() as cur:
        cur.execute("""
            WITH next_item AS (
                SELECT id
                FROM crawl_queue
                WHERE status = 'pending'
                ORDER BY priority DESC, discovered_at ASC, id ASC
                FOR UPDATE SKIP LOCKED
                LIMIT 1
            )
            UPDATE crawl_queue q
            SET status = 'processing',
                attempts = attempts + 1,
                last_attempt_at = now()
            FROM next_item
            WHERE q.id = next_item.id
            RETURNING q.id, q.url, q.source_url
        """)
        row = cur.fetchone()
    conn.commit()
    return row


def finish_queue_item(conn, queue_id, success=True):
    with conn.cursor() as cur:
        cur.execute("""
            UPDATE crawl_queue
            SET status = %s
            WHERE id = %s
        """, ("done" if success else "failed", queue_id))
    conn.commit()


def requeue_stale_items(conn, older_than_seconds=3600):
    """
    Recover URLs left in 'processing' after a crash.
    """
    with conn.cursor() as cur:
        cur.execute("""
            UPDATE crawl_queue
            SET status = 'pending'
            WHERE status = 'processing'
              AND last_attempt_at < now() - (%s * interval '1 second')
        """, (older_than_seconds,))
        changed = cur.rowcount
    conn.commit()
    if changed:
        logging.info("Requeued %d stale processing URLs.", changed)


def queue_urls(conn, links, source_url):
    if not links:
        return 0

    # Link URLs also come from the web and can carry NUL bytes.
    safe_links = clean_text_list(links)
    safe_source = clean_text(source_url)

    rows = [(url, safe_source, 0, "pending") for url in safe_links]

    with conn.cursor() as cur:
        execute_values(
            cur,
            """
            INSERT INTO crawl_queue (url, source_url, priority, status)
            VALUES %s
            ON CONFLICT (url) DO NOTHING
            """,
            rows,
            template="(%s, %s, %s, %s)",
            page_size=500,
        )
        inserted = cur.rowcount

    conn.commit()
    return inserted


# ============================================================
# TEXT SANITIZATION
# ============================================================

def clean_text(value):
    """
    Remove NUL characters from a string.

    Web content can contain a literal "\x00" anywhere in the extracted
    text. PostgreSQL rejects NUL bytes inside ordinary text columns, so a
    single dirty string would abort the INSERT and crash the worker.
    Sanitizing centrally before every DB write keeps the crawl running.
    """
    if isinstance(value, str):
        return value.replace("\x00", "")
    return value


def clean_text_list(values):
    """Sanitize every string of a list (headings, links, ...)."""
    return [clean_text(v) for v in values]


# ============================================================
# URL NORMALIZATION
# ============================================================

def normalize_url(url, base_url=None):
    if not url:
        return None

    url = unescape(url).strip()

    if base_url:
        url = urljoin(base_url, url)

    if any(url.lower().startswith(prefix) for prefix in BLOCKED_PREFIXES):
        return None

    url, _fragment = urldefrag(url)

    parsed = urlparse(url)

    if parsed.scheme.lower() not in ALLOWED_SCHEMES:
        return None

    if not parsed.netloc:
        return None

    # Lowercase scheme and hostname.
    scheme = parsed.scheme.lower()
    hostname = (parsed.hostname or "").lower()

    if not hostname:
        return None

    # Drop default ports.
    port = parsed.port
    if (scheme == "http" and port == 80) or (scheme == "https" and port == 443):
        port = None

    if port:
        netloc = f"{hostname}:{port}"
    else:
        netloc = hostname

    # Preserve path/query, but remove empty query marker.
    path = parsed.path or "/"
    query = parsed.query

    normalized = f"{scheme}://{netloc}{path}"
    if query:
        normalized += "?" + query

    # Avoid obvious files.
    path_lower = path.lower()
    for ext in BLOCKED_EXTENSIONS:
        if path_lower.endswith(ext):
            return None

    return normalized


# ============================================================
# ROBOTS
# ============================================================

class RobotsCache:
    def __init__(self):
        self.cache = {}
        self.last_checked = {}

    def allowed(self, url):
        parsed = urlparse(url)
        host = parsed.netloc.lower()

        now = time.monotonic()

        if (
            host not in self.cache
            or now - self.last_checked.get(host, 0) > ROBOTS_CACHE_SECONDS
        ):
            self._load(host, parsed.scheme)

        rp = self.cache[host]
        try:
            return rp.can_fetch(USER_AGENT, url)
        except Exception:
            return True

    def delay(self, url):
        parsed = urlparse(url)
        host = parsed.netloc.lower()

        rp = self.cache.get(host)
        if not rp:
            return DEFAULT_DELAY

        # RobotFileParser does not expose Crawl-delay consistently across
        # all Python versions, so use it when available.
        try:
            delay = rp.crawl_delay(USER_AGENT)
            if delay is None:
                delay = rp.crawl_delay("*")
            if delay is not None:
                return max(DEFAULT_DELAY, float(delay))
        except Exception:
            pass

        return DEFAULT_DELAY

    def _load(self, host, scheme):
        robots_url = f"{scheme}://{host}/robots.txt"

        rp = RobotFileParser()
        rp.set_url(robots_url)

        try:
            req = Request(
                robots_url,
                headers={"User-Agent": USER_AGENT}
            )
            with urlopen(req, timeout=REQUEST_TIMEOUT) as response:
                data = response.read(512 * 1024).decode(
                    "utf-8", errors="replace"
                )
            rp.parse(data.splitlines())
        except Exception:
            # If robots.txt cannot be retrieved, default to allowing the URL.
            # This is deliberately conservative with respect to availability.
            rp.parse([])

        self.cache[host] = rp
        self.last_checked[host] = time.monotonic()


# ============================================================
# HTTP
# ============================================================

class RateLimiter:
    def __init__(self):
        self.last_request = defaultdict(float)

    def wait(self, url, delay):
        host = urlparse(url).netloc.lower()
        now = time.monotonic()
        wait_for = delay - (now - self.last_request[host])

        if wait_for > 0:
            logging.info("[RATE] %s: %.1fs", host, wait_for)
            time.sleep(wait_for)

        self.last_request[host] = time.monotonic()


def fetch(url, limiter, robots):
    parsed = urlparse(url)

    if not robots.allowed(url):
        return None, None, "robots"

    delay = robots.delay(url)
    limiter.wait(url, delay)

    req = Request(
        url,
        headers={
            "User-Agent": USER_AGENT,
            "Accept": (
                "text/html,application/xhtml+xml,"
                "application/xml;q=0.9,*/*;q=0.1"
            ),
            "Accept-Language": "en,de;q=0.8",
        },
    )

    try:
        with urlopen(req, timeout=REQUEST_TIMEOUT) as response:
            status = getattr(response, "status", 200)
            content_type = response.headers.get("Content-Type", "")
            content_length = response.headers.get("Content-Length")

            if content_length:
                try:
                    if int(content_length) > MAX_CONTENT_LENGTH:
                        return status, None, "too_large"
                except ValueError:
                    pass

            # We only parse HTML/XHTML.
            ctype = content_type.lower()
            if ctype and not (
                "text/html" in ctype
                or "application/xhtml+xml" in ctype
            ):
                return status, None, "not_html"

            data = response.read(MAX_CONTENT_LENGTH + 1)

            if len(data) > MAX_CONTENT_LENGTH:
                return status, None, "too_large"

            charset = response.headers.get_content_charset() or "utf-8"
            text = data.decode(charset, errors="replace")

            return status, text, None

    except HTTPError as e:
        return e.code, None, f"http_{e.code}"
    except (URLError, TimeoutError, OSError) as e:
        return None, None, f"network:{e}"
    except Exception as e:
        return None, None, f"error:{e}"


# ============================================================
# HTML PARSING
# ============================================================

def clean_text_bs4(soup):
    for tag in soup([
        "script", "style", "noscript", "template",
        "svg", "canvas", "iframe", "object",
        "embed", "form"
    ]):
        tag.decompose()

    # Remove navigation/boilerplate-ish elements where possible.
    for tag in soup.find_all([
        "nav", "footer", "header", "aside"
    ]):
        tag.decompose()

    text = soup.get_text(" ", strip=True)

    # Normalize whitespace.
    text = re.sub(r"\s+", " ", text)
    return text[:MAX_TEXT_LENGTH]


def parse_html(html, base_url):
    if BeautifulSoup is None:
        raise RuntimeError(
            "BeautifulSoup fehlt. Installiere python3-bs4."
        )

    soup = BeautifulSoup(html, "html.parser")

    title = ""
    if soup.title:
        title = soup.title.get_text(" ", strip=True)

    title = re.sub(r"\s+", " ", title)[:MAX_TITLE_LENGTH]

    headings = []
    for tag in soup.find_all(re.compile(r"^h[1-6]$")):
        value = re.sub(r"\s+", " ", tag.get_text(" ", strip=True))
        if value:
            headings.append(value[:MAX_HEADING_LENGTH])
        if len(headings) >= MAX_HEADINGS:
            break

    links = []
    seen = set()

    for tag in soup.find_all("a", href=True):
        normalized = normalize_url(tag.get("href"), base_url)
        if not normalized:
            continue

        if normalized in seen:
            continue

        seen.add(normalized)
        links.append(normalized)

        if len(links) >= MAX_LINKS:
            break

    content = clean_text_bs4(soup)

    domain = urlparse(base_url).hostname or ""
    domain = domain.lower()

    # TLD is intentionally simple here; the DB can later be extended with
    # Public Suffix List handling if needed.
    tld = ""
    if "." in domain:
        tld = "." + domain.rsplit(".", 1)[-1]

    return {
        "title": title,
        "headings": headings,
        "content": content,
        "links": links,
        "domain": domain,
        "tld": tld,
    }


# ============================================================
# PAGE STORAGE
# ============================================================

def save_page(conn, url, source_url, status, content_type, parsed, crawl_ms):
    # Sanitize everything that came from the web before it reaches
    # PostgreSQL. NUL bytes are legal in HTML but illegal in text columns.
    url = clean_text(url)
    source_url = clean_text(source_url)
    content_type = clean_text(content_type)

    safe_parsed = {
        "title": clean_text(parsed["title"]),
        "headings": clean_text_list(parsed["headings"]),
        "content": clean_text(parsed["content"]),
        "links": clean_text_list(parsed["links"]),
        "domain": clean_text(parsed["domain"]),
        "tld": clean_text(parsed["tld"]),
    }

    with conn.cursor() as cur:
        cur.execute("""
            INSERT INTO pages (
                url,
                title,
                headings,
                content,
                links,
                domain,
                tld,
                source_url,
                http_status,
                content_type,
                discovered_at,
                crawled_at,
                crawl_time_ms
            )
            VALUES (
                %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
                COALESCE(
                    (SELECT discovered_at
                     FROM crawl_queue
                     WHERE url = %s),
                    now()
                ),
                now(),
                %s
            )
            ON CONFLICT (url) DO UPDATE SET
                title = EXCLUDED.title,
                headings = EXCLUDED.headings,
                content = EXCLUDED.content,
                links = EXCLUDED.links,
                domain = EXCLUDED.domain,
                tld = EXCLUDED.tld,
                source_url = EXCLUDED.source_url,
                http_status = EXCLUDED.http_status,
                content_type = EXCLUDED.content_type,
                crawled_at = now(),
                crawl_time_ms = EXCLUDED.crawl_time_ms
        """, (
            url,
            safe_parsed["title"],
            safe_parsed["headings"],
            safe_parsed["content"],
            safe_parsed["links"],
            safe_parsed["domain"],
            safe_parsed["tld"],
            source_url,
            status,
            content_type,
            url,
            crawl_ms,
        ))

    conn.commit()


# ============================================================
# WORKER
# ============================================================

def worker(worker_id):
    process_name = f"crawler-{worker_id}"
    mp.current_process().name = process_name

    conn = None
    robots = RobotsCache()
    limiter = RateLimiter()

    crawled = 0
    failed = 0
    last_stats = time.monotonic()

    try:
        conn = db_connect()

        logging.info(
            "Started. DB=%s:%s/%s",
            DB_HOST, DB_PORT, DB_NAME
        )

        while not STOP.is_set():
            item = get_next_queue_item(conn)

            if not item:
                logging.info("Queue empty. Worker waiting...")
                time.sleep(5)
                continue

            queue_id, url, source_url = item

            logging.info("=" * 70)
            logging.info("[CRAWL] %s", url)
            logging.info("[FROM ] %s", source_url or "-")

            started = time.monotonic()

            status, html, error = fetch(url, limiter, robots)

            if error:
                logging.info("[HTTP ] %s -> %s", url, error)
                finish_queue_item(conn, queue_id, success=False)
                failed += 1
                continue

            logging.info("[HTTP ] %s -> HTTP %s", url, status)

            # A redirect normally arrives as the final URL in urllib.
            # The requested URL remains the canonical queue identity.
            if html is None:
                finish_queue_item(conn, queue_id, success=False)
                failed += 1
                continue

            try:
                parsed = parse_html(html, url)
            except Exception as exc:
                logging.exception("[PARSE] %s: %s", url, exc)
                finish_queue_item(conn, queue_id, success=False)
                failed += 1
                continue

            crawl_ms = int((time.monotonic() - started) * 1000)

            logging.info("[TITLE] %s", parsed["title"][:160])
            logging.info("[DOMAIN] %s", parsed["domain"])
            logging.info("[HEADS ] %d", len(parsed["headings"]))
            logging.info("[LINKS ] %d gefunden", len(parsed["links"]))

            save_page(
                conn=conn,
                url=url,
                source_url=source_url,
                status=status,
                content_type="text/html",
                parsed=parsed,
                crawl_ms=crawl_ms,
            )

            inserted = queue_urls(
                conn,
                parsed["links"],
                url
            )

            logging.info(
                "[SORT ] %s -> %d Links, %d neue URLs",
                url,
                len(parsed["links"]),
                inserted,
            )

            finish_queue_item(conn, queue_id, success=True)

            crawled += 1

            # Periodic stats; not on every page to avoid unnecessary DB load.
            if time.monotonic() - last_stats >= 30:
                try:
                    with conn.cursor() as cur:
                        cur.execute(
                            "SELECT COUNT(*) FROM crawl_queue WHERE status='pending'"
                        )
                        queued = cur.fetchone()[0]

                    write_stats(conn, crawled, queued, failed)
                    last_stats = time.monotonic()
                except Exception:
                    conn.rollback()

    except Exception:
        logging.exception("Worker crashed.")
        raise
    finally:
        if conn is not None:
            conn.close()

        logging.info(
            "Stopped. crawled=%d failed=%d",
            crawled, failed
        )


# ============================================================
# STARTUP
# ============================================================

def main():
    print("=" * 70)
    print("     JonasSearch Crawler Gamma 0.1")
    print("=" * 70)
    print(f"Start:        {START_URL}")
    print(f"Workers:      {WORKERS}")
    print(f"Delay:        {DEFAULT_DELAY:.1f} Sekunden")
    print(f"Robots-Cache: {ROBOTS_CACHE_SECONDS // 3600} Stunden")
    print(f"DB:           {DB_HOST}:{DB_PORT}/{DB_NAME}")
    print()

    conn = db_connect()

    ensure_stats_table(conn)
    requeue_stale_items(conn)

    # Seed URL. ON CONFLICT makes restarting harmless.
    queue_urls(conn, [normalize_url(START_URL)], None)

    conn.close()

    workers = []

    for i in range(WORKERS):
        p = mp.Process(target=worker, args=(i,), name=f"crawler-{i}")
        p.start()
        workers.append(p)

    try:
        while any(p.is_alive() for p in workers):
            time.sleep(1)

            # If a worker dies unexpectedly, stop the remaining workers.
            for p in workers:
                if p.exitcode not in (None, 0):
                    logging.error(
                        "Worker %s exited with code %s.",
                        p.name,
                        p.exitcode,
                    )
                    STOP.set()

    except KeyboardInterrupt:
        STOP.set()

    finally:
        STOP.set()

        for p in workers:
            p.join(timeout=10)

        for p in workers:
            if p.is_alive():
                p.terminate()
                p.join()

    print("Crawler stopped.")


if __name__ == "__main__":
    mp.set_start_method("spawn", force=True)
    main()