script update.

This commit is contained in:
ed committed 2026-08-25 23:19:12 -04:00
1 parent d95b8d627b
commit d9bb5db8c7
4 files changed
+139 -8

No files matched your search

+1 -1
View File
@@ -25,7 +25,7 @@ python scripts/twitter/tw.py relabel --all --apply
## Capture
`refresh` is refetch + ingest + emit `--force` + media merge. `--tip` treats `--id` as the tip and walks back to the opening post. Repeat `--branch <tip-id>` to capture extra tips into the same `thread_data.json`. It does not invent replies. Captures run one at a time. gallery-dl retries are off. Extractor and request sleeps are five seconds.
`refresh` is refetch + ingest + emit `--force` + media merge. `--tip` treats `--id` as the tip and walks back to the opening post. If that tip is the OP, same-handle self-replies stay on the spine and foreign replies are branches. Repeat `--branch <tip-id>` to capture extra tips into the same `thread_data.json`. It does not invent replies. Captures run one at a time. gallery-dl retries are off. Extractor and request sleeps are five seconds. Re-emit deletes branch notes that are no longer roots.
Each thread writes one archive folder. The owner is the opening post (`reply_to_id` is empty). Other authors stay in that folder. New notes set `draft: false`. The command does not commit. Frozen ids abort.
+6
View File
@@ -703,6 +703,12 @@ def emit(
)
branch_path.write_text(branch_text, encoding="utf-8", newline="\n")
keep = {spine_path.name}
keep.update(f"{name}.md" for name in branch_names.values())
for path in note_dir.glob("*.md"):
if path.name not in keep:
path.unlink()
shutil.copy2(src, asset_dir / "thread_data.json")
media_doc = {
"schema_version": 2,
+124
View File
@@ -733,6 +733,57 @@ class TipIsOpTests(unittest.TestCase):
spine = spine_from_tip(thread, "1000")
self.assertEqual(spine, ["1000", "1001", "1002"])
def test_op_tip_stops_without_same_handle_child(self) -> None:
from twitter.tree import spine_from_tip
thread = ThreadData(
root_post_id="1000",
posts=(
make_post(
"1000",
"mike_acton",
"OP post",
"2023-01-01 00:00:00",
None,
),
make_post(
"1001",
"aras_p",
"joke reply",
"2023-01-01 00:05:00",
"1000",
),
),
source_url="https://x.com/i/status/1000",
)
spine = spine_from_tip(thread, "1000")
self.assertEqual(spine, ["1000"])
def test_spine_ids_still_promotes_earliest_foreign_child(self) -> None:
from twitter.tree import spine_ids
thread = ThreadData(
root_post_id="1000",
posts=(
make_post(
"1000",
"mike_acton",
"OP post",
"2023-01-01 00:00:00",
None,
),
make_post(
"1001",
"aras_p",
"joke reply",
"2023-01-01 00:05:00",
"1000",
),
),
source_url="https://x.com/i/status/1000",
)
self.assertEqual(spine_ids(thread), ["1000", "1001"])
def test_tip_with_parent_walks_back(self) -> None:
from twitter.tree import spine_from_tip
@@ -813,6 +864,79 @@ class TipIsOpTests(unittest.TestCase):
self.assertEqual(spine, ["2000", "2001", "2002"])
class StaleBranchPruneTests(unittest.TestCase):
"""Re-emit must delete branch notes that are no longer roots."""
def setUp(self) -> None:
self.tmp = Path(tempfile.mkdtemp(prefix="stale_branch_"))
self.input_dir = self.tmp / "input"
self.input_dir.mkdir()
(self.input_dir / "thread_data.json").write_text(
json.dumps(
{
"root_post_id": "1000",
"source_url": "https://x.com/i/status/1000",
"posts": [
{
"post_id": "1000",
"author": "alice",
"handle": "alice",
"text": "Hello",
"timestamp": "2026-01-01 00:00:00",
"media_urls": [],
"reply_to_id": None,
"quote_of_id": None,
"metrics": {
"reply_count": 0,
"repost_count": 0,
"like_count": 0,
"view_count": 0,
},
}
],
},
indent=2,
),
encoding="utf-8",
newline="\n",
)
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_emit_deletes_stale_branch_notes(self) -> None:
emit(
input_dir=self.input_dir,
vault=self.vault,
slug=None,
archived="2026-08-25",
force=True,
tip="1000",
)
note_dir = (
self.vault
/ "archive"
/ "threads"
/ "alice"
/ "2026-01-01-hello"
)
stale = note_dir / "2026-01-01-bob-noise.md"
stale.write_text("leftover\n", encoding="utf-8", newline="\n")
emit(
input_dir=self.input_dir,
vault=self.vault,
slug=None,
archived="2026-08-25",
force=True,
tip="1000",
)
self.assertFalse(stale.is_file())
self.assertTrue((note_dir / "index.md").is_file())
class ReslugReuseTests(unittest.TestCase):
"""Compact same-handle integration tests for the emit-side rename flow.
+8 -7
View File
@@ -23,9 +23,10 @@ def spine_from_tip(thread: ThreadData, tip_id: str) -> list[str]:
"""Resolve the OP chain from `tip_id`.
If `tip_id` is the OP (its `reply_to_id` is `None`), walk DOWN
from it via `children_map` — same logic as `spine_ids`. This is
the common case when `--root=<op>` and `--tip=<op>` are the same
post id; the back-walk would otherwise return a single-element
through same-handle children only. Stop when the OP has no
self-reply. Do not promote a foreign-handle child onto the
spine; those posts are branches. This is the `--id <OP> --tip`
case. The back-walk would otherwise return a single-element
chain and drop every self-reply.
If `tip_id` has a parent, walk BACK through `reply_to_id` to
@@ -36,17 +37,17 @@ def spine_from_tip(thread: ThreadData, tip_id: str) -> list[str]:
if tip_id not in ids:
raise ValueError(f"tip {tip_id} not in dump")
if ids[tip_id].reply_to_id is None:
# Tip is the OP. Walk down its same-handle chain.
# Tip is the OP. Walk down its same-handle chain only.
kids = children_map(thread)
handle = ids[tip_id].handle
out = [tip_id]
cur = tip_id
while True:
children = kids.get(cur, [])
if not children:
break
same = [c for c in children if ids[c].handle == handle]
cur = same[0] if same else children[0]
if not same:
break
cur = same[0]
out.append(cur)
return out
chain = [tip_id]