peter
bc8fe2cc6c
A subscription pulled in from YouTube left an explored channel flagged as an exploration's, so its later videos came in hidden from the Library with their watched marks waiting for good; a subscribe landing while an exploration's videos came in left those flagged too. One step, settle_explored, now holds the rule by who follows the channel now: the subscribe here, the subscription import, the watched-channel videos and the explore page all end with it. Migration 0116 clears what the gap left behind. The list's "may still grow" count no longer counts a channel it already shows from a stub row, and the comments on the cleanup's keep rule and on the in-memory daily shares say what they do.
498 lines
21 KiB
Python
498 lines
21 KiB
Python
"""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
|