"""A Google Takeout export, imported: what the user watched on YouTube becomes watched here, and their
YouTube Watch later list joins theirs. The YouTube API reads neither of them, so the export is the
only way in.

Two files of the export are read, each uploaded on its own: `history/watch-history.json` and
`playlists/Watch later-videos.csv`. Their videos are kept (`TakeoutItem`) and the import job applies
what it can; a video the catalog does not know yet waits for it. The watched videos' channels also
make the list of channels the user watches but does not follow (`queries.watched_channels`): the
job asks YouTube how long those videos are, and what the channels are, for it.

The format, measured on a real export (2026-10): the history is a JSON list, one entry per view,
newest first. A view of a video has a `titleUrl` with `watch?v=<id>` (on www.youtube.com or
music.youtube.com) and a `time`, with or without milliseconds. The channel is the first `subtitles`
entry's url, absent for a video since removed. An ad is an entry whose `details` say "From Google
Ads". The title is in the account's language, so nothing here reads it. A Short is a plain
`watch?v=` view, told apart only once the catalog has probed the video. The Watch later csv is a
header row (in the account's language) and `<id>,<ISO time added>` rows.
"""

import codecs
import csv
import io
import json
import re
import threading
from collections.abc import Iterator
from contextlib import contextmanager
from dataclasses import dataclass, field
from datetime import datetime
from itertools import chain

from sqlalchemy.orm import Session

from app import logs, quota
from app.bodylimit import TAKEOUT_MAX_BYTES
from app.db import SessionLocal
from app.models import User
from app.ratelimit import RateLimiter
from app.reader import queries, service
from app.reader.models import TakeoutKind
from app.sync.runner import get_service_user
from app.sync.subscriptions import describe_channels
from app.sync.videos import ensure_videos, parse_iso8601_duration
from app.youtube.client import YouTubeClient, YouTubeError

log = logs.get_logger(__name__)

# The largest file the upload takes (its edge cap and the route's own are one number).
MAX_BYTES = TAKEOUT_MAX_BYTES
# The most videos a Watch later file may hold: YouTube's own cap on one playlist. Each unknown one is
# looked up on the shared API budget, so a longer list is not one YouTube made.
WATCH_LATER_MAX = 5000
# The most Takeout videos one user keeps: twenty times the real export measured here (~9,000 videos
# across both files), so it only stops uploads of made-up ones piling up for good.
ROWS_MAX = 200_000

_VIDEO_ID = re.compile(r"^[A-Za-z0-9_-]{11}$")
_WATCH_URL = re.compile(
    r"^https?://(?:www\.|music\.|m\.)?youtube\.com/watch\?v=([A-Za-z0-9_-]{11})"
)
_CHANNEL_URL = re.compile(r"/channel/(UC[A-Za-z0-9_-]{22})")


class TakeoutError(ValueError):
    """The file is not one the import reads; `reason` is the client's word for why."""

    def __init__(self, reason: str):
        super().__init__(reason)
        self.reason = reason


@dataclass
class Parsed:
    """One file's videos, `{video id: (time, channel id)}`, and what it left out, counted."""

    kind: TakeoutKind
    videos: dict[str, tuple[datetime, str | None]] = field(default_factory=dict)
    entries: int = 0
    ads: int = 0
    other: int = 0  # not a view of a video (a post, a story) or a video with no link left
    malformed: int = 0


def _time(raw: object) -> datetime | None:
    if not isinstance(raw, str):
        return None
    try:
        at = datetime.fromisoformat(raw.replace("Z", "+00:00"))
    # silent-ok: the row is counted as malformed, and the count is in the upload's answer.
    except ValueError:
        return None
    return at if at.tzinfo is not None else None


def _keep(parsed: Parsed, vid: str, at: datetime, channel: str | None) -> None:
    old = parsed.videos.get(vid)
    if old is None or at > old[0]:
        parsed.videos[vid] = (at, channel or (old[1] if old else None))


def _history(data: Iterator[object]) -> Parsed:
    parsed = Parsed(TakeoutKind.WATCHED)
    for entry in data:
        parsed.entries += 1
        if not isinstance(entry, dict):
            parsed.malformed += 1
            continue
        details = entry.get("details")
        if isinstance(details, list) and any(
            isinstance(d, dict) and d.get("name") == "From Google Ads" for d in details
        ):
            parsed.ads += 1
            continue
        url = entry.get("titleUrl")
        match = _WATCH_URL.match(url) if isinstance(url, str) else None
        if match is None:
            parsed.other += 1
            continue
        at = _time(entry.get("time"))
        if at is None:
            parsed.malformed += 1
            continue
        channel = None
        subtitles = entry.get("subtitles")
        if isinstance(subtitles, list) and subtitles and isinstance(subtitles[0], dict):
            found = _CHANNEL_URL.search(str(subtitles[0].get("url", "")))
            channel = found.group(1) if found else None
        _keep(parsed, match.group(1), at, channel)
    return parsed


def _watch_later(text: str) -> Parsed:
    parsed = Parsed(TakeoutKind.WATCH_LATER)
    rows = (row for row in csv.reader(io.StringIO(text)) if any(cell.strip() for cell in row))
    first = next(rows, None)
    if first is None or len(first) < 2:
        raise TakeoutError("not_takeout")
    # The first row is the header, in the account's language, unless it is already a video.
    for row in chain([first], rows) if _VIDEO_ID.match(first[0].strip()) else rows:
        parsed.entries += 1
        vid = row[0].strip()
        at = _time(row[1].strip()) if len(row) > 1 else None
        if not _VIDEO_ID.match(vid) or at is None:
            parsed.malformed += 1
            continue
        _keep(parsed, vid, at, None)
        if len(parsed.videos) > WATCH_LATER_MAX:
            raise TakeoutError("too_many")
    return parsed


def _skip(text: str, at: int) -> int:
    """The first index at or after `at` that is not JSON whitespace."""
    while at < len(text) and text[at] in " \t\n\r":
        at += 1
    return at


# How much of the file is decoded at a time. Decoded whole, the text costs up to four bytes a
# character (one emoji in a title is enough), several times the file on top of its bytes.
_WINDOW = 1024 * 1024


class _Window:
    """The file's text, decoded a window at a time: `text` holds what is not read yet from `at` on.
    The bytes are the caller's, so only the window is this reader's memory."""

    def __init__(self, body: bytes, start: int):
        self._body, self._next = body, start
        self._decoder = codecs.getincrementaldecoder("utf-8")()  # strict: bad UTF-8 raises
        self.text, self.at = "", 0

    @property
    def done(self) -> bool:
        return self._next >= len(self._body)

    def more(self, at_least: int = 0) -> None:
        """Decode the next window (at least `at_least` bytes), dropping what is read already."""
        self.text, self.at = self.text[self.at :], 0
        size = max(_WINDOW, at_least)
        chunk = self._body[self._next : self._next + size]
        self._next += len(chunk)
        self.text += self._decoder.decode(chunk, final=self.done)

    def skip(self) -> None:
        """Past the whitespace, reading on as far as it goes."""
        while True:
            self.at = _skip(self.text, self.at)
            if self.at < len(self.text) or self.done:
                return
            self.more()

    def startswith(self, char: str) -> bool:
        self.skip()
        return self.text.startswith(char, self.at)


def _decode(decoder: json.JSONDecoder, text: str, at: int) -> tuple[object, int] | None:
    """The JSON value at `at` and where it ends, or None where none reads (yet)."""
    try:
        return decoder.raw_decode(text, at)
    # silent-ok: not a failure here — the caller reads on, or refuses the file once it has all.
    except json.JSONDecodeError:
        return None


def _json_list(body: bytes, start: int) -> Iterator[object]:
    """The entries of the JSON list at byte `start` (its "["), one at a time: the whole tree is never
    built, nor the whole text, so a file of a million tiny entries costs a window and one entry, not
    a million dicts at once."""
    decoder = json.JSONDecoder()
    window = _Window(body, start + 1)  # past the "[" the caller saw
    if window.startswith("]"):
        window.at += 1
    else:
        while True:
            window.skip()
            read = _decode(decoder, window.text, window.at)
            # An entry cut by the window's edge fails, or (a number) reads short: read on and try
            # again, with a window twice what is left, so one entry larger than any window is read
            # in a few tries, not one a window.
            if read is None or (read[1] == len(window.text) and not window.done):
                if window.done:
                    raise ValueError("not a JSON list")
                window.more(2 * (len(window.text) - window.at))
                continue
            entry, window.at = read
            yield entry
            if window.startswith(","):
                window.at += 1
            elif window.startswith("]"):
                window.at += 1
                break
            else:
                raise ValueError("not a JSON list")
    window.skip()
    if window.at != len(window.text):
        raise ValueError("text after the JSON list")


# The largest Watch later file read: YouTube's longest list is some 200 KB of rows, so a file ten
# times that is not one — and, unlike the history, it is decoded whole.
CSV_MAX_BYTES = 2 * 1024 * 1024
_BOM = codecs.BOM_UTF8


def parse(body: bytes) -> Parsed:
    """Read an uploaded Takeout file: the watch history (JSON) or the Watch later list (csv), told
    apart by their content, not their name. Raises `TakeoutError` for anything else, and for a file
    with nothing to import in it."""
    start = len(_BOM) if body.startswith(_BOM) else 0
    while start < len(body) and body[start] in b" \t\n\r\f\v":
        start += 1
    first = body[start : start + 1]
    if first in (b"{", b"<"):
        # Another JSON file of the export, or the history's HTML variant.
        raise TakeoutError("not_takeout")
    if first != b"[" and len(body) - start > CSV_MAX_BYTES:
        raise TakeoutError("too_many")
    try:
        # A bad UTF-8 sequence is a ValueError too (UnicodeDecodeError), read below as not a file.
        parsed = (
            _history(_json_list(body, start))
            if first == b"["
            else _watch_later(body[start:].decode("utf-8"))
        )
    # A file the import cannot read, however it fails: bad JSON (a ValueError, as is a number too
    # long to convert), nesting deeper than the decoder recurses, a csv field past csv's limit.
    except (ValueError, RecursionError, csv.Error) as exc:
        if isinstance(exc, TakeoutError):
            raise
        raise TakeoutError("not_takeout") from None
    if not parsed.videos:
        raise TakeoutError("no_videos")
    return parsed


# One import per user at a time: a second file uploaded while the first one's run is under way runs
# after it (the scheduler's ad-hoc queue), but the Shorts job settles the same users' rows on a
# thread of its own, and the two would settle the same rows. One API process runs the scheduler, so a
# lock in this module holds it (as the channel backfill's does).
_user_locks: dict[int, threading.Lock] = {}
_user_locks_guard = threading.Lock()


@contextmanager
def _one_import(user_id: int) -> Iterator[None]:
    with _user_locks_guard:
        lock = _user_locks.setdefault(user_id, threading.Lock())
    with lock:
        yield


def run_target(user_id: int) -> str:
    """The import run's target, `job_runs.target`: one user's import."""
    return f"user:{user_id}"


def target_title(db: Session, target: str) -> str | None:
    """Who an import run's target is, for the admin's run lists: the user's name, else their email."""
    _, _, raw = target.partition("user:")
    user = db.get(User, int(raw)) if raw.isdigit() else None
    return (user.display_name or user.email) if user is not None else None


class TooMany(Exception):
    """The upload would take the user past `ROWS_MAX` kept videos."""


def stage(db: Session, user: User, parsed: Parsed) -> None:
    """Keep an uploaded file's videos for the import, unless that would take the user past
    `ROWS_MAX` (`TooMany`). The caller commits."""
    service.lock_takeout(db, user)  # the count and the insert are one user's, not two uploads' each
    if (
        queries.takeout_rows_besides(db, user, parsed.kind, parsed.videos) + len(parsed.videos)
        > ROWS_MAX
    ):
        raise TooMany
    service.stage_takeout(db, user, parsed.kind, parsed.videos)


# How many unknown Watch later videos one user may have looked up on the shared API budget in a day:
# two full Watch later lists (two units a hundred ids). The rest wait for the next day's runs. Both
# shares pace the work, they do not bound its cost: they are held in memory, so a restart starts
# them afresh, which only brings the rest forward — every video is asked about once (a looked-up
# row is marked), and `quota.backfill_reserve_reached` still guards the shared budget.
_lookups = RateLimiter(max_events=2 * WATCH_LATER_MAX, window_seconds=24 * 3600)
# How many watched videos' lengths one user may ask about in a day for the "watched but not followed"
# list (a unit a fifty). A share of its own: a large history asks for thousands, and on the Watch
# later share it would leave the export's Watch later file waiting a day behind it.
_lengths = RateLimiter(max_events=2 * WATCH_LATER_MAX, window_seconds=24 * 3600)


def _put_in_watch_later(db: Session, user: User) -> dict:
    """Put the user's waiting Watch later videos into theirs here: the ones the catalog knows at once,
    the others once YouTube was asked for them — when a YouTube-connected account can ask, today's
    budget is above the backfill reserve, and the user's own daily share of lookups allows it. A video
    not asked about yet waits for a later run. The lookup commits what it fetched; the caller commits
    the rest."""
    waiting = queries.takeout_waiting(db, user, TakeoutKind.WATCH_LATER)
    if not waiting:
        return {}
    ids = [vid for _, vid in waiting]
    known = queries.known_video_ids(db, ids)
    ask = [vid for vid in ids if vid not in known][:WATCH_LATER_MAX]
    asked: set[str] = set()
    fetcher = get_service_user(db) if ask else None
    if (
        fetcher is not None
        and not quota.backfill_reserve_reached()
        and _lookups.room(str(user.id)) >= len(ask)
    ):
        with quota.attribute(user.id, quota.QuotaAction.VIDEOS_LOOKUP):
            with YouTubeClient(db, fetcher) as yt:
                ensure_videos(db, yt, ask)
        # Counted once done: a lookup that failed spent none of the share (one run per user at a
        # time, so the room asked above is still there).
        _lookups.allow(str(user.id), cost=len(ask))
        asked = set(ask)
    # Only the rows read above: one a second upload staged meanwhile was never asked about.
    return service.apply_takeout_watch_later(db, user, waiting, asked)


def _ask_lengths(db: Session, user: User) -> dict:
    """Ask YouTube how long the user's watched videos are that the "watched but not followed" list
    still needs, and keep the answer — on the same terms as the Watch later lookups, on a daily
    share of its own. What is not asked yet (no YouTube-connected account, the reserve reached, the
    share spent) waits for a later run, and the list says it is not complete. A YouTube failure is
    logged and left for a later run too: it is no reason to fail the import it follows. The caller
    commits."""
    ask = queries.takeout_lengths_to_ask(db, user)
    fetcher = get_service_user(db) if ask else None
    if fetcher is None or quota.backfill_reserve_reached():
        return {}
    ask = ask[: int(_lengths.room(str(user.id)))]
    if not ask:
        return {}
    try:
        with quota.attribute(user.id, quota.QuotaAction.VIDEOS_LOOKUP):
            with YouTubeClient(db, fetcher) as yt:
                items = yt.get_videos(ask)
    except YouTubeError as exc:
        log.warning(
            "Takeout: asking the watched videos' lengths failed: %s",
            exc,
            extra=logs.event("takeout.lengths.failed", user_id=user.id, reason=exc.reason),
        )
        return {}
    _lengths.allow(str(user.id), cost=len(ask))
    lengths = {
        it["id"]: parse_iso8601_duration(it.get("contentDetails", {}).get("duration"))
        for it in items
        if it.get("id")
    }
    service.record_takeout_lengths(db, user, ask, lengths)
    return {"lengths_asked": len(ask)}


def _describe_channels(db: Session, user: User) -> dict:
    """Bring the channels the user's list holds into the catalog with their details (name, picture,
    description), the ones it lacks: without them the list cannot show a channel. A YouTube failure
    is logged and left for a later run. The caller commits."""
    need = queries.watched_channels_to_describe(db, user)
    fetcher = get_service_user(db) if need else None
    if fetcher is None or quota.backfill_reserve_reached():
        return {}
    try:
        with quota.attribute(user.id, quota.QuotaAction.CHANNELS_DISCOVER):
            with YouTubeClient(db, fetcher) as yt:
                describe_channels(db, yt, need)
    except YouTubeError as exc:
        db.rollback()
        log.warning(
            "Takeout: describing the watched channels failed: %s",
            exc,
            extra=logs.event("takeout.channels.failed", user_id=user.id, reason=exc.reason),
        )
        return {}
    return {"channels_described": len(need)}


def _channel_list(db: Session, user: User) -> dict:
    """The "watched but not followed" list's part of a run: the lengths, landed, then the channels.
    Commits each."""
    result = _ask_lengths(db, user)
    db.commit()
    result |= _describe_channels(db, user)
    db.commit()
    return result


def run_import(user_id: int) -> dict:
    """The import job: apply what the user's kept Takeout rows can say now, as one recorded run.
    The watched part commits before the Watch later part asks YouTube for the videos the catalog
    lacks, so a YouTube failure there loses nothing already marked; the "watched but not followed"
    list's lookups come last. Returns the counts."""
    with _one_import(user_id), SessionLocal() as db:
        user = db.get(User, user_id)
        if user is None:
            return {"skipped": "no user"}
        result = service.apply_takeout_watched(db, user)
        db.commit()
        result |= _put_in_watch_later(db, user)
        db.commit()
        result |= _channel_list(db, user)
        outcomes = queries.takeout_outcomes(db, user)
        result["waiting"] = outcomes.get((TakeoutKind.WATCHED, None), 0)
        if outcomes.get((TakeoutKind.WATCH_LATER, None)):
            result["watch_later_waiting"] = outcomes[(TakeoutKind.WATCH_LATER, None)]
        return result


def apply_waiting(db: Session) -> dict:
    """Settle what every user's import left waiting and can be settled now, one commit each: a
    watched video the catalog has come to know and the Shorts probe has cleared (by its job or a
    search's), the Watch later videos a run could not put in yet, and what the "watched but not
    followed" list still needs asked of YouTube. Run after each run of the
    `shorts` job, which is what clears the most of them. Returns its counts, or an empty dict when
    nothing was waiting that could move."""
    marked = added = asked = failed = 0
    ready = set(queries.takeout_users_ready(db))
    later = set(queries.takeout_users_watch_later_waiting(db))
    lengths = set(queries.takeout_users_with_list_work(db))
    for user_id in sorted(ready | later | lengths):
        user = db.get(User, user_id)
        if user is None:
            continue
        # One user's failure (a lookup YouTube refused) is theirs: it neither fails the job that
        # runs this nor keeps the users after them waiting. What it did not settle waits.
        try:
            with _one_import(user_id):
                if user_id in ready:
                    marked += service.apply_takeout_watched(db, user)["watched"]
                    db.commit()
                if user_id in later:
                    added += _put_in_watch_later(db, user).get("watch_later_added", 0)
                    db.commit()
                if user_id in lengths:
                    asked += _channel_list(db, user).get("lengths_asked", 0)
        except Exception:
            db.rollback()
            failed += 1
            log.exception(
                "Takeout settling failed for user %s",
                user_id,
                extra=logs.event("takeout.settle.failed", user_id=user_id),
            )
    out = {}
    if ready:
        out["takeout_watched"] = marked
    if added:
        out["takeout_watch_later_added"] = added
    if asked:
        out["takeout_lengths_asked"] = asked
    if failed:
        out["takeout_failed"] = failed
    return out