534 lines
20 KiB
Python
534 lines
20 KiB
Python
"""Typed data model for the Twitter archive pipeline."""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any, Self
|
|
import json
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PostMetrics:
|
|
"""Post engagement counters (replies / reposts / likes / views)."""
|
|
reply_count: int
|
|
repost_count: int
|
|
like_count: int
|
|
view_count: int | None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PostData:
|
|
"""A single tweet post extracted from a Twitter archive."""
|
|
post_id: str
|
|
author: str
|
|
handle: str
|
|
text: str
|
|
timestamp: str
|
|
media_urls: tuple[str, ...]
|
|
reply_to_id: str | None
|
|
quote_of_id: str | None
|
|
metrics: PostMetrics
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ThreadData:
|
|
"""A complete Twitter thread (root post + replies)."""
|
|
root_post_id: str
|
|
posts: tuple[PostData, ...]
|
|
source_url: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class MediaLocationCheck:
|
|
"""Origin/fallback URL HEAD probe outcome stored on a location."""
|
|
status: int | None
|
|
result: str
|
|
detail: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class MediaLocation:
|
|
"""A single media location record (local / origin / fallback / mirror / derived)."""
|
|
kind: str
|
|
id: str | None = None
|
|
local_path: Path | None = None
|
|
url: str | None = None
|
|
bytes: int | None = None
|
|
sha256: str | None = None
|
|
media_type: str | None = None
|
|
integrity: str | None = None
|
|
verified_at: str | None = None
|
|
provider: str | None = None
|
|
availability: str | None = None
|
|
checked_at: str | None = None
|
|
checked_status: int | None = None
|
|
recorded_at: str | None = None
|
|
uploaded_at: str | None = None
|
|
confirmed_unavailable_at: str | None = None
|
|
check: MediaLocationCheck | None = None
|
|
|
|
@classmethod
|
|
def from_dict(cls, raw: dict[str, Any]) -> Self:
|
|
if "kind" not in raw:
|
|
raise KeyError("MediaLocation.from_dict: 'kind' is required")
|
|
# Wire uses 'path' (basename) or 'local_path' (Path string) interchangeably
|
|
path_raw = raw.get("local_path")
|
|
if path_raw is None:
|
|
path_raw = raw.get("path")
|
|
check_raw = raw.get("check")
|
|
check: MediaLocationCheck | None = None
|
|
if check_raw is not None:
|
|
if not isinstance(check_raw, dict):
|
|
raise TypeError(
|
|
f"check: expected dict, got {type(check_raw).__name__}"
|
|
)
|
|
if "result" not in check_raw or "detail" not in check_raw:
|
|
raise KeyError("MediaLocation.from_dict: 'check' requires 'result' and 'detail'")
|
|
check = MediaLocationCheck(
|
|
status=_optional_int(check_raw.get("status"), "check.status"),
|
|
result=_required_str(check_raw.get("result"), "check.result"),
|
|
detail=_required_str(check_raw.get("detail"), "check.detail"),
|
|
)
|
|
return cls(
|
|
kind=_required_str(raw["kind"], "kind"),
|
|
id=_optional_str(raw.get("id"), "id"),
|
|
local_path=_optional_path(path_raw, "local_path"),
|
|
url=_optional_str(raw.get("url"), "url"),
|
|
bytes=_optional_int(raw.get("bytes"), "bytes"),
|
|
sha256=_optional_str(raw.get("sha256"), "sha256"),
|
|
media_type=_optional_str(raw.get("media_type"), "media_type"),
|
|
integrity=_optional_str(raw.get("integrity"), "integrity"),
|
|
verified_at=_optional_str(raw.get("verified_at"), "verified_at"),
|
|
provider=_optional_str(raw.get("provider"), "provider"),
|
|
availability=_optional_str(raw.get("availability"), "availability"),
|
|
checked_at=_optional_str(raw.get("checked_at"), "checked_at"),
|
|
checked_status=_optional_int(raw.get("checked_status"), "checked_status"),
|
|
recorded_at=_optional_str(raw.get("recorded_at"), "recorded_at"),
|
|
uploaded_at=_optional_str(raw.get("uploaded_at"), "uploaded_at"),
|
|
confirmed_unavailable_at=_optional_str(
|
|
raw.get("confirmed_unavailable_at"), "confirmed_unavailable_at"
|
|
),
|
|
check=check,
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DerivedFromRef:
|
|
"""Pointer to the parent (orig) media item for derived items (crt, ocr, etc.)."""
|
|
post_id: str
|
|
media_id: str
|
|
role: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Publication:
|
|
"""Selected location for a media item; embedded as 'publication' on the wire."""
|
|
selected_location_id: str
|
|
selected_at: str
|
|
reason: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class MediaItem:
|
|
"""A single media entry attached to a post (orig / crt / crt_outline / denoise / ocr).
|
|
|
|
For legacy (pre-v2) wire shapes, ``locations`` may be empty and ``filename``
|
|
plus ``url`` carry the original on-disk filename and pre-migration fallback
|
|
URL. v2 items leave both ``None`` and put the URL inside a ``MediaLocation``.
|
|
"""
|
|
post_id: str
|
|
media_id: str
|
|
locations: tuple[MediaLocation, ...] = ()
|
|
kind: str | None = None
|
|
handle: str | None = None
|
|
role: str | None = None
|
|
filename: str | None = None
|
|
url: str | None = None
|
|
embed: bool | None = None
|
|
caption: str | None = None
|
|
derived_from: DerivedFromRef | None = None
|
|
publication: Publication | None = None
|
|
|
|
@classmethod
|
|
def from_dict(cls, raw: dict[str, Any]) -> Self:
|
|
for required in ("post_id", "media_id"):
|
|
if required not in raw:
|
|
raise KeyError(f"MediaItem.from_dict: '{required}' is required")
|
|
locations_raw = raw.get("locations")
|
|
locations: tuple[MediaLocation, ...] = ()
|
|
if locations_raw is not None:
|
|
if not isinstance(locations_raw, list):
|
|
raise TypeError(
|
|
f"MediaItem.from_dict: 'locations' must be list, got {type(locations_raw).__name__}"
|
|
)
|
|
locations = tuple(MediaLocation.from_dict(loc) for loc in locations_raw)
|
|
# Wire uses 'role' as the kind identifier; accept 'kind' for direct callers
|
|
kind_raw = raw.get("role")
|
|
if kind_raw is None:
|
|
kind_raw = raw.get("kind")
|
|
if kind_raw is None:
|
|
raise KeyError("MediaItem.from_dict: 'role' or 'kind' is required")
|
|
kind = _required_str(kind_raw, "role")
|
|
derived_from_raw = raw.get("derived_from")
|
|
derived_from: DerivedFromRef | None = None
|
|
if derived_from_raw is not None:
|
|
if not isinstance(derived_from_raw, dict):
|
|
raise TypeError(
|
|
f"derived_from: expected dict, got {type(derived_from_raw).__name__}"
|
|
)
|
|
derived_from = DerivedFromRef(
|
|
post_id=_required_str(
|
|
derived_from_raw.get("post_id"), "derived_from.post_id"
|
|
),
|
|
media_id=_required_str(
|
|
derived_from_raw.get("media_id"), "derived_from.media_id"
|
|
),
|
|
role=_required_str(derived_from_raw.get("role"), "derived_from.role"),
|
|
)
|
|
publication_raw = raw.get("publication")
|
|
publication: Publication | None = None
|
|
if publication_raw is not None:
|
|
if not isinstance(publication_raw, dict):
|
|
raise TypeError(
|
|
f"publication: expected dict, got {type(publication_raw).__name__}"
|
|
)
|
|
publication = Publication(
|
|
selected_location_id=_required_str(
|
|
publication_raw.get("selected_location_id"),
|
|
"publication.selected_location_id",
|
|
),
|
|
selected_at=_required_str(
|
|
publication_raw.get("selected_at"), "publication.selected_at"
|
|
),
|
|
reason=_required_str(publication_raw.get("reason"), "publication.reason"),
|
|
)
|
|
return cls(
|
|
post_id=_required_str(raw["post_id"], "post_id"),
|
|
media_id=_required_str(raw["media_id"], "media_id"),
|
|
kind=kind,
|
|
handle=_optional_str(raw.get("handle"), "handle"),
|
|
role=_optional_str(raw.get("role"), "role"),
|
|
filename=_optional_str(raw.get("filename"), "filename"),
|
|
url=_optional_str(raw.get("url"), "url"),
|
|
locations=locations,
|
|
embed=_optional_bool(raw.get("embed"), "embed"),
|
|
caption=_optional_str(raw.get("caption"), "caption"),
|
|
derived_from=derived_from,
|
|
publication=publication,
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class MediaManifest:
|
|
"""A whole media.json document (one thread). schema_version is the wire key."""
|
|
root_post_id: str
|
|
items: tuple[MediaItem, ...]
|
|
schema_version: int | None = None
|
|
captured_at: str | None = None
|
|
|
|
@classmethod
|
|
def from_dict(cls, raw: dict[str, Any]) -> Self:
|
|
for required in ("root_post_id", "items"):
|
|
if required not in raw:
|
|
raise KeyError(f"MediaManifest.from_dict: '{required}' is required")
|
|
items_raw = raw["items"]
|
|
if not isinstance(items_raw, list):
|
|
raise TypeError(
|
|
f"MediaManifest.from_dict: 'items' must be list, got {type(items_raw).__name__}"
|
|
)
|
|
return cls(
|
|
root_post_id=_required_str(raw["root_post_id"], "root_post_id"),
|
|
items=tuple(MediaItem.from_dict(item) for item in items_raw),
|
|
schema_version=_optional_int(raw.get("schema_version"), "schema_version"),
|
|
captured_at=_optional_str(raw.get("captured_at"), "captured_at"),
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class LegacyThreadData:
|
|
"""Pre-v2 thread_data.json shape paired with media items during migration."""
|
|
root_post_id: str
|
|
posts: tuple[PostData, ...]
|
|
source_url: str
|
|
media_items: tuple[MediaItem, ...] = ()
|
|
|
|
@classmethod
|
|
def from_dict(cls, raw: dict[str, Any]) -> Self:
|
|
for required in ("root_post_id", "posts", "source_url"):
|
|
if required not in raw:
|
|
raise KeyError(f"LegacyThreadData.from_dict: '{required}' is required")
|
|
posts_raw = raw["posts"]
|
|
if not isinstance(posts_raw, list):
|
|
raise TypeError(
|
|
f"LegacyThreadData.from_dict: 'posts' must be list, got {type(posts_raw).__name__}"
|
|
)
|
|
media_items_raw = raw.get("media_items")
|
|
media_items: tuple[MediaItem, ...] = ()
|
|
if media_items_raw is not None:
|
|
if not isinstance(media_items_raw, list):
|
|
raise TypeError(
|
|
f"LegacyThreadData.from_dict: 'media_items' must be list, got {type(media_items_raw).__name__}"
|
|
)
|
|
media_items = tuple(MediaItem.from_dict(item) for item in media_items_raw)
|
|
return cls(
|
|
root_post_id=_required_str(raw["root_post_id"], "root_post_id"),
|
|
posts=tuple(_parse_post(p) for p in posts_raw),
|
|
source_url=_required_str(raw["source_url"], "source_url"),
|
|
media_items=media_items,
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class LegacyManifest:
|
|
"""Pre-v2 corpus inventory shape used during media migration."""
|
|
threads: tuple[LegacyThreadData, ...]
|
|
schema_version: int | None = None
|
|
generated_at: str | None = None
|
|
counts: dict[str, int] | None = None
|
|
frozen_hashes: dict[str, str] | None = None
|
|
|
|
@classmethod
|
|
def from_dict(cls, raw: dict[str, Any]) -> Self:
|
|
if "threads" not in raw:
|
|
raise KeyError("LegacyManifest.from_dict: 'threads' is required")
|
|
threads_raw = raw["threads"]
|
|
if not isinstance(threads_raw, list):
|
|
raise TypeError(
|
|
f"LegacyManifest.from_dict: 'threads' must be list, got {type(threads_raw).__name__}"
|
|
)
|
|
counts: dict[str, int] | None = _coerce_str_int_map(raw.get("counts"), "counts")
|
|
frozen_hashes: dict[str, str] | None = _coerce_str_str_map(
|
|
raw.get("frozen_hashes"), "frozen_hashes"
|
|
)
|
|
return cls(
|
|
threads=tuple(LegacyThreadData.from_dict(t) for t in threads_raw),
|
|
schema_version=_optional_int(raw.get("schema_version"), "schema_version"),
|
|
generated_at=_optional_str(raw.get("generated_at"), "generated_at"),
|
|
counts=counts,
|
|
frozen_hashes=frozen_hashes,
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class OriginCheck:
|
|
"""Outcome of a HEAD probe against a media item's origin URL.
|
|
|
|
``status`` is the HTTP status code returned by the probe (or ``None`` if
|
|
the request errored before a response arrived). ``result`` is the
|
|
classifier label (``"available"`` for 2xx/3xx, ``"error"`` otherwise).
|
|
``confirms_unavailable`` distinguishes transient failures like 429 from
|
|
definitive unavailability so the caller does not flip the location's
|
|
availability state on a rate-limit blip.
|
|
"""
|
|
checked_at: str
|
|
status: int | None
|
|
result: str
|
|
detail: str
|
|
confirms_unavailable: bool
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class LegacyMediaJson:
|
|
"""Per-thread ``media.json`` shape consumed by ``media_migrate.py`` and ``media_audit.py``.
|
|
|
|
Holds both the pre-v2 (legacy) and v2 (canonical) wire shapes so the
|
|
migration and audit entry points can share one boundary type. The
|
|
``items`` field is parsed as ``MediaItem`` instances; for legacy items
|
|
``locations`` is empty and ``filename`` plus ``url`` carry the
|
|
on-disk filename and pre-migration fallback URL.
|
|
|
|
``mirrors`` hold opaque destination/auth fields that are intentionally
|
|
not yet typed. This is the only non-typed dict in the data model — it
|
|
is consumed only by iteration, never by attribute access, and we
|
|
deliberately do not commit to a structure until the backup subsystem
|
|
settles on a wire shape.
|
|
"""
|
|
schema_version: int | None
|
|
root_post_id: str
|
|
items: tuple[MediaItem, ...]
|
|
captured_at: str | None
|
|
mirrors: dict[str, Any] | None = None
|
|
frozen: bool | None = None
|
|
|
|
@classmethod
|
|
def from_dict(cls, raw: dict[str, Any]) -> Self:
|
|
if "root_post_id" not in raw:
|
|
raise KeyError("LegacyMediaJson.from_dict: 'root_post_id' is required")
|
|
items_raw = raw.get("items")
|
|
if items_raw is None:
|
|
items_raw = []
|
|
if not isinstance(items_raw, list):
|
|
raise TypeError(
|
|
f"LegacyMediaJson.from_dict: 'items' must be list, got {type(items_raw).__name__}"
|
|
)
|
|
mirrors = _coerce_mirrors(raw.get("mirrors"), "mirrors")
|
|
return cls(
|
|
schema_version=_optional_int(raw.get("schema_version"), "schema_version"),
|
|
root_post_id=_required_str(raw["root_post_id"], "root_post_id"),
|
|
items=tuple(MediaItem.from_dict(item) for item in items_raw),
|
|
captured_at=_optional_str(raw.get("captured_at"), "captured_at"),
|
|
mirrors=mirrors,
|
|
frozen=_optional_bool(raw.get("frozen"), "frozen"),
|
|
)
|
|
|
|
|
|
def _coerce_str_int_map(value: object, field: str) -> dict[str, int] | None:
|
|
if value is None:
|
|
return None
|
|
if not isinstance(value, dict):
|
|
raise TypeError(f"{field}: expected dict, got {type(value).__name__}")
|
|
out: dict[str, int] = {}
|
|
for key, item in value.items():
|
|
if isinstance(item, bool) or not isinstance(item, int):
|
|
raise TypeError(f"{field}[{key}]: expected int")
|
|
out[str(key)] = item
|
|
return out
|
|
|
|
|
|
def _coerce_str_str_map(value: object, field: str) -> dict[str, str] | None:
|
|
if value is None:
|
|
return None
|
|
if not isinstance(value, dict):
|
|
raise TypeError(f"{field}: expected dict, got {type(value).__name__}")
|
|
out: dict[str, str] = {}
|
|
for key, item in value.items():
|
|
if not isinstance(item, str):
|
|
raise TypeError(f"{field}[{key}]: expected str")
|
|
out[str(key)] = item
|
|
return out
|
|
|
|
|
|
def _coerce_mirrors(value: object, field: str) -> dict[str, Any] | None:
|
|
"""Accept the legacy ``mirrors: list[dict]`` wire and coerce to the typed mirrors shape.
|
|
|
|
Legacy producers (pre-LegacyMediaJson) wrote ``mirrors`` as a list of
|
|
destination entries; current producers may write it as a dict keyed by
|
|
destination id. Both shapes are accepted; the returned dict is keyed
|
|
by ``destination_id`` when present on each entry, falling back to the
|
|
string index for entries without one.
|
|
"""
|
|
if value is None:
|
|
return None
|
|
if isinstance(value, dict):
|
|
return value
|
|
if isinstance(value, list):
|
|
coerced: dict = {}
|
|
for index, entry in enumerate(value):
|
|
if isinstance(entry, dict) and "destination_id" in entry:
|
|
coerced[str(entry["destination_id"])] = entry
|
|
else:
|
|
coerced[str(index)] = entry
|
|
return coerced
|
|
raise TypeError(f"{field}: expected dict or list, got {type(value).__name__}")
|
|
|
|
|
|
def _norm_handle(raw: str) -> str:
|
|
h = (raw or "").strip()
|
|
return h[1:] if h.startswith("@") else h
|
|
|
|
|
|
def _norm_id(raw: object) -> str | None:
|
|
if raw is None:
|
|
return None
|
|
s = str(raw).strip()
|
|
return s or None
|
|
|
|
|
|
def _required_str(value: object, name: str) -> str:
|
|
if not isinstance(value, str):
|
|
raise TypeError(f"{name}: expected str, got {type(value).__name__}")
|
|
return value
|
|
|
|
|
|
def _optional_str(value: object, field: str) -> str | None:
|
|
if value is None:
|
|
return None
|
|
if not isinstance(value, str):
|
|
raise TypeError(f"{field}: expected str, got {type(value).__name__}")
|
|
return value
|
|
|
|
|
|
def _optional_int(value: object, field: str) -> int | None:
|
|
if value is None:
|
|
return None
|
|
if isinstance(value, bool) or not isinstance(value, int):
|
|
raise TypeError(f"{field}: expected int, got {type(value).__name__}")
|
|
return value
|
|
|
|
|
|
def _optional_bool(value: object, field: str) -> bool | None:
|
|
if value is None:
|
|
return None
|
|
if not isinstance(value, bool):
|
|
raise TypeError(f"{field}: expected bool, got {type(value).__name__}")
|
|
return value
|
|
|
|
|
|
def _optional_path(value: object, field: str) -> Path | None:
|
|
if value is None:
|
|
return None
|
|
if isinstance(value, Path):
|
|
return value
|
|
if isinstance(value, str):
|
|
return Path(value)
|
|
raise TypeError(f"{field}: expected str or Path, got {type(value).__name__}")
|
|
|
|
|
|
def _parse_post(raw: object) -> PostData:
|
|
if not isinstance(raw, dict):
|
|
raise TypeError(f"post entry: expected dict, got {type(raw).__name__}")
|
|
m = raw.get("metrics") or {}
|
|
if not isinstance(m, dict):
|
|
raise TypeError(f"post entry metrics: expected dict, got {type(m).__name__}")
|
|
view = m.get("view_count")
|
|
media_urls_raw = raw.get("media_urls") or []
|
|
if not isinstance(media_urls_raw, list):
|
|
raise TypeError(
|
|
f"post entry media_urls: expected list, got {type(media_urls_raw).__name__}"
|
|
)
|
|
return PostData(
|
|
post_id=str(raw.get("post_id") or ""),
|
|
author=str(raw.get("author") or ""),
|
|
handle=_norm_handle(str(raw.get("handle") or "")),
|
|
text=str(raw.get("text") or ""),
|
|
timestamp=str(raw.get("timestamp") or ""),
|
|
media_urls=tuple(str(u) for u in media_urls_raw),
|
|
reply_to_id=_norm_id(raw.get("reply_to_id")),
|
|
quote_of_id=_norm_id(raw.get("quote_of_id")),
|
|
metrics=PostMetrics(
|
|
reply_count=int(m.get("reply_count") or 0),
|
|
repost_count=int(m.get("repost_count") or 0),
|
|
like_count=int(m.get("like_count") or 0),
|
|
view_count=int(view) if view is not None else None,
|
|
),
|
|
)
|
|
|
|
|
|
def load_thread(path: Path) -> ThreadData:
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
posts: list[PostData] = []
|
|
for pd in data.get("posts", []):
|
|
m = pd.get("metrics") or {}
|
|
view = m.get("view_count")
|
|
posts.append(PostData(
|
|
post_id=str(pd.get("post_id") or ""),
|
|
author=str(pd.get("author") or ""),
|
|
handle=_norm_handle(str(pd.get("handle") or "")),
|
|
text=str(pd.get("text") or ""),
|
|
timestamp=str(pd.get("timestamp") or ""),
|
|
media_urls=tuple(pd.get("media_urls") or []),
|
|
reply_to_id=_norm_id(pd.get("reply_to_id")),
|
|
quote_of_id=_norm_id(pd.get("quote_of_id")),
|
|
metrics=PostMetrics(
|
|
reply_count=int(m.get("reply_count") or 0),
|
|
repost_count=int(m.get("repost_count") or 0),
|
|
like_count=int(m.get("like_count") or 0),
|
|
view_count=int(view) if view is not None else None,
|
|
),
|
|
))
|
|
return ThreadData(
|
|
root_post_id=str(data.get("root_post_id") or ""),
|
|
posts=tuple(posts),
|
|
source_url=str(data.get("source_url") or ""),
|
|
)
|