intaleq/backend/marketing_engine/client_bot_template.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# ==============================================================================
# Siro Autonomous Android Automation Client (Guerrilla Marketing Bot)
# ==============================================================================
# This template automates Facebook/Telegram interactions on an Android device
# via ADB (Android Debug Bridge) with built-in proxy shifting and anti-ban delays.
#
# Requirements:
# 1. Python 3.x
# 2. Android device with USB Debugging enabled, connected to PC.
# 3. ADB installed and added to System PATH.
# 4. Target apps (Facebook/Telegram) installed and logged in.
# ==============================================================================

import os
import sys
import time
import random
import subprocess
import requests

# --- Configuration ---
SERVER_URL = "http://jordan-siro.intaleqapp.com/backend/marketing_engine/social_worker.php"
BOT_TOKEN = "YOUR_SECRET_BOT_TOKEN" # Must match headers in social_worker.php
PLATFORM = "facebook" # 'facebook' or 'telegram'
ACCOUNT_ID = None # Set to a specific integer if manual mapping is preferred
DEVICE_ID = None # e.g. "a1b2c3d4" - Recommended: The unique Android device ID for auto Plug & Play binding
CHECK_INTERVAL_SECONDS = 60 # Cooldown between checking for new tasks

headers = {
    "X-Bot-Token": BOT_TOKEN
}

# --- ADB Helper Functions ---
def run_adb(cmd):
    """Run an ADB command and return output."""
    try:
        result = subprocess.run(f"adb {cmd}", shell=True, capture_output=True, text=True, timeout=15)
        return result.stdout.strip()
    except subprocess.TimeoutExpired:
        print("[-] ADB command timed out.")
        return ""

def set_android_proxy(ip, port, username=None, password=None):
    """Dynamically set global HTTP proxy on Android device."""
    if not ip or not port:
        print("[+] Clearing Android proxy...")
        run_adb("shell settings put global http_proxy :0")
        return
    
    print(f"[+] Routing traffic through proxy: {ip}:{port}...")
    # Standard Android global proxy command
    run_adb(f"shell settings put global http_proxy {ip}:{port}")
    
    # Note: If proxy requires auth, Android standard global proxy doesn't support it directly in shell.
    # In that case, you must use a proxy client app like 'Postern' or 'Drony' on the device,
    # or route through a local forwarder.
    time.sleep(3)

def adb_click(x, y):
    """Click on coordinates (x, y)."""
    run_adb(f"shell input tap {x} {y}")
    time.sleep(random.uniform(1.0, 2.5))

def adb_type_humanlike(text):
    """Simulate human typing with random delay between characters to evade ban engines."""
    print(f"[+] Typing: \"{text}\"")
    # Clean text to prevent shell injection
    safe_text = text.replace('"', '\\"').replace('$', '\\$').replace('`', '\\`').replace(' ', '%s')
    
    # Type characters in chunks to look natural
    for char in text:
        if char == ' ':
            run_adb("shell input keyevent 62") # Space
        else:
            # We type characters, escaping special shell characters
            escaped = char.replace('"', '\\"').replace('$', '\\$').replace('`', '\\`').replace("'", "\\'")
            run_adb(f"shell input text '{escaped}'")
        time.sleep(random.uniform(0.05, 0.25)) # Typing speed variability
        
    time.sleep(random.uniform(1.0, 2.0))

def adb_scroll_down():
    """Perform a human-like swipe down."""
    # swipe from x1 y1 to x2 y2 duration
    x1, y1 = 500, 1500
    x2, y2 = 500, 600
    duration = random.randint(300, 800)
    run_adb(f"shell input swipe {x1} {y1} {x2} {y2} {duration}")
    time.sleep(random.uniform(1.5, 3.0))

def adb_open_url(url):
    """Open a URL using Android's default browser/app handler (e.g. opens deep links)."""
    print(f"[+] Launching URL: {url}")
    run_adb(f"shell am start -a android.intent.action.VIEW -d '{url}'")
    time.sleep(random.uniform(5.0, 8.0)) # Wait for app to load

# --- Logging & Status Reporting ---
def report_status(action, params):
    try:
        requests.post(f"{SERVER_URL}?action={action}", data=params, headers=headers, timeout=10)
    except Exception as e:
        print(f"[-] Status report connection error: {e}")

# --- Main Task Loop ---
def process_task():
    print("[*] Checking for pending tasks...")
    try:
        url = f"{SERVER_URL}?action=get_task&platform={PLATFORM}"
        if ACCOUNT_ID:
            url += f"&account_id={ACCOUNT_ID}"
        elif DEVICE_ID:
            url += f"&device_id={DEVICE_ID}"
            
        res = requests.get(url, headers=headers, timeout=10)
        response_data = res.json()
    except Exception as e:
        print(f"[-] Connection failed: {e}")
        return

    if response_data.get("status") != "success" or not response_data.get("data"):
        print("[*] No tasks scheduled at this moment.")
        return

    task = response_data["data"]

    task_id = task["id"]
    task_type = task["type"]
    target_url = task["target_url"]
    comment_text = task["generated_comment"]

    print(f"[+] Task Found! ID: {task_id} | Type: {task_type}")
    print(f"[+] Account to use: {task.get('bot_username')}")

    # 1. Shifting IP via Proxy Configuration
    set_android_proxy(
        task.get("proxy_ip"),
        task.get("proxy_port"),
        task.get("proxy_username"),
        task.get("proxy_password")
    )

    success = False
    error_msg = "Unknown execution error"

    try:
        # 2. Evading Automated Scanners via Random Pre-sleep
        pre_delay = random.randint(5, 15)
        print(f"[+] Simulating human delay before opening app... sleeping {pre_delay}s")
        time.sleep(pre_delay)

        # 3. Open URL (Facebook Post/Telegram Link)
        if target_url:
            adb_open_url(target_url)

        # 4. Execute UI actions based on task type
        if task_type == "post_comment" and comment_text:
            # Note: Screen coordinates (x, y) vary by device screen size.
            # You must customize these coordinates based on your device profile.
            print("[*] Performing UI interactions to comment...")
            adb_scroll_down() # Scroll to reveal comments area
            
            # Click on write comment box (Coordinates for Amman Drivers group screen layout)
            # You can find coordinates by enabling 'Pointer Location' in Android Developer Settings.
            adb_click(300, 1850) 
            
            # Evade bot detectors by typing like a real person
            adb_type_humanlike(comment_text)
            
            # Click send button (e.g. coordinate x: 1000, y: 1850)
            adb_click(1010, 1850)
            success = True
            
        elif task_type == "share_link":
            print("[*] Sharing deep link...")
            # Click Share coordinates, type message, and click post
            # Example coordinates:
            adb_click(900, 1200) # Share button
            adb_click(500, 1500) # Write post option
            adb_type_humanlike(comment_text)
            adb_click(1000, 150) # Post button
            success = True
            
        else:
            error_msg = f"Task type {task_type} is not yet supported in Android ADB Client"
            
    except Exception as e:
        error_msg = f"Automation failure: {str(e)}"
        print(f"[-] Error: {error_msg}")

    # 5. Clear proxy setting to keep device clean
    set_android_proxy(None, None)

    # 6. Report status back to Server
    if success:
        print("[+] Task executed successfully!")
        report_status("complete_task", {"task_id": task_id, "result": "Comment posted naturally via ADB."})
    else:
        print(f"[-] Task failed: {error_msg}")
        report_status("fail_task", {"task_id": task_id, "error_message": error_msg})

if __name__ == "__main__":
    print("[*] Starting Siro Guerrilla Marketing Automation Bot...")
    # Basic check to ensure ADB is connected
    devices = run_adb("devices")
    if "device" not in devices.split("\n")[1:]:
        print("[-] Critical Error: No Android device detected via ADB! Please connect device and enable USB Debugging.")
        sys.exit(1)
        
    print("[+] Android device connected. Starting automation loop...")
    while True:
        process_task()
        time.sleep(CHECK_INTERVAL_SECONDS)