"""DB-bound ingest orchestration: parse -> normalize -> dedup -> upsert -> sighting ->

watchlist eval -> audit. Row-level resilient: a bad row is recorded, never fatal."""

from __future__ import annotations

from datetime import timezone

from sqlalchemy import select

from sqlalchemy.orm import Session

from app.core.config import settings

from app.core.storage import get_storage

from app.models import (

    Alert, AlertRule, Camera, Import, LicensePlate, Location, PlateRead, Sighting, Source,

    Vehicle, Watchlist, WatchlistEntry, WatchlistHit, AuditLog,

)

from app.services.normalize import (

    NormalizedRead, build_alias_map, dedup_hash, normalize_read,

)

from app.services.parsers import get_parser

from app.services.watchlist import WatchEntry, match_all

def _get_or_create_camera(db: Session, source_id: str | None, n: NormalizedRead) -> Camera | None:

    external_id = n.external_camera_id

    if not external_id:

        return None

    cam = db.scalar(select(Camera).where(

        Camera.source_id == source_id, Camera.external_camera_id == external_id))

    if cam:

        if cam.location_id is None and n.lat is not None:

            loc = Location(label=external_id, lat=n.lat, lon=n.lon)

            db.add(loc)

            db.flush()

            cam.location_id = loc.id

        return cam

    location_id = None

    if n.lat is not None:

        loc = Location(label=external_id, lat=n.lat, lon=n.lon)

        db.add(loc)

        db.flush()

        location_id = loc.id

    cam = Camera(source_id=source_id, external_camera_id=external_id, name=external_id,

                 direction_facing=n.direction, location_id=location_id)

    db.add(cam)

    db.flush()

    return cam

def _get_or_create_plate(db: Session, n: NormalizedRead) -> LicensePlate:

    plate = db.scalar(select(LicensePlate).where(

        LicensePlate.plate_norm == n.plate_norm, LicensePlate.state == n.state))

    if plate:

        return plate

    plate = LicensePlate(plate_raw=n.plate_raw, plate_norm=n.plate_norm, state=n.state)

    db.add(plate)

    db.flush()

    return plate

def _get_or_create_vehicle(db: Session, plate: LicensePlate, n: NormalizedRead) -> Vehicle:

    veh = db.scalar(select(Vehicle).where(Vehicle.primary_plate_id == plate.id))

    if veh:

        for attr, val in (("make", n.make), ("model", n.model), ("color", n.color), ("body_type", n.body_type)):

            if val and not getattr(veh, attr):

                setattr(veh, attr, val)

        return veh

    veh = Vehicle(primary_plate_id=plate.id, make=n.make, model=n.model,

                  color=n.color, body_type=n.body_type)

    db.add(veh)

    db.flush()

    return veh

def _active_watch_entries(db: Session) -> list[WatchEntry]:

    rows = db.scalars(

        select(WatchlistEntry).join(Watchlist).where(Watchlist.active.is_(True))

    ).all()

    return [WatchEntry(id=e.id, plate_norm=e.plate_norm, partial_plate=e.partial_plate,

                       vehicle_desc=e.vehicle_desc) for e in rows]

def process_import(db: Session, import_id: str) -> dict:

    imp = db.get(Import, import_id)

    if imp is None:

        return {"error": "import not found"}

    imp.status = "processing"

    db.flush()

    org = (db.get(Source, imp.source_id).organization if imp.source_id else None) or "default"

    try:

        data = get_storage().get(imp.raw_blob_ref)

    except Exception as e:

        imp.status, imp.error = "error", f"cannot read blob: {e}"

        db.commit()

        return {"error": imp.error}

    parser = get_parser(imp.filename, imp.content_type)

    result = parser.parse(data)

    alias = build_alias_map(result.headers)

    return _ingest_rows(db, result.rows, alias, imp, org, list(result.errors))

def _ingest_rows(db: Session, rows, alias: dict, imp: Import, org: str,

                 errors: list[str]) -> dict:

    """Shared per-row pipeline: normalize -> dedup -> upsert plate/vehicle/camera ->

    plate_read + sighting -> watchlist eval. Used by file imports, connectors, and live stream."""

    watch = _active_watch_entries(db)

    inserted = duplicates = skipped = hits = 0

    for raw in rows:

        try:

            n = normalize_read(raw, alias, source_tz=timezone.utc)

        except Exception as e:

            errors.append(f"row {raw.source_row}: normalize error {e}")

            skipped += 1

            continue

        if n is None:

            skipped += 1

            continue

        cam = _get_or_create_camera(db, imp.source_id, n)

        dh = dedup_hash(org, n.plate_norm, cam.id if cam else None, n.read_ts,

                        settings.dedup_window_seconds)

        existing = db.scalar(select(PlateRead).where(PlateRead.dedup_hash == dh))

        if existing:

            existing.burst_count += 1

            duplicates += 1

            continue

        plate = _get_or_create_plate(db, n)

        veh = _get_or_create_vehicle(db, plate, n)

        read = PlateRead(

            import_id=imp.id, source_id=imp.source_id, camera_id=cam.id if cam else None,

            plate_id=plate.id, plate_norm=n.plate_norm, read_ts=n.read_ts,

            direction=n.direction, confidence=n.confidence, lane=n.lane,

            lat=n.lat, lon=n.lon, hotlist_hit=n.hotlist_hit, incident_ref=n.incident_ref,

            raw_read=n.raw, dedup_hash=dh,

        )

        db.add(read)

        db.flush()

        sighting = Sighting(plate_read_id=read.id, vehicle_id=veh.id,

                            camera_id=cam.id if cam else None, read_ts=n.read_ts,

                            confidence=n.confidence)

        db.add(sighting)

        db.flush()

        inserted += 1

        vehicle_desc = {"make": veh.make, "color": veh.color, "body_type": veh.body_type}

        for m in match_all(watch, n.plate_norm, vehicle_desc):

            wh = WatchlistHit(watchlist_entry_id=m["entry_id"], plate_read_id=read.id,

                              sighting_id=sighting.id, matched_on=m["matched_on"])

            db.add(wh)

            db.flush()

            hits += 1

            _maybe_alert(db, m["entry_id"], wh, n)

    imp.row_count = inserted

    imp.status = "done"

    imp.error = "\n".join(errors[:50]) if errors else None

    db.add(AuditLog(actor_user_id=imp.created_by, action="import.process",

                    entity_type="import", entity_id=imp.id,

                    meta={"inserted": inserted, "duplicates": duplicates,

                          "skipped": skipped, "watchlist_hits": hits}))

    db.commit()

    return {"import_id": imp.id, "inserted": inserted, "duplicates": duplicates,

            "skipped": skipped, "watchlist_hits": hits, "errors": errors[:50]}

def ingest_live_reads(db: Session, source_id: str | None, reads: list[dict],

                      label: str = "live", created_by: str | None = None) -> dict:

    """Ingest already-structured read dicts (from a stream POST or an API connector) through the

    same normalize/dedup/watchlist pipeline as file imports. Creates a synthetic Import record

    so every live read is still traceable to a batch."""

    import hashlib

    import json as _json

    from app.services.normalize import RawRead

    org = (db.get(Source, source_id).organization if source_id else None) or "default"

    payload = _json.dumps(reads, sort_keys=True, default=str).encode()

    sha = hashlib.sha256(payload + label.encode()).hexdigest()

    imp = db.scalar(select(Import).where(Import.sha256 == sha))

    if imp:

        return {"import_id": imp.id, "deduplicated": True, "row_count": imp.row_count}

    imp = Import(source_id=source_id, filename=f"{label}.json", content_type="application/json",

                 sha256=sha, raw_blob_ref=None, status="processing", created_by=created_by)

    db.add(imp)

    db.flush()

    raw_rows = [RawRead(fields=r, source_row=i) for i, r in enumerate(reads, start=1)]

    headers = sorted({k for r in reads for k in r.keys()})

    alias = build_alias_map(headers)

    return _ingest_rows(db, raw_rows, alias, imp, org, [])

def _maybe_alert(db: Session, entry_id: str, hit: WatchlistHit, n: NormalizedRead) -> None:

    entry = db.get(WatchlistEntry, entry_id)

    if not entry:

        return

    rules = db.scalars(select(AlertRule).where(

        AlertRule.watchlist_id == entry.watchlist_id, AlertRule.active.is_(True))).all()

    for rule in rules or [None]:

        cond = (rule.condition if rule else {}) or {}

        min_conf = cond.get("min_confidence")

        if min_conf is not None and (n.confidence or 0) < float(min_conf):

            continue

        db.add(Alert(alert_rule_id=rule.id if rule else None, watchlist_hit_id=hit.id,

                     status="new", payload={"plate": n.plate_norm, "matched_on": hit.matched_on,

                                            "read_ts": n.read_ts.isoformat()}))