"""Ingest klipów (feed Swipe): upsert do `clips` + `clip_tags`, joby RedGifs.

Potok celowo prosty, bez `_ingest_iter_into_run`: klipy nie mają dedupu, matchera
performerów ani fingerprintów, a `(source, source_id)` jednoznacznie identyfikuje rekord.
Upsert `ON CONFLICT ... DO UPDATE` odświeża lajki/widoki/adresy i **zdejmuje `dead_at`**
(klip, który źródło znów podaje, żyje). Tagi tylko dokładamy (`DO NOTHING`), źródło ich
nie odbiera.

Dwa tryby:
- `run_redgifs_latest(pages)`: globalne „najnowsze", co godzinę z schedulera. 5 stron =
  ~500 klipów/run, RedGifs publikuje kilkaset na godzinę, więc to łapie prawie wszystko.
- `run_redgifs_niches(...)`: backfill po niszach (społecznościach), ręcznie ze skryptu
  albo z `docker compose run` (patrz notatka o restarcie workera). Każde zapytanie widzi
  maks. 10 000 wyników, dlatego katalog chodzi się per nisza, nie jednym listingiem.
"""
from __future__ import annotations

import logging
from collections.abc import Iterable

from sqlalchemy import text
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.orm import Session

from app.connectors.clips.redgifs import RateLimited, RawClip, RedgifsClient, map_gif
from app.db import session_scope
from app.models.clip import Clip, ClipTag

log = logging.getLogger(__name__)


def upsert_clips(session: Session, clips: Iterable[RawClip]) -> dict[str, int]:
    seen = new = 0
    for c in clips:
        seen += 1
        stmt = (
            pg_insert(Clip)
            .values(
                source=c.source,
                source_id=c.source_id,
                title=c.title,
                creator=c.creator,
                creator_verified=c.creator_verified,
                duration_sec=c.duration_sec,
                width=c.width,
                height=c.height,
                has_audio=c.has_audio,
                content_type=c.content_type,
                niches=c.niches,
                hd_url=c.hd_url,
                sd_url=c.sd_url,
                poster_url=c.poster_url,
                thumb_url=c.thumb_url,
                hls=c.hls,
                likes=c.likes,
                views=c.views,
                published_at=c.published_at,
            )
            .on_conflict_do_update(
                constraint="uq_clips_source_source_id",
                set_={
                    "title": c.title,
                    "creator": c.creator,
                    "creator_verified": c.creator_verified,
                    "duration_sec": c.duration_sec,
                    "has_audio": c.has_audio,
                    "niches": c.niches,
                    "hd_url": c.hd_url,
                    "sd_url": c.sd_url,
                    "poster_url": c.poster_url,
                    "thumb_url": c.thumb_url,
                    "likes": c.likes,
                    "views": c.views,
                    "dead_at": None,
                    "dead_reason": None,
                    "updated_at": text("now()"),
                },
            )
            # xmax = 0 tylko dla świeżo wstawionego wiersza (sztuczka Postgresa na
            # rozróżnienie INSERT od UPDATE w jednym zapytaniu).
            .returning(Clip.id, text("(xmax = 0) AS inserted"))
        )
        row = session.execute(stmt).one()
        clip_id, inserted = row[0], bool(row[1])
        if inserted:
            new += 1
        if c.tags:
            session.execute(
                pg_insert(ClipTag)
                .values([{"clip_id": clip_id, "tag": t} for t in c.tags])
                .on_conflict_do_nothing()
            )
    return {"seen": seen, "new": new, "updated": seen - new}


def _merge(total: dict[str, int], part: dict[str, int]) -> None:
    for k, v in part.items():
        total[k] = total.get(k, 0) + v


def run_redgifs_latest(*, pages: int = 5) -> dict[str, int]:
    """Globalne „najnowsze" RedGifs, `pages` stron po 100. Zatrzymuje się na pierwszej
    stronie w całości już znanej (kolejne są starsze), żeby godzinny run nie mielił
    w kółko tych samych 500 klipów."""
    total: dict[str, int] = {"seen": 0, "new": 0, "updated": 0, "errors": 0}
    client = RedgifsClient()
    try:
        for page in range(1, pages + 1):
            try:
                gifs = client.latest(page=page)
            except Exception as e:
                log.warning("clips/redgifs latest: strona %d padla: %s", page, e)
                total["errors"] += 1
                break
            clips = [c for c in (map_gif(g) for g in gifs) if c is not None]
            if not clips:
                break
            with session_scope() as session:
                part = upsert_clips(session, clips)
            _merge(total, part)
            if part["new"] == 0 and page > 1:
                break
    finally:
        client.close()
    log.info("clips/redgifs latest: %s", total)
    return total


def run_redgifs_niches(
    *, niche_pages: int = 1, per_niche_pages: int = 1, order: str = "latest", skip: int = 0
) -> dict[str, int]:
    """Backfill po niszach: `niche_pages` stron listy nisz (100 na stronę, wg API
    najpopularniejsze pierwsze), dla każdej niszy `per_niche_pages` stron gifów."""
    total: dict[str, int] = {"seen": 0, "new": 0, "updated": 0, "errors": 0, "niches": 0}
    client = RedgifsClient()
    try:
        niche_ids: list[str] = []
        for p in range(1, niche_pages + 1):
            try:
                data = client.niches(page=p)
            except Exception as e:
                log.warning("clips/redgifs niches: lista strona %d padla: %s", p, e)
                total["errors"] += 1
                break
            niche_ids.extend(str(n.get("id")) for n in data.get("niches") or [] if n.get("id"))
        for nid in niche_ids[skip:]:
            total["niches"] += 1
            for page in range(1, per_niche_pages + 1):
                try:
                    gifs = client.niche_gifs(nid, order=order, page=page)
                except RateLimited as e:
                    # Limit trwaly: koniec przebiegu, wznowienie przez --skip <niches>.
                    # Pierwszy backfill (2026-09-14) mielil 40 nisz po kolei w 429.
                    log.warning("clips/redgifs niches: %s; przerywam po %d niszach", e, total["niches"])
                    total["errors"] += 1
                    return total
                except Exception as e:
                    log.warning("clips/redgifs niche %s strona %d padla: %s", nid, page, e)
                    total["errors"] += 1
                    break
                clips = [c for c in (map_gif(g) for g in gifs) if c is not None]
                if not clips:
                    break
                with session_scope() as session:
                    _merge(total, upsert_clips(session, clips))
            if total["niches"] % 20 == 0:
                log.info("clips/redgifs niches: %d nisz, %s", total["niches"], total)
    finally:
        client.close()
    log.info("clips/redgifs niches: %s", total)
    return total


# ---- konserwacja (raz na dobe) ---------------------------------------------------------
_EVENTS_RETENTION_DAYS = 90
_DEAD_CODES = {403, 404, 410}
_REAP_UA = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/152.0.0.0 Safari/537.36"}


def _head_dead(http, url: str) -> str | None:
    """'http 403' itp., gdy plik klipu zniknal; None, gdy zyje albo nie wiadomo (blad sieci)."""
    import httpx

    try:
        r = http.head(url)
    except httpx.HTTPError:
        return None
    return f"http {r.status_code}" if r.status_code in _DEAD_CODES else None


def check_clips_now(clip_ids: list) -> int:
    """Natychmiastowe sprawdzenie klipow zgloszonych przez apke jako zepsute (zdarzenie
    `error`). Ten sam test co reaper, tylko bez czekania na kolejke. Zwraca ile oznaczono."""
    import httpx
    from sqlalchemy import select

    if not clip_ids:
        return 0
    with session_scope() as session:
        rows = session.execute(
            select(Clip.id, Clip.hd_url, Clip.sd_url).where(Clip.id.in_(clip_ids), Clip.dead_at.is_(None))
        ).all()
    marked_ids: list = []
    with httpx.Client(headers=_REAP_UA, timeout=15, follow_redirects=True) as http:
        for cid, hd, sd in rows:
            url = hd or sd
            reason = _head_dead(http, url) if url else None
            if reason is None:
                continue
            with session_scope() as session:
                session.execute(
                    text("UPDATE clips SET dead_at = now(), dead_reason = :r WHERE id = :id AND dead_at IS NULL"),
                    {"r": f"{reason} (apka)", "id": cid},
                )
            marked_ids.append(cid)
    marked = len(marked_ids)
    if marked:
        log.info("clips: %d zgloszonych z apki oznaczone jako martwe", marked)
        sweep_creators(marked_ids)
    return marked


#: Ile klipow jednego przebiegu zamiatania i jak dawno sprawdzony klip bierzemy ponownie.
_SWEEP_CAP = 600
_SWEEP_RECHECK_HOURS = 6


def sweep_creators(dead_clip_ids: list, *, cap: int = _SWEEP_CAP, sleep: float = 0.1) -> int:
    """Po znalezieniu skasowanego klipu sprawdza od razu pozostale zywe klipy jego tworcy.

    Tworcy kasuja zwykle cale konto naraz, a przemial przy 1000 klipow na godzine robi
    okrazenie katalogu w okolo tydzien. 2026-10-06 @thenaughtywoman mial 115 martwych
    i 120 "zywych", z ktorych kazdy oddawal 403, wiec strona tworcy pokazywala puste
    kafelki, a feed podsuwal czarne klipy. Kazdy klip sprawdzamy osobno (HEAD), nic nie
    oznaczamy w ciemno. Sprawdzone zywe dostaja swiezy updated_at, wiec ten sam tworca
    zgloszony z kilku telefonow naraz nie jest mielony w kolko. Zwraca ile oznaczono.
    """
    import time as _time
    from datetime import UTC, datetime, timedelta

    import httpx
    from sqlalchemy import select

    if not dead_clip_ids:
        return 0
    recheck_before = datetime.now(UTC) - timedelta(hours=_SWEEP_RECHECK_HOURS)
    with session_scope() as session:
        creators = [
            c for (c,) in session.execute(
                select(Clip.creator).where(Clip.id.in_(dead_clip_ids), Clip.creator.is_not(None)).distinct()
            ).all()
        ]
        if not creators:
            return 0
        rows = session.execute(
            select(Clip.id, Clip.hd_url, Clip.sd_url)
            .where(
                Clip.creator.in_(creators),
                Clip.dead_at.is_(None),
                Clip.updated_at < recheck_before,
            )
            .order_by(Clip.updated_at.asc())
            .limit(cap)
        ).all()
    dead: list[tuple] = []
    alive: list = []
    with httpx.Client(headers=_REAP_UA, timeout=15, follow_redirects=True) as http:
        for cid, hd, sd in rows:
            url = hd or sd
            if not url:
                continue
            reason = _head_dead(http, url)
            if reason:
                dead.append((cid, reason))
            else:
                alive.append(cid)
            _time.sleep(sleep)
    with session_scope() as session:
        for cid, reason in dead:
            session.execute(
                text("UPDATE clips SET dead_at = now(), dead_reason = :r WHERE id = :id AND dead_at IS NULL"),
                {"r": f"{reason} (tworca)", "id": cid},
            )
        if alive:
            session.execute(text("UPDATE clips SET updated_at = now() WHERE id = ANY(:ids)"), {"ids": alive})
    if rows:
        log.info("clips: zamiatanie %d tworcow, sprawdzone %d, martwe %d", len(creators), len(rows), len(dead))
    return len(dead)


def reap_dead(*, limit: int = 2000, apply: bool = True, sleep: float = 0.2) -> dict[str, int]:
    """Reaper martwych klipow: tworcy kasuja klipy, RedGifs oddaje wtedy 404/410 na pliku.

    HEAD na `hd_url` (albo `sd_url`) dla najdawniej sprawdzanych zywych klipow
    (`updated_at`); 403/404/410 → `dead_at`. Bledy sieci NIE oznaczaja smierci (patrz
    notatka o dwoch kontrolach przy przemiale zrodel). Klip, ktory zrodlo znow poda
    w listingu, upsert odzywia. Sprawdzone i zywe dostaja swiezy `updated_at`, zeby
    kolejny przebieg bral inne.
    """
    import time as _time

    import httpx
    from sqlalchemy import select

    checked = dead = errors = 0
    dead_ids: list[tuple] = []
    with session_scope() as session:
        rows = session.execute(
            select(Clip.id, Clip.hd_url, Clip.sd_url)
            .where(Clip.dead_at.is_(None))
            .order_by(Clip.updated_at.asc())
            .limit(limit)
        ).all()
    with httpx.Client(headers=_REAP_UA, timeout=20, follow_redirects=True) as http:
        for cid, hd, sd in rows:
            url = hd or sd
            if not url:
                continue
            checked += 1
            try:
                r = http.head(url)
            except httpx.HTTPError:
                errors += 1
                continue
            if r.status_code in _DEAD_CODES:
                dead += 1
                dead_ids.append((cid, f"http {r.status_code}"))
            _time.sleep(sleep)
    if apply and rows:
        dead_set = {d[0] for d in dead_ids}
        with session_scope() as session:
            for cid, reason in dead_ids:
                session.execute(
                    text("UPDATE clips SET dead_at = now(), dead_reason = :r WHERE id = :id AND dead_at IS NULL"),
                    {"r": reason, "id": cid},
                )
            alive = [cid for cid, _hd, _sd in rows if cid not in dead_set]
            if alive:
                session.execute(text("UPDATE clips SET updated_at = now() WHERE id = ANY(:ids)"), {"ids": alive})
        dead += sweep_creators(list(dead_set))
    out = {"checked": checked, "dead": dead, "errors": errors, "applied": int(apply)}
    log.info("clips reaper: %s", out)
    return out


def decay_affinity() -> int:
    """Nocne zapominanie: wszystkie wagi ×0,97, |w| < 0,2 znika. Zwraca liczbe urzadzen."""
    from app.clips_feed import decay_weights
    from app.models.clip_feed import ClipAffinity

    n = 0
    with session_scope() as session:
        for row in session.query(ClipAffinity).all():
            row.tag_weights = decay_weights(dict(row.tag_weights or {}))
            row.niche_weights = decay_weights(dict(row.niche_weights or {}))
            row.creator_weights = decay_weights(dict(row.creator_weights or {}))
            n += 1
    return n


def run_clips_maintenance() -> dict[str, int]:
    out: dict[str, int] = {}
    out["affinity_rows"] = decay_affinity()
    with session_scope() as session:
        res = session.execute(
            text("DELETE FROM clip_feed_events WHERE created_at < now() - make_interval(days => :d)"),
            {"d": _EVENTS_RETENTION_DAYS},
        )
        out["events_pruned"] = int(res.rowcount or 0)
    out.update({f"reap_{k}": v for k, v in reap_dead(limit=2000, apply=True).items()})
    log.info("clips maintenance: %s", out)
    return out