"""Sensor auto-repair (spec §60, §189): failing / stale / redirected / PAGE_REMOVED sensors.

    retry ─▶ (redirect target) ─▶ re-fetch sitemap ─▶ rediscover navigation ─▶ candidate URL for the same surface
          ─▶ content-identity check (simhash / fuzzy ratio against the last snapshot text) ─▶ migrate (old retired with
          config.successor_id, new sensor with config.predecessor_id, history kept) or review_queue(kind='sensor_migration').

Registered as a periodic task (every 30 min, bounded batch). Bounded requests per sensor; never raises out of the batch.
"""
from __future__ import annotations

import logging
import re
from datetime import UTC, datetime
from typing import Any
from urllib.parse import urlparse

from rapidfuzz import fuzz

from companyatlas import archive
from companyatlas.config import settings
from companyatlas.db import execute, fetch_all, fetch_one, jsonb, transaction
from companyatlas.fetch import BlockedError, Fetcher, FetchError, FetchResult, NotModified
from companyatlas.ids import new_id
from companyatlas.sdk import connector as connectors
from companyatlas.sdk import normalize
from companyatlas.sdk.normalize import hamming, normalized_text, simhash
from companyatlas.services.periodic import periodic
from companyatlas.taxonomy import SensorStatus, Surface
from companyatlas.urls import canonicalize_url, classify_url, registrable_domain, same_company_host

log = logging.getLogger(__name__)

REPAIR_VERSION = "repair-v1"
MAX_REQUESTS_PER_SENSOR = 6
IDENTITY_HAMMING = 12
IDENTITY_RATIO = 0.6
MIN_CANDIDATE_CONF = 0.7
SOFT_404_RE = re.compile(r"(page not found|404|not be found|doesn'?t exist|no longer available)", re.IGNORECASE)


class _Session:
    def __init__(self, fetcher: Fetcher):
        self.fetcher = fetcher
        self.requests = 0

    async def get(self, url: str, *, max_bytes: int = 2 * 1024 * 1024, accept: str | None = None) -> FetchResult | None:
        if self.requests >= MAX_REQUESTS_PER_SENSOR:
            return None
        self.requests += 1
        try:
            return await self.fetcher.get(url, max_bytes=max_bytes, retries=0, accept=accept)
        except NotModified:
            return None
        except (FetchError, BlockedError) as exc:
            log.debug("repair fetch failed", extra={"url": url, "failure": str(exc.failure)})
            return None
        except Exception as exc:  # noqa: BLE001
            log.debug("repair fetch error", extra={"url": url, "error": str(exc)[:200]})
            return None


def _soft_404(res: FetchResult) -> bool:
    m = re.search(r"<title[^>]*>(.*?)</title>", res.text[:4000], re.IGNORECASE | re.DOTALL)
    return bool(m and SOFT_404_RE.search(m.group(1)))


async def _last_text(sensor: dict[str, Any]) -> str:
    sid = sensor.get("last_snapshot_id")
    if not sid:
        return ""
    async with transaction() as conn:
        row = await fetch_one(conn, "select text_key from snapshots where id = :id", id=sid)
    if row and row.get("text_key") and archive.exists(row["text_key"]):
        try:
            return archive.get_text(row["text_key"])
        except OSError:
            return ""
    return ""


def content_identity(previous_text: str, candidate_text: str) -> tuple[bool, float]:
    """(same_content, score). simhash distance first, fuzzy ratio second — both on noise-normalised text."""
    if not previous_text or not candidate_text:
        return False, 0.0
    d = hamming(simhash(previous_text), simhash(candidate_text))
    ratio = fuzz.ratio(normalized_text(previous_text)[:20000], normalized_text(candidate_text)[:20000]) / 100.0
    return (d <= IDENTITY_HAMMING or ratio >= IDENTITY_RATIO), round(max(ratio, 1 - d / 64), 3)


async def _candidates(session: _Session, sensor: dict[str, Any], company: dict[str, Any]) -> list[tuple[str, float, str]]:
    """(url, confidence, method) for the sensor's surface: redirect target, sitemap entries, navigation links."""
    surface = str(sensor["surface"])
    canonical = str(company.get("canonical_domain") or sensor["domain"])
    out: list[tuple[str, float, str]] = []
    cfg = sensor.get("config") or {}
    if cfg.get("redirect_url"):
        out.append((str(cfg["redirect_url"]), 0.9, "redirect"))
    origin = f"https://{urlparse(str(company.get('website') or 'https://' + canonical)).netloc or canonical}"
    # sitemap
    async with transaction() as conn:
        sm = await fetch_one(conn, "select url from sensors where company_id = :c and surface = 'sitemap' and status <> 'retired' order by quality_score desc limit 1",
                             c=company["id"])
    sm_url = str(sm["url"]) if sm else f"{origin}/sitemap.xml"
    res = await session.get(sm_url, max_bytes=4 * 1024 * 1024, accept="application/xml,text/xml,*/*;q=0.5")
    if res is not None and not res.is_html:
        from companyatlas.connectors.sitemap import _decode, parse_sitemap

        pages, children = parse_sitemap(_decode(res.content))
        if children and not pages:
            child = await session.get(children[0][0], max_bytes=4 * 1024 * 1024, accept="application/xml,text/xml,*/*;q=0.5")
            if child is not None:
                pages, _ = parse_sitemap(_decode(child.content))
        for loc, _lm in pages[: settings.discovery_max_sitemap_urls]:
            if not same_company_host(loc, canonical):
                continue
            s, c = classify_url(loc, canonical_domain=canonical)
            if s == surface and c >= MIN_CANDIDATE_CONF:
                out.append((loc, c * 0.9, "sitemap"))
    # navigation
    home = await session.get(origin + "/")
    if home is not None and home.is_html:
        page = normalize.parse(home.text, url=home.final_url)
        for ln in page.links:
            if not same_company_host(ln.url, canonical):
                continue
            s, c = classify_url(ln.url, anchor=ln.anchor, canonical_domain=canonical)
            if s == surface and c >= MIN_CANDIDATE_CONF:
                out.append((ln.url, c, "nav"))
    # dedupe, exclude the broken URL itself
    seen = {canonicalize_url(str(sensor["url"]))}
    uniq: list[tuple[str, float, str]] = []
    for url, conf, method in sorted(out, key=lambda x: -x[1]):
        cu = canonicalize_url(url)
        if cu in seen:
            continue
        seen.add(cu)
        uniq.append((url, conf, method))
    return uniq[:5]


async def migrate_sensor(sensor: dict[str, Any], new_url: str, *, reason: str, confidence: float, identity_score: float) -> str:
    """Retire the old sensor (successor_id) and create the replacement (predecessor_id). Returns the new sensor id."""
    now = datetime.now(UTC)
    surface = str(sensor["surface"])
    connector = connectors.for_surface(surface, new_url)
    new_id_ = new_id("sensor")
    old_cfg = dict(sensor.get("config") or {})
    new_cfg = {k: v for k, v in old_cfg.items() if k in ("canonical_domain", "vendor", "token")}
    new_cfg.update({"predecessor_id": sensor["id"], "repair": {"version": REPAIR_VERSION, "reason": reason, "confidence": confidence, "identity": identity_score,
                                                                 "at": now.isoformat(), "from_url": sensor["url"]}})
    async with transaction() as conn:
        existing = await fetch_one(conn, "select id from sensors where company_id = :c and canonical_url = :u", c=sensor["company_id"], u=canonicalize_url(new_url))
        if existing is not None:
            new_id_ = str(existing["id"])
            await execute(conn, "update sensors set status = case when status = 'retired' then 'pending' else status end, next_run_at = now(), updated_at = now() where id = :id",
                          id=new_id_)
        else:
            await execute(conn, """
                insert into sensors (id, company_id, surface, connector_id, url, canonical_url, domain, discovery_confidence, discovery_method, quality_score, status, tier,
                                     base_interval_s, current_interval_s, next_run_at, priority, config)
                values (:id, :c, :surface, :conn, :url, :curl, :domain, :conf, 'repair', :q, 'pending', :tier, :base, :base, now(), :prio, cast(:cfg as jsonb))
            """, id=new_id_, c=sensor["company_id"], surface=surface, conn=connector.connector_id, url=new_url, curl=canonicalize_url(new_url),
                          domain=registrable_domain(new_url), conf=round(confidence, 3), q=max(30.0, float(sensor.get("quality_score") or 50) * 0.9),
                          tier=sensor.get("tier") or "D", base=int(sensor.get("base_interval_s") or 86400), prio=float(sensor.get("priority") or 0.5), cfg=jsonb(new_cfg))
        await execute(conn, """update sensors set status = 'retired', retired_at = coalesce(retired_at, now()), claimed_by = null, claimed_at = null,
                               config = config || cast(:cfg as jsonb), updated_at = now() where id = :id""",
                      id=sensor["id"], cfg=jsonb({"successor_id": new_id_, "retired_reason": reason}))
        await execute(conn, "update review_queue set status = 'resolved', resolution = :r, resolved_at = now() where ref_id = :id and status = 'open'",
                      id=sensor["id"], r=f"migrated to {new_id_}")
    return new_id_


async def repair_sensor(sensor: dict[str, Any], *, fetcher: Fetcher, dry_run: bool = False) -> dict[str, Any]:
    session = _Session(fetcher)
    sid = str(sensor["id"])
    result: dict[str, Any] = {"sensor_id": sid, "surface": sensor["surface"], "url": sensor["url"], "action": "none", "requests": 0}
    async with transaction() as conn:
        company = await fetch_one(conn, "select id, canonical_domain, website from companies where id = :id", id=sensor["company_id"])
    if company is None:
        result["action"] = "orphan"
        return result
    # 1. retry the URL itself
    res = await session.get(str(sensor["url"]))
    if res is not None and res.status == 200 and registrable_domain(res.final_url) == registrable_domain(str(sensor["url"])) and not (res.is_html and _soft_404(res)):
        result["action"] = "recovered"
        if not dry_run:
            async with transaction() as conn:
                await execute(conn, """update sensors set status = 'active', consecutive_failures = 0, next_run_at = now(), claimed_by = null, claimed_at = null,
                                       config = config - 'redirect_url', updated_at = now() where id = :id""", id=sid)
        result["requests"] = session.requests
        return result
    if res is not None and registrable_domain(res.final_url) != registrable_domain(str(sensor["url"])):
        sensor = {**sensor, "config": {**(sensor.get("config") or {}), "redirect_url": res.final_url}}
    # 2–4. candidates + identity check
    previous = await _last_text(sensor)
    cands = await _candidates(session, sensor, company)
    result["candidates"] = [{"url": u, "confidence": c, "method": m} for u, c, m in cands]
    best: tuple[str, float, str, float] | None = None
    for url, conf, method in cands[:3]:
        r = await session.get(url)
        if r is None or r.status != 200 or (r.is_html and _soft_404(r)):
            continue
        text = normalize.parse(r.text, url=r.final_url).text if r.is_html else r.text
        same, score = content_identity(previous, text)
        result.setdefault("checked", []).append({"url": url, "identity": score, "same": same})
        if same and (best is None or score > best[3]):
            best = (r.final_url, conf, method, score)
        elif best is None and method == "redirect" and conf >= 0.9 and not previous:
            best = (r.final_url, conf, method, 0.0)
    result["requests"] = session.requests
    if best is not None:
        result["action"] = "migrated" if not dry_run else "would_migrate"
        result["new_url"] = best[0]
        if not dry_run:
            result["new_sensor_id"] = await migrate_sensor(sensor, best[0], reason=best[2], confidence=best[1], identity_score=best[3])
        return result
    # uncertain: one strong candidate → human review; none → leave the sensor to its failure policy
    if cands and not dry_run:
        async with transaction() as conn:
            existing = await fetch_one(conn, "select id from review_queue where kind = 'sensor_migration' and ref_id = :r and status = 'open'", r=sid)
            if existing is None:
                await execute(conn, "insert into review_queue (id, kind, ref_id, company_id, payload) values (:id, 'sensor_migration', :r, :c, cast(:p as jsonb))",
                              id=new_id("review"), r=sid, c=sensor["company_id"], p=jsonb({"sensor_id": sid, "url": sensor["url"], "surface": sensor["surface"],
                                                                                           "candidates": result["candidates"], "checked": result.get("checked", [])}))
        result["action"] = "review"
    elif cands:
        result["action"] = "would_review"
    async with transaction() as conn:
        await execute(conn, "update sensors set config = config || cast(:c as jsonb), updated_at = now() where id = :id", id=sid,
                      c=jsonb({"last_repair_at": datetime.now(UTC).isoformat(), "last_repair_action": result["action"]}))
    return result


async def repair_batch(limit: int = 50, *, fetcher: Fetcher | None = None, dry_run: bool = False) -> dict[str, Any]:
    own = fetcher is None
    fetcher = fetcher or Fetcher()
    if own:
        await fetcher.open()
    stats: dict[str, Any] = {"examined": 0, "recovered": 0, "migrated": 0, "review": 0, "none": 0, "results": []}
    try:
        async with transaction() as conn:
            rows = await fetch_all(conn, """
                select * from sensors
                where (status in ('failing', 'stale', 'redirected') or (status = 'active' and last_failure_class = 'PAGE_REMOVED' and consecutive_failures >= 2))
                  and (config->>'last_repair_at' is null or (config->>'last_repair_at')::timestamptz < now() - interval '1 day')
                  and surface <> :sitemap
                order by quality_score desc, last_run_at nulls first limit :n
            """, n=limit, sitemap=str(Surface.SITEMAP))
        for row in rows:
            stats["examined"] += 1
            try:
                r = await repair_sensor(row, fetcher=fetcher, dry_run=dry_run)
            except Exception as exc:
                log.exception("repair failed", extra={"sensor_id": row.get("id")})
                r = {"sensor_id": row.get("id"), "action": "error", "error": str(exc)[:200]}
            key = r["action"].removeprefix("would_")
            stats[key] = stats.get(key, 0) + 1
            stats["results"].append(r)
    finally:
        if own:
            await fetcher.close()
    return stats


@periodic("sensor-repair", every_s=1800, initial_delay_s=120)
async def repair_task() -> None:
    stats = await repair_batch(limit=50)
    if stats["examined"]:
        log.info("repair batch", extra={k: v for k, v in stats.items() if k != "results"})


def sensor_status_values() -> tuple[str, ...]:
    return tuple(s.value for s in SensorStatus)


__all__ = ["REPAIR_VERSION", "content_identity", "migrate_sensor", "repair_batch", "repair_sensor", "repair_task"]