mmintel/instagram-bot
1
0
Fork 0
Code Issues Pull Requests 4 Actions Packages Projects Releases Wiki Activity

fix(navigation): enforce strict unread badge detection in structural fast paths

Browse Source
This commit is contained in:
Marc Mintel
2026-04-27 14:13:00 +02:00
parent 41450c4eaf
commit b4bafb59be
1 changed files with 31 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
GramAddict/core/telepathic_engine.py
View File

@@ -78,12 +78,11 @@ class TelepathicEngine:
logger.warning("🛡️ [Comment Guard] Comments are disabled on this post.")
return {"skip": True, "semantic": "comments disabled"}
# 1.25 Grid Fast-Path (Deterministically bypass VLM for first grid item)
if "first image in explore grid" in intent_description.lower():
nodes_dicts = self._extract_semantic_nodes(xml_string)
fast_node = self._grid_fast_path(intent_description, nodes_dicts, kwargs.get("skip_positions"))
if fast_node:
return fast_node
# 1.25 Structural Fast-Paths (Deterministically bypass VLM for fixed UI elements)
nodes_dicts = self._extract_semantic_nodes(xml_string)
fast_node = self._structural_fast_path(intent_description, nodes_dicts, kwargs.get("skip_positions"), xml_string)
if fast_node:
return fast_node
# 1. Parse into Spatial Topology
root = self._parser.parse(xml_string)
@@ -149,7 +148,7 @@ class TelepathicEngine:
nodes = self._parser.get_clickable_nodes(root)
return [self._translate_node(n) for n in nodes]
def _grid_fast_path(self, intent_description: str, nodes: list, skip_positions: set = None) -> Optional[dict]:
def _structural_fast_path(self, intent_description: str, nodes: list, skip_positions: set = None, xml_string: str = "") -> Optional[dict]:
if skip_positions is None:
skip_positions = set()
@@ -173,42 +172,51 @@ class TelepathicEngine:
# --- DM Engine Structural Fast Paths ---
if "find the message input text field" in intent_lower:
for n in nodes:
if "row_thread_composer_edittext" in n.get("resource_id", ""):
if "row_thread_composer_edittext" in n.get("id", "") or "row_thread_composer_edittext" in n.get("resource_id", ""):
return n
if "find the send message button" in intent_lower:
for n in nodes:
if "row_thread_composer_button_send" in n.get("resource_id", ""):
if "row_thread_composer_button_send" in n.get("id", "") or "row_thread_composer_button_send" in n.get("resource_id", ""):
return n
if "find unread message threads" in intent_lower:
# Unread threads usually have a specific indicator or are the top rows in the recyclerview
# We must be extremely strict here: It's only unread if it has the "unread" text or indicator dot
unread_candidates = []
# Check for explicitly unread containers
for n in nodes:
# Check if it has a direct structural unread marker (though IG usually relies on styling)
# But we can fallback to the row_inbox_container
if (
"row_inbox_container" in n.get("resource_id", "")
and (n.get("x", -1), n.get("y", -1)) not in skip_positions
):
# Avoid the 'Search' or header elements, typically thread rows start below y=200
is_unread = False
res_id = n.get("id", "") or n.get("resource_id", "")
content_desc = (n.get("description", "") or "").lower()
semantic = (n.get("semantic_string", "") or "").lower()
# Check for explicit 'unread' in description
if "unread" in content_desc or "unread" in semantic:
is_unread = True
# Some versions might use the explicit unread dot
if "thread_indicator_status_dot" in xml_string:
# Let's fallback to any row if we know an unread dot exists in the xml
if "row_inbox_container" in res_id:
is_unread = True
if is_unread and "row_inbox_container" in res_id and (n.get("x", -1), n.get("y", -1)) not in skip_positions:
if n.get("y", 0) > 200:
unread_candidates.append(n)
if unread_candidates:
# Return the highest y (first thread in list)
unread_candidates.sort(key=lambda n: n.get("y", 9999))
return unread_candidates[0]
if "find the last received message text" in intent_lower:
msg_candidates = []
for n in nodes:
res_id = n.get("id", "") or n.get("resource_id", "")
# The actual message text bubble
if "direct_text_message_text_view" in n.get("resource_id", "") or "message_content" in n.get(
"resource_id", ""
):
if "direct_text_message_text_view" in res_id or "message_content" in res_id:
msg_candidates.append(n)
if msg_candidates:
# Sort by y descending (bottom-most message is the last one)
msg_candidates.sort(key=lambda n: n.get("y", 0), reverse=True)