"""MariaDB cache for crawled pages: domains, files, link maps, timestamps, discovery, interactions."""

from __future__ import annotations

import hashlib
import logging
import os
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any

import pymysql

log = logging.getLogger("crawler.db")

DB_CONFIG = {
    "host": os.getenv("CRAWLER_DB_HOST", "mariadb"),
    "port": int(os.getenv("CRAWLER_DB_PORT", "3306")),
    "user": os.getenv("CRAWLER_DB_USER", "crawler"),
    "password": os.getenv("CRAWLER_DB_PASS", "crawler-secret"),
    "database": os.getenv("CRAWLER_DB_NAME", "crawler_cache"),
    "charset": "utf8mb4",
    "connect_timeout": 5,
}

CACHE_TTL_SECONDS = int(os.getenv("CRAWLER_CACHE_TTL_SECONDS", "86400"))

_connection: pymysql.Connection | None = None


def _connect() -> pymysql.Connection:
    global _connection
    try:
        if _connection is not None and _connection.open:
            _connection.ping(reconnect=True)
            return _connection
    except Exception:
        _connection = None
    _connection = pymysql.connect(**DB_CONFIG)
    return _connection


def _ensure_schema() -> None:
    conn = _connect()
    with conn.cursor() as cur:
        cur.execute("""
            CREATE TABLE IF NOT EXISTS domains (
                id INT AUTO_INCREMENT PRIMARY KEY,
                domain VARCHAR(255) NOT NULL UNIQUE,
                first_seen_at DOUBLE NOT NULL,
                last_crawled_at DOUBLE NOT NULL,
                crawl_count INT NOT NULL DEFAULT 0,
                total_bytes BIGINT NOT NULL DEFAULT 0,
                INDEX idx_domain (domain)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
        """)
        cur.execute("""
            CREATE TABLE IF NOT EXISTS pages (
                id INT AUTO_INCREMENT PRIMARY KEY,
                url VARCHAR(2048) NOT NULL,
                url_hash CHAR(32) NOT NULL,
                domain_id INT NOT NULL,
                stem VARCHAR(512) NOT NULL,
                status VARCHAR(32) NOT NULL DEFAULT 'ok',
                content_digest CHAR(64),
                markdown_bytes INT NOT NULL DEFAULT 0,
                html_bytes INT NOT NULL DEFAULT 0,
                image_count INT NOT NULL DEFAULT 0,
                images_bytes BIGINT NOT NULL DEFAULT 0,
                crawled_at DOUBLE NOT NULL,
                http_status INT DEFAULT NULL,
                elapsed_seconds DOUBLE DEFAULT NULL,
                consent_status VARCHAR(64) DEFAULT NULL,
                canonical_url VARCHAR(2048) NULL,
                canonical_url_hash CHAR(64) NULL,
                page_type VARCHAR(32) NOT NULL DEFAULT 'url',
                parent_canonical_url_hash CHAR(64) NULL,
                action_key CHAR(64) NULL,
                action_label VARCHAR(512) NULL,
                cache_policy VARCHAR(32) NULL,
                first_seen_at DOUBLE NULL,
                last_seen_at DOUBLE NULL,
                etag VARCHAR(512) NULL,
                last_modified VARCHAR(512) NULL,
                UNIQUE INDEX idx_url_hash (url_hash),
                UNIQUE INDEX idx_pages_canonical_type_action (canonical_url_hash, page_type, action_key),
                INDEX idx_domain_id (domain_id),
                INDEX idx_crawled_at (crawled_at),
                INDEX idx_pages_content_digest (content_digest),
                INDEX idx_pages_parent_hash (parent_canonical_url_hash)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
        """)
        cur.execute("""
            CREATE TABLE IF NOT EXISTS page_links (
                id INT AUTO_INCREMENT PRIMARY KEY,
                page_url_hash CHAR(32) NOT NULL,
                source_url VARCHAR(2048) NOT NULL,
                target_url VARCHAR(2048) NOT NULL,
                link_text VARCHAR(1024) NOT NULL DEFAULT '',
                target_is_crawled BOOL NOT NULL DEFAULT FALSE,
                target_stem VARCHAR(512) DEFAULT NULL,
                INDEX idx_page_url_hash (page_url_hash),
                INDEX idx_target_url (target_url(255))
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
        """)
        cur.execute("""
            CREATE TABLE IF NOT EXISTS page_images (
                id INT AUTO_INCREMENT PRIMARY KEY,
                page_url_hash CHAR(32) NOT NULL,
                source_url VARCHAR(2048) NOT NULL,
                local_path VARCHAR(1024) NOT NULL,
                content_type VARCHAR(128) NOT NULL DEFAULT '',
                size_bytes INT NOT NULL DEFAULT 0,
                from_cache BOOL NOT NULL DEFAULT FALSE,
                INDEX idx_page_url_hash (page_url_hash)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
        """)
        cur.execute("""
            CREATE TABLE IF NOT EXISTS discovered_targets (
                id INT AUTO_INCREMENT PRIMARY KEY,
                discovery_key CHAR(71) NOT NULL,
                target_type VARCHAR(32) NOT NULL DEFAULT 'url',
                raw_url VARCHAR(2048) NULL,
                canonical_url VARCHAR(2048) NULL,
                canonical_url_hash CHAR(64) NULL,
                parent_canonical_url_hash CHAR(64) NULL,
                source_url VARCHAR(2048) NULL,
                link_text VARCHAR(1024) NOT NULL DEFAULT '',
                action_signature VARCHAR(2048) NULL,
                first_seen_at DOUBLE NOT NULL,
                last_seen_at DOUBLE NOT NULL,
                seen_count INT NOT NULL DEFAULT 1,
                crawl_status VARCHAR(32) NOT NULL DEFAULT 'discovered',
                reserved_at DOUBLE NULL,
                crawled_at DOUBLE NULL,
                error_text TEXT NULL,
                UNIQUE INDEX idx_discovered_key (discovery_key),
                INDEX idx_discovered_canonical_hash (canonical_url_hash),
                INDEX idx_discovered_parent_hash (parent_canonical_url_hash),
                INDEX idx_discovered_status (crawl_status)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
        """)
        cur.execute("""
            CREATE TABLE IF NOT EXISTS page_actions (
                id INT AUTO_INCREMENT PRIMARY KEY,
                parent_canonical_url_hash CHAR(64) NOT NULL,
                action_key CHAR(64) NOT NULL,
                action_type VARCHAR(32) NOT NULL,
                selector VARCHAR(2048) NOT NULL DEFAULT '',
                label VARCHAR(1024) NOT NULL DEFAULT '',
                href VARCHAR(2048) NULL,
                before_url VARCHAR(2048) NULL,
                after_url VARCHAR(2048) NULL,
                result_type VARCHAR(32) NOT NULL DEFAULT 'unknown',
                virtual_url VARCHAR(2048) NULL,
                content_digest CHAR(64) NULL,
                created_at DOUBLE NOT NULL,
                UNIQUE INDEX idx_page_actions_key (action_key),
                INDEX idx_page_actions_parent (parent_canonical_url_hash)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
        """)
    conn.commit()


# ---------------------------------------------------------------------------
def _url_hash(url: str) -> str:
    return hashlib.md5(url.strip().lower().encode()).hexdigest()


def canonical_hash(canonical_url: str) -> str:
    return hashlib.sha256(canonical_url.encode("utf-8")).hexdigest()


def _ensure_domain(domain: str) -> int:
    conn = _connect()
    now = time.time()
    with conn.cursor() as cur:
        cur.execute("SELECT id FROM domains WHERE domain = %s", (domain,))
        row = cur.fetchone()
        if row:
            cur.execute(
                "UPDATE domains SET last_crawled_at = %s, crawl_count = crawl_count + 1 WHERE id = %s",
                (now, row[0]),
            )
            conn.commit()
            return row[0]
        cur.execute(
            "INSERT INTO domains (domain, first_seen_at, last_crawled_at, crawl_count) VALUES (%s, %s, %s, 1)",
            (domain, now, now),
        )
        conn.commit()
        return cur.lastrowid


# ---------------------------------------------------------------------------
# Cache entry + lookup
# ---------------------------------------------------------------------------
@dataclass
class CacheEntry:
    found: bool
    url: str
    url_hash: str
    canonical_url: str = ""
    canonical_url_hash: str = ""
    stem: str = ""
    domain_id: int | None = None
    content_digest: str | None = None
    markdown_bytes: int = 0
    crawled_at: float | None = None
    elapsed_seconds: float | None = None
    page_type: str = "url"

    @property
    def is_fresh(self) -> bool:
        if self.crawled_at is None:
            return False
        return (time.time() - self.crawled_at) < CACHE_TTL_SECONDS


def check_cache(url: str, canonical: str = "") -> CacheEntry | None:
    try:
        _ensure_schema()
    except Exception as exc:
        log.warning("DB schema init failed: %s", exc)
        return None

    uh = _url_hash(url)
    conn = _connect()
    try:
        with conn.cursor(pymysql.cursors.DictCursor) as cur:
            cur.execute(
                "SELECT url_hash, canonical_url, canonical_url_hash, stem, domain_id, "
                "content_digest, markdown_bytes, crawled_at, elapsed_seconds, page_type "
                "FROM pages WHERE url_hash = %s OR canonical_url_hash = %s",
                (uh, canonical_hash(canonical) if canonical else ""),
            )
            row = cur.fetchone()
            if row:
                return CacheEntry(
                    found=True, url=url, url_hash=row["url_hash"],
                    canonical_url=row["canonical_url"] or "",
                    canonical_url_hash=row["canonical_url_hash"] or "",
                    stem=row["stem"], domain_id=row["domain_id"],
                    content_digest=row["content_digest"],
                    markdown_bytes=row["markdown_bytes"],
                    crawled_at=row["crawled_at"],
                    elapsed_seconds=row["elapsed_seconds"],
                    page_type=row.get("page_type", "url"),
                )
    except Exception as exc:
        log.warning("DB cache check failed: %s", exc)
        return None
    return None


def get_page_by_canonical_hash(chash: str, page_type: str = "url") -> CacheEntry | None:
    conn = _connect()
    try:
        with conn.cursor(pymysql.cursors.DictCursor) as cur:
            cur.execute(
                "SELECT * FROM pages WHERE canonical_url_hash = %s AND page_type = %s",
                (chash, page_type),
            )
            row = cur.fetchone()
            if row:
                return CacheEntry(
                    found=True, url=row["url"], url_hash=row["url_hash"],
                    canonical_url=row["canonical_url"] or "",
                    canonical_url_hash=row["canonical_url_hash"] or "",
                    stem=row["stem"], domain_id=row["domain_id"],
                    content_digest=row["content_digest"],
                    markdown_bytes=row["markdown_bytes"],
                    crawled_at=row["crawled_at"],
                    elapsed_seconds=row["elapsed_seconds"],
                    page_type=row.get("page_type", "url"),
                )
    except Exception as exc:
        log.warning("DB canonical lookup failed: %s", exc)
    return None


# ---------------------------------------------------------------------------
# Discovered targets
# ---------------------------------------------------------------------------
def is_known_target(discovery_key: str, canonical_chash: str = "") -> bool:
    conn = _connect()
    try:
        with conn.cursor() as cur:
            cur.execute(
                "SELECT 1 FROM pages WHERE canonical_url_hash = %s AND page_type = 'url'",
                (canonical_chash,),
            )
            if cur.fetchone():
                return True
            cur.execute(
                "SELECT 1 FROM discovered_targets WHERE discovery_key = %s",
                (discovery_key,),
            )
            return cur.fetchone() is not None
    except Exception:
        return False


def mark_discovered_target(
    discovery_key: str,
    target_type: str = "url",
    raw_url: str = "",
    canonical_url: str = "",
    canonical_url_hash: str = "",
    parent_canonical_url_hash: str = "",
    source_url: str = "",
    link_text: str = "",
    action_signature: str = "",
) -> None:
    try:
        _ensure_schema()
    except Exception:
        return
    now = time.time()
    conn = _connect()
    try:
        with conn.cursor() as cur:
            cur.execute(
                "INSERT INTO discovered_targets "
                "(discovery_key, target_type, raw_url, canonical_url, canonical_url_hash, "
                "parent_canonical_url_hash, source_url, link_text, action_signature, "
                "first_seen_at, last_seen_at, seen_count) "
                "VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,1) "
                "ON DUPLICATE KEY UPDATE "
                "last_seen_at=VALUES(last_seen_at), seen_count=seen_count+1, "
                "link_text=VALUES(link_text)",
                (discovery_key, target_type, raw_url, canonical_url, canonical_url_hash,
                 parent_canonical_url_hash, source_url, link_text, action_signature,
                 now, now),
            )
        conn.commit()
    except Exception as exc:
        log.debug("DB mark_discovered_target failed: %s", exc)


def reserve_target_for_crawl(discovery_key: str) -> bool:
    now = time.time()
    conn = _connect()
    try:
        with conn.cursor() as cur:
            cur.execute(
                "SELECT crawl_status, reserved_at FROM discovered_targets WHERE discovery_key = %s",
                (discovery_key,),
            )
            row = cur.fetchone()
            if row:
                status = row[0]
                reserved = row[1]
                if status == "reserved" and reserved and (now - reserved) < 300:
                    return False
                if status == "crawled":
                    return False
            cur.execute(
                "UPDATE discovered_targets SET crawl_status='reserved', reserved_at=%s "
                "WHERE discovery_key = %s",
                (now, discovery_key),
            )
        conn.commit()
        return True
    except Exception:
        return False


def mark_target_crawled(discovery_key: str) -> None:
    now = time.time()
    conn = _connect()
    try:
        with conn.cursor() as cur:
            cur.execute(
                "UPDATE discovered_targets SET crawl_status='crawled', crawled_at=%s WHERE discovery_key = %s",
                (now, discovery_key),
            )
        conn.commit()
    except Exception:
        pass


def mark_target_failed(discovery_key: str, error_text: str = "") -> None:
    now = time.time()
    conn = _connect()
    try:
        with conn.cursor() as cur:
            cur.execute(
                "UPDATE discovered_targets SET crawl_status='failed', crawled_at=%s, error_text=%s "
                "WHERE discovery_key = %s",
                (now, error_text[:1000], discovery_key),
            )
        conn.commit()
    except Exception:
        pass


# ---------------------------------------------------------------------------
# Save page record with canonical URL
# ---------------------------------------------------------------------------
def save_page_record(
    url: str,
    domain: str,
    stem: str,
    status: str,
    content_digest: str | None,
    markdown_bytes: int,
    html_bytes: int,
    image_count: int,
    images_bytes: int,
    crawled_at: float,
    http_status: int | None,
    elapsed_seconds: float | None,
    consent_status: str | None,
    canonical_url: str = "",
    page_type: str = "url",
    parent_canonical_url_hash: str = "",
    action_key: str = "",
    action_label: str = "",
    cache_policy: str = "",
) -> None:
    try:
        _ensure_schema()
    except Exception as exc:
        log.warning("DB schema init failed: %s", exc)
        return

    domain_id = _ensure_domain(domain)
    uh = _url_hash(url)
    cch = canonical_hash(canonical_url) if canonical_url else ""

    conn = _connect()
    try:
        with conn.cursor() as cur:
            cur.execute(
                "INSERT INTO pages "
                "(url, url_hash, domain_id, stem, status, content_digest, "
                "markdown_bytes, html_bytes, image_count, images_bytes, crawled_at, "
                "http_status, elapsed_seconds, consent_status, "
                "canonical_url, canonical_url_hash, page_type, "
                "parent_canonical_url_hash, action_key, action_label, cache_policy, "
                "first_seen_at, last_seen_at) "
                "VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) "
                "ON DUPLICATE KEY UPDATE "
                "url_hash=VALUES(url_hash), status=VALUES(status), content_digest=VALUES(content_digest), "
                "markdown_bytes=VALUES(markdown_bytes), html_bytes=VALUES(html_bytes), "
                "image_count=VALUES(image_count), images_bytes=VALUES(images_bytes), "
                "crawled_at=VALUES(crawled_at), http_status=VALUES(http_status), "
                "elapsed_seconds=VALUES(elapsed_seconds), consent_status=VALUES(consent_status), "
                "last_seen_at=VALUES(last_seen_at)",
                (
                    url, uh, domain_id, stem, status, content_digest,
                    markdown_bytes, html_bytes, image_count, images_bytes, crawled_at,
                    http_status, elapsed_seconds, consent_status,
                    canonical_url, cch, page_type,
                    parent_canonical_url_hash, action_key, action_label, cache_policy,
                    crawled_at, crawled_at,
                ),
            )
            cur.execute(
                "UPDATE domains SET total_bytes = (SELECT COALESCE(SUM(markdown_bytes + html_bytes + images_bytes), 0) FROM pages WHERE domain_id = %s) WHERE id = %s",
                (domain_id, domain_id),
            )
        conn.commit()
    except Exception as exc:
        log.warning("DB save page record failed: %s", exc)


# ---------------------------------------------------------------------------
# Links and images (unchanged)
# ---------------------------------------------------------------------------
def save_page_links(page_url: str, links: list[dict[str, str]], out_dir: Path | None = None) -> None:
    if not links:
        return
    try:
        _ensure_schema()
    except Exception:
        return
    uh = _url_hash(page_url)
    conn = _connect()
    try:
        with conn.cursor() as cur:
            for link in links:
                target = link.get("url", "")
                text = link.get("text", "")
                target_uh = _url_hash(target)
                cur.execute("SELECT stem FROM pages WHERE url_hash = %s", (target_uh,))
                stem_row = cur.fetchone()
                target_crawled = stem_row is not None
                target_stem = stem_row[0] if stem_row else None
                cur.execute(
                    "INSERT INTO page_links (page_url_hash, source_url, target_url, link_text, target_is_crawled, target_stem) "
                    "VALUES (%s,%s,%s,%s,%s,%s)",
                    (uh, page_url, target, text, target_crawled, target_stem),
                )
        conn.commit()
    except Exception as exc:
        log.warning("DB save links failed: %s", exc)


def save_page_images(page_url: str, manifest: list[dict]) -> None:
    if not manifest:
        return
    try:
        _ensure_schema()
    except Exception:
        return
    uh = _url_hash(page_url)
    conn = _connect()
    try:
        with conn.cursor() as cur:
            cur.execute("DELETE FROM page_images WHERE page_url_hash = %s", (uh,))
            for img in manifest:
                cur.execute(
                    "INSERT INTO page_images (page_url_hash, source_url, local_path, content_type, size_bytes, from_cache) "
                    "VALUES (%s,%s,%s,%s,%s,%s)",
                    (uh, img.get("source_url", ""), img.get("relative_path", img.get("local_path", "")),
                     img.get("content_type", ""), img.get("size_bytes", 0), img.get("from_cache", False)),
                )
        conn.commit()
    except Exception as exc:
        log.warning("DB save images failed: %s", exc)


def get_link_map(url: str) -> list[dict[str, str]]:
    uh = _url_hash(url)
    conn = _connect()
    try:
        with conn.cursor(pymysql.cursors.DictCursor) as cur:
            cur.execute(
                "SELECT source_url, target_url, link_text, target_is_crawled, target_stem "
                "FROM page_links WHERE page_url_hash = %s",
                (uh,),
            )
            return [
                {"source_url": r["source_url"], "target_url": r["target_url"],
                 "link_text": r["link_text"], "target_is_crawled": bool(r["target_is_crawled"]),
                 "target_stem": r["target_stem"] or ""}
                for r in cur.fetchall()
            ]
    except Exception:
        return []


# ---------------------------------------------------------------------------
# Product discovery CRUD
# ---------------------------------------------------------------------------
def product_key(canonical_url: str) -> str:
    return hashlib.sha256(f"product:{canonical_url}".encode()).hexdigest()[:64]


def upsert_product(
    domain: str,
    canonical_product_url: str = "",
    product_name: str = "",
    product_code: str = "",
    product_family: str = "",
    product_type: str = "",
    product_page_role: str = "",
    identity_confidence: float | None = None,
    identity_source: str = "",
    run_id: str = "",
) -> str:
    key = product_key(canonical_product_url or product_name)
    now = time.time()
    conn = _connect()
    try:
        with conn.cursor() as cur:
            cur.execute(
                "INSERT INTO products "
                "(product_key, domain, canonical_product_url, product_name, product_code, "
                "product_family, product_type, product_page_role, identity_confidence, "
                "identity_source, first_seen_run_id, last_seen_run_id) "
                "VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) "
                "ON DUPLICATE KEY UPDATE "
                "product_name=COALESCE(NULLIF(VALUES(product_name),''), product_name), "
                "product_code=COALESCE(NULLIF(VALUES(product_code),''), product_code), "
                "product_family=COALESCE(NULLIF(VALUES(product_family),''), product_family), "
                "last_seen_run_id=VALUES(last_seen_run_id)",
                (key, domain, canonical_product_url, product_name, product_code,
                 product_family, product_type, product_page_role,
                 identity_confidence, identity_source, run_id, run_id),
            )
        conn.commit()
    except Exception as exc:
        log.debug("upsert_product failed: %s", exc)
    return key


def add_page_product_reference(
    source_page_url: str = "",
    source_page_canonical: str = "",
    product_key: str = "",
    reference_type: str = "",
    context_type: str = "",
    context_label: str = "",
    visible_product_name: str = "",
    visible_product_code: str = "",
    product_url: str = "",
    canonical_product_url: str = "",
    position_index: int | None = None,
    run_id: str = "",
) -> str:
    page_key = canonical_hash(source_page_canonical) if source_page_canonical else hashlib.sha256(source_page_url.encode()).hexdigest()[:64]
    ref_key = hashlib.sha256(
        f"page_product:{page_key}:{product_key}:{context_type}:{context_label}".encode()
    ).hexdigest()[:64]
    conn = _connect()
    try:
        with conn.cursor() as cur:
            cur.execute(
                "INSERT INTO page_product_references "
                "(reference_key, page_key, source_page_url, source_page_canonical, product_key, "
                "reference_type, context_type, context_label, visible_product_name, "
                "visible_product_code, product_url, canonical_product_url, position_index, "
                "first_seen_run_id, last_seen_run_id) "
                "VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) "
                "ON DUPLICATE KEY UPDATE "
                "visible_product_name=COALESCE(NULLIF(VALUES(visible_product_name),''), visible_product_name), "
                "last_seen_run_id=VALUES(last_seen_run_id)",
                (ref_key, page_key, source_page_url, source_page_canonical, product_key,
                 reference_type, context_type, context_label, visible_product_name,
                 visible_product_code, product_url, canonical_product_url, position_index,
                 run_id, run_id),
            )
        conn.commit()
    except Exception as exc:
        log.debug("add_page_product_reference failed: %s", exc)
    return ref_key


def upsert_product_variant(
    product_key: str,
    article_number: str = "",
    article_number_raw: str = "",
    variant_name: str = "",
    type_description: str = "",
    dimming: str = "",
    status: str = "",
    luminous_flux: str = "",
    power: str = "",
    efficacy: str = "",
    color_temperature: str = "",
    cri: str = "",
    ugr: str = "",
    ip_rate: str = "",
    source_url: str = "",
    source_section: str = "",
    run_id: str = "",
) -> str:
    variant_key = hashlib.sha256(
        f"variant:{article_number or article_number_raw}".encode()
    ).hexdigest()[:64]
    conn = _connect()
    try:
        with conn.cursor() as cur:
            cur.execute(
                "INSERT INTO product_variants "
                "(variant_key, product_key, article_number, article_number_raw, variant_name, "
                "type_description, dimming, status, luminous_flux, power, efficacy, "
                "color_temperature, cri, ugr, ip_rate, source_url, source_section, "
                "first_seen_run_id, last_seen_run_id) "
                "VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) "
                "ON DUPLICATE KEY UPDATE "
                "article_number=COALESCE(NULLIF(VALUES(article_number),''), article_number), "
                "variant_name=COALESCE(NULLIF(VALUES(variant_name),''), variant_name), "
                "last_seen_run_id=VALUES(last_seen_run_id)",
                (variant_key, product_key, article_number, article_number_raw, variant_name,
                 type_description, dimming, status, luminous_flux, power, efficacy,
                 color_temperature, cri, ugr, ip_rate, source_url, source_section,
                 run_id, run_id),
            )
        conn.commit()
    except Exception as exc:
        log.debug("upsert_product_variant failed: %s", exc)
    return variant_key


def add_product_download(
    product_key: str,
    url: str,
    download_label: str = "",
    download_type: str = "",
    variant_key: str = "",
    canonical_url: str = "",
    source_section: str = "",
    run_id: str = "",
) -> str:
    download_key = hashlib.sha256(
        f"download:{product_key}:{url}".encode()
    ).hexdigest()[:64]
    conn = _connect()
    try:
        with conn.cursor() as cur:
            cur.execute(
                "INSERT INTO product_downloads "
                "(download_key, product_key, variant_key, download_label, download_type, "
                "url, canonical_url, source_section, first_seen_run_id, last_seen_run_id) "
                "VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) "
                "ON DUPLICATE KEY UPDATE "
                "last_seen_run_id=VALUES(last_seen_run_id)",
                (download_key, product_key, variant_key, download_label, download_type,
                 url, canonical_url, source_section, run_id, run_id),
            )
        conn.commit()
    except Exception as exc:
        log.debug("add_product_download failed: %s", exc)
    return download_key


def product_exists(canonical_product_url: str) -> str | None:
    """Return product_key if product already known, else None."""
    key = product_key(canonical_product_url)
    conn = _connect()
    try:
        with conn.cursor() as cur:
            cur.execute("SELECT product_key FROM products WHERE product_key = %s", (key,))
            row = cur.fetchone()
            return row[0] if row else None
    except Exception:
        return None


def search_products(domain: str = "", query: str = "", limit: int = 50) -> list[dict]:
    conn = _connect()
    try:
        with conn.cursor(pymysql.cursors.DictCursor) as cur:
            if query:
                cur.execute(
                    "SELECT * FROM products WHERE (domain = %s OR %s = '') "
                    "AND (product_name LIKE %s OR product_code LIKE %s OR product_family LIKE %s) LIMIT %s",
                    (domain, domain, f"%{query}%", f"%{query}%", f"%{query}%", limit),
                )
            else:
                cur.execute(
                    "SELECT * FROM products WHERE domain = %s OR %s = '' LIMIT %s",
                    (domain, domain, limit),
                )
            return [dict(r) for r in cur.fetchall()]
    except Exception:
        return []


def get_product_contexts(product_key: str) -> list[dict]:
    conn = _connect()
    try:
        with conn.cursor(pymysql.cursors.DictCursor) as cur:
            cur.execute(
                "SELECT * FROM page_product_references WHERE product_key = %s",
                (product_key,),
            )
            return [dict(r) for r in cur.fetchall()]
    except Exception:
        return []


def get_product_variants(product_key: str) -> list[dict]:
    conn = _connect()
    try:
        with conn.cursor(pymysql.cursors.DictCursor) as cur:
            cur.execute(
                "SELECT * FROM product_variants WHERE product_key = %s",
                (product_key,),
            )
            return [dict(r) for r in cur.fetchall()]
    except Exception:
        return []


def get_product_downloads(product_key: str) -> list[dict]:
    conn = _connect()
    try:
        with conn.cursor(pymysql.cursors.DictCursor) as cur:
            cur.execute(
                "SELECT * FROM product_downloads WHERE product_key = %s",
                (product_key,),
            )
            return [dict(r) for r in cur.fetchall()]
    except Exception:
        return []


# ---------------------------------------------------------------------------
# Parser projects CRUD
# ---------------------------------------------------------------------------
def _ensure_parser_table() -> None:
    conn = _connect()
    with conn.cursor() as cur:
        cur.execute("""
            CREATE TABLE IF NOT EXISTS parser_projects (
                id INT AUTO_INCREMENT PRIMARY KEY,
                project_key CHAR(64) NOT NULL UNIQUE,
                domain VARCHAR(255) NOT NULL,
                parser_name VARCHAR(255) NOT NULL,
                parser_type VARCHAR(64) NOT NULL DEFAULT 'product_discovery',
                page_classifiers JSON NULL,
                actions JSON NULL,
                extractors JSON NULL,
                validations JSON NULL,
                test_urls JSON NULL,
                validation_status VARCHAR(32) NOT NULL DEFAULT 'draft',
                validation_errors JSON NULL,
                created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
                updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
                INDEX idx_pp_domain (domain)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
        """)
    conn.commit()


def create_parser_project(domain: str, parser_name: str, parser_type: str = "product_discovery") -> dict:
    _ensure_parser_table()
    key = hashlib.sha256(f"{domain}:{parser_name}".encode()).hexdigest()[:64]
    conn = _connect()
    with conn.cursor(pymysql.cursors.DictCursor) as cur:
        cur.execute(
            "INSERT INTO parser_projects (project_key, domain, parser_name, parser_type) VALUES (%s,%s,%s,%s) "
            "ON DUPLICATE KEY UPDATE parser_type=VALUES(parser_type)",
            (key, domain, parser_name, parser_type),
        )
        conn.commit()
        cur.execute("SELECT * FROM parser_projects WHERE project_key = %s", (key,))
        return dict(cur.fetchone())


def get_parser_project(project_id: str) -> dict | None:
    conn = _connect()
    try:
        with conn.cursor(pymysql.cursors.DictCursor) as cur:
            cur.execute("SELECT * FROM parser_projects WHERE project_key = %s OR id = %s",
                        (project_id, int(project_id) if project_id.isdigit() else 0))
            row = cur.fetchone()
            return dict(row) if row else None
    except Exception:
        return None


def save_parser_manifest(project_id: str, manifest: dict) -> bool:
    conn = _connect()
    try:
        with conn.cursor() as cur:
            cur.execute(
                "UPDATE parser_projects SET page_classifiers=%s, actions=%s, extractors=%s, "
                "validations=%s, test_urls=%s WHERE project_key = %s",
                (
                    json.dumps(manifest.get("page_classifiers", {})),
                    json.dumps(manifest.get("actions", {})),
                    json.dumps(manifest.get("extractors", {})),
                    json.dumps(manifest.get("validations", {})),
                    json.dumps(manifest.get("test_urls", [])),
                    project_id,
                ),
            )
        conn.commit()
        return True
    except Exception:
        return False


def update_parser_validation(project_id: str, status: str, errors: list | None = None) -> None:
    conn = _connect()
    try:
        with conn.cursor() as cur:
            cur.execute(
                "UPDATE parser_projects SET validation_status=%s, validation_errors=%s WHERE project_key = %s",
                (status, json.dumps(errors or []), project_id),
            )
        conn.commit()
    except Exception:
        pass
