another fix
This commit is contained in:
ed committed
1 parent
26f04e6cb0
commit
49d1afaa26
4 files changed
+281
-13
No files matched your search
@@ -15,7 +15,7 @@ import unittest
|
||||
from pathlib import Path
|
||||
|
||||
from twitter.emit_archive import emit
|
||||
from twitter.models import PostData, PostMetrics
|
||||
from twitter.models import PostData, PostMetrics, ThreadData
|
||||
|
||||
|
||||
def make_post(
|
||||
@@ -648,5 +648,169 @@ class SpinePrefersSameAuthorTests(unittest.TestCase):
|
||||
self.assertIn("early reply from outsider", branch_text)
|
||||
|
||||
|
||||
class TipIsOpTests(unittest.TestCase):
|
||||
"""`--tip=<op>` where the tip's reply_to_id is None. The walker
|
||||
must walk DOWN from the OP via children_map, not stop at the
|
||||
OP because the back-walk has no parent."""
|
||||
|
||||
def test_tip_with_no_parent_walks_down(self) -> None:
|
||||
from twitter.tree import spine_from_tip
|
||||
|
||||
thread_data = {
|
||||
"root_post_id": "1000",
|
||||
"source_url": "https://x.com/i/status/1000",
|
||||
"posts": [
|
||||
{
|
||||
"post_id": "1000",
|
||||
"author": "X",
|
||||
"handle": "X",
|
||||
"text": "OP post",
|
||||
"timestamp": "2023-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,
|
||||
},
|
||||
},
|
||||
{
|
||||
"post_id": "1001",
|
||||
"author": "X",
|
||||
"handle": "X",
|
||||
"text": "self-reply 1",
|
||||
"timestamp": "2023-01-01 00:05:00",
|
||||
"media_urls": [],
|
||||
"reply_to_id": "1000",
|
||||
"quote_of_id": None,
|
||||
"metrics": {
|
||||
"reply_count": 0,
|
||||
"repost_count": 0,
|
||||
"like_count": 0,
|
||||
"view_count": 0,
|
||||
},
|
||||
},
|
||||
{
|
||||
"post_id": "1002",
|
||||
"author": "X",
|
||||
"handle": "X",
|
||||
"text": "self-reply 2",
|
||||
"timestamp": "2023-01-01 00:10:00",
|
||||
"media_urls": [],
|
||||
"reply_to_id": "1001",
|
||||
"quote_of_id": None,
|
||||
"metrics": {
|
||||
"reply_count": 0,
|
||||
"repost_count": 0,
|
||||
"like_count": 0,
|
||||
"view_count": 0,
|
||||
},
|
||||
},
|
||||
],
|
||||
}
|
||||
thread = ThreadData(
|
||||
root_post_id="1000",
|
||||
posts=tuple(
|
||||
PostData(
|
||||
post_id=p["post_id"],
|
||||
author=p["author"],
|
||||
handle=p["handle"],
|
||||
text=p["text"],
|
||||
timestamp=p["timestamp"],
|
||||
media_urls=tuple(p["media_urls"]),
|
||||
reply_to_id=p["reply_to_id"],
|
||||
quote_of_id=p["quote_of_id"],
|
||||
metrics=PostMetrics(0, 0, 0, 0),
|
||||
)
|
||||
for p in thread_data["posts"]
|
||||
),
|
||||
source_url="https://x.com/i/status/1000",
|
||||
)
|
||||
# --tip=OP, which has reply_to=None. Must walk DOWN.
|
||||
spine = spine_from_tip(thread, "1000")
|
||||
self.assertEqual(spine, ["1000", "1001", "1002"])
|
||||
|
||||
def test_tip_with_parent_walks_back(self) -> None:
|
||||
from twitter.tree import spine_from_tip
|
||||
|
||||
thread_data = {
|
||||
"root_post_id": "2000",
|
||||
"source_url": "https://x.com/i/status/2000",
|
||||
"posts": [
|
||||
{
|
||||
"post_id": "2000",
|
||||
"author": "X",
|
||||
"handle": "X",
|
||||
"text": "OP",
|
||||
"timestamp": "2023-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,
|
||||
},
|
||||
},
|
||||
{
|
||||
"post_id": "2001",
|
||||
"author": "X",
|
||||
"handle": "X",
|
||||
"text": "self-reply",
|
||||
"timestamp": "2023-01-01 00:05:00",
|
||||
"media_urls": [],
|
||||
"reply_to_id": "2000",
|
||||
"quote_of_id": None,
|
||||
"metrics": {
|
||||
"reply_count": 0,
|
||||
"repost_count": 0,
|
||||
"like_count": 0,
|
||||
"view_count": 0,
|
||||
},
|
||||
},
|
||||
{
|
||||
"post_id": "2002",
|
||||
"author": "X",
|
||||
"handle": "X",
|
||||
"text": "tip reply",
|
||||
"timestamp": "2023-01-01 00:10:00",
|
||||
"media_urls": [],
|
||||
"reply_to_id": "2001",
|
||||
"quote_of_id": None,
|
||||
"metrics": {
|
||||
"reply_count": 0,
|
||||
"repost_count": 0,
|
||||
"like_count": 0,
|
||||
"view_count": 0,
|
||||
},
|
||||
},
|
||||
],
|
||||
}
|
||||
thread = ThreadData(
|
||||
root_post_id="2000",
|
||||
posts=tuple(
|
||||
PostData(
|
||||
post_id=p["post_id"],
|
||||
author=p["author"],
|
||||
handle=p["handle"],
|
||||
text=p["text"],
|
||||
timestamp=p["timestamp"],
|
||||
media_urls=tuple(p["media_urls"]),
|
||||
reply_to_id=p["reply_to_id"],
|
||||
quote_of_id=p["quote_of_id"],
|
||||
metrics=PostMetrics(0, 0, 0, 0),
|
||||
)
|
||||
for p in thread_data["posts"]
|
||||
),
|
||||
source_url="https://x.com/i/status/2000",
|
||||
)
|
||||
# --tip=middle post (has parent). Walks back.
|
||||
spine = spine_from_tip(thread, "2002")
|
||||
self.assertEqual(spine, ["2000", "2001", "2002"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
+28
-11
@@ -20,26 +20,43 @@ def children_map(thread: ThreadData) -> dict[str, list[str]]:
|
||||
|
||||
|
||||
def spine_from_tip(thread: ThreadData, tip_id: str) -> list[str]:
|
||||
"""Walk back from `tip_id` through `reply_to_id` to the OP (the
|
||||
post with `reply_to_id == None` or a missing parent). Returns
|
||||
all post ids on the linear chain in chronological order: OP at
|
||||
index 0, tip at last.
|
||||
"""Resolve the OP chain from `tip_id`.
|
||||
|
||||
The walk crosses handle boundaries. For a chain that alternates
|
||||
authors, every post is included. The archive dir for the
|
||||
thread is owned by the OP at index 0.
|
||||
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
|
||||
chain and drop every self-reply.
|
||||
|
||||
If `tip_id` has a parent, walk BACK through `reply_to_id` to
|
||||
the OP and return the chain in chronological order (OP at
|
||||
index 0, tip at last). The walk crosses handle boundaries.
|
||||
"""
|
||||
ids = by_id(thread)
|
||||
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.
|
||||
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]
|
||||
out.append(cur)
|
||||
return out
|
||||
chain = [tip_id]
|
||||
cur = tip_id
|
||||
while True:
|
||||
parent = ids[cur].reply_to_id
|
||||
if not parent or parent not in ids:
|
||||
parent_id = ids[cur].reply_to_id
|
||||
if not parent_id or parent_id not in ids:
|
||||
break
|
||||
chain.append(parent)
|
||||
cur = parent
|
||||
chain.append(parent_id)
|
||||
cur = parent_id
|
||||
chain.reverse()
|
||||
return chain
|
||||
|
||||
|
||||
Reference in new issue
Block a user