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

Hardened E2E integrity, purged synthetic mocks, and implemented proactive device discovery.

Browse Source
This commit is contained in:
Marc Mintel
2026-04-29 15:42:03 +02:00
parent 6abb519e3b
commit 0ed12303ac
9 changed files with 442 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

37
GramAddict/core/device_facade.py
View File

@@ -36,15 +36,38 @@ def create_device(device_id, app_id, args=None):
try:
return DeviceFacade(device_id, app_id, args)
except Exception as e:
str(e)
err_msg = str(e)
err_type = str(type(e))
if (
"ConnectError" in err_type
or "ConnectionRefusedError" in err_type
or "ConnectionError" in err_type
or "Timeout" in err_type
if any(
keyword in err_type or keyword in err_msg
for keyword in ["ConnectError", "ConnectionRefused", "ConnectionError", "Timeout"]
):
logger.error(f"⚠️ [ADB ConnectError] Could not connect to device '{device_id}'.")
# Proactive Discovery
try:
import subprocess
result = subprocess.run(["adb", "devices"], capture_output=True, text=True, timeout=2)
lines = [
line.strip()
for line in result.stdout.split("\n")
if line.strip() and not line.startswith("List of devices")
]
devices = [line.split("\t")[0] for line in lines if "device" in line]
if devices:
logger.info("🔍 Proactive Discovery: I found the following devices connected:")
for d in devices:
if d.split(":")[0] == device_id.split(":")[0]:
logger.info(f" 👉 {d} (MATCHING IP - Is this the same device with a different port?)")
else:
logger.info(f" - {d}")
else:
logger.warning("🔍 Proactive Discovery: No ADB devices found. Is your phone authorized?")
except Exception as discovery_err:
logger.debug(f"Proactive discovery failed: {discovery_err}")
logger.error("👉 Please verify:")
logger.error(" 1. Your phone is connected via USB or Wi-Fi.")
logger.error(" 2. 'USB Debugging' is enabled in Developer Options.")
@@ -359,6 +382,8 @@ class DeviceFacade:
from io import BytesIO
img = self.deviceV2.screenshot()
if img is None:
return None
buffered = BytesIO()
img.save(buffered, format="JPEG", quality=70) # Compressed for target latency
return base64.b64encode(buffered.getvalue()).decode("utf-8")

27
GramAddict/core/perception/action_memory.py
View File

@@ -73,7 +73,7 @@ class ActionMemory:
logger.info(
f"✅ [ActionMemory] Confirming success for '{ctx['intent']}'. Boosting confidence.",
extra={"color": "\x1b[32m"}
extra={"color": "\x1b[32m"},
)
# Store or boost in Qdrant
@@ -99,8 +99,7 @@ class ActionMemory:
return
logger.warning(
f"❌ [ActionMemory] Click failed for '{ctx['intent']}'. Applying penalty.",
extra={"color": "\x1b[31m"}
f"❌ [ActionMemory] Click failed for '{ctx['intent']}'. Applying penalty.", extra={"color": "\x1b[31m"}
)
try:
@@ -121,9 +120,17 @@ class ActionMemory:
# Specific check for opening a post (from explore/profile grid)
if "view a post" in intent_lower or "first image" in intent_lower or "grid item" in intent_lower:
if "row_feed_photo_imageview" in post_xml_lower or "row_feed_button_like" in post_xml_lower or "clips_viewer_view_pager" in post_xml_lower:
if (
"row_feed_photo_imageview" in post_xml_lower
or "row_feed_button_like" in post_xml_lower
or "clips_viewer_view_pager" in post_xml_lower
):
return True
if "explore_action_bar" in post_xml_lower and "row_feed_button_like" not in post_xml_lower and "clips_viewer" not in post_xml_lower:
if (
"explore_action_bar" in post_xml_lower
and "row_feed_button_like" not in post_xml_lower
and "clips_viewer" not in post_xml_lower
):
return None # Still on grid, inconclusive
state_toggles = ["like", "save", "follow", "heart"]
@@ -178,6 +185,8 @@ class ActionMemory:
try:
screenshot = device.get_screenshot_b64()
if not screenshot:
raise ValueError("No screenshot available from device")
response = evaluator._query_vlm(prompt, screenshot)
if response and "yes" in response.lower() and "no" not in response.lower():
@@ -202,7 +211,9 @@ class ActionMemory:
return False
# Fallback to structural delta
logger.info(f"DEBUG: len(pre_click_xml)={len(pre_click_xml)} len(post_click_xml)={len(post_click_xml)}")
diff = abs(len(pre_click_xml) - len(post_click_xml))
logger.info(f"DEBUG: diff={diff}")
if is_toggle:
if diff > 1000:
@@ -222,7 +233,13 @@ class ActionMemory:
from GramAddict.core.screen_topology import ScreenTopology
# We don't have screen type here, so we just check if it's in the HD Map keys
logger.info(f"DEBUG: intent is '{intent}'")
logger.info(
f"DEBUG: TRANSITIONS keys are: {[list(t.keys()) for t in ScreenTopology.TRANSITIONS.values()]}"
)
is_standard = any(intent in transitions for transitions in ScreenTopology.TRANSITIONS.values())
logger.info(f"DEBUG: is_standard={is_standard}")
if is_standard:
logger.debug(
f"🧠 [ActionMemory] Structural change detected for known navigation '{intent}'. Verification PASS."

67
GramAddict/core/perception/intent_resolver.py
View File

@@ -53,13 +53,47 @@ class IntentResolver:
if intent_lower in abstract_goals:
return None
# --- Semantic Match Guard ---
# If the intent explicitly quotes a target (e.g., "tap 'New Message'"),
# we strictly filter candidates to those whose text or content_desc contains the quote.
import re
quotes = re.findall(r"['\"](.*?)['\"]", intent_description)
if quotes:
target_text = quotes[0].lower()
pattern = r"\b" + re.escape(target_text) + r"\b"
semantic_candidates = []
for node in candidates:
n_text = (node.text or "").lower()
n_desc = (node.content_desc or "").lower()
if re.search(pattern, n_text) or re.search(pattern, n_desc):
semantic_candidates.append(node)
if semantic_candidates:
if len(semantic_candidates) == 1:
logger.info(f"🎯 [Semantic Guard] Exact match found for '{target_text}', skipping VLM.")
return semantic_candidates[0]
else:
logger.info(
f"🎯 [Semantic Guard] {len(semantic_candidates)} matches found for '{target_text}'. Reducing candidates for VLM."
)
candidates = semantic_candidates
else:
logger.warning(
f"⚠️ [Semantic Guard] No candidates found containing '{target_text}'. Returning None to prevent hallucination."
)
return None
# ── PRIMARY PATH: Visual Discovery ──
# If we have a device, the VLM SEES the screen and decides.
if device is not None and (
hasattr(device, "screenshot") or hasattr(getattr(device, "deviceV2", None), "screenshot")
):
logger.info("📸 Device screenshot capability detected. Enforcing visual discovery.")
return self._visual_discovery(intent_description, candidates, device)
visual_res = self._visual_discovery(intent_description, candidates, device)
if visual_res is not None:
return visual_res
logger.warning("👁️ [IntentResolver] Visual discovery yielded None. Falling back to text-based resolution.")
# --- Strict VLM Hallucination Guard (Text-only Fallback) ---
# For known structural targets that the text-based VLM frequently hallucinates when they are missing,
@@ -254,37 +288,6 @@ class IntentResolver:
logger.info(f"🎯 [Grid Guard] Filtered to {len(grid_candidates)} actual grid candidates.")
candidates = grid_candidates
# --- Semantic Match Guard ---
# If the intent explicitly quotes a target (e.g., "tap 'New Message'"),
# we strictly filter candidates to those whose text or content_desc contains the quote.
import re
quotes = re.findall(r"['\"](.*?)['\"]", intent_description)
if quotes:
target_text = quotes[0].lower()
pattern = r"\b" + re.escape(target_text) + r"\b"
semantic_candidates = []
for node in candidates:
n_text = (node.text or "").lower()
n_desc = (node.content_desc or "").lower()
if re.search(pattern, n_text) or re.search(pattern, n_desc):
semantic_candidates.append(node)
if semantic_candidates:
if len(semantic_candidates) == 1:
logger.info(f"🎯 [Semantic Guard] Exact match found for '{target_text}', skipping VLM.")
return semantic_candidates[0]
else:
logger.info(
f"🎯 [Semantic Guard] {len(semantic_candidates)} matches found for '{target_text}'. Reducing candidates for VLM."
)
candidates = semantic_candidates
else:
logger.warning(
f"⚠️ [Semantic Guard] No candidates found containing '{target_text}'. Returning None to prevent hallucination."
)
return None
try:
annotated_b64, box_map = self._annotate_screenshot_with_candidates(device, candidates)
except Exception as e:

12
GramAddict/core/screen_topology.py
View File

@@ -33,6 +33,7 @@ class ScreenTopology:
"tap profile tab": ScreenType.OWN_PROFILE,
"tap reels tab": ScreenType.REELS_FEED,
"tap messages tab": ScreenType.DM_INBOX,
"tap story ring avatar": ScreenType.STORY_VIEW,
},
ScreenType.EXPLORE_GRID: {
"tap home tab": ScreenType.HOME_FEED,
@@ -57,6 +58,9 @@ class ScreenTopology:
ScreenType.FOLLOW_LIST: {
"press back": ScreenType.OWN_PROFILE,
},
ScreenType.STORY_VIEW: {
"press back": ScreenType.HOME_FEED,
},
ScreenType.OTHER_PROFILE: {
"press back": ScreenType.HOME_FEED,
"tap home tab": ScreenType.HOME_FEED,
@@ -88,7 +92,9 @@ class ScreenTopology:
}
@classmethod
def find_route(cls, from_screen: ScreenType, to_screen: ScreenType, avoid_actions: set = None) -> Optional[List[Tuple[str, ScreenType]]]:
def find_route(
cls, from_screen: ScreenType, to_screen: ScreenType, avoid_actions: set = None
) -> Optional[List[Tuple[str, ScreenType]]]:
"""
BFS shortest path from from_screen to to_screen.
@@ -99,7 +105,7 @@ class ScreenTopology:
"""
if from_screen == to_screen:
return []
avoid_actions = avoid_actions or set()
queue: deque = deque()
@@ -113,7 +119,7 @@ class ScreenTopology:
for action, next_screen in transitions.items():
if action in avoid_actions or action.replace(" ", "_") in avoid_actions:
continue
if next_screen == to_screen:
return path + [(action, next_screen)]