zzstoatzz.io
a digital entity named phi that roams bsky
refactor: extract thread traversal to reusable utility
- created bot.utils.thread module with traverse_thread() and extract_posts_chronological()
- refactored MessageHandler to use utility instead of custom recursion
- well-defined problem solved once, reusable in multiple contexts
- can be used in tests, analysis scripts, viewing tools, etc.
+16
-20
src/bot/services/message_handler.py
+16
-20
src/bot/services/message_handler.py
···
9
9
from bot.core.atproto_client import BotClient
10
10
from bot.database import thread_db
11
11
from bot.status import bot_status
12
+
from bot.utils.thread import traverse_thread
12
13
13
14
logger = logging.getLogger("bot.handler")
14
15
···
21
22
self.agent = PhiAgent()
22
23
23
24
async def _store_thread_messages(self, thread_node, thread_uri: str):
24
-
"""Recursively extract and store all messages from a thread."""
25
-
if not thread_node or not hasattr(thread_node, "post"):
26
-
return
27
-
28
-
post = thread_node.post
25
+
"""Extract and store all messages from a thread."""
29
26
30
-
# Store this message
31
-
thread_db.add_message(
32
-
thread_uri=thread_uri,
33
-
author_handle=post.author.handle,
34
-
author_did=post.author.did,
35
-
message_text=post.record.text,
36
-
post_uri=post.uri,
37
-
)
27
+
def store_post(node):
28
+
"""Store a single post from the thread."""
29
+
if not hasattr(node, "post"):
30
+
return
38
31
39
-
# Recursively store replies
40
-
if hasattr(thread_node, "replies") and thread_node.replies:
41
-
for reply in thread_node.replies:
42
-
await self._store_thread_messages(reply, thread_uri)
32
+
post = node.post
33
+
thread_db.add_message(
34
+
thread_uri=thread_uri,
35
+
author_handle=post.author.handle,
36
+
author_did=post.author.did,
37
+
message_text=post.record.text,
38
+
post_uri=post.uri,
39
+
)
43
40
44
-
# Also check for parent if this is a reply
45
-
if hasattr(thread_node, "parent") and thread_node.parent:
46
-
await self._store_thread_messages(thread_node.parent, thread_uri)
41
+
# Use utility to traverse and store all posts
42
+
traverse_thread(thread_node, store_post)
47
43
48
44
async def handle_mention(self, notification):
49
45
"""Process a mention or reply notification."""
+60
src/bot/utils/thread.py
+60
src/bot/utils/thread.py
···
1
+
"""Thread utilities for ATProto thread operations."""
2
+
3
+
from collections.abc import Callable
4
+
5
+
6
+
def traverse_thread(
7
+
thread_node,
8
+
visit: Callable[[any], None],
9
+
*,
10
+
include_parent: bool = True,
11
+
include_replies: bool = True,
12
+
):
13
+
"""Recursively traverse a thread structure and call visit() on each post.
14
+
15
+
Args:
16
+
thread_node: ATProto thread node with optional .post, .parent, .replies
17
+
visit: Callback function called for each post node
18
+
include_parent: Whether to traverse up to parent posts
19
+
include_replies: Whether to traverse down to reply posts
20
+
21
+
Example:
22
+
posts = []
23
+
traverse_thread(thread_data.thread, lambda node: posts.append(node.post))
24
+
"""
25
+
if not thread_node or not hasattr(thread_node, "post"):
26
+
return
27
+
28
+
# Visit this node
29
+
visit(thread_node)
30
+
31
+
# Traverse parent chain (moving up the thread)
32
+
if include_parent and hasattr(thread_node, "parent") and thread_node.parent:
33
+
traverse_thread(thread_node.parent, visit, include_parent=True, include_replies=False)
34
+
35
+
# Traverse replies (moving down the thread)
36
+
if include_replies and hasattr(thread_node, "replies") and thread_node.replies:
37
+
for reply in thread_node.replies:
38
+
traverse_thread(reply, visit, include_parent=False, include_replies=True)
39
+
40
+
41
+
def extract_posts_chronological(thread_node) -> list[any]:
42
+
"""Extract all posts from a thread in chronological order.
43
+
44
+
Args:
45
+
thread_node: ATProto thread node
46
+
47
+
Returns:
48
+
List of post objects sorted by timestamp
49
+
"""
50
+
posts = []
51
+
52
+
def collect(node):
53
+
if hasattr(node, "post"):
54
+
posts.append(node.post)
55
+
56
+
traverse_thread(thread_node, collect)
57
+
58
+
# Sort by indexed timestamp
59
+
posts.sort(key=lambda p: p.indexed_at if hasattr(p, "indexed_at") else "")
60
+
return posts