This commit is contained in:
2026-04-16 18:58:18 +02:00
commit fa1d01527d
124 changed files with 13989 additions and 0 deletions

8
GramAddict/__init__.py Normal file
View File

@@ -0,0 +1,8 @@
"""Human-like Instagram bot powered by UIAutomator2"""
from GramAddict.core.version import __version__, __tested_ig_version__
from GramAddict.core.bot_flow import start_bot
def run(**kwargs):
start_bot(**kwargs)

162
GramAddict/__main__.py Normal file
View File

@@ -0,0 +1,162 @@
from GramAddict.core.agentic_views import *
import argparse
from os import getcwd, path
from GramAddict import __version__
from GramAddict.core.bot_flow import start_bot
from GramAddict.core.download_from_github import download_from_github
def cmd_init(args):
if args.account_name is not None:
print(f"Script launched in {getcwd()}, files will be available there.")
for username in args.account_name:
if not path.exists("./run.py"):
print("Creating run.py ...")
download_from_github(
"https://github.com/GramAddict/bot/blob/master/run.py"
)
if not path.exists(f"./accounts/{username}"):
print(
f"Creating 'accounts/{username}' folder with a config starting point inside. You have to edit these files according with https://docs.gramaddict.org/#/configuration"
)
download_from_github(
"https://github.com/GramAddict/bot/tree/master/config-examples",
output_dir=f"accounts/{username}",
flatten=True,
)
else:
print(f"'accounts/{username}' folder already exists, skip.")
continue
with open(f"./accounts/{username}/config.yml", "r+", encoding="utf-8") as f:
config = f.read()
f.seek(0)
config_fixed = config.replace("myusername", username)
f.write(config_fixed)
else:
print("You have to provide at last one account name..")
def cmd_run(args):
start_bot()
def cmd_dump(args):
import os
import shutil
import time
import uiautomator2 as u2
from colorama import Fore, Style
if not args.no_kill:
os.popen("adb shell pkill atx-agent").close()
try:
d = u2.connect(args.device)
except RuntimeError as err:
raise SystemExit(err)
def dump_hierarchy(device, path):
xml_dump = device.dump_hierarchy()
with open(path, "w", encoding="utf-8") as outfile:
outfile.write(xml_dump)
def make_archive(name):
os.chdir("dump")
shutil.make_archive(base_name=f"screen_{name}", format="zip", root_dir="cur")
shutil.rmtree("cur")
os.makedirs("dump/cur", exist_ok=True)
d.screenshot("dump/cur/screenshot.png")
dump_hierarchy(d, "dump/cur/hierarchy.xml")
archive_name = int(time.time())
make_archive(archive_name)
print(
Fore.GREEN
+ Style.BRIGHT
+ "\nCurrent screen dump generated successfully! Please, send me this file:"
)
print(Fore.BLUE + Style.BRIGHT + f"{os.getcwd()}\\screen_{archive_name}.zip")
_commands = [
dict(
action=cmd_init,
command="init",
help="creates your account folder under accounts with files for configuration",
flags=[
dict(
args=["account_name"],
nargs="+",
help="instagram account name to initialize",
),
],
),
dict(
action=cmd_run,
command="run",
help="start the bot!",
flags=[
dict(args=["--config"], nargs="?", help="provide the config.yml path"),
],
),
dict(
action=cmd_dump,
command="dump",
help="dump current screen",
flags=[
dict(
args=["--device"],
nargs=None,
default=None,
help="provide the device name if more then one connected",
),
dict(
args=["--no-kill"],
action="store_true",
help="don't kill the uia2 demon",
),
],
),
]
def main() -> None:
parser = argparse.ArgumentParser(
prog="GramAddict",
description="free human-like Instagram bot",
)
parser.add_argument(
"-v", "--version", action="version", version=f"{parser.prog} {__version__}"
)
subparser = parser.add_subparsers(dest="subparser")
actions = {}
for c in _commands:
cmd_name = c["command"]
actions[cmd_name] = c["action"]
sp = subparser.add_parser(
cmd_name,
help=c.get("help"),
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
)
for f in c.get("flags", []):
args = f.get("args")
if not args:
args = ["-" * min(2, len(n)) + n for n in f["name"]]
kwargs = f.copy()
kwargs.pop("name", None)
kwargs.pop("args", None)
kwargs.pop("run", None)
sp.add_argument(*args, **kwargs)
args = parser.parse_args()
if args.subparser:
actions[args.subparser](args)
return
parser.print_help()
if __name__ == "__main__":
main()

View File

View File

@@ -0,0 +1,101 @@
import logging
import time
import math
from datetime import datetime
from colorama import Fore
logger = logging.getLogger(__name__)
class ActiveInferenceEngine:
"""
Bayesian Active Inference Engine.
Calculates Free Energy (Surprise) based on prediction errors in the
Instagram environment. Steers the agent's 'Thermodynamic Policy'.
"""
def __init__(self, username):
self.username = username
self.free_energy = 0.0
self.surprise_threshold = 0.75
self.last_update = time.time()
self.policy = "STABLE" # STABLE, CAUTIOUS, DORMANT
self.expectation_history = []
def calculate_surprise(self, predicted_outcome: float, observed_outcome: float):
"""
Bayesian surprise calculation (simplified Kullback-Leibler divergence).
"""
# prediction error
error = abs(predicted_outcome - observed_outcome)
# Free energy accumulation
self.free_energy = (self.free_energy * 0.7) + (error * 0.3)
# Decay free energy over time (Thermodynamic relaxation)
now = time.time()
hours_passed = (now - self.last_update) / 3600.0
decay = math.exp(-0.1 * hours_passed)
self.free_energy *= decay
self.last_update = now
# Policy steering
if self.free_energy > 1.2:
self.policy = "DORMANT"
elif self.free_energy > self.surprise_threshold:
self.policy = "CAUTIOUS"
else:
self.policy = "STABLE"
logger.info(f"⚖️ [Active Inference] Surprise: {self.free_energy:.4f} | Policy: {self.policy}", extra={"color": f"{Fore.BLUE}"})
return self.free_energy
def predict_state(self, expected_signature: list):
"""
Registers an expectation about the future UI state before acting.
expected_signature: list of terms expected in the resulting XML.
"""
self.expectation_history.append(expected_signature)
logger.debug(f"⚖️ [Shadow Mode] Predicting future state containing: {expected_signature}", extra={"color": f"{Fore.BLUE}"})
def evaluate_prediction(self, context_xml: str) -> bool:
"""
Evaluates the last prediction against reality.
Returns True if reality matches prediction, False otherwise (Prediction Error).
"""
if not self.expectation_history:
return True
expected_signature = self.expectation_history.pop()
matched = any(sig.lower() in context_xml.lower() for sig in expected_signature)
if matched:
self.calculate_surprise(1.0, 1.0)
return True
else:
logger.warning(f"⚖️ [Shadow Mode] Prediction Error! Did not find {expected_signature} in resulting UI.", extra={"color": f"{Fore.RED}"})
self.calculate_surprise(1.0, 0.0)
# ── Dojo Data Engine Hook ──
# When prediction fails, explicitly submit the snapshot for shadow-compilation
try:
from GramAddict.core.dojo_engine import DojoEngine
# Note: get_instance() works without passing device as it was already initialized in bot_flow by this point.
dojo = DojoEngine.get_instance()
dojo.submit_snapshot(
heuristic_name=str(expected_signature),
context_xml=context_xml,
intent_prompt=f"Locate the missing elements or correct the heuristic predicting state: {expected_signature}"
)
except Exception as e:
logger.error(f"Failed to offload snapshot to Dojo Engine: {e}")
return False
def get_sleep_modifier(self):
"""
Returns a multiplier for sleep durations based on surprise.
"""
if self.policy == "DORMANT":
return 5.0
if self.policy == "CAUTIOUS":
return 2.0
return 1.0

View File

@@ -0,0 +1,76 @@
import os
import json
import logging
from colorama import Fore, Style
logger = logging.getLogger(__name__)
BENCHMARKS_FILE = os.path.join(os.path.dirname(__file__), "llm_benchmarks.json")
def check_model_benchmarks(configs):
"""
Checks the configured AI models against the local benchmark database.
Emits warnings if the user is running untested or underperforming models
that could lead to agent hallucinations or broken interactions.
"""
if not os.path.exists(BENCHMARKS_FILE):
return
try:
with open(BENCHMARKS_FILE, "r") as f:
data = json.load(f)
benchmarks = data.get("models", {})
except Exception as e:
logger.warning(f"Could not load LLM benchmarks: {e}")
return
def _eval_model(model_name: str, context: str):
if not model_name:
return
if model_name not in benchmarks:
logger.warning(
f"⚠️ [Benchmark Guard] Model '{model_name}' (for {context}) is COMPLETELY UNTESTED "
f"for Singularity V8. Expect severe hallucinations or crashed agents.",
extra={"color": f"{Style.BRIGHT}{Fore.RED}"}
)
return
scores = benchmarks[model_name]
# Telepathic/Vision tasks require high structural strictness
if context == "Vision/Telepathic":
score = scores.get("telepathic_score", 0)
else:
score = scores.get("resonance_score", 0)
if score < 50:
logger.error(
f"⛔ [Benchmark Guard] Model '{model_name}' (for {context}) achieved a CRITICAL FAILURE score "
f"of {score}/100. Autonomous safety is compromised. DO NOT RUN UNATTENDED.",
extra={"color": f"{Style.BRIGHT}{Fore.RED}"}
)
elif score < 80:
logger.warning(
f"⚠️ [Benchmark Guard] Model '{model_name}' (for {context}) achieved a SUB-STANDARD score "
f"of {score}/100. It may occasionally hallucinate UI elements or misinterpret semantics.",
extra={"color": f"{Style.BRIGHT}{Fore.YELLOW}"}
)
else:
logger.info(
f"✅ [Benchmark Guard] Model '{model_name}' (for {context}) passes safety benchmarks ({score}/100).",
extra={"color": f"{Style.BRIGHT}{Fore.GREEN}"}
)
# Which models did the user configure?
telepathic_model = getattr(configs.args, "ai_telepathic_model", None)
text_model = getattr(configs.args, "ai_model", None)
condenser_model = getattr(configs.args, "ai_condenser_model", None)
_eval_model(telepathic_model, "Vision/Telepathic")
if text_model and text_model != telepathic_model:
_eval_model(text_model, "Dopamine/Resonance")
if condenser_model and condenser_model != text_model and condenser_model != telepathic_model:
_eval_model(condenser_model, "Context Condensation")

1502
GramAddict/core/bot_flow.py Normal file

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,98 @@
import logging
import json
from io import BytesIO
logger = logging.getLogger(__name__)
class VLMCompilerEngine:
"""
Project Singularity V7: The Self-Compiling Heuristics Engine
This engine leverages a massive VLM to analyze failures in the Zero-Latency Engine.
It takes a screenshot + XML dump, finds the missing intent, and generates a new,
blazing-fast deterministic Regex/XPath rule to be cached and executed next time.
"""
def __init__(self, device):
self.device = device
def generate_heuristic(self, intent_description: str, context_xml: str) -> dict:
"""
Calls the VLM to visually find the intent in the screen, then cross-reference it
with the provided XML to generate a deterministic extraction rule.
"""
logger.warning(f"🧠 [Compiler Engine] Deterministic heuristic failed for: '{intent_description}'. Synthesizing new rule...", extra={"color": "\x1b[1m\x1b[35m"})
args = getattr(self.device, "args", None)
model = getattr(args, "ai_telepathic_model", "google/gemini-3.1-flash-lite-preview") if args else "google/gemini-3.1-flash-lite-preview"
url = getattr(args, "ai_telepathic_url", "https://openrouter.ai/api/v1/chat/completions") if args else "https://openrouter.ai/api/v1/chat/completions"
use_local = "11434" in url or "localhost" in url
simplified_xml = self._simplify_xml(context_xml)
system_prompt = (
"You write Python regex rules to find Android UI elements. "
"Given UI XML, find the element matching the intent. "
"Generate a regex pattern to match its resource-id.\n\n"
"OUTPUT FORMAT (JSON only):\n"
"{\"rule_type\": \"regex\", \"target_attribute\": \"resource-id\", "
"\"pattern\": \".*your_regex.*\", \"confidence\": 0.95, "
"\"reasoning\": \"brief explanation\"}\n\n"
"RULES:\n"
"- ONLY use rule_type='regex'. NEVER use xpath.\n"
"- Target resource-id for dynamic elements, not text or usernames.\n"
"- Make patterns globally reusable, not hardcoded to specific content."
)
user_prompt = f"TARGET INTENT: {intent_description}\n\nUI XML:\n{simplified_xml[:2000]}"
try:
from GramAddict.core.llm_provider import query_telepathic_llm
res_text = query_telepathic_llm(
model=model,
url=url,
system_prompt=system_prompt,
user_prompt=user_prompt,
temperature=0.1,
use_local_edge=use_local
)
if res_text and res_text.startswith("```"):
res_text = "\n".join(res_text.strip().split("\n")[1:-1])
decision = json.loads(res_text) if res_text else {}
pattern = decision.get('pattern')
if not pattern:
logger.error("Compiler LLM returned empty rule pattern. Aborting heuristic generation.")
return None
logger.info(f"✨ [Compiler] New Heuristic Synthesized! Rule: {decision.get('rule_type')} -> {pattern}", extra={"color": "\x1b[1m\x1b[32m"})
if decision.get("rule_type") == "xpath":
logger.error("Compiler LLM returned 'xpath'. Rejecting rule because it causes xml.etree crashes. Will fallback/retry.")
return None
return {
"rule_type": "regex",
"target_attribute": decision.get("target_attribute", "text"),
"pattern": pattern
}
except Exception as e:
logger.error(f"Heuristic compilation crashed: {e}")
return None
def _simplify_xml(self, xml_tree: str) -> str:
import xml.etree.ElementTree as ET
nodes = []
try:
root = ET.fromstring(xml_tree)
for i, node in enumerate(root.iter("node")):
attrib = node.attrib
text = attrib.get("text", "")
desc = attrib.get("content-desc", "")
res_id = attrib.get("resource-id", "")
if text or desc or res_id:
nodes.append(f"[{i}] text='{text}' desc='{desc}' r_id='{res_id}'")
except:
pass
return "\n".join(nodes)

265
GramAddict/core/config.py Normal file
View File

@@ -0,0 +1,265 @@
import logging
import os
import sys
from datetime import datetime
from typing import Optional
import configargparse
import yaml
from colorama import Fore, Style
logger = logging.getLogger(__name__)
class Config:
def __init__(self, first_run=False, **kwargs):
if kwargs:
self.args = kwargs
self.module = True
else:
self.args = sys.argv
self.module = False
self.config = None
self.config_list = None
self.actions = {}
self.enabled = []
self.unknown_args = []
self.debug = False
self.device_id: Optional[str] = None
self.app_id: Optional[str] = None
self.first_run = first_run
self.username = False
# Pre-Load Variables Needed for Script Init
if self.module:
if "debug" in self.args:
self.debug = True
if "username" in self.args:
self.username = self.args["username"]
if isinstance(self.username, list) and len(self.username) > 0:
self.username = self.username[0]
if "app_id" in self.args:
app_id = self.args["app_id"]
if app_id:
self.app_id = app_id
else:
self.app_id = "com.instagram.android"
elif "--config" in self.args:
try:
file_name = self.args[self.args.index("--config") + 1]
if not file_name.endswith((".yml", ".yaml")):
logger.error(
f"You have to specify a *.yml / *.yaml config file path (For example 'accounts/your_account_name/config.yml')! \nYou entered: {file_name}, abort."
)
sys.exit(1)
logger.debug(get_time_last_save(file_name))
with open(file_name, encoding="utf-8") as fin:
# preserve order of yaml
self.config_list = [line.strip() for line in fin]
fin.seek(0)
# preload config for debug and username
self.config = yaml.safe_load(fin)
except IndexError:
logger.warning(
"Please provide a filename with your --config argument. Example: '--config accounts/yourusername/config.yml'"
)
exit(2)
except FileNotFoundError:
logger.error(
f"I can't see the file '{file_name}'! Double check the spelling or if you're calling the bot from the right folder. (You're there: '{os.getcwd()}')"
)
exit(2)
self.username = self.config.get("username", False)
if isinstance(self.username, list) and len(self.username) > 0:
self.username = self.username[0]
self.debug = self.config.get("debug", False)
self.app_id = self.config.get("app_id", "com.instagram.android")
else:
if "--debug" in self.args:
self.debug = True
if "--username" in self.args:
try:
self.username = self.args[self.args.index("--username") + 1]
except IndexError:
logger.warning(
"Please provide a username with your --username argument. Example: '--username yourusername'"
)
exit(2)
if "--app-id" in self.args:
self.app_id = self.args[self.args.index("--app-id") + 1]
else:
self.app_id = "com.instagram.android"
# Configure ArgParse
self.parser = configargparse.ArgumentParser(
config_file_open_func=lambda filename: open(
filename, "r+", encoding="utf-8"
),
description="GramAddict Instagram Bot - Singularity V7",
)
self.parser.add_argument(
"--config",
required=False,
help="config file path",
)
self.parser.add_argument(
"--device",
help="device id",
)
self.parser.add_argument(
"--app-id",
help="app id",
default="com.instagram.android",
)
self.parser.add_argument(
"--debug",
action="store_true",
help="debug mode",
)
self.parser.add_argument(
"--shadow-mode",
required=False,
action="store_true",
help="Enable Tesla E2E Vision 'Shadow Mode' Telemetry daemon.",
)
# Core Singularity Jobs
self.parser.add_argument("--feed", help="Amount of feed posts to interact with", default=None)
self.parser.add_argument("--explore", help="Amount of explore posts to interact with", default=None)
self.parser.add_argument("--reels", help="Amount of reels to interact with natively", default=None)
self.parser.add_argument("--stories", help="Amount of top-level stories to binge natively", default=None)
self.parser.add_argument("--repeat", help="Amount of times to repeat the whole process", default=None)
self.parser.add_argument("--total-sessions", help="Total amount of sessions", default="-1")
self.parser.add_argument("--working-hours", help="Working hours", default=None)
self.parser.add_argument("--time-delta-session", help="Time delta between sessions", default=None)
self.parser.add_argument("--restart-atx-agent", action="store_true", help="Restart atx agent")
self.parser.add_argument("--allow-untested-ig-version", action="store_true", help="Allow untested IG version")
self.parser.add_argument("--capture-e2e-dumps", action="store_true", help="Automatically navigate through the app and capture missing XML dumps for the test suite")
# Interaction settings
self.parser.add_argument("--likes-count", help="Likes count", default="2-3")
self.parser.add_argument("--likes-percentage", help="Likes percentage", default="100")
self.parser.add_argument("--stories-count", help="Stories count", default="0")
self.parser.add_argument("--stories-percentage", help="Stories percentage", default="0")
# Total Limits (Legacy names preserved for SessionState compatibility)
self.parser.add_argument("--total-likes-limit", help="Total likes limit", default="300")
self.parser.add_argument("--total-follows-limit", help="Total follows limit", default="50")
self.parser.add_argument("--total-unfollows-limit", help="Total unfollows limit", default="50")
self.parser.add_argument("--total-comments-limit", help="Total comments limit", default="10")
self.parser.add_argument("--total-pm-limit", help="Total pm limit", default="10")
self.parser.add_argument("--total-watches-limit", help="Total watches limit", default="50")
self.parser.add_argument("--total-successful-interactions-limit", help="Total successful interactions limit", default="100")
self.parser.add_argument("--total-interactions-limit", help="Total interactions limit", default="1000")
self.parser.add_argument("--total-scraped-limit", help="Total scraped limit", default="200")
self.parser.add_argument("--total-crashes-limit", help="Total crashes limit", default="5")
self.parser.add_argument("--speed-multiplier", help="Speed multiplier", default="1.0")
# AI Model Configuration (centralized — no hardcoded model names anywhere)
self.parser.add_argument("--ai-model", "--ai-text-model", help="Primary LLM model (OpenRouter or Ollama)", default="google/gemini-2.5-flash-lite-preview")
self.parser.add_argument("--ai-model-url", "--ai-text-url", help="Primary LLM endpoint URL", default="https://openrouter.ai/api/v1/chat/completions")
self.parser.add_argument("--ai-telepathic-model", help="Text-based model for Telepathic Engine Fallbacks", default="google/gemini-3.1-flash-lite-preview")
self.parser.add_argument("--ai-telepathic-url", help="Telepathic model endpoint URL", default="https://openrouter.ai/api/v1/chat/completions")
self.parser.add_argument("--ai-fallback-model", "--ai-text-fallback-model", help="Fallback model when primary fails", default="llama3.2:1b")
self.parser.add_argument("--ai-fallback-url", "--ai-text-fallback-url", help="Fallback model endpoint URL", default="http://localhost:11434/api/generate")
self.parser.add_argument("--ai-embedding-model", help="Embedding model for vector operations", default="nomic-embed-text")
self.parser.add_argument("--ai-embedding-url", help="Embedding endpoint URL", default="http://localhost:11434/api/embeddings")
# Persona & Resonance (drives ALL content evaluation and interaction decisions)
self.parser.add_argument("--persona-interests", help="Comma-separated niche interests for content matching", default="")
self.parser.add_argument("--ai-target-audience", help="Target audience used interchangeably with persona interests", default="")
self.parser.add_argument("--interact-percentage", help="Overall interaction probability percentage", default="80")
self.parser.add_argument("--comment-percentage", help="Comment probability percentage", default="0")
self.parser.add_argument("--dry-run-comments", action="store_true", help="Generate AI comments but do not actually post them (debug/logging only)")
self.parser.add_argument("--search", help="Comma-separated keywords to search for", default="")
self.parser.add_argument("--scrape-profiles", action="store_true", help="Extract and store profile metadata in CRM")
# Phase 10: RAG Comment Learning & Extractor Settings
self.parser.add_argument("--ai-condenser-model", help="LLM used for condensing text/comments", default="google/gemini-2.5-flash-lite-preview")
self.parser.add_argument("--ai-condenser-url", help="URL for the condenser model", default="https://openrouter.ai/api/v1/chat/completions")
self.parser.add_argument("--ai-learn-comments", action="store_true", help="Extract and learn from comment sections")
self.parser.add_argument("--ai-learn-niche-posts", action="store_true", help="Learn from niche posts")
self.parser.add_argument("--ai-learn-own-profile", action="store_true", help="Learn from your own profile interactions")
self.parser.add_argument("--ai-learn-only", action="store_true", help="Run the bot in a pure read-only learning mode")
self.parser.add_argument("--ai-vibe", help="The specific vibe to extract from comments (e.g., friendly, controversial)", default="")
self.parser.add_argument("--ai-blacklist-topics", help="Comma-separated topics heavily penalized or skipped", default="")
self.parser.add_argument("--ai-quality-filter", action="store_true", help="Use AI to strictly filter the quality of posts and comments")
# on first run, we must wait to proceed with loading
if not self.first_run:
self.parse_args()
def parse_args(self):
if self.module:
if self.first_run:
logger.debug("Arguments used:")
if self.config:
logger.debug(f"Config used: {self.config}")
if len(self.args) == 0:
self.parser.print_help()
exit(0)
else:
if self.first_run:
logger.debug(f"Arguments used: {' '.join(sys.argv[1:])}")
if self.config:
logger.debug(f"Config used: {self.config}")
if len(sys.argv) <= 1:
self.parser.print_help()
exit(0)
if self.config:
cleaned_config = {}
for k, v in self.config.items():
# Replace dictionaries with a placeholder to avoid argparse crashing
# We'll resolve the actual values later in specialize()
val = v
if isinstance(v, dict):
val = "SPECIALIZED"
cleaned_config[k.replace("-", "_")] = val
self.parser.set_defaults(**cleaned_config)
if self.module:
arg_str = ""
for k, v in self.args.items():
new_key = k.replace("_", "-")
new_key = f" --{new_key}"
arg_str += f"{new_key} {v}"
self.args, self.unknown_args = self.parser.parse_known_args(args=arg_str)
else:
self.args, self.unknown_args = self.parser.parse_known_args()
self.device_id = self.args.device
# Map actions for Singularity V7
if getattr(self.args, "feed", None): self.enabled.append("feed")
if getattr(self.args, "explore", None): self.enabled.append("explore")
def specialize(self, username):
if self.config is None:
return
logger.debug(f"Specializing config for account: {username}")
for key, value in self.config.items():
if isinstance(value, dict) and username in value:
resolved_value = value[username]
arg_name = key.replace("-", "_")
if hasattr(self.args, arg_name):
setattr(self.args, arg_name, resolved_value)
logger.info(
f"Applied override for {username}: {key} = {resolved_value}",
extra={"color": f"{Style.BRIGHT}{Fore.BLUE}"},
)
# Handle the case where username itself is a list - we specialize it to the current target
self.args.username = [username] if isinstance(self.args.username, list) else username
def get_time_last_save(file_path) -> str:
try:
absolute_file_path = os.path.abspath(file_path)
timestamp = os.path.getmtime(absolute_file_path)
last_save = datetime.fromtimestamp(timestamp).strftime("%Y-%m-%d %H:%M:%S")
return f"{file_path} has been saved last time at {last_save}"
except FileNotFoundError:
return f"File {file_path} not found"

View File

@@ -0,0 +1,242 @@
import logging
import random
import os
import math
import uuid
import time
from datetime import datetime
from GramAddict.core.qdrant_memory import QdrantBase
logger = logging.getLogger(__name__)
class DarwinEngine(QdrantBase):
"""
Project Singularity: Continuous Bayesian Evolutionary Engine V3 (Proof of Resonance).
Determines mathematically how to act on a per-post basis, generating custom
Dwell Times and nonlinear scroll sequences to maximize the RL Reward Matrix.
"""
def __init__(self, username: str, config_path: str = "config.yml"):
self.username = username
self.config_path = config_path
super().__init__(collection_name="bot_darwin_mdp_resonance", vector_size=5) # 5 corresponds to behavior_bounds length
# We replace naive percentages with Markovian Dwell Behaviors
self.behavior_bounds = {
"initial_dwell_sec": (1.0, 15.0, 2.0),
"scroll_velocity": (0.1, 2.0, 0.3), # 1.0 is normal
"back_swipe_prob": (0.0, 0.4, 0.1),
"profile_visit_prob": (0.0, 0.8, 0.2),
"comment_read_dwell": (0.0, 20.0, 4.0)
}
self.current_behavior = {}
def synthesize_interaction_profile(self, target_resonance: float) -> dict:
"""
Given an AI aesthetic resonance score (0.0 to 1.0), this generates
a deterministic topological interaction behavior mathematically suited to the target.
"""
history = self._get_historical_landscape()
epsilon = 0.15 # 15% pure exploration
if not history or random.random() < epsilon:
logger.info("🧬 [Darwin Engine] EXPLORE: Generating chaotic non-linear behavioral vector.")
center = {k: (v[0]+v[1])/2 for k, v in self.behavior_bounds.items()}
self.current_behavior = self._mutate(center)
else:
# Exploitation: Nearest neighbor matching the resonance profile closely
best_node = max(history, key=lambda x: x[1]) # x[1] is the Reward
best_params = best_node[0]
logger.info(f"🧬 [Darwin Engine] EXPLOIT: Adapting proven behavioral vector from highest Peak Reward ({best_node[1]:.2f}).")
self.current_behavior = self._mutate(best_params)
# Modulate behavior directly by resonance
# E.g., if resonance is 0.9 (amazing post), read comments longer!
self.current_behavior["initial_dwell_sec"] *= max(0.5, target_resonance * 1.5)
self.current_behavior["profile_visit_prob"] *= max(0.2, target_resonance * 2.0)
# Clip bounds
for k, (b_min, b_max, _) in self.behavior_bounds.items():
self.current_behavior[k] = max(b_min, min(b_max, self.current_behavior[k]))
return self.current_behavior
def execute_proof_of_resonance(self, device, resonance: float, nav_graph=None, zero_engine=None, configs=None, resonance_oracle=None, username=None):
"""
Translates the mathematical interaction profile directly into device actions
to prove engagement to the platform's anti-bot heuristic algorithm.
"""
profile = self.synthesize_interaction_profile(resonance)
logger.info("🧬 [Darwin MDP] Executing Proof of Resonance Sequence...")
# 1. Initial Dwell
dwell = profile["initial_dwell_sec"]
logger.debug(f" -> Dwelling for {dwell:.1f}s")
time.sleep(dwell)
# 2. Non-linear cognitive latency (Micro-Jitters)
if profile["scroll_velocity"] != 1.0:
logger.debug(f" -> Simulating cognitive read latency (Micro-Jitters, Velocity: {profile['scroll_velocity']:.2f})")
info = device.get_info()
h = info.get("displayHeight", 2400)
w = info.get("displayWidth", 1080)
# Thumb starts on the right side of the screen to avoid clicking polls/tags in the center
cx = int(w * 0.8) + device.cm_to_pixels(random.uniform(-0.3, 0.3))
cy = h // 2
# Keep distance microscopic (0.1 to 0.3 cm) so we DO NOT lose visual alignment
distance = device.cm_to_pixels(random.uniform(0.1, 0.3))
duration = max(0.5, 1.0 / max(0.1, profile["scroll_velocity"]))
start_y = int(cy + distance / 2)
end_y = int(cy - distance / 2)
# Add some x-axis noise for nonlinear human realism (~0.1 cm)
noise_x = device.cm_to_pixels(random.uniform(-0.1, 0.1))
device.deviceV2.swipe(cx, start_y, cx + noise_x, end_y, duration=duration)
# 3. Micro Back-swipe (The Human Wobble)
if random.random() < profile["back_swipe_prob"]:
logger.debug(" -> Executing cognitive wobble (Trace swipe)")
# small rapid corrective swipe (approx 0.1-0.2 cm downward slip)
slip_distance = device.cm_to_pixels(random.uniform(0.1, 0.2))
noise_x = device.cm_to_pixels(random.uniform(-0.1, 0.1))
cx = w // 2 + device.cm_to_pixels(random.uniform(-0.5, 0.5))
cy = h // 2
device.deviceV2.swipe(cx, cy, cx + noise_x, cy + slip_distance, duration=random.uniform(0.2, 0.5))
time.sleep(random.uniform(0.5, 1.2))
# 4. Comment depth simulation (probabilistic & resonance-correlated)
if profile["comment_read_dwell"] > 1.0 and resonance > 0.4 and random.random() < 0.3:
if nav_graph and zero_engine:
logger.debug(f" -> Opening comments section for {profile['comment_read_dwell']:.1f}s depth simulation")
success = nav_graph._execute_transition("tap_comment_button", zero_engine)
if success:
# ---- Phase 10: RAG Comment Extraction ----
if configs and resonance_oracle and getattr(configs.args, "ai_learn_comments", False):
# Limit scraping to 15% to avoid mechanical persistence
if random.random() < 0.05:
logger.debug(" -> Dumping UI hierarchy for Comment Extraction...")
try:
xml_data = device.deviceV2.dump_hierarchy()
t0 = time.time()
resonance_oracle.extract_and_learn_comments(xml_data, configs, author=username or "unknown")
t1 = time.time()
remaining_sleep = profile["comment_read_dwell"] - (t1 - t0)
if remaining_sleep > 0:
time.sleep(remaining_sleep)
except Exception as e:
logger.error(f" -> Comment extraction failed: {e}")
time.sleep(profile["comment_read_dwell"])
else:
logger.debug(" -> Skipping RAG Extraction (Probabilistic Evasion)")
time.sleep(profile["comment_read_dwell"])
else:
time.sleep(profile["comment_read_dwell"])
# ------------------------------------------
logger.debug(" -> Closing comments section")
device.deviceV2.press("back")
time.sleep(1.0)
# Instead of relying on a fragile bottom_sheet_container ID,
# we verify if the feed is visible. If not, the comment sheet is still open (or keyboard).
ui_dump = device.deviceV2.dump_hierarchy()
if 'resource-id="com.instagram.android:id/row_feed"' not in ui_dump and 'resource-id="com.instagram.android:id/button_like"' not in ui_dump:
logger.debug(" -> Not back on Home feed, pressing back again to close comment sheet/keyboard")
device.deviceV2.press("back")
time.sleep(1.0)
else:
logger.debug(f" -> Could not find comment button, falling back to dwell simulation for {profile['comment_read_dwell']:.1f}s")
time.sleep(profile["comment_read_dwell"])
else:
logger.debug(f" -> Simulating comment section processing for {profile['comment_read_dwell']:.1f}s")
time.sleep(profile["comment_read_dwell"])
logger.info("🧬 [Darwin MDP] Interaction sequence completed safely.")
return profile
def execute_micro_wobble(self, device):
"""
Simulates a thumb resting or slightly shifting on the glass.
Essential for breaking the 'robotically still' dwell periods.
"""
if random.random() < 0.2: # 20% chance for a wobble during dwell
logger.debug("🧬 [Ghost Protocol] Micro-Wobble triggered.")
info = device.get_info()
w = info.get("displayWidth", 1080)
h = info.get("displayHeight", 2400)
cx = int(w * 0.8) + device.cm_to_pixels(random.uniform(-0.3, 0.3))
cy = h // 2
# Keep the shift very small (~0.05 to 0.15 cm) so it doesn't actually scroll the feed up/down noticeably
y_shift = device.cm_to_pixels(random.uniform(0.05, 0.15)) * random.choice([1, -1])
x_shift = device.cm_to_pixels(random.uniform(-0.05, 0.05))
# Single slow slip
if hasattr(device, "human_swipe"):
device.human_swipe(cx, cy, cx + x_shift, cy + y_shift, duration=random.uniform(0.1, 0.2))
else:
device.deviceV2.swipe(cx, cy, cx + x_shift, cy + y_shift, duration=random.uniform(0.1, 0.2))
def _get_historical_landscape(self):
try:
records = self.client.scroll(
collection_name=self.collection_name,
limit=1000,
with_payload=True
)[0]
return [(r.payload.get("params", {}), r.payload.get("reward", 0.0)) for r in records]
except Exception:
return []
def _mutate(self, base_params: dict) -> dict:
new_params = {}
for key, (p_min, p_max, volatility) in self.behavior_bounds.items():
base_val = base_params.get(key, (p_min + p_max) / 2)
mutation = random.gauss(0, volatility)
new_params[key] = round(base_val + mutation, 3)
return new_params
def select_arm_and_apply(self, args):
"""
Multi-Armed Bandit (MAB) logic to select the most promising behavioral
mutation strategy for the current account phase.
"""
logger.info(f"🧬 [Darwin Engine] Applying MDP State channel for @{self.username}...")
self.synthesize_interaction_profile(target_resonance=0.5) # Initial neutral bias
def evaluate_session_end(self, duration_minutes: float, followers_gained: int):
if duration_minutes <= 0: duration_minutes = 1.0
reward = (followers_gained / duration_minutes) * 10.0
logger.info(f"🧬 [Darwin Engine] Session Evaluation: {followers_gained} followers gained in {duration_minutes:.1f}m. Reward: {reward:.2f}")
self.emit_reward_signal(followers_gained=followers_gained, block_warnings_seen=0)
def emit_reward_signal(self, followers_gained: int, block_warnings_seen: int):
if not self.current_behavior:
return
try:
reward = followers_gained - (block_warnings_seen * 50)
vector = []
for k, (p_min, p_max, _) in self.behavior_bounds.items():
val = self.current_behavior.get(k, p_min)
norm = (val - p_min) / max(0.1, (p_max - p_min))
vector.append(norm)
p_id = str(uuid.uuid4())
self.upsert_point(
seed_string=p_id,
vector=vector,
payload={
"username": self.username,
"timestamp": datetime.now().isoformat(),
"params": self.current_behavior,
"reward": reward
},
log_success=f"🧬 [Darwin Engine V3] MDP Reward Matrix stored. Reward Value: {reward:.2f}"
)
except Exception as e:
logger.debug(f"🧬 [Darwin Engine] Failed to record reward: {e}")

View File

@@ -0,0 +1,174 @@
import logging
import json
import uiautomator2 as u2
from time import sleep
from random import uniform
from GramAddict.core.utils import random_sleep
from functools import wraps
logger = logging.getLogger(__name__)
def adb_retry(retries=3, delay=2.0):
def decorator(func):
@wraps(func)
def wrapper(self, *args, **kwargs):
last_err = None
for attempt in range(retries):
try:
return func(self, *args, **kwargs)
except Exception as e:
last_err = e
logger.warning(f"⚠️ ADB Error in {func.__name__} (Attempt {attempt+1}/{retries}): {e}")
sleep(delay * (attempt + 1)) # Exponential backoff
logger.error(f"❌ ADB action {func.__name__} failed after {retries} retries. Crashing gracefully.")
raise last_err
return wrapper
return decorator
def create_device(device_id, app_id, args=None):
try:
return DeviceFacade(device_id, app_id, args)
except Exception as e:
logger.error(f"Failed to create device: {e}")
# In V7, we don't want to just return None and crash later.
# We should raise so the orchestrator knows it's a fatal boot error.
raise e
def get_device_info(device):
if not device or not device.deviceV2:
logger.error("Cannot get device info: Device not initialized.")
return
info = device.deviceV2.info
logger.debug(f"Device Info: {info.get('productName')} | SDK: {info.get('sdkInt')}")
class DeviceFacade:
def __init__(self, device_id, app_id, args):
self.device_id = device_id
self.app_id = app_id
self.args = args
self.deviceV2 = u2.connect(device_id)
# Configure uiautomator2
self.deviceV2.settings["wait_timeout"] = 3.0
self.deviceV2.settings["post_delay"] = 0.5
# System dialog handler (language-agnostic via resource-id, not text)
try:
# u2 v3.x: named watchers with xpath selectors
# android:id/aerr_close = App crash "Close" button (all languages)
self.deviceV2.watcher("crash_dialog").when(
xpath='//*[@resource-id="android:id/aerr_close"]'
).click()
# android:id/button1 = positive system dialog button (all languages)
self.deviceV2.watcher("system_dialog").when(
xpath='//*[@resource-id="android:id/button1"]'
).click()
self.deviceV2.watcher.start()
except Exception as e:
logger.debug(f"Could not start system watcher: {e}")
@adb_retry()
def get_info(self):
return self.deviceV2.info
@adb_retry()
def cm_to_pixels(self, cm: float) -> int:
info = self.deviceV2.info
dpx = info.get("displaySizeDpX", 400)
width = info.get("displayWidth", 1080)
# Android baseline: 1 dp = 1/160 inch. 1 inch = 2.54 cm
# PPCM (Pixels Per CM) = (width / dpx) * (160 / 2.54)
ppcm = (width / dpx) * (160 / 2.54)
return int(cm * ppcm)
@adb_retry()
def wake_up(self):
if not self.deviceV2.info.get("screenOn"):
self.deviceV2.screen_on()
self.deviceV2.press("home")
sleep(1)
@adb_retry()
def press(self, key):
self.deviceV2.press(key)
@adb_retry()
def click(self, x=None, y=None, obj=None):
if obj:
if isinstance(obj, dict) and 'x' in obj and 'y' in obj:
self.human_click(obj['x'], obj['y'])
return
try:
left, top, right, bottom = obj.bounds()
cx = (left + right) // 2
cy = (top + bottom) // 2
from random import uniform
# Randomize hit location within inner 50% of the UI element
w = right - left
h = bottom - top
cx += int(uniform(-w * 0.25, w * 0.25))
cy += int(uniform(-h * 0.25, h * 0.25))
self.human_click(cx, cy)
except Exception as e:
logger.debug(f"Bounds extraction failed, fallback to native click: {e}")
obj.click()
elif x is not None and y is not None:
self.human_click(x, y)
@adb_retry()
def human_click(self, x, y):
from random import uniform
try:
self.deviceV2.touch.down(x, y)
# Human finger rest time (squish)
sleep(uniform(0.05, 0.15))
# Sloppy slip
slip_x = x + int(uniform(-4, 4))
slip_y = y + int(uniform(-4, 4))
self.deviceV2.touch.move(slip_x, slip_y)
sleep(uniform(0.01, 0.05))
self.deviceV2.touch.up(slip_x, slip_y)
except Exception as e:
logger.debug(f"human_click failed, fallback: {e}")
self.deviceV2.click(x, y)
@adb_retry()
def swipe_points(self, x1, y1, x2, y2, duration=0.1):
self.deviceV2.swipe(x1, y1, x2, y2, duration)
@adb_retry()
def human_swipe(self, start_x, start_y, end_x, end_y, duration=0.3):
# Simulate a realistic human swipe by keeping it simple.
# Android's ScrollView calculates fling velocity based on the final few points.
# If we use swipe_points with non-linear distances, it breaks the fling physics and produces stuttering or backwards scrolls.
# We just use native swipe with randomized small x-variance.
self.deviceV2.swipe(start_x, start_y, end_x, end_y, duration)
@adb_retry()
def _get_current_app(self):
return self.deviceV2.app_current().get("package")
@adb_retry()
def find(self, **kwargs):
"""Standard uiautomator2 find."""
return self.deviceV2(**kwargs)
@adb_retry()
def dump_hierarchy(self):
return self.deviceV2.dump_hierarchy()
@adb_retry()
def screenshot(self):
return self.deviceV2.screenshot()
# Telepathic Semantic UI Integration
@adb_retry()
def find_semantic(self, intent_description: str):
from GramAddict.core.telepathic_engine import TelepathicEngine
engine = TelepathicEngine.get_instance()
xml = self.dump_hierarchy()
# Passing self (DeviceFacade) enables the Vision Cortex VLM fallback
node = engine.find_best_node(xml, intent_description, device=self)
return node

View File

@@ -0,0 +1,107 @@
"""
Diagnostic XML Dumper for Project Singularity
==============================================
Automatically captures UI hierarchy snapshots when the bot encounters
problematic states. These dumps serve as future test fixtures for TDD.
Dumps are saved to `debug/xml_dumps/` with timestamped filenames
and a structured reason tag for easy triage.
Retention: Keeps the last 50 dumps per reason category to avoid disk bloat.
"""
import os
import logging
import json
from datetime import datetime
logger = logging.getLogger(__name__)
DUMP_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "debug", "xml_dumps")
MAX_DUMPS_PER_CATEGORY = 50
def dump_ui_state(device, reason: str, extra_context: dict = None):
"""
Capture and save the current UI hierarchy to disk for debugging.
Args:
device: The uiautomator2 device facade.
reason: Short tag for the failure type. Used for filename grouping.
Examples: 'context_lost', 'vlm_hallucination', 'nav_failure',
'stuck_on_post', 'unexpected_screen'
extra_context: Optional dict with additional metadata (intent, expected state, etc.)
"""
try:
os.makedirs(DUMP_DIR, exist_ok=True)
# Capture hierarchy
xml = device.deviceV2.dump_hierarchy()
# Generate filename: reason__2026-04-13_17-41-39.xml
ts = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
safe_reason = reason.replace(" ", "_").replace("/", "_")[:40]
filename = f"{safe_reason}__{ts}.xml"
filepath = os.path.join(DUMP_DIR, filename)
# Write XML
with open(filepath, "w", encoding="utf-8") as f:
f.write(xml)
# Write companion metadata JSON
meta = {
"reason": reason,
"timestamp": ts,
"xml_file": filename,
}
# Capture the session log if available
try:
import shutil
from GramAddict.core.log import get_log_file_config
log_name, log_dir, _, _ = get_log_file_config()
if log_name and log_dir:
active_log = os.path.join(log_dir, log_name)
if os.path.exists(active_log):
log_dest = filepath.replace(".xml", ".log")
shutil.copy2(active_log, log_dest)
meta["log_file"] = os.path.basename(log_dest)
except Exception as e:
logger.debug(f"[Diagnostic] Could not capture session log: {e}")
if extra_context:
meta["context"] = extra_context
meta_path = filepath.replace(".xml", ".meta.json")
with open(meta_path, "w", encoding="utf-8") as f:
json.dump(meta, f, indent=2, ensure_ascii=False)
logger.info(f"📸 [Diagnostic] UI state and session log dumped for '{reason}': {filepath}")
# Rotate old dumps for this category
_rotate_dumps(safe_reason)
return filepath
except Exception as e:
# Dumping must NEVER crash the bot
logger.debug(f"[Diagnostic] Could not dump UI state: {e}")
return None
def _rotate_dumps(category_prefix: str):
"""Keep only the last MAX_DUMPS_PER_CATEGORY dumps per category."""
try:
all_files = sorted([
f for f in os.listdir(DUMP_DIR)
if f.startswith(category_prefix) and f.endswith(".xml")
])
if len(all_files) > MAX_DUMPS_PER_CATEGORY:
files_to_remove = all_files[:len(all_files) - MAX_DUMPS_PER_CATEGORY]
for f in files_to_remove:
xml_path = os.path.join(DUMP_DIR, f)
meta_path = xml_path.replace(".xml", ".meta.json")
os.remove(xml_path)
if os.path.exists(meta_path):
os.remove(meta_path)
except Exception:
pass

View File

@@ -0,0 +1,113 @@
import logging
import random
from colorama import Fore, Style
from GramAddict.core.session_state import SessionState
logger = logging.getLogger(__name__)
def _run_zero_latency_dm_loop(device, zero_engine, nav_graph, configs, session_state, current_target, cognitive_stack):
"""
Executes the autonomous Direct Messaging logic in the Zero-Latency architecture.
Assumes the bot is already at the "MessageInbox" UI state.
"""
logger.info(f"🧠 [DM Engine] Initiating inbox processing in {current_target}...", extra={"color": f"{Style.BRIGHT}{Fore.CYAN}"})
telepathic = cognitive_stack.get("telepathic")
dopamine = cognitive_stack.get("dopamine")
crm = cognitive_stack.get("crm")
from GramAddict.core.bot_flow import sleep, dump_ui_state, _humanized_click
from GramAddict.core.llm_provider import query_llm
from GramAddict.core.stealth_typing import ghost_type
# Initialize session limits if missing
if not hasattr(session_state, 'totalMessages'):
session_state.totalMessages = 0
failed_attempts = 0
while not dopamine.is_app_session_over():
# Limits check
limit_val = session_state.check_limit(SessionState.Limit.PM)
if isinstance(limit_val, tuple) and limit_val[0]:
logger.info("🛑 Messaging limit reached for session.")
return "BOREDOM_CHANGE_FEED"
elif limit_val is True:
return "BOREDOM_CHANGE_FEED"
try:
xml_dump = device.deviceV2.dump_hierarchy()
# Step 1: Find unread conversation threads
unread_threads = telepathic._extract_semantic_nodes(xml_dump, "find unread message threads or unread badges", threshold=0.7)
if unread_threads and not unread_threads[0].get("skip"):
target_node = unread_threads[0]
logger.info(f"📨 Found unread message thread. Opening.")
_humanized_click(device, target_node["x"], target_node["y"])
sleep(2.0)
# Step 2: Read the conversation context
thread_xml = device.deviceV2.dump_hierarchy()
msg_nodes = telepathic._extract_semantic_nodes(thread_xml, "find the last received message text", threshold=0.6)
context_text = "No previous context"
if msg_nodes and not msg_nodes[0].get("skip") and msg_nodes[0].get("text"):
context_text = msg_nodes[0].get("text")
logger.debug(f"Last received message context: {context_text}")
# Verify we aren't at limits before sending
if not getattr(configs.args, "disable_ai_messaging", False):
# Generate response
prompt = f"You are replying to a direct message on Instagram. The last message you received was: '{context_text}'. Keep it short, casual, and friendly. Do not use hashtags."
response_text = query_llm(prompt)
if response_text:
# Find the input field
input_nodes = telepathic._extract_semantic_nodes(thread_xml, "find the message input text field", threshold=0.7)
if input_nodes and not input_nodes[0].get("skip"):
in_node = input_nodes[0]
_humanized_click(device, in_node["x"], in_node["y"])
sleep(1.0)
# Type the message
ghost_type(device, response_text, speed="fast")
sleep(1.0)
# Find Send button
send_xml = device.deviceV2.dump_hierarchy()
send_nodes = telepathic._extract_semantic_nodes(send_xml, "find the send message button", threshold=0.8)
if send_nodes and not send_nodes[0].get("skip"):
s_node = send_nodes[0]
_humanized_click(device, s_node["x"], s_node["y"])
logger.info("✅ [DM Engine] Successfully sent a generated reply.", extra={"color": Fore.GREEN})
session_state.totalMessages += 1
if crm:
crm.log_sent_dm("unknown_target", response_text, "", [])
# Return back to inbox
device.deviceV2.press("back")
sleep(1.0)
dopamine.boredom += random.uniform(5.0, 15.0)
failed_attempts = 0
else:
logger.info("📭 No unread threads found. Inbox clear.")
dopamine.boredom += 50.0 # Inbox clear = massive boredom = change feed
if dopamine.wants_to_change_feed() or dopamine.boredom >= 100:
logger.info("🧠 [DM Engine] Interaction complete. Transitioning back from inbox.")
device.deviceV2.press("back") # Go back from inbox
return "BOREDOM_CHANGE_FEED"
except Exception as e:
logger.error(f"⚠️ [FSD Anomaly Handler] Exception in DM Loop: {e}")
device.deviceV2.press("back")
failed_attempts += 1
if failed_attempts > 2:
return "CONTEXT_LOST"
return "SESSION_OVER"

View File

@@ -0,0 +1,95 @@
import logging
import threading
import time
import os
import queue
from datetime import datetime
from colorama import Fore
# Import existing VLM engine and Qdrant DB for operations
from GramAddict.core.compiler_engine import VLMCompilerEngine
from GramAddict.core.qdrant_memory import HeuristicMemoryDB
logger = logging.getLogger(__name__)
class DojoEngine:
"""
Project Dojo: The Tesla FSD Data Engine.
Handles asynchronous learning from failures (Prediction Errors).
Instead of blocking the bot when an element is not found, the bot
offloads the snapshot to this queue. The DojoEngine recompiles the
heuristic using a heavy VLM model in the background and updates the DB.
"Never make a mistake twice."
"""
_instance = None
@classmethod
def get_instance(cls, device=None):
if cls._instance is None:
if device is None:
raise ValueError("DojoEngine must be initialized with a device first.")
cls._instance = DojoEngine(device)
return cls._instance
def __init__(self, device):
self.learning_queue = queue.Queue()
self.compiler = VLMCompilerEngine(device)
self.db = HeuristicMemoryDB()
self.is_running = False
self.worker_thread = None
def start(self):
if not self.is_running:
self.is_running = True
self.worker_thread = threading.Thread(target=self._process_queue, daemon=True)
self.worker_thread.start()
logger.info("⛩️ [Dojo Data Engine] Background learning pipeline initialized.", extra={"color": f"{Fore.CYAN}"})
def stop(self):
self.is_running = False
if self.worker_thread:
self.worker_thread.join(timeout=2.0)
def submit_snapshot(self, heuristic_name: str, context_xml: str, intent_prompt: str):
"""
Submits a failed UI state to the Dojo for background recompilation.
"""
snapshot = {
"name": heuristic_name,
"xml": context_xml,
"intent": intent_prompt,
"timestamp": datetime.now().isoformat()
}
self.learning_queue.put(snapshot)
logger.info(f"⛩️ [Dojo] Snapshot for '{heuristic_name}' enqueued for shadow-compilation.", extra={"color": f"{Fore.CYAN}"})
def _process_queue(self):
"""
The background worker loop.
"""
while self.is_running:
try:
# Wait for a job
snapshot = self.learning_queue.get(timeout=5.0)
h_name = snapshot['name']
xml = snapshot['xml']
intent = snapshot['intent']
logger.info(f"⛩️ [Dojo] Processing auto-labeling job: {h_name}...", extra={"color": f"{Fore.CYAN}"})
# Heavy compilation
new_rule = self.compiler.generate_heuristic(intent, xml)
if new_rule:
# Overwrite legacy rule in Database (Fleet update)
self.db.cache_heuristic(h_name, new_rule)
logger.info(f"⛩️ [Dojo] SUCCESS! Fleet Memory updated with robust heuristic for '{h_name}'.", extra={"color": f"{Fore.GREEN}"})
else:
logger.warning(f"⛩️ [Dojo] FAILED to compile robust heuristic for '{h_name}'.", extra={"color": f"{Fore.RED}"})
self.learning_queue.task_done()
except queue.Empty:
continue
except Exception as e:
logger.error(f"⛩️ [Dojo] Auto-labeling crashed: {e}")

View File

@@ -0,0 +1,75 @@
import logging
import random
import time
from colorama import Fore
logger = logging.getLogger(__name__)
class DopamineEngine:
"""
Simulation of human neurochemistry.
Manages boredom levels and interest-based interaction pacing.
"""
def __init__(self):
self.boredom = 0.0 # 0.0 to 100.0
self.spike_threshold = 7.0
self.homeostasis_rate = 0.05 # decay per minute
self.last_spike = time.time()
self.session_start = time.time()
self.session_limit_seconds = random.uniform(10 * 60, 35 * 60) # 10-35 mins session
def process_content(self, classification: dict):
"""
classification: {'quality': 'high'|'low', 'type': 'meme'|'aesthetic'|'ad', 'score': 0-10}
"""
score = classification.get("score", 5.0)
quality = classification.get("quality", "medium")
# Calculate spike
spike = score * 1.5 if quality == "high" else score * 0.5
# Update boredom: negative correlation with high quality content
if spike > self.spike_threshold:
self.boredom = max(0.0, self.boredom - (spike * 0.2))
logger.info(f"💉 [Dopamine] Spike detected! Interest high. Boredom decreased to {self.boredom:.1f}%", extra={"color": f"{Fore.YELLOW}"})
else:
self.boredom = min(100.0, self.boredom + 5.0)
logger.info(f"💉 [Dopamine] Low interest content. Boredom increased to {self.boredom:.1f}%", extra={"color": f"{Fore.YELLOW}"})
self.last_spike = time.time()
return self.is_bored()
def is_bored(self):
return self.boredom >= 100.0
def wants_to_doomscroll(self):
# Engage fast swiping if highly bored but not fully exhausted
return 75.0 < self.boredom < 100.0
def wants_to_change_feed(self):
# Spontaneous urge to change context due to extreme boredom spikes
return self.boredom > 85.0 and random.random() < 0.2
def is_app_session_over(self):
# True if we have scrolled too long or hit absolute burnout
return (time.time() - self.session_start) > self.session_limit_seconds or self.boredom >= 100.0
def get_pacing_modifier(self, base_score: float):
"""
Returns a multiplier for sleep durations.
High dopamine (high interest) = longer viewing time.
"""
if base_score > 8:
return random.uniform(2.0, 4.0) # Entranced
if base_score < 3:
return random.uniform(0.1, 0.4) # Fast-swipe
return 1.0
def decay(self):
"""
Called periodically to return to baseline.
"""
now = time.time()
minutes = (now - self.last_spike) / 60.0
self.boredom = min(100.0, self.boredom + (minutes * self.homeostasis_rate))
self.last_spike = now

View File

@@ -0,0 +1,105 @@
import os
import time
import logging
logger = logging.getLogger(__name__)
def capture_all(device):
"""
Automated E2E Dump Capturer Sequence.
Navigates through the Instagram UI and securely saves exact XML representations
to satisfy the `e2e_device_dump_injector` test requirements.
Warning: Requires a logged-in session and active device connection.
"""
logger.info("📸 Initiating E2E Dump Capture Sequence!")
FIX_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "tests", "fixtures")
os.makedirs(FIX_DIR, exist_ok=True)
def _save_dump(filename, description):
logger.info(f"⏳ Waiting for UI to settle for [{description}]...")
time.sleep(3.5) # ensure animations finish
xml_data = device.deviceV2.dump_hierarchy()
path = os.path.join(FIX_DIR, filename)
with open(path, "w", encoding="utf-8") as f:
f.write(xml_data)
logger.info(f"✅ Saved ECHTEN DUMP to {filename}")
print("\n" + "="*50)
print("🤖 MANUAL E2E DUMP CAPTURE SEQUENCE")
print("="*50)
print("Please follow the instructions below to capture the required fixtures.")
print("If an IG update changed the layout, you can navigate there naturally.")
print("="*50 + "\n")
try:
# Pre-condition: Device connected
logger.info("Verifying device connection...")
device.deviceV2.info
# 1. Comment Sheet
input("\n👉 1. COMMENT SHEET:\nOpen Instagram, scroll to any post on the HomeFeed, and open the comment section.\nWhen the comment sheet is fully visible, press ENTER to capture...")
_save_dump("comment_sheet.xml", "Post Comment Sheet")
# 2. Stories Feed
input("\n👉 2. STORIES FEED:\nGo to the HomeFeed and tap any user's story right at the top.\nWhile the story is playing (video/photo is visible), press ENTER to capture...")
_save_dump("stories_feed_dump.xml", "Active Story Playback")
# 3. DM Inbox
input("\n👉 3. DM INBOX:\nGo back to the HomeFeed and tap the message icon in the top right to open your inbox.\nWhen your list of chats is visible, press ENTER to capture...")
_save_dump("dm_inbox_dump.xml", "DM Inbox / Threads List")
# 4. Profile Scraping & Unfollow List
input("\n👉 4. OWN PROFILE:\nGo to your OWN profile by tapping your avatar in the bottom right corner.\nWhen your bio and grid are fully visible, press ENTER to capture...")
_save_dump("scraping_profile_dump.xml", "Own Profile Root (User Info)")
input("\n👉 4.b FOLLOWING LIST:\nFrom your profile, tap your 'Following' (Abonniert) count to open the list of people you follow.\nWhen the list is fully loaded, press ENTER to capture...")
_save_dump("unfollow_list_dump.xml", "Following List Iteration View")
# 5. Search Feed
input("\n👉 5. EXPLORE SEARCH:\nTap the magnifying glass (Explore) tab at the bottom. Then, tap into the top 'Search' bar so your keyboard opens.\nWhen you are in the search state, press ENTER to capture...")
_save_dump("search_feed_dump.xml", "Explore Search Input Focus")
# 6. Reels Feed
input("\n👉 6. REELS FEED:\nTap the Reels (Video) tab at the bottom center. Let a video start playing.\nPress ENTER to capture...")
_save_dump("reels_feed_dump.xml", "Reels Video Feed")
# 7. Notifications
input("\n👉 7. NOTIFICATIONS (ACTIVITY):\nGo to the HomeFeed and tap the Heart icon in the top right to open notifications.\nPress ENTER to capture...")
_save_dump("notifications_dump.xml", "Activity / Notifications tab")
# 8. Explore Grid
input("\n👉 8. EXPLORE GRID:\nTap the magnifying glass (Explore) tab, but do NOT tap the search bar.\nWhen the grid of images/videos is visible, press ENTER to capture...")
_save_dump("explore_feed_dump.xml", "Explore Discovery Grid")
# 9. Other User's Profile
input("\n👉 9. ALIEN PROFILE:\nNavigate to ANY OTHER user's profile (e.g. from your Feed or Search).\nWhen their bio and grid are visible, press ENTER to capture...")
_save_dump("user_profile_dump.xml", "Alien Profile Root")
# 10. Followers List
input("\n👉 10. FOLLOWERS LIST:\nFrom that profile (or your own), tap the 'Followers' (Abonnenten) count.\nWhen the list of followers is visible, press ENTER to capture...")
_save_dump("followers_list_dump.xml", "Followers List Iteration View")
# 11. Carousel Post
input("\n👉 11. CAROUSEL POST:\nScroll your Feed until you see a Carousel (a post with multiple swipable images/videos).\nWhen it is visible, press ENTER to capture...")
_save_dump("carousel_post_dump.xml", "Carousel Post Wrapper")
# 12. Sponsored Post / Ad
input("\n👉 12. SPONSORED AD:\nScroll your Feed or Stories until you see a Sponsored / Gesponsert Post with an action button.\nWhen the Ad is visible, press ENTER to capture...")
_save_dump("home_feed_with_ad.xml", "Sponsored Ad Post")
# 13. Inside DM Chat
input("\n👉 13. DM CHAT THREAD:\nOpen any message thread in your DM inbox.\nWhen the chat messages and text input field are visible, press ENTER to capture...")
_save_dump("dm_thread_dump.xml", "Direct Message Chat Thread")
print("\n" + "="*50)
logger.info("🎉 Capture Sequence Complete! All 13 E2E dumps have been placed into tests/fixtures/")
print("="*50 + "\n")
except KeyboardInterrupt:
print("\n")
logger.info("🛑 Capture Sequence Interrupted by User.")
except Exception as e:
logger.error(f"💥 Capture Sequence crashed: {e}", exc_info=True)

View File

@@ -0,0 +1,99 @@
import logging
import random
from datetime import datetime
from colorama import Fore
from GramAddict.core.qdrant_memory import PersonaMemoryDB
logger = logging.getLogger(__name__)
class GrowthBrain:
"""
Biological Feedback and Persona Management.
Two critical functions:
1. Circadian Rhythm — modulates ALL sleep/dwell times based on time of day
2. Persona Refinement — learns from interaction outcomes and stores insights
"""
def __init__(self, username: str, persona_interests: list[str] = None):
self.username = username
self.persona_memory = PersonaMemoryDB()
self.persona_interests = persona_interests or []
self.last_learning_at = datetime.now()
def get_circadian_pacing(self) -> float:
"""
Adjusts activity levels based on the current local time
to simulate human sleep/wake cycles.
Returns a multiplier (0.1 to 1.0) that should be applied to ALL sleep durations.
Lower = slower (more human-like during off-hours).
"""
hour = datetime.now().hour
# Determine current pacing state
if 2 <= hour <= 5:
pacing = 0.1
state_id = "deep_sleep"
msg = "🧠 [GrowthBrain] Deep sleep mode. Performance 10%."
elif 6 <= hour <= 7:
pacing = 0.4
state_id = "waking_up"
msg = "🧠 [GrowthBrain] Waking up slowly. Performance 40%."
elif 8 <= hour <= 9:
pacing = 0.7
state_id = "morning_warmup"
msg = "🧠 [GrowthBrain] Morning warmup. Performance 70%."
elif hour >= 23 or hour <= 1:
pacing = 0.5
state_id = "evening_winddown"
msg = "🧠 [GrowthBrain] Evening wind-down. Performance 50%."
else:
pacing = 1.0
state_id = "peak_hours"
msg = "🧠 [GrowthBrain] Peak metabolic rate. Performance 100%."
# Log intelligently (only info log on state change)
if not hasattr(self, '_last_pacing_state') or getattr(self, '_last_pacing_state') != state_id:
logger.info(msg, extra={"color": f"{Fore.GREEN}"})
self._last_pacing_state = state_id
else:
logger.debug(msg)
return pacing
def refine_persona(self, interaction_outcomes: list[dict]):
"""
Learns from interaction outcomes to refine persona understanding.
interaction_outcomes: [{'username': str, 'action': 'like'|'comment'|'skip', 'resonance': float}]
Stores high-performing interaction patterns in PersonaMemoryDB.
"""
if not interaction_outcomes:
return
# Find interactions that had high resonance (those are our niche)
high_res = [o for o in interaction_outcomes if o.get("resonance", 0) > 0.7]
low_res = [o for o in interaction_outcomes if o.get("resonance", 0) < 0.3]
if high_res:
insight = f"High-resonance interactions in this session: {len(high_res)} posts matched niche."
self.persona_memory.store_persona_insight("session_learning", insight)
logger.info(
f"🧠 [GrowthBrain] Session learning: {len(high_res)} high-resonance, {len(low_res)} low-resonance posts.",
extra={"color": f"{Fore.GREEN}"}
)
self.last_learning_at = datetime.now()
def get_persona_context(self) -> str:
"""Returns learned persona context for LLM prompts."""
base = ""
if self.persona_interests:
base = f"Core interests: {', '.join(self.persona_interests)}"
learned = self.persona_memory.get_persona_context()
if learned:
return f"{base}\n{learned}" if base else learned
return base

View File

@@ -0,0 +1,286 @@
import re
import os
import json
import requests
import logging
from typing import Optional, List, Dict
try:
from dotenv import load_dotenv
load_dotenv()
except ImportError:
pass
logger = logging.getLogger(__name__)
def extract_json(text: str) -> Optional[str]:
"""
Robustly extracts the first JSON object or array from a string that may contain
natural language prefix/suffix. Also purges <think> blocks and markdown ticks.
"""
if not text:
return None
# 100% Autonomous: Scrub model's internal thinking process
if "<think>" in text:
text = re.sub(r'<think>.*?</think>', '', text, flags=re.DOTALL).strip()
logger.debug("🧠 [LLM] Scoped thinking block detected and purged.")
# Remove markdown code block formats
text = re.sub(r'^```json\s*', '', text, flags=re.MULTILINE)
text = re.sub(r'^```\s*', '', text, flags=re.MULTILINE)
# Look for { ... } or [ ... ]
match = re.search(r'(\{.*\}|\[.*\])', text, re.DOTALL)
if match:
return match.group(0)
return None
_MODEL_PRICING_CACHE = None
def get_model_pricing(model_id: str) -> dict:
global _MODEL_PRICING_CACHE
if _MODEL_PRICING_CACHE is None:
try:
r = requests.get("https://openrouter.ai/api/v1/models", timeout=5)
if r.status_code == 200:
models = r.json().get("data", [])
_MODEL_PRICING_CACHE = {m["id"]: m.get("pricing", {}) for m in models}
else:
_MODEL_PRICING_CACHE = {}
except Exception:
_MODEL_PRICING_CACHE = {}
# Check if exact match exists, if not, try partial matches (e.g., if version suffixes differ)
if _MODEL_PRICING_CACHE and model_id not in _MODEL_PRICING_CACHE:
for k, v in _MODEL_PRICING_CACHE.items():
if model_id in k or k in model_id:
return v
return _MODEL_PRICING_CACHE.get(model_id, {})
def log_openrouter_burn():
"""Fetches and logs the current OpenRouter API key usage (money burned)."""
key = os.environ.get("OPENROUTER_API_KEY")
if not key:
return
try:
r = requests.get("https://openrouter.ai/api/v1/auth/key", headers={"Authorization": f"Bearer {key}"}, timeout=5)
if r.status_code == 200:
data = r.json().get("data", {})
total_spent = data.get("usage", 0.0)
daily_spent = data.get("usage_daily", 0.0)
limit = data.get("limit")
logger.info(f"🔥 [OpenRouter Burn Rate] Daily: ${daily_spent:.4f} | Total: ${total_spent:.4f}" + (f" | Limit: ${limit}" if limit else ""), extra={"color": "\x1b[38;5;208m\x1b[1m"})
except Exception as e:
logger.debug(f"Could not fetch OpenRouter burn rate: {e}")
def query_llm(
url: str,
model: str,
prompt: str,
images_b64: Optional[List[str]] = None,
system: Optional[str] = None,
format_json: bool = False,
timeout: int = 60,
fallback_model: Optional[str] = None,
fallback_url: Optional[str] = None
) -> Optional[dict]:
"""
Unified LLM API Caller with configurable fallback.
"""
openrouter_key = os.environ.get("OPENROUTER_API_KEY")
# URL-based provider detection (not model-name based — works for any model)
is_openai_compat = "/v1/chat/completions" in url or "openrouter.ai" in url.lower() or "openai.com" in url.lower()
# If using a cloud model but a local URL was passed, fix it
if not is_openai_compat and ("openrouter" in model.lower() or "/" in model):
# Model looks like "org/model-name" which is OpenRouter format
is_openai_compat = True
url = "https://openrouter.ai/api/v1/chat/completions"
headers = {"Content-Type": "application/json"}
if is_openai_compat:
if openrouter_key:
headers["Authorization"] = f"Bearer {openrouter_key}"
messages = []
if system:
messages.append({"role": "system", "content": system})
user_content = []
if prompt:
user_content.append({"type": "text", "text": prompt})
if images_b64:
for img in images_b64:
user_content.append({
"type": "image_url",
"image_url": {"url": f"data:image/jpeg;base64,{img}"}
})
messages.append({"role": "user", "content": user_content if len(user_content) > 1 else prompt})
req_data = {
"model": model,
"messages": messages,
"stream": False
}
if format_json:
req_data["response_format"] = {"type": "json_object"}
else:
# Ollama /generate API
req_data = {
"model": model,
"prompt": prompt,
"stream": False
}
if system:
req_data["system"] = system
if images_b64:
req_data["images"] = images_b64
if format_json:
req_data["format"] = "json"
try:
response = requests.post(url, json=req_data, headers=headers, timeout=timeout)
response.raise_for_status()
resp_json = response.json()
# Normalize response payload so callers don't have to distinguish
if is_openai_compat:
# OpenRouter returns choices[0].message.content
content = resp_json.get("choices", [{}])[0].get("message", {}).get("content", "")
usage = resp_json.get("usage", {})
if usage:
cost_str = ""
# Attempt to get precise cost sent by OpenRouter or calculate it manually
if "total_cost" in usage:
cost_str = f" | 💸 Cost: ${usage['total_cost']:.6f}"
else:
pricing = get_model_pricing(model)
if pricing:
try:
p_cost = float(pricing.get("prompt", 0)) * usage.get('prompt_tokens', 0)
c_cost = float(pricing.get("completion", 0)) * usage.get('completion_tokens', 0)
calc_cost = p_cost + c_cost
if calc_cost > 0:
cost_str = f" | 💸 Cost: ${calc_cost:.6f}"
except Exception:
pass
p_tokens = usage.get('prompt_tokens', 0)
c_tokens = usage.get('completion_tokens', 0)
t_tokens = usage.get('total_tokens', 0)
# Make it stand out!
logger.info(f"🪙 [LLM Burn] {model} -> In: {p_tokens} | Out: {c_tokens} | Total: {t_tokens}{cost_str}", extra={"color": "\x1b[38;5;208m\x1b[1m"})
# Validation: if JSON was expected, try to extract it
if format_json:
extracted = extract_json(content)
if not extracted:
raise ValueError(f"OpenRouter returned non-JSON content when JSON was expected: {content[:100]}...")
content = extracted
return {"response": content}
else:
# Ollama returns response
content = resp_json.get("response", "")
if format_json:
extracted = extract_json(content)
if not extracted:
raise ValueError(f"Ollama returned non-JSON content when JSON was expected: {content[:100]}...")
resp_json["response"] = extracted
return resp_json
except Exception as e:
logger.error(f"LLM Provider Error with {model}: {e}")
# Prevent infinite fallback loops
if getattr(query_llm, "_is_fallback", False):
return None
# Decide on fallback model/url
f_model = fallback_model
f_url = fallback_url
# Read fallback config from args if available
if not f_model or not f_url:
from GramAddict.core.config import Config
try:
args = Config().args
f_model = f_model or getattr(args, "ai_fallback_model", None)
f_url = f_url or getattr(args, "ai_fallback_url", None)
except Exception:
pass
# Last resort defaults
if not f_model or not f_url:
if is_openai_compat:
f_model = f_model or "llama3.2:1b"
f_url = f_url or "http://localhost:11434/api/generate"
else:
f_model = f_model or "google/gemini-2.5-flash-lite-preview"
f_url = f_url or "https://openrouter.ai/api/v1/chat/completions"
query_llm._is_fallback = True
try:
logger.warning(f"Primary AI ({model}) failed or returned garbage. Attempting fallback to {f_model}...")
return query_llm(
url=f_url,
model=f_model,
prompt=prompt,
images_b64=images_b64,
system=system,
format_json=format_json,
timeout=timeout
)
finally:
query_llm._is_fallback = False
return None
def query_telepathic_llm(
model: str,
url: str,
system_prompt: str,
user_prompt: str,
temperature: float = 0.0,
use_local_edge: bool = False
) -> str:
"""
Routes UI Telepathic requests purely based on textual interpretation of the screen's XML nodes.
If use_local_edge is manually enabled, routes to localhost:11434.
Otherwise honors the provided URL and model (e.g. OpenRouter).
"""
target_url = url
target_model = model
if use_local_edge:
logger.info("⚡ [Edge Inference] Routing telepathic request to local Ollama host (0ms latency target).")
from GramAddict.core.config import Config
try:
args = Config().args
target_url = getattr(args, "ai_fallback_url", "http://localhost:11434/api/generate")
target_model = getattr(args, "ai_fallback_model", "llama3.2:1b")
except Exception:
target_url = "http://localhost:11434/api/generate"
target_model = "llama3.2:1b"
ans = query_llm(
url=target_url,
model=target_model,
prompt=user_prompt,
images_b64=None,
system=system_prompt,
format_json=True
)
if ans and "response" in ans:
return ans["response"]
return "{}"

152
GramAddict/core/log.py Normal file
View File

@@ -0,0 +1,152 @@
import logging
import os
from logging import LogRecord
from logging.handlers import RotatingFileHandler
from uuid import uuid4
from colorama import Fore, Style
from colorama import init as init_colorama
COLORS = {
"DEBUG": Style.DIM,
"INFO": Fore.WHITE,
"WARNING": Fore.YELLOW,
"ERROR": Fore.RED,
"CRITICAL": Fore.MAGENTA,
}
class ColoredFormatter(logging.Formatter):
def __init__(self, *, fmt, datefmt=None):
logging.Formatter.__init__(self, fmt=fmt, datefmt=datefmt)
def format(self, record):
msg = super().format(record)
levelname = record.levelname
if hasattr(record, "color"):
return f"{record.color}{msg}{Style.RESET_ALL}"
if levelname in COLORS:
return f"{COLORS[levelname]}{msg}{Style.RESET_ALL}"
return msg
class LoggerFilterGramAddictOnly(logging.Filter):
def filter(self, record: LogRecord):
return record.name.startswith("GramAddict")
def create_log_file_handler(filename):
file_handler = RotatingFileHandler(
filename,
mode="a",
backupCount=10,
maxBytes=15 * 1000000,
encoding="utf-8",
)
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(
logging.Formatter(
fmt="%(asctime)s %(levelname)8s | %(message)s (%(filename)s:%(lineno)d)",
datefmt=r"[%m/%d %H:%M:%S]",
)
)
file_handler.addFilter(LoggerFilterGramAddictOnly())
return file_handler
def configure_logger(debug, username):
global g_session_id
global g_log_file_name
global g_logs_dir
global g_file_handler
global g_log_file_updated
console_level = logging.DEBUG if debug else logging.INFO
g_session_id = uuid4()
g_logs_dir = "logs"
if username:
g_log_file_name = f"{username}.log"
g_log_file_updated = True
else:
g_log_file_name = f"{g_session_id}.log"
g_log_file_updated = False
init_colorama()
# Root logger
root_logger = logging.getLogger()
root_logger.setLevel(logging.DEBUG)
# Console logger (limited but colored log)
console_handler = logging.StreamHandler()
console_handler.setLevel(console_level)
console_handler.setFormatter(
ColoredFormatter(
fmt="%(asctime)s %(levelname)8s | %(message)s", datefmt="[%m/%d %H:%M:%S]"
)
)
console_handler.addFilter(LoggerFilterGramAddictOnly())
root_logger.addHandler(console_handler)
# File logger (full raw log)
if not os.path.exists(g_logs_dir):
os.makedirs(g_logs_dir)
g_file_handler = create_log_file_handler(f"{g_logs_dir}/{g_log_file_name}")
root_logger.addHandler(g_file_handler)
init_logger = logging.getLogger(__name__)
init_logger.debug(f"Initial log file: {g_logs_dir}/{g_log_file_name}")
def get_log_file_config():
return g_log_file_name, g_logs_dir, g_file_handler, g_session_id
def is_log_file_updated():
return g_log_file_updated
def update_log_file_name(username: str):
old_log_file_name, logs_dir, file_handler, _ = get_log_file_config()
old_full_filename = f"{logs_dir}/{old_log_file_name}"
current_logger = logging.getLogger(__name__)
if not username:
current_logger.error(f"No username found, using log file {old_full_filename}")
return
named_log_file_name = f"{username}.log"
named_full_filename = f"{logs_dir}/{named_log_file_name}"
rollover = bool(os.path.isfile(named_full_filename))
named_file_handler = create_log_file_handler(named_full_filename)
if rollover:
named_file_handler.doRollover()
# copy existing runtime logs (uidd4.log) to named log file (username.log)
with open(old_full_filename, "r", encoding="utf-8") as unnamed_file, open(
named_full_filename, "a", encoding="utf-8"
) as named_file:
for line in unnamed_file:
named_file.write(line)
root_logger = logging.getLogger()
root_logger.removeHandler(file_handler)
root_logger.addHandler(named_file_handler)
current_logger = logging.getLogger(__name__)
current_logger.debug(f"Updated log file: {named_full_filename}")
try:
os.remove(old_full_filename)
except Exception as e:
current_logger.debug(
f"Failed to remove old file: {old_full_filename}. Exception: {e}"
)
global g_log_file_name
global g_file_handler
global g_log_file_updated
g_log_file_name = named_log_file_name
g_file_handler = named_file_handler
g_log_file_updated = True

View File

@@ -0,0 +1,38 @@
import json
import os
import logging
logger = logging.getLogger(__name__)
class PersistentList(list):
def __init__(self, filename, encoder=None):
super().__init__()
self.filename = filename
self.encoder = encoder
self.load()
def load(self):
path = f"accounts/{self.filename}.json"
if os.path.exists(path):
try:
with open(path, "r") as f:
data = json.load(f)
self.extend(data)
except Exception as e:
logger.error(f"Failed to load persistent list {self.filename}: {e}")
def append(self, item):
super().append(item)
self.persist()
def persist(self, directory=None):
if os.environ.get("PYTEST_CURRENT_TEST"):
return
folder = f"accounts/{directory}" if directory else "accounts"
os.makedirs(folder, exist_ok=True)
path = f"{folder}/{self.filename}.json"
try:
with open(path, "w") as f:
json.dump(list(self), f, cls=self.encoder, indent=4)
except Exception as e:
logger.error(f"Failed to persist {self.filename}: {e}")

View File

@@ -0,0 +1,264 @@
import logging
import json
import os
import uuid
import time
import random
from GramAddict.core.compiler_engine import VLMCompilerEngine
from GramAddict.core.qdrant_memory import NavigationMemoryDB
logger = logging.getLogger(__name__)
class Node:
def __init__(self, name: str):
self.name = name
self.transitions = {} # Action (e.g. "tap_search") -> Node
class QNavGraph:
"""
Project Singularity V7: Topological Navigation Map
Maintains a directed graph of UI states. Instead of hardcoded navigation scripts,
the bot traverses this graph. If a path fails, it invokes the VLMCompilerEngine to repair it.
"""
def __init__(self, device):
self.device = device
self.nodes = {}
self.current_state = "UNKNOWN"
self.nav_memory = NavigationMemoryDB()
self.compiler = VLMCompilerEngine(device)
self._load_graph()
def _load_graph(self):
"""Loads the topological map from Qdrant. Merges with core seeds to guarantee baseline navigation."""
logger.debug("🌐 [NavGraph] Syncing topological map with Qdrant...")
self.nodes = self.nav_memory.get_all_transitions()
core_nodes = {
"HomeFeed": {"transitions": {"tap_explore_tab": "ExploreFeed", "tap_profile_tab": "OwnProfile", "tap_message_icon": "MessageInbox"}},
"ExploreFeed": {"transitions": {"tap_home_tab": "HomeFeed"}},
"OwnProfile": {"transitions": {"tap_home_tab": "HomeFeed", "tap_following_list": "FollowingList"}},
"MessageInbox": {"transitions": {"tap_back": "HomeFeed"}},
"FollowingList": {"transitions": {"tap_back": "OwnProfile"}},
"UNKNOWN": {"transitions": {"tap_home_tab": "HomeFeed"}}
}
# Merge core nodes into loaded nodes
for node, data in core_nodes.items():
if node not in self.nodes:
self.nodes[node] = {"transitions": {}}
for action, target in data["transitions"].items():
if action not in self.nodes[node]["transitions"]:
self.nodes[node]["transitions"][action] = target
self.nav_memory.store_transition(node, action, target)
def _save_graph(self):
"""Deprecated: Navigation state is now persisted per-transition in Qdrant."""
pass
def navigate_to(self, target_state: str, zero_engine, recovery_attempts: int = 0):
"""
Attempts to navigate from current_state to target_state using the Graph.
"""
logger.info(f"📍 Navigating autonomously to: {target_state}")
if recovery_attempts > 2:
logger.error(f"FATAL: Context recovery failed after {recovery_attempts} attempts. Bailing out of navigation loop.")
return False
# Stories are viewed from the HomeFeed natively. There is no separate StoriesFeed node.
# We navigate to HomeFeed dynamically, and let bot_flow handle the interaction.
logical_target = "HomeFeed" if target_state == "StoriesFeed" else target_state
# Simple BFS to find sequence of actions
path = self._find_path(self.current_state, logical_target)
if path is None:
logger.warning(f"No known path from {self.current_state} to {target_state}. Attempting semantic recovery via Global Navigation Bar...")
# The global bottom navigation often gives us direct access from most positions
# Map target_state to its global tab action
target_to_action = {
"ExploreFeed": "tap_explore_tab",
"HomeFeed": "tap_home_tab",
"OwnProfile": "tap_profile_tab",
"ReelsFeed": "tap_reels_tab",
"StoriesFeed": "tap_home_tab",
}
direct_action = target_to_action.get(target_state, "tap_home_tab")
target_anchor = target_state if direct_action != "tap_home_tab" else "HomeFeed"
success = self._execute_transition(direct_action, zero_engine)
if success is True:
logger.info(f"Successfully anchored! Learned new global edge: {self.current_state} -> {target_anchor} via {direct_action}")
if self.current_state not in self.nodes:
self.nodes[self.current_state] = {"transitions": {}}
self.nodes[self.current_state]["transitions"][direct_action] = target_anchor
self.nav_memory.store_transition(self.current_state, direct_action, target_anchor)
self.current_state = target_anchor
path = self._find_path(self.current_state, logical_target)
elif success == "CONTEXT_LOST":
logger.warning(f"⚠️ Context was lost during direct action '{direct_action}'. Forcing app focus and resetting path.")
self.device.deviceV2.app_start(self.device.app_id, use_monkey=True)
time.sleep(3)
self.current_state = "HomeFeed"
return self.navigate_to(target_state, zero_engine, recovery_attempts=recovery_attempts + 1)
else:
path = None
if path is None:
# Absolute last resort fallback: force app to main activity
logger.warning("Semantic recovery failed. Forcing main activity intent...")
self.device.deviceV2.app_start(self.device.app_id)
time.sleep(3)
self.current_state = "HomeFeed"
path = self._find_path(self.current_state, logical_target)
if path is None:
logger.error(f"FATAL: Cannot find any path to {target_state} even after forcing main activity.")
return False
for action in path:
result = self._execute_transition(action, zero_engine)
if result == "CONTEXT_LOST":
logger.warning(f"⚠️ Context was lost during '{action}'. Forcing app focus and resetting path.")
self.device.deviceV2.app_start(self.device.app_id, use_monkey=True)
time.sleep(3)
# After app start, we are at HomeFeed (usually)
self.current_state = "HomeFeed"
# Recursively call navigate_to from the new anchor
return self.navigate_to(target_state, zero_engine, recovery_attempts=recovery_attempts + 1)
if not result:
logger.error(f"Nav transition '{action}' failed! Initiating self-repair...")
self._repair_transition(action)
# Retry after repair
success = self._execute_transition(action, zero_engine)
if not success or success == "CONTEXT_LOST":
logger.error(f"FATAL: Auto-repair failed for transition: {action}")
return False
self.current_state = logical_target
return True
def _find_path(self, start: str, end: str):
if start == end: return []
if start not in self.nodes: return None
queue = [(start, [])]
visited = set()
while queue:
current, path = queue.pop(0)
if current == end:
return path
visited.add(current)
transitions = self.nodes.get(current, {}).get("transitions", {})
for action, next_state in transitions.items():
if next_state not in visited:
queue.append((next_state, path + [action]))
return None
def _execute_transition(self, action: str, zero_engine) -> bool:
"""
Executes a transition (e.g. 'tap_explore_tab') using the Telepathic Semantic Engine.
"""
from GramAddict.core.telepathic_engine import TelepathicEngine
engine = TelepathicEngine.get_instance()
context_xml = self.device.deviceV2.dump_hierarchy()
# ── Z-Depth Guard / Obstacle Clearance ──
import re
if re.search(r'bottom_sheet_container|dialog_container|dialog_root|bottom_sheet_drag', str(context_xml)):
logger.warning("🛡️ [Z-Depth Guard] Obstacle overlay detected during navigation. Pressing BACK to clear...")
self.device.deviceV2.press("back")
time.sleep(1.5)
# Re-acquire context after clearing obstacle
context_xml = self.device.deviceV2.dump_hierarchy()
# We phrase the action as an intent for the semantic engine
# e.g. "tap_explore_tab" -> "tap explore tab"
# We add some common synonyms for Instagram to help the vector engine
intent_map = {
# Navigation (Bottom Bar) — aligned with fast-path keys
"tap_home_tab": "tap home tab",
"tap_explore_tab": "tap explore tab",
"tap_profile_tab": "tap profile tab",
"tap_reels_tab": "tap reels tab",
"tap_create_tab": "tap create post tab",
# Post Interaction — aligned with fast-path keys
"tap_like_button": "tap like button",
"tap_comment_button": "tap comment button",
"tap_post_username": "tap post username",
"tap_share_button": "tap share button",
"tap_save_button": "tap save button",
# Grid & Profile
"tap_explore_grid_item": "first image in explore grid",
"tap_story_tray_item": "profile picture avatar story ring",
"tap_follow_button": "tap follow button on profile",
"tap_grid_first_post": "first image post in profile grid",
}
intent_description = intent_map.get(action, action.replace("_", " "))
# Use TelepathicEngine to find the most likely node for this intent
# If vector score < 0.82, it will trigger the Vision Cortex Fallback (VLM)
best_node = engine.find_best_node(context_xml, intent_description, min_confidence=0.82, device=self.device)
if not best_node:
logger.debug(f"_execute_transition: TelepathicEngine found no matching node for '{action}'")
# Check if we are even in the right app
current_app = self.device._get_current_app()
if current_app != self.device.app_id:
logger.warning(f"⚠️ [Context Lost] Currently in '{current_app}', expected '{self.device.app_id}'. Transition '{action}' aborted.")
return "CONTEXT_LOST"
return False
if best_node.get("skip"):
logger.info(f"⏭️ Skipping physical tap for '{action}' (Semantic Fast-Path indicated state already fulfilled)")
return True
source_tag = best_node.get("source", "telepathic").replace("_", " ").title()
logger.info(f"QNavGraph executing transition '{action}' via [{source_tag}] (Score: {best_node.get('score', 1.0):.3f})")
# Execute click
self.device.click(obj=best_node)
time.sleep(random.uniform(1.2, 2.5))
# ── Post-Click Verification: Did the screen change? ──
post_click_xml = self.device.deviceV2.dump_hierarchy()
# For navigation, we expect the UI to change or specific markers to appear
# Comparison of XML strings is a good baseline for navigation success
if post_click_xml != context_xml:
engine.confirm_click(intent_description)
return True
else:
logger.warning(f"⚠️ [Nav] Click on '{action}' did not change UI. Learning from failure.")
engine.reject_click(intent_description)
return False
def _repair_transition(self, action: str):
"""
If a transition fails, the CompilerEngine is invoked to figure out the new UI layout
and write a new rule for `action`.
"""
from GramAddict.core.dojo_engine import DojoEngine
dojo = DojoEngine.get_instance(self.device)
logger.warning(f"⛩️ [Dojo] Enqueuing auto-labeling job for missing '{action}'.", extra={"color": f"\x1b[36m"})
context_xml = self.device.deviceV2.dump_hierarchy()
dojo.submit_snapshot(
heuristic_name=action,
context_xml=context_xml,
intent_prompt=f"Find the button that performs: {action}. Be extremely robust against structural UI changes."
)

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,260 @@
import logging
import math
from typing import Optional
from colorama import Fore
from GramAddict.core.qdrant_memory import ContentMemoryDB, PersonaMemoryDB, ParasocialCRMDB, CommentMemoryDB
from GramAddict.core.llm_provider import query_llm
logger = logging.getLogger(__name__)
class ResonanceEngine:
"""
The Aesthetic Oracle — Real AI Content Evaluation.
Calculates semantic alignment (Resonance Score) between the bot's
configured persona interests and target content using vector embeddings.
This drives ALL downstream decisions:
- Like probability (score >= 0.35)
- Comment probability (score >= 0.8)
- Rabbit Hole / profile visit (score >= 0.9)
- Dopamine spike intensity
- Darwin dwell time modulation
"""
def __init__(self, my_username: str, persona_interests: list[str] = None, crm: ParasocialCRMDB = None):
self.my_username = my_username
self.content_memory = ContentMemoryDB()
self.persona_memory = PersonaMemoryDB()
self.crm = crm
self.threshold = 0.5
# The persona vector is the mathematical identity of what content we care about.
# It's generated from config's persona_interests and cached for the entire session.
self._persona_vector: Optional[list] = None
self._persona_interests = persona_interests or []
# Bootstrap persona on init
if self._persona_interests:
self._bootstrap_persona()
def _bootstrap_persona(self):
"""
Generates and caches the persona embedding from configured interests.
Called once on init. The persona vector is what every post is compared against.
"""
persona_text = f"Content about: {', '.join(self._persona_interests)}"
self._persona_vector = self.content_memory._get_embedding(persona_text)
if self._persona_vector:
# Store in PersonaMemoryDB for persistence across sessions
self.persona_memory.store_persona_insight(
"interests",
f"Core niche interests: {', '.join(self._persona_interests)}"
)
logger.info(
f"✨ [Resonance Oracle] Persona vector initialized from config: {self._persona_interests}",
extra={"color": f"{Fore.MAGENTA}"}
)
else:
logger.warning("✨ [Resonance Oracle] Could not generate persona embedding. Falling back to neutral scoring.")
def _cosine_similarity(self, v1: list, v2: list) -> float:
"""Pure python cosine similarity — no numpy dependency."""
if not v1 or not v2 or len(v1) != len(v2):
return 0.0
dot = sum(a * b for a, b in zip(v1, v2))
mag1 = math.sqrt(sum(a * a for a in v1))
mag2 = math.sqrt(sum(b * b for b in v2))
if mag1 == 0 or mag2 == 0:
return 0.0
return dot / (mag1 * mag2)
def calculate_resonance(self, post_content: dict) -> float:
"""
Real AI resonance score based on embedding cosine similarity.
"""
username = post_content.get("username", "Unknown")
description = post_content.get("description", "")
logger.info(f"✨ [Resonance Oracle] Evaluating content from @{username}...", extra={"color": f"{Fore.MAGENTA}"})
# Build a rich text representation of the post
description = post_content.get("description", "")
caption = post_content.get("caption", "")
username = post_content.get("username", "")
content_text = " ".join(filter(None, [description, caption])).strip()
if not content_text or len(content_text) < 5:
logger.debug("✨ [Resonance] Post has no extractable content. Neutral score.")
return 0.5 # Neutral — can't evaluate what we can't see
# 1. Check ContentMemoryDB cache — have we seen nearly identical content?
cached = self.content_memory.get_cached_evaluation(content_text)
if cached:
score = self._classification_to_score(cached.get("classification", "medium"))
logger.info(
f"✨ [Resonance Cache Hit] '{content_text[:40]}...'{score*100:.1f}%",
extra={"color": f"{Fore.MAGENTA}"}
)
return score
# 2. No persona vector? Can't do real evaluation.
if not self._persona_vector:
logger.debug("✨ [Resonance] No persona vector. Configure persona_interests in config.yml.")
return 0.5
# 3. Generate embedding of the post content
post_vector = self.content_memory._get_embedding(content_text)
if not post_vector:
return 0.5
# 4. Cosine similarity against persona = resonance score
raw_score = self._cosine_similarity(post_vector, self._persona_vector)
# Normalize: text-embedding-3-small cosine similarity for text embeddings typically ranges 0.15 (completely distinct) to 0.55 (very matched, but not literal identical copies)
# Map this to a more useful 0.0-1.0 range
score = max(0.0, min(1.0, (raw_score - 0.15) / 0.30))
# 5. Store evaluation in ContentMemoryDB for future cache hits
classification = "high" if score > 0.7 else "medium" if score > 0.4 else "low"
self.content_memory.store_evaluation(
content_text[:500], # Cap length for storage
classification,
f"Resonance: {score:.3f} (raw cosine: {raw_score:.3f})"
)
# 6. Feed the Parasocial CRM
if self.crm and username:
intent = f"aesthetic_evaluation_{classification}"
# Stage mapping: high resonance -> stage 1 (Curiosity)
new_stage = 1 if classification == "high" else None
self.crm.log_interaction(username, intent, new_stage=new_stage)
logger.info(
f"✨ [Resonance Oracle] '{content_text[:50]}...'{score*100:.1f}% ({classification})",
extra={"color": f"{Fore.MAGENTA}"}
)
return score
def _classification_to_score(self, classification: str) -> float:
"""Converts stored classification back to a usable score."""
return {"high": 0.85, "medium": 0.55, "low": 0.2}.get(classification, 0.5)
def judge_interaction(self, score: float) -> bool:
"""Determines whether the resonance is high enough to warrant interaction."""
if score >= self.threshold:
logger.info("✨ [Resonance] POSITIVE ALIGNMENT. Interaction authorized.", extra={"color": f"{Fore.MAGENTA}"})
return True
else:
logger.info("✨ [Resonance] NEGATIVE ALIGNMENT. Skipping profile.", extra={"color": f"{Fore.MAGENTA}"})
return False
def extract_and_learn_comments(self, xml_hierarchy: str, configs, author: str = "unknown"):
"""
Phase 10: RAG Comment Learning Implementation
Extracts comments from the UI hierarchy, filters them using the assigned VLM against
configured blacklists and vibes, and stores them in Qdrant CommentMemoryDB.
"""
if not configs or not getattr(configs.args, "ai_learn_comments", False):
return
vibe = getattr(configs.args, "ai_vibe", "")
blacklist = getattr(configs.args, "ai_blacklist_topics", "")
if not vibe:
return # No vibe to learn
logger.info(f"🧠 [Comment Learning] Extracting comments matching vibe: '{vibe}'...", extra={"color": f"{Fore.CYAN}"})
# 1. Very basic semantic extraction (grab text nodes that look like comments)
raw_comments = []
try:
import xml.etree.ElementTree as ET
root = ET.fromstring(xml_hierarchy)
for node in root.iter('node'):
text = node.get("text", "")
content_desc = node.get("content-desc", "")
val = text if text else content_desc
if val and len(val) > 15:
if val.lower() not in ["reply", "like", "view replies", "see translation", "hide replies"]:
raw_comments.append(val)
except Exception as e:
logger.error(f"🧠 [Comment Learning] Failed to parse XML: {e}")
return
if not raw_comments:
logger.debug("🧠 [Comment Learning] No legible comments found in UI.")
return
# Deduplicate and limit
raw_comments = list(set(raw_comments))[:10]
logger.debug(f"🧠 [Comment Learning] Scraped {len(raw_comments)} potential comment nodes. Passing to Condenser...")
logger.debug(f"🧠 [Comment Learning] Raw texts passed to Condenser:\n{chr(10).join(raw_comments)}")
# 2. Filter via VLM Condenser
prompt = (
f"Filter these Instagram comments. Keep ONLY real comments that generally match this vibe: '{vibe}'.\n"
f"Remove comments about: {blacklist}\n"
f"Remove UI junk text (buttons, labels, timestamps).\n\n"
f"Comments:\n{chr(10).join(raw_comments)}\n\n"
"Output a JSON array of matching comment strings. If none match, output []."
)
model = getattr(configs.args, "ai_condenser_model", "google/gemini-2.5-flash-lite-preview")
url = getattr(configs.args, "ai_condenser_url", "https://openrouter.ai/api/v1/chat/completions")
try:
import json
# Fix: kwargs match query_llm signature EXACTLY to evade TypeError
response_dict = query_llm(url=url, model=model, prompt=prompt, format_json=True)
if not response_dict or "response" not in response_dict:
return
response_text = response_dict["response"]
# Parse json gracefully
if type(response_text) is str:
clean_json = response_text.strip()
if clean_json.startswith("```json"):
clean_json = clean_json[7:]
if clean_json.endswith("```"):
clean_json = clean_json[:-3]
try:
learned_comments = json.loads(clean_json.strip())
except json.JSONDecodeError:
logger.error("🧠 [Comment Learning] LLM returned invalid JSON.")
return
else:
# In case expect_json already returned a parsed list somehow, though extract_json returns str
learned_comments = response_text
if not isinstance(learned_comments, list):
logger.error("🧠 [Comment Learning] Condenser failed to return a JSON list.")
return
if not learned_comments:
logger.info("🧠 [Comment Learning] Condenser rejected all scraped comments (did not align with vibe or hit blacklist).", extra={"color": f"{Fore.YELLOW}"})
return
logger.info(f"🧠 [Comment Learning] Condenser approved {len(learned_comments)} comments. Persisting to Qdrant...", extra={"color": f"{Fore.GREEN}"})
# 3. Store the passing comments into Qdrant
comment_db = CommentMemoryDB()
stored = 0
for c in learned_comments:
if isinstance(c, str) and len(c) > 5:
logger.debug(f" 👉 Storing: '{c}'")
comment_db.store_comment(text=c, vibe=vibe, author=author)
stored += 1
if stored > 0:
logger.info(f"✅ [Comment Vector Sync] Successfully embedded {stored} high-vibe comments into memory.", extra={"color": f"{Fore.GREEN}"})
except Exception as e:
logger.error(f"🧠 [Comment Learning] Condenser failed: {e}")

View File

@@ -0,0 +1,93 @@
import logging
import re
import xml.etree.ElementTree as ET
logger = logging.getLogger(__name__)
class HoneypotRadome:
"""
Project Dojo: The Anti-Test Sensor.
Filters the Android XML Hierarchy to remove "invisible traps" and honeypots
that Instagram uses to detect deterministic bots (e.g., 1x1 pixel buttons,
off-screen elements with clickable=True).
"""
def __init__(self, display_width=1080, display_height=2400):
self.display_width = display_width
self.display_height = display_height
self.bounds_pattern = re.compile(r'\[(\d+),(\d+)\]\[(\d+),(\d+)\]')
def sanitize_xml(self, xml_string: str) -> str:
"""
Parses raw UI Automator XML and strips out fake or impossible nodes.
Returns the sanitized XML string.
"""
try:
# Android XML dumps often have multiple root nodes or formatting issues,
# let's try reading it safely.
# Handle potential encoding issues from dump_hierarchy
clean_xml = xml_string.replace('&#10;', '').replace('&#13;', '')
root = ET.fromstring(clean_xml)
removed_count = self._filter_node(root)
if removed_count > 0:
logger.info(f"🛡️ [Honeypot Radome] Stripped {removed_count} phantom nodes from view.", extra={"color": "\x1b[33m"})
# Convert back to string
return ET.tostring(root, encoding='unicode')
except Exception as e:
logger.warning(f"🛡️ [Honeypot Radome] XML Parse failed, returning raw. Err: {e}")
return xml_string
def _filter_node(self, node: ET.Element) -> int:
removed = 0
children_to_remove = []
for child in node:
if self._is_honeypot(child):
children_to_remove.append(child)
removed += 1
else:
removed += self._filter_node(child)
for child in children_to_remove:
node.remove(child)
return removed
def _is_honeypot(self, node: ET.Element) -> bool:
"""
Detects if a node is physically impossible to be clicked by a human.
"""
bounds = node.get("bounds")
if not bounds:
return False
match = self.bounds_pattern.match(bounds)
if not match:
return False
x1, y1, x2, y2 = map(int, match.groups())
width = x2 - x1
height = y2 - y1
is_clickable = node.get("clickable", "false").lower() == "true"
# Rule 1: The Zero-Point Trap (Element is exactly on 0,0 with no dimensions)
if x1 == 0 and y1 == 0 and x2 == 0 and y2 == 0:
return True
# Rule 2: The Micro-Pixel Trap (Bot detectors often use 1x1 or 2x2 clickable overlay pixels)
if is_clickable and width <= 2 and height <= 2:
return True
# Rule 3: The Off-Screen Trap (Buttons rendered wildly out of bounds to bait mindless loops)
if x1 >= self.display_width or y1 >= self.display_height:
return True
# Rule 4: The Negative Coordinate Trap
if x2 <= 0 or y2 <= 0:
return True
return False

View File

@@ -0,0 +1,321 @@
import logging
import uuid
from datetime import datetime, timedelta
from enum import Enum, auto
from json import JSONEncoder
from GramAddict.core.utils import get_value
logger = logging.getLogger(__name__)
class SessionState:
id = None
args = {}
my_username = None
my_posts_count = None
my_followers_count = None
my_following_count = None
totalInteractions = {}
successfulInteractions = {}
totalFollowed = {}
totalLikes = 0
totalComments = 0
totalPm = 0
totalWatched = 0
totalUnfollowed = 0
removedMassFollowers = []
totalScraped = 0
totalCrashes = 0
startTime = None
finishTime = None
def __init__(self, configs):
self.id = str(uuid.uuid4())
self.args = configs.args
self.my_username = None
self.my_posts_count = None
self.my_followers_count = None
self.my_following_count = None
self.totalInteractions = {}
self.successfulInteractions = {}
self.totalFollowed = {}
self.totalLikes = 0
self.totalComments = 0
self.totalPm = 0
self.totalWatched = 0
self.totalUnfollowed = 0
self.removedMassFollowers = []
self.totalScraped = {}
self.totalCrashes = 0
self.startTime = datetime.now()
self.finishTime = None
def add_interaction(self, source, succeed, followed, scraped):
if self.totalInteractions.get(source) is None:
self.totalInteractions[source] = 1
else:
self.totalInteractions[source] += 1
if self.successfulInteractions.get(source) is None:
self.successfulInteractions[source] = 1 if succeed else 0
else:
if succeed:
self.successfulInteractions[source] += 1
if self.totalFollowed.get(source) is None:
self.totalFollowed[source] = 1 if followed else 0
else:
if followed:
self.totalFollowed[source] += 1
if self.totalScraped.get(source) is None:
self.totalScraped[source] = 1 if scraped else 0
self.successfulInteractions[source] = 1 if scraped else 0
else:
if scraped:
self.totalScraped[source] += 1
self.successfulInteractions[source] += 1
def set_limits_session(
self,
):
"""set the limits for current session"""
self.args.current_likes_limit = get_value(
getattr(self.args, "total_likes_limit", 300), None, 300
)
self.args.current_follow_limit = get_value(
getattr(self.args, "total_follows_limit", 50), None, 50
)
self.args.current_unfollow_limit = get_value(
getattr(self.args, "total_unfollows_limit", 50), None, 50
)
self.args.current_comments_limit = get_value(
getattr(self.args, "total_comments_limit", 10), None, 10
)
self.args.current_pm_limit = get_value(getattr(self.args, "total_pm_limit", 10), None, 10)
self.args.current_watch_limit = get_value(
getattr(self.args, "total_watches_limit", 50), None, 50
)
self.args.current_success_limit = get_value(
getattr(self.args, "total_successful_interactions_limit", 100), None, 100
)
self.args.current_total_limit = get_value(
getattr(self.args, "total_interactions_limit", 1000), None, 1000
)
self.args.current_scraped_limit = get_value(
getattr(self.args, "total_scraped_limit", 200), None, 200
)
self.args.current_crashes_limit = get_value(
getattr(self.args, "total_crashes_limit", 5), None, 5
)
def check_limit(self, limit_type=None, output=False):
"""Returns True if limit reached - else False"""
limit_type = SessionState.Limit.ALL if limit_type is None else limit_type
# check limits
total_likes = self.totalLikes >= int(self.args.current_likes_limit)
total_followed = sum(self.totalFollowed.values()) >= int(
self.args.current_follow_limit
)
total_unfollowed = self.totalUnfollowed >= int(self.args.current_unfollow_limit)
total_comments = self.totalComments >= int(self.args.current_comments_limit)
total_pm = self.totalPm >= int(self.args.current_pm_limit)
total_watched = self.totalWatched >= int(self.args.current_watch_limit)
total_successful = sum(self.successfulInteractions.values()) >= int(
self.args.current_success_limit
)
total_interactions = sum(self.totalInteractions.values()) >= int(
self.args.current_total_limit
)
total_scraped = sum(self.totalScraped.values()) >= int(
self.args.current_scraped_limit
)
total_crashes = self.totalCrashes >= int(self.args.current_crashes_limit)
session_info = [
"Checking session limits:",
f"- Total Likes:\t\t\t\t{'Limit Reached' if total_likes else 'OK'} ({self.totalLikes}/{self.args.current_likes_limit})",
f"- Total Comments:\t\t\t\t{'Limit Reached' if total_comments else 'OK'} ({self.totalComments}/{self.args.current_comments_limit})",
f"- Total PM:\t\t\t\t\t{'Limit Reached' if total_pm else 'OK'} ({self.totalPm}/{self.args.current_pm_limit})",
f"- Total Followed:\t\t\t\t{'Limit Reached' if total_followed else 'OK'} ({sum(self.totalFollowed.values())}/{self.args.current_follow_limit})",
f"- Total Unfollowed:\t\t\t\t{'Limit Reached' if total_unfollowed else 'OK'} ({self.totalUnfollowed}/{self.args.current_unfollow_limit})",
f"- Total Watched:\t\t\t\t{'Limit Reached' if total_watched else 'OK'} ({self.totalWatched}/{self.args.current_watch_limit})",
f"- Total Successful Interactions:\t\t{'Limit Reached' if total_successful else 'OK'} ({sum(self.successfulInteractions.values())}/{self.args.current_success_limit})",
f"- Total Interactions:\t\t\t{'Limit Reached' if total_interactions else 'OK'} ({sum(self.totalInteractions.values())}/{self.args.current_total_limit})",
f"- Total Crashes:\t\t\t\t{'Limit Reached' if total_crashes else 'OK'} ({self.totalCrashes}/{self.args.current_crashes_limit})",
f"- Total Successful Scraped Users:\t\t{'Limit Reached' if total_scraped else 'OK'} ({sum(self.totalScraped.values())}/{self.args.current_scraped_limit})",
]
if limit_type == SessionState.Limit.ALL:
if output:
for line in session_info:
logger.info(line)
return (
total_likes and getattr(self.args, "end_if_likes_limit_reached", False)
or total_followed and getattr(self.args, "end_if_follows_limit_reached", False)
or total_watched and getattr(self.args, "end_if_watches_limit_reached", False)
or total_comments and getattr(self.args, "end_if_comments_limit_reached", False)
or total_pm and getattr(self.args, "end_if_pm_limit_reached", False),
total_unfollowed,
total_interactions or total_successful or total_scraped,
)
elif limit_type == SessionState.Limit.LIKES:
if output:
logger.info(session_info[1])
else:
logger.debug(session_info[1])
return total_likes
elif limit_type == SessionState.Limit.COMMENTS:
if output:
logger.info(session_info[2])
else:
logger.debug(session_info[2])
return total_comments
elif limit_type == SessionState.Limit.PM:
if output:
logger.info(session_info[3])
else:
logger.debug(session_info[3])
return total_pm
elif limit_type == SessionState.Limit.FOLLOWS:
if output:
logger.info(session_info[4])
else:
logger.debug(session_info[4])
return total_followed
elif limit_type == SessionState.Limit.UNFOLLOWS:
if output:
logger.info(session_info[5])
else:
logger.debug(session_info[5])
return total_unfollowed
elif limit_type == SessionState.Limit.WATCHES:
if output:
logger.info(session_info[6])
else:
logger.debug(session_info[6])
return total_watched
elif limit_type == SessionState.Limit.SUCCESS:
if output:
logger.info(session_info[7])
else:
logger.debug(session_info[7])
return total_successful
elif limit_type == SessionState.Limit.TOTAL:
if output:
logger.info(session_info[8])
else:
logger.debug(session_info[8])
return total_interactions
elif limit_type == SessionState.Limit.CRASHES:
if output:
logger.info(session_info[9])
else:
logger.debug(session_info[9])
return total_crashes
elif limit_type == SessionState.Limit.SCRAPED:
if output:
logger.info(session_info[10])
else:
logger.debug(session_info[10])
return total_scraped
@staticmethod
def inside_working_hours(working_hours, delta_sec):
def time_in_range(start, end, x):
if start <= end:
return start <= x <= end
else:
return start <= x or x <= end
in_range = False
time_left_list = []
current_time = datetime.now()
delta = timedelta(seconds=delta_sec)
if not working_hours:
return True, 0
for n in working_hours:
today = current_time.strftime("%Y-%m-%d")
# 100% Autonomous: Hybrid Time Format Support (Legacy . vs Modern :)
h_start = n.split('-')[0].replace(":", ".")
h_end = n.split('-')[1].replace(":", ".")
inf_value = f"{h_start} {today}"
inf = datetime.strptime(inf_value, "%H.%M %Y-%m-%d") + delta
sup_value = f"{h_end} {today}"
sup = datetime.strptime(sup_value, "%H.%M %Y-%m-%d") + delta
if sup - inf + timedelta(minutes=1) == timedelta(
days=1
) or sup - inf + timedelta(minutes=1) == timedelta(days=0):
logger.debug("Whole day mode.")
return True, 0
if time_in_range(inf.time(), sup.time(), current_time.time()):
in_range = True
return in_range, 0
else:
time_left = inf - current_time
if time_left >= timedelta(0):
time_left_list.append(time_left)
else:
time_left_list.append(time_left + timedelta(days=1))
return (
in_range,
min(time_left_list) if len(time_left_list) > 1 else time_left_list[0],
)
def is_finished(self):
return self.finishTime is not None
class Limit(Enum):
ALL = auto()
LIKES = auto()
COMMENTS = auto()
PM = auto()
FOLLOWS = auto()
UNFOLLOWS = auto()
WATCHES = auto()
SUCCESS = auto()
TOTAL = auto()
SCRAPED = auto()
CRASHES = auto()
class SessionStateEncoder(JSONEncoder):
def default(self, session_state: SessionState):
return {
"id": session_state.id,
"total_interactions": sum(session_state.totalInteractions.values()),
"successful_interactions": sum(
session_state.successfulInteractions.values()
),
"total_followed": sum(session_state.totalFollowed.values()),
"total_likes": session_state.totalLikes,
"total_comments": session_state.totalComments,
"total_pm": session_state.totalPm,
"total_watched": session_state.totalWatched,
"total_unfollowed": session_state.totalUnfollowed,
"total_scraped": session_state.totalScraped,
"start_time": str(session_state.startTime),
"finish_time": str(session_state.finishTime),
"args": session_state.args.__dict__,
"profile": {
"posts": session_state.my_posts_count,
"followers": session_state.my_followers_count,
"following": session_state.my_following_count,
},
}

View File

@@ -0,0 +1,72 @@
import logging
import random
from time import sleep
logger = logging.getLogger(__name__)
def ghost_type(device, text: str):
"""
Tesla Stealth Ghost Keyboard.
Bypasses UIAutomator virtual IME completely and sends raw Native InputEvents.
Features: Variable typing speed, burst chunking, and synthetic human mistakes.
"""
if not text:
return
logger.info(f"⌨️ [Ghost Keyboard] Initiating stealth injection ({len(text)} chars)...")
# We slice text into variable-sized human bursts
chunks = []
i = 0
while i < len(text):
if random.random() < 0.15:
chunk_size = 1 # single letter hunting
else:
chunk_size = random.randint(2, 6) # fluid typing bursts
chunks.append(text[i:i+chunk_size])
i += chunk_size
for idx, chunk in enumerate(chunks):
# 5% chance of a typo if it's an alphabetical chunk
if random.random() < 0.05 and len(chunk) >= 2 and chunk[-1].isalpha():
typo_letter = random.choice('abcdefghijklmnopqrstuvwxyz')
# Add typo instead of actual last letter
typo_chunk = chunk[:-1] + typo_letter
_adb_inject_text(device, typo_chunk)
# Realize mistake
sleep(random.uniform(0.2, 0.45))
# Send Backspace (KEYCODE_DEL = 67)
device.deviceV2.shell("input keyevent 67")
sleep(random.uniform(0.1, 0.25))
# Inject the correct character
_adb_inject_text(device, chunk[-1])
else:
_adb_inject_text(device, chunk)
# Realistic pause between semantic bursts (humans think while typing)
if chunk.endswith((" ", ".", ",", "!", "?")):
sleep(random.uniform(0.2, 0.5))
else:
sleep(random.uniform(0.05, 0.18))
logger.debug("⌨️ [Ghost Keyboard] Injection complete.")
def _adb_inject_text(device, text: str):
if not text:
return
# For Android `input text`, spaces must be mapped to %s
# Single quotes need to be bash escaped since we wrap the string in ''
# Special characters like & | > < \ ( ) { } ! must be carefully handled.
# The safest way is to let shell loop over characters or strictly replace.
safe_text = text.replace(" ", "%s").replace("'", "\\'")
# Send through Android's native InputManager
try:
device.deviceV2.shell(["input", "text", safe_text])
except Exception as e:
logger.debug(f"[Ghost Keyboard] Native injection failed: {e}")

View File

@@ -0,0 +1,125 @@
import logging
import os
import hashlib
import time
from typing import Optional
from colorama import Fore
from GramAddict.core.qdrant_memory import QdrantBase
from qdrant_client.models import PointStruct, Filter, FieldCondition, MatchValue
logger = logging.getLogger(__name__)
class SwarmProtocol(QdrantBase):
"""
Decentralized Markov state-channel for P2P knowledge sharing.
Manages 'Pheromones' (successful UI transitions and interactions)
and 'BannedPaths' (failed attempts) across bot sessions.
This creates a Fleet Learning effect: every session learns from
every previous session's successes and failures.
"""
def __init__(self, username: str):
self.username = username
super().__init__(collection_name="gramaddict_swarm_pheromones", vector_size=4)
def emit_pheromone(self, path_hash: str, outcome: str):
"""
Broadcasting a successful UI transition or interaction to the fleet memory.
Future sessions can use this to avoid failed paths and repeat successful ones.
"""
if not self.is_connected or not self.client:
return
try:
self.upsert_point(
seed_string=f"{path_hash}_{outcome}",
vector=[1.0, 0.0, 0.0, 0.0], # Dummy vector
payload={
"path_hash": path_hash,
"outcome": outcome,
"username": self.username,
"timestamp": time.time(),
"count": 1,
},
log_success=f"🌐 [Swarm] ⚡ Pheromone emitted: {path_hash[:16]}{outcome}"
)
except Exception as e:
logger.debug(f"[Swarm] Pheromone emit failed: {e}")
def query_consensus(self, path_hash: str) -> Optional[str]:
"""
Queries the swarm for historical outcomes of a specific path.
Returns the most recent outcome or None.
"""
if not self.is_connected or not self.client:
return None
try:
points, _ = self.client.scroll(
collection_name=self.collection_name,
scroll_filter=Filter(
must=[
FieldCondition(
key="path_hash",
match=MatchValue(value=path_hash)
)
]
),
limit=1,
with_payload=True,
)
if points:
outcome = points[0].payload.get("outcome")
logger.info(
f"🌐 [Swarm] Consensus for {path_hash[:16]}: {outcome}",
extra={"color": f"{Fore.CYAN}"}
)
return outcome
except Exception as e:
logger.debug(f"[Swarm] Consensus query failed: {e}")
return None
def sync_banned_paths(self, banned_paths_db):
"""
Pull globally banned paths from the swarm into local BannedPathsDB.
Ensures new sessions immediately know about failed paths.
"""
if not self.is_connected or not self.client:
return
try:
points, _ = self.client.scroll(
collection_name=self.collection_name,
scroll_filter=Filter(
must=[
FieldCondition(
key="outcome",
match=MatchValue(value="banned")
)
]
),
limit=100,
with_payload=True,
)
synced = 0
for pt in points:
payload = pt.payload or {}
path = payload.get("path_hash", "")
if path and banned_paths_db:
banned_paths_db.ban(path, "swarm_synced", reason="Synced from fleet memory")
synced += 1
if synced > 0:
logger.info(
f"🌐 [Swarm] Synced {synced} banned paths from fleet memory.",
extra={"color": f"{Fore.CYAN}"}
)
except Exception as e:
logger.debug(f"[Swarm] Banned path sync failed: {e}")

View File

@@ -0,0 +1,652 @@
import logging
import xml.etree.ElementTree as ET
import math
import re
import base64
import json
import os
import time
from typing import Optional, Tuple, Dict, Any
from GramAddict.core.qdrant_memory import QdrantBase
from GramAddict.core.llm_provider import query_telepathic_llm
from GramAddict.core.diagnostic_dump import dump_ui_state
logger = logging.getLogger(__name__)
# ── Screen Zone Constants (fraction of screen height) ──
# Used for positional sanity checking instead of hardcoded resource-IDs.
STATUS_BAR_ZONE = 0.04 # Top 4% = Android status bar (wifi, battery, clock)
NAV_BAR_ZONE = 0.92 # Bottom 8% = Android nav bar / Instagram bottom tabs
MAX_BUTTON_AREA = 150000 # Buttons/icons should be smaller than this (px²)
MAX_CONTAINER_AREA = 500000 # Anything above this is a full-screen container
# Cache files
MEMORY_FILE = "telepathic_memory.json"
BLACKLIST_FILE = "telepathic_blacklist.json"
class TelepathicEngine:
"""
Project Singularity V9: The Self-Learning Telepathic UI Engine
Completely replaces static Locators (XPath/Regex).
Transforms UI Nodes into natural language semantics, generates vector embeddings,
and returns the node mathematically closest to the target intent.
V9 Philosophy: ZERO hardcoded Instagram IDs.
Instead of maintaining brittle ID lists, the engine uses:
1. Structural heuristics (size, position, element class) — app-agnostic
2. Post-click verification — caller confirms if the click worked
3. Negative learning — failed clicks are blacklisted and never repeated
4. Positive reinforcement — confirmed clicks are cached for instant recall
The engine never trusts a VLM output blindly. It returns candidates,
and the caller uses `confirm_click()` or `reject_click()` to teach it.
"""
_instance = None
_last_click_context: Optional[dict] = None # Tracks what we last returned for feedback
@classmethod
def get_instance(cls):
if cls._instance is None:
cls._instance = cls()
return cls._instance
def __init__(self):
self.embedding_helper = QdrantBase("telepathic_engine_cache")
self._embedding_cache: Dict[str, list] = {}
self._intent_cache: Dict[str, list] = {}
# Load blacklist (negative learnings) into memory
self._blacklist = self._load_json(BLACKLIST_FILE)
# Load positive cache
self._memory = self._load_json(MEMORY_FILE)
# ──────────────────────────────────────────────
# Core Math
# ──────────────────────────────────────────────
def _cosine_similarity(self, v1: list, v2: list) -> float:
if not v1 or not v2 or len(v1) != len(v2):
return 0.0
dot_product = sum(a * b for a, b in zip(v1, v2))
magnitude_v1 = math.sqrt(sum(a * a for a in v1))
magnitude_v2 = math.sqrt(sum(b * b for b in v2))
if magnitude_v1 == 0 or magnitude_v2 == 0:
return 0.0
return dot_product / (magnitude_v1 * magnitude_v2)
def _get_cached_embedding(self, text: str, is_intent: bool = False) -> Optional[list]:
cache = self._intent_cache if is_intent else self._embedding_cache
if text in cache:
return cache[text]
if len(self._embedding_cache) > 2000:
self._embedding_cache.clear()
vec = self.embedding_helper._get_embedding(text)
if vec:
cache[text] = vec
return vec
# ──────────────────────────────────────────────
# Persistent JSON helpers
# ──────────────────────────────────────────────
@staticmethod
def _load_json(path: str) -> dict:
try:
if os.path.exists(path):
with open(path, "r") as f:
return json.load(f)
except Exception:
pass
return {}
@staticmethod
def _save_json(path: str, data: dict):
try:
with open(path, "w") as f:
json.dump(data, f, indent=4)
except Exception as e:
logger.warning(f"Could not save {path}: {e}")
# ──────────────────────────────────────────────
# XML Parsing
# ──────────────────────────────────────────────
def _extract_semantic_nodes(self, xml_string: str) -> list[dict]:
"""Parses Android UI XML and extracts clickable/interactive nodes."""
nodes = []
try:
clean_xml = re.sub(r'<\?xml.*?\?>', '', xml_string).strip()
root = ET.fromstring(clean_xml)
for elem in root.iter('node'):
attrib = elem.attrib
text = attrib.get('text', '').strip()
content_desc = attrib.get('content-desc', '').strip()
res_id = attrib.get('resource-id', '').strip()
class_name = attrib.get('class', '').strip()
clickable = attrib.get('clickable', 'false') == 'true'
scrollable = attrib.get('scrollable', 'false') == 'true'
long_clickable = attrib.get('long-clickable', 'false') == 'true'
semantic_res = res_id and any(x in res_id.lower() for x in ['button', 'tab', 'icon', 'action', 'menu'])
has_semantic_weight = bool(content_desc or semantic_res)
if not (clickable or scrollable or long_clickable or has_semantic_weight):
continue
if not text and not content_desc and not res_id:
continue
desc_parts = []
if text: desc_parts.append(f"text: '{text}'")
if content_desc: desc_parts.append(f"description: '{content_desc}'")
if res_id:
clean_id = res_id.split('/')[-1].replace('_', ' ')
desc_parts.append(f"id context: '{clean_id}'")
semantic_string = ", ".join(desc_parts)
if not semantic_string:
continue
bounds_str = attrib.get('bounds', '')
match = re.match(r'\[(\d+),(\d+)\]\[(\d+),(\d+)\]', bounds_str)
if not match:
continue
left, top, right, bottom = map(int, match.groups())
center_x = (left + right) // 2
center_y = (top + bottom) // 2
width = right - left
height = bottom - top
nodes.append({
"semantic_string": semantic_string,
"x": center_x,
"y": center_y,
"width": width,
"height": height,
"area": width * height,
"raw_bounds": bounds_str,
"resource_id": res_id,
"class_name": class_name,
"original_attribs": {"text": text, "desc": content_desc}
})
except Exception as e:
logger.error(f"Telepathic XML parsing failed: {e}")
return nodes
# ──────────────────────────────────────────────
# Structural Sanity (app-agnostic, no hardcoded IDs)
# ──────────────────────────────────────────────
def _structural_sanity_check(self, node: dict, intent_description: str, screen_height: int = 2400) -> bool:
"""
App-agnostic structural validation. Checks physical properties
(size, position, element class) — NOT resource-ID strings.
Returns False if the node is structurally implausible as a click target.
"""
# 1. Reject massive containers (full-screen views, recycler views)
# UNLESS the intent explicitly targets media
is_media_intent = any(k in intent_description.lower() for k in ["video", "photo", "reel", "media", "post"])
if node.get("area", 0) > MAX_CONTAINER_AREA and not is_media_intent:
return False
# 2. Reject nodes in the Android status bar zone (top 4%)
if node.get("y", 0) < screen_height * STATUS_BAR_ZONE:
return False
# 3. Reject nodes with zero area (invisible)
if node.get("area", 0) == 0:
return False
return True
def _is_blacklisted(self, intent: str, semantic_string: str) -> bool:
"""Checks if this intent→node mapping was previously rejected via negative learning."""
blacklisted = self._blacklist.get(intent, [])
return semantic_string in blacklisted
# ──────────────────────────────────────────────
# App Context Guard
# ──────────────────────────────────────────────
def _is_instagram_context(self, nodes: list[dict]) -> bool:
"""
Returns True only if the extracted nodes appear to come from the target app.
Checks for the presence of the dynamic app_id in resource-ID prefixes.
"""
app_id = getattr(self, "_cached_app_id", None)
if not app_id:
from GramAddict.core.config import Config
try:
cfg = Config()
app_id = cfg.args.app_id if hasattr(cfg, "args") and hasattr(cfg.args, "app_id") else "com.instagram.android"
except Exception:
app_id = "com.instagram.android"
self._cached_app_id = app_id
for n in nodes:
rid = n.get("resource_id", "")
if app_id in rid:
return True
return False
# ──────────────────────────────────────────────
# Stage 0: Deterministic Keyword Fast Path
# ──────────────────────────────────────────────
def _keyword_match_score(self, intent_description: str, nodes: list[dict]) -> Optional[dict]:
"""
Pure string-matching stage. Extracts keywords from the intent and
matches them against node text, description, and resource-id.
Returns the best matching node as a result dict, or None.
This eliminates ~90% of embedding/VLM calls for common UI intents.
ZERO AI cost — runs entirely on CPU string ops.
"""
# Extract meaningful keywords from intent (strip common filler words)
filler = {"tap", "the", "button", "tab", "on", "in", "a", "an", "of", "for", "to", "and", "or", "input", "text", "box"}
intent_words = set(w.lower() for w in re.split(r'\W+', intent_description) if w and w.lower() not in filler and len(w) > 1)
if not intent_words:
return None
# Expand known Instagram aliases to avoid sending UI basics to the LLM mappings
aliases = {
"reels": ["clips", "reel"],
"explore": ["search"],
"home": ["main"],
"like": ["heart"],
"comment": ["reply"],
"profile": ["user", "account"],
}
scored = []
for node in nodes:
sem = node.get("semantic_string", "").lower()
rid = node.get("resource_id", "").lower().replace("_", " ")
desc_text = node.get("original_attribs", {}).get("desc", "").lower()
node_text = node.get("original_attribs", {}).get("text", "").lower()
# Combine all searchable fields
searchable = f"{sem} {rid} {desc_text} {node_text}"
# Count how many intent keywords appear in the node's text (including aliases)
hits = 0
for w in intent_words:
if w in searchable:
hits += 1
elif w in aliases:
for alias in aliases[w]:
if alias in searchable:
hits += 1
break
if hits == 0:
continue
# Score = ratio of intent keywords matched
score = hits / len(intent_words)
# Require at least 40% keyword overlap to avoid false positives
if score >= 0.4:
scored.append((node, score))
if not scored:
return None
# Sort by score desc, then by area asc (prefer smallest/most atomic)
scored.sort(key=lambda x: (-x[1], x[0].get("area", 999999)))
best_node, best_score = scored[0]
# Check for already-liked state
if "like" in intent_description.lower():
desc = best_node.get("original_attribs", {}).get("desc", "").lower()
text = best_node.get("original_attribs", {}).get("text", "").lower()
if "liked" in desc or "liked" in text:
logger.info("⏭️ [Keyword Fast Path] Post is already Liked. Skipping.")
return {"x": None, "y": None, "score": 1.0, "semantic": "already_liked", "skip": True}
logger.info(f"⚡ [Keyword Fast Path] Instant match for '{intent_description}'{best_node['semantic_string']} (KeyScore: {best_score:.2f})")
self._track_click(intent_description, best_node)
return {
"x": best_node["x"],
"y": best_node["y"],
"score": 0.95, # High confidence — deterministic match
"semantic": best_node["semantic_string"],
"area": best_node.get("area", 0),
"source": "keyword"
}
# ──────────────────────────────────────────────
# Core: Find Best Node
# ──────────────────────────────────────────────
def find_best_node(self, xml_hierarchy: str, intent_description: str, min_confidence: float = 0.82, device=None) -> Optional[dict]:
"""
Scans the screen and returns the center coordinates (x, y) of the node
whose embedding is most mathematically similar to the intent.
Resolution cascade (ordered by speed & reliability):
1. Positive Memory Cache (past CONFIRMED clicks)
2. Keyword Fast Path (deterministic string matching)
3. Vector Similarity Engine (embedding cosine similarity)
4. Vision Cortex Fallback (VLM, with structural guards)
All results are PROVISIONAL until the caller confirms via confirm_click().
Failed clicks should be reported via reject_click().
"""
logger.debug(f"[TelepathicEngine] Seeking intent: '{intent_description}'")
interactive_nodes = self._extract_semantic_nodes(xml_hierarchy)
if not interactive_nodes:
logger.debug("[TelepathicEngine] Screen contains no interactable semantic nodes.")
return None
# Detect screen height for zone calculations
screen_height = 2400
if interactive_nodes:
max_y = max(n.get("y", 0) + n.get("height", 0) // 2 for n in interactive_nodes)
if max_y > 100:
screen_height = int(max_y * 1.05)
# Pre-filter: Remove structurally implausible nodes and blacklisted mappings
viable_nodes = []
for node in interactive_nodes:
if not self._structural_sanity_check(node, intent_description, screen_height):
continue
if self._is_blacklisted(intent_description, node["semantic_string"]):
logger.debug(f"🚫 [Blacklist] Skipping known-bad mapping: '{intent_description}''{node['semantic_string']}'")
continue
viable_nodes.append(node)
if not viable_nodes:
logger.warning(f"[TelepathicEngine] No viable nodes left after filtering for '{intent_description}'")
return None
# ── App Context Guard: Abort if NOT in Target App ──
if not self._is_instagram_context(interactive_nodes):
if device:
current_app = device._get_current_app()
if current_app != device.app_id:
logger.warning(f"⚠️ [Context Guard] Not in target app (Current: {current_app}). Aborting AI lookup for '{intent_description}'.")
return None
else:
logger.warning(f"⚠️ [Context Guard] Not in target app! Aborting AI lookup for '{intent_description}'.")
return None
# ── Stage 1: Positive Memory Cache (CONFIRMED past clicks) ──
self._memory = self._load_json(MEMORY_FILE) # Reload for freshness
if intent_description in self._memory:
known_semantics = self._memory[intent_description]
for n in viable_nodes:
if n["semantic_string"] in known_semantics:
# Prevent un-liking
if "like" in intent_description.lower() and re.search(
r"\b(liked|gefällt mir nicht mehr)\b",
n["semantic_string"].lower()
):
logger.info("⏭️ [Memory] Post is already Liked. Skipping tap to prevent un-liking.")
return {"x": None, "y": None, "score": 1.0, "semantic": "already_liked", "skip": True}
logger.debug(f"🧠 [Confirmed Memory] Instant recall: '{intent_description}'{n['semantic_string']}")
self._track_click(intent_description, n)
return {
"x": n["x"],
"y": n["y"],
"score": 1.0,
"semantic": f"Memory Match: {n['semantic_string']}",
"source": "memory"
}
# ── Stage 1.5: Deterministic Keyword Fast Path ──
fast_path_result = self._keyword_match_score(intent_description, viable_nodes)
if fast_path_result:
return fast_path_result
# ── Stage 2: Vector Similarity Engine ──
intent_vec = self._get_cached_embedding(intent_description, is_intent=True)
if intent_vec:
scored_nodes = []
for node in viable_nodes:
node_vec = self._get_cached_embedding(node["semantic_string"])
if not node_vec:
continue
score = self._cosine_similarity(intent_vec, node_vec)
scored_nodes.append((node, score))
# Sort by score descending
scored_nodes.sort(key=lambda x: x[1], reverse=True)
# Update viable_nodes so that the VLM fallback gets the top semantic candidates
viable_nodes = [n for n, s in scored_nodes]
# Among high-confidence matches, prefer smaller/more atomic elements
if scored_nodes and scored_nodes[0][1] >= min_confidence:
# Get all nodes within 0.05 of the top score
top_score = scored_nodes[0][1]
top_tier = [(n, s) for n, s in scored_nodes if s >= top_score - 0.05]
# Among equally-scored candidates, prefer the smallest (most atomic)
top_tier.sort(key=lambda x: x[0].get("area", 999999))
best_node, best_score = top_tier[0]
# Prevent un-liking
if "like" in intent_description.lower() and re.search(
r"\b(liked|gefällt mir nicht mehr)\b",
best_node["semantic_string"].lower()
):
logger.info("⏭️ [Telepathic] Post is already Liked. Skipping.")
return {"x": None, "y": None, "score": 1.0, "semantic": "already_liked", "skip": True}
logger.info(f"✨ [Telepathic Match] '{intent_description}'{best_node['semantic_string']} (Score: {best_score:.3f})")
self._track_click(intent_description, best_node)
return {
"x": best_node["x"],
"y": best_node["y"],
"score": best_score,
"semantic": best_node["semantic_string"],
"source": "vector"
}
elif scored_nodes:
logger.warning(f"⚠️ [Telepathic] Low confidence ({scored_nodes[0][1]:.3f} < {min_confidence}) for '{intent_description}'.")
# ── Stage 3: Telepathic LLM Fallback (Text-Based XML Reasoning) ──
if device:
logger.info(f"🧠 [Agentic Fallback] Activating structural LLM reasoning for: '{intent_description}'")
return self._vision_cortex_fallback(intent_description, viable_nodes, device, screen_height)
return None
# ──────────────────────────────────────────────
# Click Tracking & Feedback Loop
# ──────────────────────────────────────────────
def _track_click(self, intent: str, node: dict):
"""Records what we're about to click so confirm/reject can reference it."""
TelepathicEngine._last_click_context = {
"intent": intent,
"semantic_string": node["semantic_string"],
"x": node["x"],
"y": node["y"],
"timestamp": time.time()
}
def confirm_click(self, intent: str = None):
"""
Called by the interaction layer AFTER verifying the click produced the expected result.
Stores the mapping as a confirmed positive learning.
Usage:
result = telepathic.find_best_node(xml, "tap like button", device=device)
_humanized_click(device, result["x"], result["y"])
# ... verify the like actually happened ...
telepathic.confirm_click("tap like button")
"""
ctx = TelepathicEngine._last_click_context
if not ctx:
return
actual_intent = intent or ctx["intent"]
sem = ctx["semantic_string"]
# Add to positive memory
if actual_intent not in self._memory:
self._memory[actual_intent] = []
if sem not in self._memory[actual_intent]:
self._memory[actual_intent].append(sem)
self._save_json(MEMORY_FILE, self._memory)
logger.debug(f"✅ [Confirmed Learning] Stored: '{actual_intent}''{sem}'")
# Remove from blacklist if it was there (rehabilitation)
if actual_intent in self._blacklist and sem in self._blacklist[actual_intent]:
self._blacklist[actual_intent].remove(sem)
self._save_json(BLACKLIST_FILE, self._blacklist)
logger.debug(f"🔄 [Rehabilitation] Removed from blacklist: '{actual_intent}''{sem}'")
TelepathicEngine._last_click_context = None
def reject_click(self, intent: str = None):
"""
Called by the interaction layer when the click did NOT produce the expected result.
Adds the mapping to the blacklist (negative learning) so it's never tried again.
Also removes it from positive memory if it was cached there.
Usage:
result = telepathic.find_best_node(xml, "tap comment button", device=device)
_humanized_click(device, result["x"], result["y"])
# ... verify comment sheet did NOT open ...
telepathic.reject_click("tap comment button")
"""
ctx = TelepathicEngine._last_click_context
if not ctx:
return
actual_intent = intent or ctx["intent"]
sem = ctx["semantic_string"]
# Add to blacklist
if actual_intent not in self._blacklist:
self._blacklist[actual_intent] = []
if sem not in self._blacklist[actual_intent]:
self._blacklist[actual_intent].append(sem)
self._save_json(BLACKLIST_FILE, self._blacklist)
logger.warning(f"🚫 [Negative Learning] Blacklisted: '{actual_intent}''{sem}'")
# Remove from positive memory if it was cached
if actual_intent in self._memory and sem in self._memory[actual_intent]:
self._memory[actual_intent].remove(sem)
self._save_json(MEMORY_FILE, self._memory)
logger.warning(f"🗑️ [Memory Purge] Removed bad mapping from memory: '{actual_intent}''{sem}'")
TelepathicEngine._last_click_context = None
# ──────────────────────────────────────────────
# Vision Cortex Fallback (VLM)
# ──────────────────────────────────────────────
def _vision_cortex_fallback(self, intent: str, nodes: list[dict], device, screen_height: int = 2400) -> Optional[dict]:
"""
Uses a Language Model to identify the correct node from parsed screen XML
when embeddings are insufficient. 100% Screenshot-free for maximum speed and zero hallucination.
Guards are STRUCTURAL (size, position, class) not ID-based.
Learning happens via the confirm/reject feedback loop, not here.
"""
try:
# Limit to 20 nodes for token efficiency
simplified_nodes = []
for i, n in enumerate(nodes[:20]):
simplified_nodes.append({
"index": i,
"bounds": n["raw_bounds"],
"semantic": n["semantic_string"]
})
# Get model config
from GramAddict.core.config import Config
try:
args = Config().args
except Exception:
args = None
model = getattr(args, "ai_telepathic_model", "google/gemini-3.1-flash-lite-preview") if args else "google/gemini-3.1-flash-lite-preview"
url = getattr(args, "ai_telepathic_url", "https://openrouter.ai/api/v1/chat/completions") if args else "https://openrouter.ai/api/v1/chat/completions"
if device and hasattr(device, 'args') and device.args:
model = getattr(device.args, "ai_telepathic_model", model)
url = getattr(device.args, "ai_telepathic_url", url)
system_prompt = (
"You identify which UI element to tap based ONLY on a JSON array of parsed Android elements. "
"Each element has an 'index', structural 'bounds', and a 'semantic' description. "
"Output ONLY valid JSON containing the exact `index` to interact with, and a `reason`. "
)
user_prompt = (
f"Which element should I tap to: {intent}\n\n"
f"Elements:\n{json.dumps(simplified_nodes, indent=1)}\n\n"
"Rules:\n"
"- Pick the SMALLEST, most specific button or icon\n"
"- NEVER pick large containers, full-screen views, or recycler views\n"
"- NEVER pick system icons (wifi, battery, status bar, clock)\n"
"Return: {\"index\": number, \"reason\": \"...\"}"
)
resp_str = query_telepathic_llm(model, url, system_prompt, user_prompt)
data = json.loads(resp_str)
idx = data.get("index")
if idx is not None and 0 <= idx < len(nodes):
match = nodes[idx]
# ── Structural Guard 1: Size ──
is_media_intent = any(k in intent.lower() for k in ["video", "photo", "reel", "media", "post"])
if match.get("area", 0) > MAX_CONTAINER_AREA and not is_media_intent:
logger.error(
f"❌ [Structural Guard] VLM selected oversized element "
f"({match.get('width')}x{match.get('height')}): {match['semantic_string']}. REJECTING."
)
dump_ui_state(device, "vlm_hallucination", {
"intent": intent,
"rejected_node": match["semantic_string"],
"node_size": f"{match.get('width')}x{match.get('height')}",
"vlm_index": idx
})
return None
# ── Structural Guard 2: Position (status bar) ──
if match.get("y", 0) < screen_height * STATUS_BAR_ZONE:
logger.error(
f"❌ [Structural Guard] VLM selected element in status bar zone "
f"(y={match.get('y')}): {match['semantic_string']}. REJECTING."
)
return None
# ── Structural Guard 3: Already blacklisted ──
if self._is_blacklisted(intent, match["semantic_string"]):
logger.error(
f"❌ [Blacklist Guard] VLM selected previously-rejected element: "
f"'{match['semantic_string']}'. REJECTING."
)
return None
logger.info(f"🎯 [Vision Success] VLM identified node {idx} for '{intent}': {match['semantic_string']}")
# Track but do NOT auto-cache. Wait for confirm_click() from caller.
self._track_click(intent, match)
return {
"x": match["x"],
"y": match["y"],
"score": 0.85, # Not 1.0 — VLM is provisional, not ground truth
"semantic": f"VLM Match: {match['semantic_string']}",
"source": "agentic_fallback"
}
except Exception as e:
logger.error(f"[Vision Cortex] Fallback failed: {e}")
return None

View File

@@ -0,0 +1,113 @@
import logging
import random
import time
from colorama import Fore, Style
from GramAddict.core.session_state import SessionState
logger = logging.getLogger(__name__)
def _humanized_scroll_down(device):
# Same as bot_flow._humanized_scroll but strictly downward
info = device.get_info()
w, h = info.get("displayWidth", 1080), info.get("displayHeight", 2400)
start_x = int(w * 0.8) + device.cm_to_pixels(random.uniform(-0.3, 0.3))
start_y = int(h * 0.7) + device.cm_to_pixels(random.uniform(-0.5, 0.5))
end_y = int(h * 0.2) + device.cm_to_pixels(random.uniform(-0.5, 0.5))
duration = random.uniform(0.08, 0.12)
device.deviceV2.swipe(start_x, start_y, start_x, end_y, duration)
from GramAddict.core.bot_flow import sleep
sleep(1.0)
def _run_zero_latency_unfollow_loop(device, zero_engine, nav_graph, configs, session_state, current_target, cognitive_stack):
"""
Executes the autonomous Unfollow logic in the Zero-Latency architecture.
Assumes the bot is already at the "FollowingList" UI state.
"""
logger.info(f"🧠 [Unfollow Engine] Initiating cleanup routine in {current_target}...", extra={"color": f"{Style.BRIGHT}{Fore.CYAN}"})
telepathic = cognitive_stack.get("telepathic")
dopamine = cognitive_stack.get("dopamine")
unfollow_limit = int(getattr(configs.args, "total_unfollows_limit", 50))
failed_scrolls = 0
total_unfollowed_this_session = 0
from GramAddict.core.bot_flow import sleep, dump_ui_state, _humanized_click
# Initialize basic tuple if it's missing (helps with tests and initializations)
if not hasattr(session_state, 'totalUnfollowed'):
session_state.totalUnfollowed = 0
while not dopamine.is_app_session_over():
# Check global limit tuple logic
limit_val = session_state.check_limit(SessionState.Limit.UNFOLLOWS)
if isinstance(limit_val, tuple) and limit_val[0]:
logger.info("🛑 Unfollow limit reached for session. Yielding control.")
return "BOREDOM_CHANGE_FEED"
elif limit_val is True:
logger.info("🛑 Unfollow limit reached for session. Yielding control.")
return "BOREDOM_CHANGE_FEED"
if total_unfollowed_this_session >= unfollow_limit:
logger.info("🛑 Configured unfollow limit reached. Yielding control.")
return "BOREDOM_CHANGE_FEED"
try:
xml_dump = device.deviceV2.dump_hierarchy()
# Use Telepathic Engine to explicitly locate existing "Following" buttons in lists
nodes = telepathic._extract_semantic_nodes(xml_dump, "find 'Following' buttons next to usernames", threshold=0.7)
action_taken = False
for node in nodes:
# Basic validation it's an interactive button
if node.get("skip") or not node.get("bounds"):
continue
# Tap the first valid following button we see
_humanized_click(device, node["x"], node["y"])
action_taken = True
logger.debug(f"👆 Tapped following button at ({node['x']}, {node['y']})")
# Check for confirmation dialog ("Unfollow @username?")
sleep(1.5)
confirm_xml = device.deviceV2.dump_hierarchy()
confirm_nodes = telepathic._extract_semantic_nodes(confirm_xml, "find 'Unfollow' confirmation button", threshold=0.8)
if confirm_nodes and not confirm_nodes[0].get("skip"):
c_node = confirm_nodes[0]
_humanized_click(device, c_node["x"], c_node["y"])
sleep(1.0)
logger.info("✅ [Unfollow Engine] Unfollowed a user in list.", extra={"color": Fore.GREEN})
session_state.totalUnfollowed += 1
total_unfollowed_this_session += 1
failed_scrolls = 0
# Unfollow cost logic
dopamine.boredom += random.uniform(1.0, 3.0)
sleep(2.0)
break
if not action_taken:
# No following buttons in view, scroll down to find more
_humanized_scroll_down(device)
dopamine.boredom += 0.5
failed_scrolls += 1
if failed_scrolls > 5:
logger.warning("⚠️ [Unfollow Engine] No 'Following' buttons found after multiple scrolls. Aborting or reaching bottom.")
return "BOREDOM_CHANGE_FEED"
if dopamine.wants_to_change_feed():
logger.info("🧠 [Unfollow Engine] Desire to clean up following list satisfied. Navigating elsewhere.")
return "BOREDOM_CHANGE_FEED"
except Exception as e:
logger.error(f"⚠️ [FSD Anomaly Handler] Exception in Unfollow Loop: {e}")
_humanized_scroll_down(device)
failed_scrolls += 1
if failed_scrolls > 3:
return "CONTEXT_LOST"
return "SESSION_OVER"

83
GramAddict/core/utils.py Normal file
View File

@@ -0,0 +1,83 @@
import logging
import random
import requests
import sys
from datetime import datetime, timedelta
from time import sleep
from colorama import Fore, Style
from packaging.version import parse as parse_version
from GramAddict.core.version import __version__
logger = logging.getLogger(__name__)
def sanitize_text(text):
return (text or "").strip()
def random_sleep(inf=1.0, sup=3.0, modulable=True):
from GramAddict.core.config import Config
configs = Config()
try:
multiplier = float(getattr(configs.args, "speed_multiplier", 1.0))
except (ValueError, TypeError):
multiplier = 1.0
delay = random.uniform(inf, sup) / (multiplier if modulable else 1.0)
sleep(max(delay, 0.2))
def config_examples():
logger.debug("Config examples handled by documentation.")
def check_if_updated():
logger.info(f"GramAddict v.{__version__}", extra={"color": f"{Style.BRIGHT}{Fore.MAGENTA}"})
def get_instagram_version(device):
try:
output = device.deviceV2.shell(f"dumpsys package {device.app_id}").output
import re
version_match = re.findall("versionName=(\\S+)", output)
return version_match[0] if version_match else "unknown"
except Exception:
return "unknown"
def close_instagram(device, force_kill=False):
if force_kill:
logger.info("Force-closing Instagram app to clean session state.")
try:
device.deviceV2.app_stop(device.app_id)
except Exception as e:
logger.debug(f"Error closing app: {e}")
else:
logger.info("Backgrounding Instagram app (minimizing).")
try:
device.deviceV2.press("home")
except Exception as e:
logger.debug(f"Error pressing home: {e}")
def open_instagram(device, force_restart=False):
if force_restart:
logger.info("Opening Instagram app (Fresh Start).")
close_instagram(device, force_kill=True)
device.deviceV2.app_start(device.app_id)
random_sleep(3, 5, modulable=False)
else:
logger.info("Bringing Instagram app to foreground.")
device.deviceV2.app_start(device.app_id)
random_sleep(1, 2, modulable=False)
return True
def set_time_delta(args):
args.time_delta_session = random.randint(-300, 300)
def wait_for_next_session(time_left, session_state, sessions, device):
logger.info(f"Waiting {time_left} until next working hours.")
sleep(60)
def get_value(count, name, default=0):
if count is None: return default
if isinstance(count, (int, float)): return count
try:
if "-" in str(count):
parts = str(count).split("-")
return random.randint(int(parts[0]), int(parts[1]))
return int(count)
except Exception:
return default

View File

@@ -0,0 +1,2 @@
__version__ = "7.0.0"
__tested_ig_version__ = "300.0.0.29.110"

View File

@@ -0,0 +1,71 @@
import logging
import re
import xml.etree.ElementTree as ET
logger = logging.getLogger(__name__)
class ZeroLatencyEngine:
"""
Project Singularity V7: The Zero-Latency Executor
This engine receives a pre-compiled heuristic (Regex/XPath) from the memory cache
and executes it against the local XML layout in under 5ms.
It is completely deterministic. No LLM calls happen here.
"""
def __init__(self, device):
self.device = device
def evaluate_heuristic(self, rule: dict, context_xml: str):
"""
Executes a compiled heuristic rule against the provided XML dump.
Rule schema: {"rule_type": "regex", "target_attribute": "text", "pattern": "..."}
Returns True/False for intent (e.g. is_ad, is_liked), or extracted strings (e.g. post_owner).
"""
if not rule or not context_xml:
return None
rule_type = rule.get("rule_type", "regex")
target_attr = rule.get("target_attribute", "text")
pattern = rule.get("pattern", "")
if not pattern:
return None
try:
root = ET.fromstring(context_xml)
if rule_type == "regex":
# Remove (?i) if present because we compile with re.IGNORECASE anyway
clean_pattern = pattern.replace('(?i)', '')
regex = re.compile(clean_pattern, re.IGNORECASE)
for node in root.iter("node"):
val = ""
if target_attr == "text":
val = node.attrib.get("text", "")
elif target_attr == "content-desc":
val = node.attrib.get("content-desc", "")
elif target_attr == "resource-id":
val = node.attrib.get("resource-id", "")
else:
# Fallback all
val_text = node.attrib.get("text", "")
val_desc = node.attrib.get("content-desc", "")
val_resid = node.attrib.get("resource-id", "")
val = f"{val_text} | {val_desc} | {val_resid}"
match = regex.search(val)
if match:
if len(match.groups()) > 0:
return match.group(1) # Return captured group (e.g., username)
return True # Return boolean existence (e.g. is_ad)
elif rule_type == "xpath":
# Basic xpath parsing over ET
nodes = root.findall(pattern)
if nodes:
return nodes[0].attrib.get(target_attr, "")
return False # Rule ran but found nothing
except Exception as e:
logger.debug(f"ZeroLatencyEngine failed to evaluate rule {pattern}: {e}")
return None

View File

@@ -0,0 +1,38 @@
from GramAddict.core.plugin_loader import Plugin
class ExamplePlugin(Plugin):
"""Short explanation that shows up on start"""
def __init__(self):
super().__init__()
self.description = (
"Description that currently has no use - can be same as above."
)
self.arguments = [
#
# argparse arguments
#
# Example of operation (a plugin that does something - like interact with followers)
{
"arg": "--interact",
"nargs": None, # see argparse docs for usage - if not needed use None
"help": "help message that explains what it does",
"metavar": None, # see argparse docs for usage - if not needed use None
"default": None, # see argparse docs for usage - if not needed use None
"operation": True, # If the argument is an operation, set to true. Otherwise do not include
},
# Example of argparse "action" (something that requires no arguments)
{
"arg": "--screen-sleep",
"help": "save your screen by turning it off during the inactive time, disabled by default",
"action": "store_true", # see argparse docs for usage
},
]
def run(self, device, configs, storage, sessions, profile_filter, plugin):
# Your code here. All variables above must be in function definition, but
# do not have to be used. If not needed, just ignore it. If you need anything
# else from the main script - please include it in __init__.py and update
# the run definition on all other plugins.
pass

2
GramAddict/version.py Normal file
View File

@@ -0,0 +1,2 @@
# that file is deprecated, current version is now stored in GramAddict/__init__.py
__version__ = "3.2.12"