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

test(core): enforce 100% TDD parity, eliminate mocks, and harden VLM hallucination guards

Browse Source
This commit is contained in:
Marc Mintel
2026-04-28 10:26:11 +02:00
parent bd9148e6e9
commit cd64794f55
17 changed files with 236 additions and 165 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
GramAddict/core/behaviors/comment.py
View File

@@ -34,11 +34,16 @@ class CommentPlugin(BehaviorPlugin):
# Safety Guard: Do not comment on stories or grids
xml_lower = (ctx.context_xml or "").lower()
STORY_MARKERS = ("reel_viewer_media_layout", "reel_viewer_header", "reel_viewer_progress_bar", "reel_viewer_root")
STORY_MARKERS = (
"reel_viewer_media_layout",
"reel_viewer_header",
"reel_viewer_progress_bar",
"reel_viewer_root",
)
if any(marker in xml_lower for marker in STORY_MARKERS):
return False
if "explore_grid" in xml_lower or "profile_tabs_container" in xml_lower:
if "explore_action_bar" in xml_lower or "profile_tabs_container" in xml_lower:
return False
config = self.get_config(ctx)

6
GramAddict/core/bot_flow.py
View File

@@ -51,7 +51,6 @@ from GramAddict.core.physics.timing import (
wait_for_story_loaded as _wait_for_story_loaded_impl,
)
from GramAddict.core.q_nav_graph import QNavGraph
from GramAddict.core.resonance_engine import ResonanceEngine
from GramAddict.core.sensors.honeypot_radome import HoneypotRadome
from GramAddict.core.session_state import SessionState, SessionStateEncoder
from GramAddict.core.swarm_protocol import SwarmProtocol
@@ -177,10 +176,9 @@ def start_bot(**kwargs):
)
persona_interests = [p.strip() for p in persona_raw.split(",") if p.strip()] if persona_raw else []
from GramAddict.core.qdrant_memory import DMMemoryDB, ParasocialCRMDB
from GramAddict.core.resonance_engine import ResonanceEngine
from GramAddict.core.interaction import LLMWriter
from GramAddict.core.qdrant_memory import DMMemoryDB, ParasocialCRMDB
from GramAddict.core.resonance_engine import ResonanceEngine
dopamine = DopamineEngine()
crm_db = ParasocialCRMDB()

26
GramAddict/core/interaction.py
View File

@@ -1,13 +1,15 @@
import logging
from typing import Dict, Optional
from typing import Dict
from GramAddict.core.llm_provider import query_llm
logger = logging.getLogger(__name__)
class LLMWriter:
"""
The Creative Engine — Content Generation for Interactions.
Generates high-fidelity, persona-aligned comments and messages.
Replaces legacy static 'comment_list' with dynamic, contextual resonance.
"""
@@ -29,7 +31,7 @@ class LLMWriter:
caption = post_data.get("caption", "")
description = post_data.get("description", "")
target_username = post_data.get("username", "the user")
# Build context for the LLM
context = f"Post by @{target_username}\n"
if caption:
@@ -38,7 +40,7 @@ class LLMWriter:
context += f"Visual Description: {description}\n"
interests_str = ", ".join(self.persona_interests) if self.persona_interests else "general interesting things"
prompt = (
f"You are an Instagram user interested in: {interests_str}.\n"
f"You want to leave a brief, friendly, and authentic comment on the following post:\n\n"
@@ -53,10 +55,12 @@ class LLMWriter:
)
model = getattr(self.args, "ai_writer_model", getattr(self.args, "ai_model", "llama3.2:1b"))
url = getattr(self.args, "ai_writer_url", getattr(self.args, "ai_model_url", "http://localhost:11434/api/generate"))
url = getattr(
self.args, "ai_writer_url", getattr(self.args, "ai_model_url", "http://localhost:11434/api/generate")
)
logger.info(f"✍️ [Writer] Generating comment for @{target_username} using {model}...")
try:
response_dict = query_llm(
url=url,
@@ -65,18 +69,18 @@ class LLMWriter:
system="You are a friendly Instagram user. You write short, authentic comments.",
format_json=False,
timeout=60,
temperature=0.7 # Add some variety to avoid 'the to the' loops
temperature=0.7, # Add some variety to avoid 'the to the' loops
)
if response_dict and "response" in response_dict:
comment = response_dict["response"].strip().strip('"')
# Basic cleaning to remove LLM artifacts
comment = comment.split("\n")[0] # Take only first line
comment = comment.split("\n")[0] # Take only first line
if not comment:
return "Nice!"
return comment
except Exception as e:
logger.error(f"✍️ [Writer] Failed to generate comment: {e}")
return "Great post! 🔥"

2
GramAddict/core/navigation/planner.py
View File

@@ -105,7 +105,7 @@ class GoalPlanner:
# The user explicitly wants the AI to be the primary driver of goals.
from GramAddict.core.navigation.brain import ask_brain_for_action
brain_action = ask_brain_for_action(goal, screen_type.name, available, explored_nav_actions)
brain_action = ask_brain_for_action(goal, screen_type.name, available, avoid_actions)
if brain_action:
logger.info(f"🧠 [Brain] Decided dynamically to execute: '{brain_action}'")
return brain_action

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

@@ -128,10 +128,10 @@ class ActionMemory:
if "follow" in intent_lower:
FOLLOW_SUCCESS_MARKERS = ["following", "requested", "abonniert", "angefragt", "gefolgt"]
if any(m in post_xml_lower for m in FOLLOW_SUCCESS_MARKERS):
logger.info(f"✅ [ActionMemory] Structural check confirmed follow success.")
logger.info("✅ [ActionMemory] Structural check confirmed follow success.")
return True
else:
logger.warning(f"⚠️ [ActionMemory] Follow success markers NOT found in post-click XML.")
logger.warning("⚠️ [ActionMemory] Follow success markers NOT found in post-click XML.")
# We don't return False immediately because it might take a second to update
# If we are highly confident (e.g. pulled from Qdrant memory), bypass heavy VLM

6
GramAddict/core/perception/screen_identity.py
View File

@@ -199,12 +199,12 @@ class ScreenIdentity:
# Stories hide the navigation tab bar, so selected_tab is always None.
# Must be checked BEFORE tab-based fallbacks to prevent UNKNOWN classification.
STORY_MARKERS = (
"reel_viewer_media_layout",
"reel_viewer_header",
"reel_viewer_media_layout",
"reel_viewer_header",
"reel_viewer_progress_bar",
"reel_viewer_root",
"story_viewer_container",
"reel_viewer_content_layout"
"reel_viewer_content_layout",
)
if any(marker in ids for marker in STORY_MARKERS):
return ScreenType.STORY_VIEW

33
GramAddict/core/physics/timing.py
View File

@@ -135,15 +135,15 @@ def align_active_post(device):
"""
aligned = False
attempts = 0
max_attempts = 5 # Increased for structural retry loop
max_attempts = 5 # Increased for structural retry loop
# Intents for structural discovery
intents = [
"post author header profile",
"post username name",
"row_feed_photo_profile_name", # ID fallback
"clips_viewer_author_container", # Reels fallback
"feed post content" # Final desperation
"row_feed_photo_profile_name", # ID fallback
"clips_viewer_author_container", # Reels fallback
"feed post content", # Final desperation
]
while not aligned and attempts < max_attempts:
@@ -151,16 +151,15 @@ def align_active_post(device):
try:
xml = device.dump_hierarchy()
from GramAddict.core.telepathic_engine import TelepathicEngine
telepath = TelepathicEngine.get_instance()
target_node = None
for intent in intents:
target_node = telepath.find_best_node(
xml, intent, min_confidence=0.35, device=device, track=False
)
target_node = telepath.find_best_node(xml, intent, min_confidence=0.35, device=device, track=False)
if target_node:
break
if target_node:
original_attribs = target_node.get("original_attribs", {})
bounds = original_attribs.get("bounds")
@@ -178,7 +177,7 @@ def align_active_post(device):
else:
logger.warning(f"📐 [Alignment] Could not parse bounds: {bounds}")
continue
# Check if this is a false positive (e.g. bottom bar item misclassified)
# Post headers should be in the top half usually, or at least not at the very bottom
info = device.get_info()
@@ -188,7 +187,7 @@ def align_active_post(device):
continue
header_y = (t + b) // 2
target_y = 250 # Top margin for headers
target_y = 250 # Top margin for headers
diff = header_y - target_y
# If target is off-center (> 50px for higher precision), execute precise correction swipe
@@ -198,7 +197,7 @@ def align_active_post(device):
cx = w // 2
max_safe_swipe = int(h * 0.4)
# Calculate movement
dist = min(abs(diff), max_safe_swipe)
if diff > 0:
@@ -214,9 +213,9 @@ def align_active_post(device):
# Duration 1.5s = ultra-precise mechanical drag with ZERO momentum
device.swipe(cx, start_y, cx, end_y, duration=1.5)
sleep(1.0)
# Refresh XML for next iteration check
continue
continue
else:
logger.info(f"🎯 [Alignment] Perfect snap achieved after {attempts} attempts.")
aligned = True
@@ -227,9 +226,9 @@ def align_active_post(device):
if attempts < 3:
info = device.get_info()
w, h = info.get("displayWidth", 1080), info.get("displayHeight", 2400)
device.swipe(w//2, h//2, w//2, h//2 - 20, duration=0.2)
device.swipe(w // 2, h // 2, w // 2, h // 2 - 20, duration=0.2)
sleep(0.5)
device.swipe(w//2, h//2 - 20, w//2, h//2, duration=0.2)
device.swipe(w // 2, h // 2 - 20, w // 2, h // 2, duration=0.2)
sleep(1.0)
else:
break

14
GramAddict/core/unfollow_engine.py
View File

@@ -66,17 +66,23 @@ def _run_zero_latency_unfollow_loop(
xml_dump = device.dump_hierarchy()
import re
# Smart Unfollow Phase 1: Find user rows via structural UI markers, not LLM (too prone to hallucinate headers)
nodes = []
# Find all nodes with resource-id="com.instagram.android:id/follow_list_username"
for match in re.finditer(r'resource-id="com\.instagram\.android:id/follow_list_username".*?bounds="\[(\d+),(\d+)\]\[(\d+),(\d+)\]"', xml_dump):
for match in re.finditer(
r'resource-id="com\.instagram\.android:id/follow_list_username".*?bounds="\[(\d+),(\d+)\]\[(\d+),(\d+)\]"',
xml_dump,
):
x1, y1, x2, y2 = map(int, match.groups())
nodes.append({"x": (x1 + x2) // 2, "y": (y1 + y2) // 2, "bounds": True})
# Also try com.instagram.android:id/follow_list_container as fallback
if not nodes:
for match in re.finditer(r'resource-id="com\.instagram\.android:id/follow_list_container".*?bounds="\[(\d+),(\d+)\]\[(\d+),(\d+)\]"', xml_dump):
for match in re.finditer(
r'resource-id="com\.instagram\.android:id/follow_list_container".*?bounds="\[(\d+),(\d+)\]\[(\d+),(\d+)\]"',
xml_dump,
):
x1, y1, x2, y2 = map(int, match.groups())
nodes.append({"x": (x1 + x2) // 2, "y": (y1 + y2) // 2, "bounds": True})