diff --git a/GramAddict/core/perception/feed_analysis.py b/GramAddict/core/perception/feed_analysis.py index 70d4bb3..7a5a955 100644 --- a/GramAddict/core/perception/feed_analysis.py +++ b/GramAddict/core/perception/feed_analysis.py @@ -63,18 +63,15 @@ def extract_post_content(context_xml: str, device=None) -> dict: # 1. Learn/extract post author dynamically # 🛡️ Structural Fast-Path: Prioritize deterministic IDs over AI guesses + # Try structural ID fast-path first (100% deterministic) author_node = None try: root = ET.fromstring(context_xml) - for node in root.iter(): + for node in root.iter("node"): res_id = node.attrib.get("resource-id", "") - if "row_feed_photo_profile_name" in res_id or "profile_header_name" in res_id: - author_node = { - "original_attribs": { - "text": node.attrib.get("text", ""), - "content_desc": node.attrib.get("content-desc", ""), - } - } + if "row_feed_photo_profile_name" in res_id or "clips_author_username" in res_id or "profile_header_name" in res_id: + author_node = {"original_attribs": node.attrib} + logger.debug(f"Identified author_node via structural ID: {res_id}") break except Exception as e: logger.debug(f"XML parse error in author structural fast-path: {e}") @@ -87,10 +84,55 @@ def extract_post_content(context_xml: str, device=None) -> dict: logger.debug(f"Telepathic fallback for author_node: {author_node}") # 🛡️ Anti-Hallucination Guard: Ensure we actually found text. - if author_node and author_node.get("original_attribs", {}).get("text"): - result["username"] = author_node["original_attribs"]["text"].strip() - elif author_node and author_node.get("original_attribs", {}).get("content_desc"): - result["username"] = author_node["original_attribs"]["content_desc"].strip() + if author_node: + attribs = author_node.get("original_attribs", {}) + text = attribs.get("text", "").strip() + desc = attribs.get("content_desc", "").strip() + + if text: + result["username"] = text + elif desc: + result["username"] = desc + else: + # If the VLM selected a container (like clips_author_info_component), + # extract text from its children. + logger.debug("Author node lacks text/desc. Searching children for username...") + bounds = attribs.get("bounds") + if bounds: + try: + # Re-parse to find children within bounds + import re + match = re.match(r"\[(\d+),(\d+)\]\[(\d+),(\d+)\]", bounds) + if match: + l, t, r, b = map(int, match.groups()) + + # Fallback: scan all nodes in XML and see if they are inside these bounds + possible_texts = [] + possible_descs = [] + for n in ET.fromstring(context_xml).iter("node"): + child_bounds = n.attrib.get("bounds") + child_text = n.attrib.get("text", "").strip() + child_desc = n.attrib.get("content-desc", "").strip() + + if child_bounds and (child_text or child_desc): + cm = re.match(r"\[(\d+),(\d+)\]\[(\d+),(\d+)\]", child_bounds) + if cm: + cl, ct, cr, cb = map(int, cm.groups()) + # Check if child is strictly inside the container + if cl >= l and ct >= t and cr <= r and cb <= b: + if child_text: + possible_texts.append(child_text) + if child_desc and "Profile picture" not in child_desc: + possible_descs.append(child_desc) + + if possible_texts: + result["username"] = possible_texts[0] + logger.debug(f"Extracted username '{result['username']}' from child node text.") + elif possible_descs: + result["username"] = possible_descs[0] + logger.debug(f"Extracted username '{result['username']}' from child node desc.") + except Exception as e: + logger.debug(f"Failed to extract username from children: {e}") # 2. Learn/extract post media description dynamically media_node = telepath.find_best_node( diff --git a/tests/e2e/test_post_data_extraction.py b/tests/e2e/test_post_data_extraction.py new file mode 100644 index 0000000..1e1494b --- /dev/null +++ b/tests/e2e/test_post_data_extraction.py @@ -0,0 +1,105 @@ +""" +Tests for PostDataExtractionPlugin +================================== +Ensures that data extraction robustly extracts the author username +from different UI layouts (Home Feed, Reels Feed, etc.) without failing. +""" + +from GramAddict.core.behaviors.post_data_extraction import PostDataExtractionPlugin +from GramAddict.core.behaviors import BehaviorContext +from tests.e2e.conftest import load_fixture_xml +import pytest + +class DummyDevice: + def dump_hierarchy(self): + return "" + +def test_extract_username_from_home_feed(): + plugin = PostDataExtractionPlugin() + xml = load_fixture_xml("home_feed_real.xml") + + ctx = BehaviorContext( + device=DummyDevice(), + configs=None, + session_state=None, + cognitive_stack=None, + context_xml=xml, + post_data={}, + username="", + ) + + result = plugin.execute(ctx) + assert result.executed is True + assert ctx.username != "" + assert ctx.username is not None + assert ctx.post_data.get("username_missing") is not True + +def test_extract_username_from_reels_feed(): + plugin = PostDataExtractionPlugin() + xml = load_fixture_xml("reels_feed_real.xml") + + ctx = BehaviorContext( + device=DummyDevice(), + configs=None, + session_state=None, + cognitive_stack=None, + context_xml=xml, + post_data={}, + username="", + ) + + result = plugin.execute(ctx) + assert result.executed is True, "PostDataExtraction failed to execute on Reels Feed" + assert ctx.username != "", "Username extracted from Reels Feed is empty!" + assert ctx.username is not None + assert ctx.post_data.get("username_missing") is not True + +def test_extract_username_with_vlm_fallback(monkeypatch): + """ + Tests that if the structural fast path fails, the VLM fallback correctly + descends into ViewGroups if the VLM selects a container instead of the text node. + """ + plugin = PostDataExtractionPlugin() + xml = load_fixture_xml("reels_feed_real.xml") + + # Mock TelepathicEngine to return the container node (like in the logs) + class FakeTelepath: + def find_best_node(self, xml_str, prompt, **kwargs): + if "author username" in prompt: + # Return the clips_author_info_component (a ViewGroup with no text) + return { + "original_attribs": { + "resource-id": "com.instagram.android:id/clips_author_info_component", + "text": "", + "content_desc": "", + "bounds": "[42,1957][591,2098]" + } + } + if "media content" in prompt: + return {"original_attribs": {"content_desc": "Test media"}} + return None + + from GramAddict.core.telepathic_engine import TelepathicEngine + monkeypatch.setattr(TelepathicEngine, "get_instance", lambda *args, **kwargs: FakeTelepath()) + + # We must also mock the structural fast path so it falls through to VLM + # by temporarily removing 'clips_author_username' from the fast path list during this test. + # We will just mutate the xml to not have the structural IDs + xml_no_fastpath = xml.replace("row_feed_photo_profile_name", "hidden_id") + xml_no_fastpath = xml_no_fastpath.replace("clips_author_username", "hidden_id") + + ctx = BehaviorContext( + device=DummyDevice(), + configs=None, + session_state=None, + cognitive_stack=None, + context_xml=xml_no_fastpath, + post_data={}, + username="", + ) + + result = plugin.execute(ctx) + assert result.executed is True + # The VLM selected the container. The engine should have parsed the children + # and found "aditnugrahh" inside it. + assert ctx.username == "aditnugrahh", f"Expected 'aditnugrahh' but got {ctx.username}"