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

test(e2e): eliminate all legacy mocks and establish real-world sim suite

Browse Source
This commit is contained in:
Marc Mintel
2026-04-27 01:11:47 +02:00
parent b916b86bc5
commit 42a11107fd
51 changed files with 1007 additions and 3113 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
GramAddict/core/behaviors/__init__.py
View File

@@ -240,6 +240,7 @@ from GramAddict.core.behaviors.profile_visit import ProfileVisitPlugin # noqa:
from GramAddict.core.behaviors.rabbit_hole import RabbitHolePlugin # noqa: E402
from GramAddict.core.behaviors.repost import RepostPlugin # noqa: E402
from GramAddict.core.behaviors.resonance_evaluator import ResonanceEvaluatorPlugin # noqa: E402
from GramAddict.core.behaviors.scrape_profile import ScrapeProfilePlugin # noqa: E402
# Note: We do not automatically instantiate all of them globally here to avoid circular
# dependencies during initial load. The bot_flow.py engine should explicitly register them.
@@ -265,3 +266,4 @@ def load_all_plugins():
registry.register(RabbitHolePlugin())
registry.register(RepostPlugin())
registry.register(ResonanceEvaluatorPlugin())
registry.register(ScrapeProfilePlugin())

2
GramAddict/core/behaviors/anomaly_handler.py
View File

@@ -31,7 +31,7 @@ class AnomalyHandlerPlugin(BehaviorPlugin):
return getattr(self, "_enabled", True)
def execute(self, ctx: BehaviorContext) -> BehaviorResult:
telepathic = TelepathicEngine.get_instance()
telepathic = ctx.cognitive_stack.get("telepathic") or TelepathicEngine.get_instance()
xml = ctx.context_xml if ctx.context_xml else ctx.device.dump_hierarchy()
nodes = telepathic._extract_semantic_nodes(xml)

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

@@ -78,7 +78,7 @@ class CommentPlugin(BehaviorPlugin):
# 4. Type and post
if nav_graph.do("type and post comment", text=text):
logger.info(f"💬 [Comment] Posted to @{ctx.username} ✓")
ctx.session_state.add_interaction(source=ctx.username, succeed=True, followed=False, liked=False)
ctx.session_state.add_interaction(source=ctx.username, succeed=True, followed=False, scraped=False)
ctx.session_state.totalComments += 1
return BehaviorResult(executed=True, interactions=1, metadata={"text": text})

4
GramAddict/core/behaviors/like.py
View File

@@ -30,6 +30,7 @@ class LikePlugin(BehaviorPlugin):
from GramAddict.core.session_state import SessionState
if ctx.session_state.check_limit(SessionState.Limit.LIKES):
logger.error("LikePlugin: limit check failed")
return False
config = self.get_config(ctx)
@@ -54,7 +55,8 @@ class LikePlugin(BehaviorPlugin):
if nav_graph.do("tap like button"):
logger.info(f"❤️ [Like] Liked post by @{ctx.username} ✓")
ctx.session_state.add_interaction(source=ctx.username, succeed=True, followed=False, liked=True)
ctx.session_state.add_interaction(source=ctx.username, succeed=True, followed=False, scraped=False)
ctx.session_state.totalLikes += 1
return BehaviorResult(executed=True, interactions=1)
return BehaviorResult(executed=False)

78
GramAddict/core/behaviors/scrape_profile.py Normal file
View File

@@ -0,0 +1,78 @@
import logging
from GramAddict.core.behaviors import BehaviorContext, BehaviorPlugin, BehaviorResult
from GramAddict.core.telepathic_engine import TelepathicEngine
logger = logging.getLogger(__name__)
class ScrapeProfilePlugin(BehaviorPlugin):
"""
Extracts profile metadata (followers, following, bio) when visiting a profile.
Priority: 45. (Runs after ProfileGuard, before deep interactions like GridLike)
"""
def __init__(self):
super().__init__()
self._enabled = True
@property
def name(self) -> str:
return "scrape_profile"
@property
def priority(self) -> int:
return 45
def can_activate(self, ctx: BehaviorContext) -> bool:
if not getattr(self, "_enabled", True):
return False
# Only activate if scrape_profiles is True in config
if not getattr(ctx.configs.args, "scrape_profiles", False):
return False
# Only activate when we are actively visiting a profile (via ProfileVisitPlugin)
nav_graph = ctx.cognitive_stack.get("nav_graph")
if not nav_graph or nav_graph.current_state != "ProfileView":
return False
return True
def execute(self, ctx: BehaviorContext) -> BehaviorResult:
from colorama import Fore
logger.info(f"📊 [Scraping] Extracting metadata for @{ctx.username}...", extra={"color": f"{Fore.CYAN}"})
telepathic = ctx.cognitive_stack.get("telepathic") or TelepathicEngine.get_instance()
crm = ctx.cognitive_stack.get("crm")
xml_check = ctx.context_xml or ctx.device.dump_hierarchy()
f_node = telepathic.find_best_node(xml_check, "Followers count text or number", device=ctx.device)
fg_node = telepathic.find_best_node(xml_check, "Following count text or number", device=ctx.device)
bio_node = telepathic.find_best_node(xml_check, "User biography or description text", device=ctx.device)
scraped_data = {
"username": ctx.username,
"followers": f_node.get("text") if f_node else "unknown",
"following": fg_node.get("text") if fg_node else "unknown",
"bio": bio_node.get("text") if bio_node else "No bio",
}
logger.info(
f"✅ [Scraping] Data acquired: {scraped_data['followers']} followers, {scraped_data['following']} following."
)
ctx.session_state.add_interaction(source=ctx.username, succeed=False, followed=False, scraped=True)
if crm:
try:
crm.enrich_lead(ctx.username, scraped_data)
logger.info(f"💾 [CRM] Enriched lead @{ctx.username} in database.")
except Exception as e:
logger.error(f"❌ [CRM] Failed to enrich lead @{ctx.username}: {e}")
# Return executed=True, but we don't return interactions=1 since it's just data extraction
return BehaviorResult(executed=True)

32
GramAddict/core/bot_flow.py
View File

@@ -9,21 +9,6 @@ except ImportError:
from datetime import datetime
from time import sleep
def log_metabolic_rate():
if psutil is None:
logging.getLogger(__name__).debug("🧬 [Metabolism] psutil not installed. Skipping memory log.")
return
try:
process = psutil.Process(os.getpid())
mem_info = process.memory_info()
logging.getLogger(__name__).info(
f"🧬 [Metabolism] RSS: {mem_info.rss / 1024 / 1024:.2f} MB | VMS: {mem_info.vms / 1024 / 1024:.2f} MB"
)
except Exception as e:
logging.getLogger(__name__).debug(f"🧬 [Metabolism] Failed to log memory: {e}")
from colorama import Fore, Style
from GramAddict.core.account_switcher import verify_and_switch_account
@@ -84,6 +69,21 @@ from GramAddict.core.utils import (
)
from GramAddict.core.zero_latency_engine import ZeroLatencyEngine
def log_metabolic_rate():
if psutil is None:
logging.getLogger(__name__).debug("🧬 [Metabolism] psutil not installed. Skipping memory log.")
return
try:
process = psutil.Process(os.getpid())
mem_info = process.memory_info()
logging.getLogger(__name__).info(
f"🧬 [Metabolism] RSS: {mem_info.rss / 1024 / 1024:.2f} MB | VMS: {mem_info.vms / 1024 / 1024:.2f} MB"
)
except Exception as e:
logging.getLogger(__name__).debug(f"🧬 [Metabolism] Failed to log memory: {e}")
logger = logging.getLogger(__name__)
@@ -256,6 +256,7 @@ def start_bot(**kwargs):
from GramAddict.core.behaviors.rabbit_hole import RabbitHolePlugin
from GramAddict.core.behaviors.repost import RepostPlugin
from GramAddict.core.behaviors.resonance_evaluator import ResonanceEvaluatorPlugin
from GramAddict.core.behaviors.scrape_profile import ScrapeProfilePlugin
from GramAddict.core.behaviors.story_view import StoryViewPlugin
PluginRegistry.reset()
@@ -279,6 +280,7 @@ def start_bot(**kwargs):
plugin_registry.register(CommentPlugin())
plugin_registry.register(RepostPlugin())
plugin_registry.register(PostInteractionPlugin())
plugin_registry.register(ScrapeProfilePlugin())
cognitive_stack["plugin_registry"] = plugin_registry

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

@@ -145,47 +145,52 @@ def align_active_post(device):
telepath = TelepathicEngine.get_instance()
target_node = telepath.find_best_node(xml, "post author header profile", min_confidence=0.4, device=device)
if target_node:
original_attribs = target_node.get("original_attribs", {})
bounds = original_attribs.get("bounds", "")
if not bounds:
bounds = target_node.get("bounds", "")
bounds = original_attribs.get("bounds")
m = re.match(r"\[(\d+),(\d+)\]\[(\d+),(\d+)\]", bounds)
if m:
l, t, r, b = map(int, m.groups())
header_y = (t + b) // 2
# Instagram's optimal top margin for a snapped post is ~200-280px
target_y = 250
diff = header_y - target_y
# If target is off-center (> 100px), execute precise correction swipe
if abs(diff) > 100:
info = device.get_info()
w, h = info.get("displayWidth", 1080), info.get("displayHeight", 2400)
cx = w // 2
max_safe_swipe = int(h * 0.4)
if diff > 0:
# Content is too LOW. Move it UP.
dist = min(diff, max_safe_swipe)
start_y = int(h * 0.7)
end_y = start_y - dist
else:
# Content is too HIGH. Move it DOWN.
dist = min(abs(diff), max_safe_swipe)
start_y = int(h * 0.3)
end_y = start_y + dist
# Duration 1.0s = precise mechanical drag with ZERO momentum
device.swipe(cx, start_y, cx, end_y, duration=1.0)
sleep(1.0)
logger.debug(f"📐 [Alignment] Snapping attempt {attempts}: Shifted {diff}px.")
# If bounds is a tuple from SpatialNode.to_dict()
if isinstance(bounds, tuple) and len(bounds) == 4:
left, t, r, b = bounds
else:
# Fallback to string parsing
if not bounds:
bounds = target_node.get("bounds", "")
m = re.match(r"\[(\d+),(\d+)\]\[(\d+),(\d+)\]", str(bounds))
if m:
left, t, r, b = map(int, m.groups())
else:
aligned = True
break # Cannot parse bounds
header_y = (t + b) // 2
target_y = 250
diff = header_y - target_y
# If target is off-center (> 100px), execute precise correction swipe
if abs(diff) > 100:
info = device.get_info()
w, h = info.get("displayWidth", 1080), info.get("displayHeight", 2400)
cx = w // 2
max_safe_swipe = int(h * 0.4)
if diff > 0:
# Content is too LOW. Move it UP.
dist = min(diff, max_safe_swipe)
start_y = int(h * 0.7)
end_y = start_y - dist
else:
# Content is too HIGH. Move it DOWN.
dist = min(abs(diff), max_safe_swipe)
start_y = int(h * 0.3)
end_y = start_y + dist
# Duration 1.0s = precise mechanical drag with ZERO momentum
device.swipe(cx, start_y, cx, end_y, duration=1.0)
sleep(1.0)
logger.debug(f"📐 [Alignment] Snapping attempt {attempts}: Shifted {diff}px.")
else:
aligned = True
else:
break # No header found, cannot align
except Exception as e:

19
GramAddict/core/qdrant_memory.py
View File

@@ -1188,6 +1188,25 @@ class ParasocialCRMDB(QdrantBase):
log_success=f"🧠 [ParasocialCRM] Updated @{username} into Qdrant. Stage {stage} ({intent_type})",
)
def enrich_lead(self, username: str, data: dict):
"""
Enriches a lead with scraped data.
"""
if not self.is_connected:
return
current = self.get_relationship_stage(username)
current.update(data)
vector = self._get_embedding(f"User: {username}")
if vector:
self.upsert_point(
seed_string=f"User_{username}",
vector=vector,
payload=current,
log_success=f"🧠 [ParasocialCRM] Enriched @{username} data.",
)
def log_generated_comment(self, username: str, comment_text: str):
"""Phase 10: RAG memory point for specific users."""
if not self.is_connected:

2
GramAddict/core/telepathic_engine.py
View File

@@ -54,6 +54,8 @@ class TelepathicEngine:
# ──────────────────────────────────────────────
def find_best_node(self, xml_string: str, intent_description: str, device=None, **kwargs) -> Optional[dict]:
print("FIND_BEST_NODE CALLED")
"""
Public facade for resolving a node.
Translates Android UI bounds into standard GramAddict node dicts.