fix(perception): structural fast-path and VLM bounds extraction for Reels
This commit is contained in:
@@ -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(
|
||||
|
||||
105
tests/e2e/test_post_data_extraction.py
Normal file
105
tests/e2e/test_post_data_extraction.py
Normal file
@@ -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}"
|
||||
Reference in New Issue
Block a user