from __future__ import annotations

import hashlib
import re
import time as time_module
from contextlib import contextmanager
from dataclasses import asdict, dataclass
from datetime import date, datetime, time, timezone

import httpx
from sqlalchemy import select, text
from sqlalchemy.orm import Session
from rf4_research.official_parser import RecordsContractError, parse_official_records

from .models import (
    Bait, BaitKind, CatchReport, Fish, ImportRecordEvent, ImportStatus, ModerationStatus,
    OfficialRecordImport, SourceType, Waterbody,
)


USER_AGENT = "RF4-Spotter/0.1 (public records importer)"


class ImportSourceError(ValueError):
    pass


class ImportAlreadyRunning(RuntimeError):
    pass


@dataclass(frozen=True, slots=True)
class RawRecord:
    region: str
    category: str
    player: str | None
    fish: str
    weight_g: int
    waterbody: str
    bait: str | None
    record_date: date


@dataclass(frozen=True, slots=True)
class FetchResult:
    records: list[RawRecord] | None
    status_code: int
    etag: str | None
    last_modified: str | None
    content_type: str | None
    response_bytes: int


def normalize(value: str) -> str:
    return " ".join(value.replace("\xa0", " ").replace("–", "-").replace("—", "-").split()).casefold()


def slugify(value: str) -> str:
    compact = re.sub(r"[^a-z0-9а-яё]+", "-", normalize(value), flags=re.IGNORECASE).strip("-")
    return compact or hashlib.sha256(value.encode()).hexdigest()[:16]


def external_id(record: RawRecord) -> str:
    parts = [record.region, record.category, record.player or "", record.fish, str(record.weight_g), record.waterbody, record.bait or "", record.record_date.isoformat()]
    return hashlib.sha256("|".join(normalize(part) for part in parts).encode()).hexdigest()


def parse_html(html: str, *, region: str, category: str) -> list[RawRecord]:
    try:
        rows = parse_official_records(html)
    except RecordsContractError as exc:
        raise ImportSourceError(str(exc)) from exc
    return [RawRecord(
        region=region.upper(), category=category, player=row.player, fish=row.fish,
        weight_g=row.weight_g, waterbody=row.waterbody, bait=row.bait,
        record_date=row.record_date,
    ) for row in rows]


def fetch_records(
    url: str, *, region: str, category: str,
    etag: str | None = None, last_modified: str | None = None,
) -> FetchResult:
    headers = {"User-Agent": USER_AGENT, "Accept": "text/html"}
    if etag:
        headers["If-None-Match"] = etag
    if last_modified:
        headers["If-Modified-Since"] = last_modified
    with httpx.Client(timeout=20, follow_redirects=True, headers=headers) as client:
        for attempt in range(3):
            try:
                response = client.get(url)
                metadata = {
                    "status_code": response.status_code,
                    "etag": response.headers.get("etag"),
                    "last_modified": response.headers.get("last-modified"),
                    "content_type": response.headers.get("content-type"),
                    "response_bytes": len(response.content),
                }
                if response.status_code == 304:
                    return FetchResult(records=None, **metadata)
                response.raise_for_status()
                if "text/html" not in response.headers.get("content-type", ""):
                    raise ImportSourceError("source did not return HTML")
                return FetchResult(records=parse_html(response.text, region=region, category=category), **metadata)
            except (httpx.HTTPError, ImportSourceError):
                if attempt == 2:
                    raise
                time_module.sleep(2 ** attempt)
    raise AssertionError("unreachable")


def _lock_key(url: str, region: str, category: str) -> int:
    digest = hashlib.sha256(f"{url}|{region.upper()}|{category}".encode()).digest()
    return int.from_bytes(digest[:8], byteorder="big", signed=True)


@contextmanager
def _official_import_lock(session: Session, *, url: str, region: str, category: str):
    bind = session.get_bind()
    if bind.dialect.name != "postgresql":
        yield
        return
    connection = bind.connect()
    key = _lock_key(url, region, category)
    try:
        acquired = bool(connection.scalar(text("SELECT pg_try_advisory_lock(:key)"), {"key": key}))
    except Exception:
        connection.close()
        raise
    if not acquired:
        connection.close()
        raise ImportAlreadyRunning("official import is already running for this source and category")
    try:
        yield
    finally:
        try:
            connection.execute(text("SELECT pg_advisory_unlock(:key)"), {"key": key})
        finally:
            connection.close()


def import_records(session: Session, *, url: str, region: str, category: str, html: str | None = None) -> OfficialRecordImport:
    with _official_import_lock(session, url=url, region=region, category=category):
        return _import_records_locked(session, url=url, region=region, category=category, html=html)


def _import_records_locked(session: Session, *, url: str, region: str, category: str, html: str | None = None) -> OfficialRecordImport:
    run = OfficialRecordImport(started_at=datetime.now(timezone.utc), status=ImportStatus.running, source_url=url, rows_seen=0, rows_created=0, rows_updated=0)
    session.add(run)
    session.commit()
    try:
        if html is not None:
            records = parse_html(html, region=region, category=category)
        else:
            previous = session.scalar(
                select(OfficialRecordImport).where(
                    OfficialRecordImport.source_url == url,
                    OfficialRecordImport.status == ImportStatus.success,
                ).order_by(OfficialRecordImport.started_at.desc()).limit(1)
            )
            fetched = fetch_records(
                url, region=region, category=category,
                etag=previous.response_etag if previous else None,
                last_modified=previous.response_last_modified if previous else None,
            )
            run.response_status = fetched.status_code
            run.response_etag = fetched.etag or (previous.response_etag if previous else None)
            run.response_last_modified = fetched.last_modified or (previous.response_last_modified if previous else None)
            run.response_content_type = fetched.content_type
            run.response_bytes = fetched.response_bytes
            if fetched.records is None:
                run.not_modified = True
                run.status = ImportStatus.success
                run.finished_at = datetime.now(timezone.utc)
                session.commit()
                return run
            records = fetched.records
        run.rows_seen = len(records)
        for raw in records:
            key = external_id(raw)
            report = session.scalar(select(CatchReport).where(CatchReport.source_external_id == key))
            fish = _fish(session, raw.fish)
            waterbody = _waterbody(session, raw.waterbody)
            bait = _bait(session, raw.bait) if raw.bait else None
            payload = asdict(raw) | {"record_date": raw.record_date.isoformat()}
            caught = datetime.combine(raw.record_date, time(), tzinfo=timezone.utc)
            now = datetime.now(timezone.utc)
            if report is None:
                report = CatchReport(fish=fish, waterbody=waterbody, bait=bait, spot=None, weight_g=raw.weight_g, caught_at=caught, reported_at=now, player_name=raw.player, source_type=SourceType.official_record, source_url=url, source_external_id=key, source_confidence=100, moderation_status=ModerationStatus.approved, raw_payload=payload)
                session.add(report)
                session.add(ImportRecordEvent(catch_report=report, import_run=run, event_type="created", created_at=now))
                run.rows_created += 1
            else:
                report.raw_payload = payload
                report.source_url = url
                session.add(ImportRecordEvent(catch_report=report, import_run=run, event_type="updated", created_at=now))
                run.rows_updated += 1
        run.status = ImportStatus.success
        run.finished_at = datetime.now(timezone.utc)
        session.commit()
        return run
    except Exception as exc:
        session.rollback()
        run = session.get(OfficialRecordImport, run.id)
        run.status = ImportStatus.failed
        run.finished_at = datetime.now(timezone.utc)
        run.error_summary = str(exc)[:1000]
        session.commit()
        raise


def _fish(session: Session, name: str) -> Fish:
    item = session.scalar(select(Fish).where(Fish.name_ru == name))
    if item is None:
        item = Fish(slug=_unique_slug(session, Fish, name), name_ru=name, trophy_weight_g=None)
        session.add(item)
    return item


def _waterbody(session: Session, name: str) -> Waterbody:
    item = session.scalar(select(Waterbody).where(Waterbody.name_ru == name))
    if item is None:
        item = Waterbody(slug=_unique_slug(session, Waterbody, name), name_ru=name, unlock_level=None)
        session.add(item)
    return item


def _bait(session: Session, name: str) -> Bait:
    normalized = normalize(name)
    item = session.scalar(select(Bait).where(Bait.normalized_name == normalized))
    if item is None:
        item = Bait(name=name, normalized_name=normalized, kind=BaitKind.unknown)
        session.add(item)
    return item


def _unique_slug(session: Session, model: type[Fish] | type[Waterbody], name: str) -> str:
    base = slugify(name)
    candidate = base
    index = 2
    while session.scalar(select(model.id).where(model.slug == candidate)) is not None:
        candidate = f"{base}-{index}"
        index += 1
    return candidate