ed/Threadwell
1
0 Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

another test of refresh with diff op.

Browse Source
This commit is contained in:
ed committed 2026-08-24 15:45:13 -04:00
1 parent 7e4ed4b303
commit 3805685d8f
10 files changed
+640 -73
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

scripts/twitter/emit_archive.py
+10 -3
View File
@@ -544,11 +544,18 @@ def emit(
if reuse_dir is None and force:
_existing_ids, spines = collect_existing_ids(vault)
reuse_dir = spines.get(thread.root_post_id)
# Only accept a candidate whose parent handle matches the
# OP handle. A thread previously archived at a different
# author's dir (e.g. an old tip-as-root archive of a cross-
# author conversation) is treated as orphan, not a reuse.
candidate = spines.get(thread.root_post_id)
if candidate is not None and candidate.parent.name == handle:
reuse_dir = candidate
if reuse_dir is None:
for post in thread.posts:
reuse_dir = spines.get(post.post_id)
if reuse_dir is not None:
candidate = spines.get(post.post_id)
if candidate is not None and candidate.parent.name == handle:
reuse_dir = candidate
break
if reuse_dir is None and slug:
candidate = vault / "archive" / "threads" / handle / slug
scripts/twitter/models.py
+211
View File
@@ -2,11 +2,13 @@ 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
@@ -15,6 +17,7 @@ class PostMetrics:
@dataclass(frozen=True)
class PostData:
"""A single tweet post extracted from a Twitter archive."""
post_id: str
author: str
handle: str
@@ -28,11 +31,149 @@ class PostData:
@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 MediaLocation:
"""A single media location record (local / remote / fallback / mirror / derived)."""
kind: str
local_path: Path | None = None
url: str | None = None
bytes: int | None = None
sha256: str | None = None
checked_at: str | None = None
checked_status: int | 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")
return cls(
kind=_required_str(raw["kind"], "kind"),
local_path=_optional_path(raw.get("local_path"), "local_path"),
url=_optional_str(raw.get("url"), "url"),
bytes=_optional_int(raw.get("bytes"), "bytes"),
sha256=_optional_str(raw.get("sha256"), "sha256"),
checked_at=_optional_str(raw.get("checked_at"), "checked_at"),
checked_status=_optional_int(raw.get("checked_status"), "checked_status"),
)
@dataclass(frozen=True)
class MediaItem:
"""A single media entry attached to a post (image / crt / crt_outline / denoise / ocr)."""
post_id: str
media_id: str
kind: str
locations: tuple[MediaLocation, ...]
embed: bool | None = None
caption: str | None = None
@classmethod
def from_dict(cls, raw: dict[str, Any]) -> Self:
for required in ("post_id", "media_id", "kind", "locations"):
if required not in raw:
raise KeyError(f"MediaItem.from_dict: '{required}' is required")
locations_raw = raw["locations"]
if not isinstance(locations_raw, list):
raise TypeError(
f"MediaItem.from_dict: 'locations' must be list, got {type(locations_raw).__name__}"
)
return cls(
post_id=_required_str(raw["post_id"], "post_id"),
media_id=_required_str(raw["media_id"], "media_id"),
kind=_required_str(raw["kind"], "kind"),
locations=tuple(MediaLocation.from_dict(loc) for loc in locations_raw),
embed=_optional_bool(raw.get("embed"), "embed"),
caption=_optional_str(raw.get("caption"), "caption"),
)
@dataclass(frozen=True)
class MediaManifest:
"""A whole media.json document (one thread)."""
root_post_id: str
items: tuple[MediaItem, ...]
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),
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
@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__}"
)
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"),
)
def _norm_handle(raw: str) -> str:
h = (raw or "").strip()
return h[1:] if h.startswith("@") else h
@@ -45,6 +186,76 @@ def _norm_id(raw: object) -> str | None:
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] = []
scripts/twitter/tests/test_emit_op_owned.py
+177
View File
@@ -339,5 +339,182 @@ class OpOwnedEmitTipModeTests(unittest.TestCase):
self.assertEqual(len(thread_data["posts"]), 7)
class OrphansNotReusedTests(unittest.TestCase):
"""An old archive keyed by a non-OP handle must not be reused.
A pre-existing tip-as-root archive of a cross-author conversation
is treated as orphan; the OP-owned emit produces a fresh dir at
the OP handle."""
def setUp(self) -> None:
self.tmp = Path(tempfile.mkdtemp(prefix="op_owned_orphan_"))
self.input_dir = self.tmp / "input"
self.input_dir.mkdir()
self.vault = self.tmp / "vault"
(self.vault / "archive" / "threads").mkdir(parents=True)
(self.vault / "assets" / "threads").mkdir(parents=True)
def tearDown(self) -> None:
shutil.rmtree(self.tmp, ignore_errors=True)
def test_old_archive_at_non_op_handle_is_not_reused(self) -> None:
"""Pre-existing dir keyed by the tip's handle (lottes) must
not be reused when the OP is a different handle (kenpex).
The emit creates a fresh dir at the OP handle."""
# Simulate a pre-existing tip-as-root archive at NOTimothyLottes
legacy_archive = (
self.vault
/ "archive"
/ "threads"
/ "NOTimothyLottes"
/ "2023-04-26-legacy-slug"
)
legacy_archive.mkdir(parents=True)
(legacy_archive / "index.md").write_text(
(
"---\n"
"title: legacy\n"
"handle: NOTimothyLottes\n"
'post_id: "1651268028795961344"\n'
"draft: false\n"
"---\n\n"
"old single-post archive\n"
),
encoding="utf-8",
newline="\n",
)
(legacy_archive.parent / "index.md").write_text(
(
"---\n"
"title: NOTimothyLottes\n"
"---\n\n"
"- [[archive/threads/NOTimothyLottes/2023-04-26-legacy-slug]]\n"
),
encoding="utf-8",
newline="\n",
)
# Write the cross-author thread data (5 posts, kenpex OP)
thread_data = {
"root_post_id": "1651282559287042048",
"source_url": "https://x.com/i/status/1651282559287042048",
"posts": [
{
"post_id": "1650678968255913985",
"author": "c0de517e",
"handle": "kenpex",
"text": "Nerds are crazy",
"timestamp": "2023-04-25 01:51:48",
"media_urls": [],
"reply_to_id": None,
"quote_of_id": None,
"metrics": {
"reply_count": 0,
"repost_count": 0,
"like_count": 0,
"view_count": 0,
},
},
{
"post_id": "1651056827839180800",
"author": "wadetb",
"handle": "wadetb",
"text": "Forth is asm-level",
"timestamp": "2023-04-26 02:53:17",
"media_urls": [],
"reply_to_id": "1650678968255913985",
"quote_of_id": None,
"metrics": {
"reply_count": 0,
"repost_count": 0,
"like_count": 0,
"view_count": 0,
},
},
{
"post_id": "1651253961524142081",
"author": "kenpex",
"handle": "kenpex",
"text": "Yes of course",
"timestamp": "2023-04-26 15:56:37",
"media_urls": [],
"reply_to_id": "1651056827839180800",
"quote_of_id": None,
"metrics": {
"reply_count": 0,
"repost_count": 0,
"like_count": 0,
"view_count": 0,
},
},
{
"post_id": "1651268028795961344",
"author": "NOTimothyLottes",
"handle": "NOTimothyLottes",
"text": "Custom forth in few K",
"timestamp": "2023-04-26 16:52:31",
"media_urls": [],
"reply_to_id": "1651253961524142081",
"quote_of_id": None,
"metrics": {
"reply_count": 0,
"repost_count": 0,
"like_count": 0,
"view_count": 0,
},
},
{
"post_id": "1651282559287042048",
"author": "wvo",
"handle": "wvo",
"text": "Why not 1K?",
"timestamp": "2023-04-26 17:50:16",
"media_urls": [],
"reply_to_id": "1651268028795961344",
"quote_of_id": None,
"metrics": {
"reply_count": 0,
"repost_count": 0,
"like_count": 0,
"view_count": 0,
},
},
],
}
(self.input_dir / "thread_data.json").write_text(
json.dumps(thread_data, indent=2),
encoding="utf-8",
newline="\n",
)
emit(
input_dir=self.input_dir,
vault=self.vault,
slug=None,
archived="2026-08-24",
force=True,
tip="1651282559287042048",
)
archive_root = self.vault / "archive" / "threads"
# The OP-owned archive lives under kenpex, not NOTimothyLottes
self.assertTrue(
(archive_root / "kenpex").is_dir(),
"OP-owned archive must live under the OP handle's dir",
)
kenpex_thread_dirs = sorted(
d.name
for d in (archive_root / "kenpex").iterdir()
if d.is_dir()
)
self.assertEqual(len(kenpex_thread_dirs), 1)
# The legacy NOTimothyLottes dir is left as orphan; this
# test does not auto-remove it (manual cleanup is operator
# responsibility).
self.assertTrue(
(archive_root / "NOTimothyLottes" / "2023-04-26-legacy-slug").is_dir(),
)
if __name__ == "__main__":
unittest.main()
scripts/twitter/tests/test_models_boundary.py
+183
View File
@@ -0,0 +1,183 @@
from __future__ import annotations
import unittest
from pathlib import Path
from twitter.models import (
LegacyManifest,
LegacyThreadData,
MediaItem,
MediaLocation,
MediaManifest,
PostData,
)
class MediaLocationTests(unittest.TestCase):
"""Tests for the MediaLocation boundary dataclass."""
def test_local_round_trip_with_path_bytes_sha(self) -> None:
loc = MediaLocation.from_dict({
"kind": "local",
"local_path": "asset.png",
"bytes": 100,
"sha256": "abc",
})
self.assertEqual(loc.kind, "local")
self.assertEqual(loc.local_path, Path("asset.png"))
self.assertEqual(loc.bytes, 100)
self.assertEqual(loc.sha256, "abc")
self.assertIsNone(loc.url)
self.assertIsNone(loc.checked_at)
self.assertIsNone(loc.checked_status)
def test_remote_round_trip_with_url(self) -> None:
loc = MediaLocation.from_dict({
"kind": "remote",
"url": "https://example.com/asset.png",
})
self.assertEqual(loc.kind, "remote")
self.assertEqual(loc.url, "https://example.com/asset.png")
self.assertIsNone(loc.local_path)
self.assertIsNone(loc.bytes)
self.assertIsNone(loc.sha256)
def test_empty_dict_raises_keyerror(self) -> None:
with self.assertRaises(KeyError):
MediaLocation.from_dict({})
def test_bytes_bool_raises_typeerror(self) -> None:
with self.assertRaisesRegex(TypeError, "bytes"):
MediaLocation.from_dict({
"kind": "local",
"local_path": "/tmp/x",
"bytes": True,
"sha256": "abc",
})
class MediaItemTests(unittest.TestCase):
"""Tests for the MediaItem boundary dataclass."""
def test_image_round_trip_with_locations_tuple(self) -> None:
item = MediaItem.from_dict({
"post_id": "100",
"media_id": "AAA",
"kind": "image",
"locations": [
{"kind": "local", "local_path": "asset.png", "bytes": 100, "sha256": "abc"},
{"kind": "remote", "url": "https://example.com/asset.png"},
],
})
self.assertEqual(item.post_id, "100")
self.assertEqual(item.media_id, "AAA")
self.assertEqual(item.kind, "image")
self.assertIsInstance(item.locations, tuple)
self.assertEqual(len(item.locations), 2)
self.assertEqual(item.locations[0].kind, "local")
self.assertEqual(item.locations[1].kind, "remote")
self.assertIsNone(item.embed)
self.assertIsNone(item.caption)
def test_locations_not_list_raises_typeerror(self) -> None:
with self.assertRaisesRegex(TypeError, "locations"):
MediaItem.from_dict({
"post_id": "p",
"media_id": "m",
"kind": "image",
"locations": "not a list",
})
def test_embed_not_bool_raises_typeerror(self) -> None:
with self.assertRaisesRegex(TypeError, "embed"):
MediaItem.from_dict({
"post_id": "p",
"media_id": "m",
"kind": "image",
"locations": [],
"embed": "true",
})
class MediaManifestTests(unittest.TestCase):
"""Tests for the MediaManifest boundary dataclass."""
def test_minimal_manifest_round_trip(self) -> None:
manifest = MediaManifest.from_dict({
"root_post_id": "100",
"items": [
{
"post_id": "100",
"media_id": "AAA",
"kind": "image",
"locations": [
{"kind": "local", "local_path": "x.png", "bytes": 1, "sha256": "a"},
],
}
],
})
self.assertEqual(manifest.root_post_id, "100")
self.assertIsInstance(manifest.items, tuple)
self.assertEqual(len(manifest.items), 1)
self.assertEqual(manifest.items[0].media_id, "AAA")
self.assertIsNone(manifest.captured_at)
def test_items_not_list_raises_typeerror(self) -> None:
with self.assertRaisesRegex(TypeError, "items"):
MediaManifest.from_dict({"root_post_id": "p", "items": {}})
class LegacyManifestTests(unittest.TestCase):
"""Tests for the LegacyManifest boundary dataclass."""
def test_round_trip_with_threads(self) -> None:
manifest = LegacyManifest.from_dict({
"threads": [
{
"root_post_id": "100",
"posts": [],
"source_url": "https://x.com/example/status/100",
}
],
})
self.assertIsInstance(manifest.threads, tuple)
self.assertEqual(len(manifest.threads), 1)
self.assertEqual(manifest.threads[0].root_post_id, "100")
class LegacyThreadDataTests(unittest.TestCase):
"""Tests for the LegacyThreadData boundary dataclass."""
def test_round_trip_with_posts_tuple(self) -> None:
thread = LegacyThreadData.from_dict({
"root_post_id": "100",
"source_url": "https://x.com/example/status/100",
"posts": [
{
"post_id": "100",
"author": "Example",
"handle": "example",
"text": "Hello",
"timestamp": "2026-08-24 12:00:00",
"media_urls": [],
"reply_to_id": None,
"quote_of_id": None,
"metrics": {
"reply_count": 0,
"repost_count": 0,
"like_count": 0,
"view_count": None,
},
}
],
})
self.assertEqual(thread.root_post_id, "100")
self.assertEqual(thread.source_url, "https://x.com/example/status/100")
self.assertIsInstance(thread.posts, tuple)
self.assertEqual(len(thread.posts), 1)
self.assertIsInstance(thread.posts[0], PostData)
self.assertEqual(thread.posts[0].post_id, "100")
if __name__ == "__main__":
unittest.main()