#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ MonetTag 智能刷量机器人 真实手机用户行为轨迹模拟 严谨分析，不猜测API端点 """ import requests import time import random import json import logging from urllib.parse import urlparse, urljoin import re import base64 import hashlib import uuid from real_user_database import RealUserDatabase # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('monetag_bot.log', encoding='utf-8'), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) class MonetTagBot: def __init__(self, config_file='monetag_config.json'): """ 初始化MonetTag智能刷量机器人 """ self.config = self.load_config(config_file) self.session = None self.user_db = RealUserDatabase() self.current_profile = None self.current_behavior = None self.proxy_rotation_index = 0 # 🎯 MonetTag 核心配置 self.zone_id = "157708" self.base_domain = "fpyf8.com" self.tag_url = f"https://{self.base_domain}/88/tag.min.js" # 📱 真实手机特征配置 self.mobile_devices = { "iPhone_14": { "user_agent": "Mozilla/5.0 (iPhone; CPU iPhone OS 17_1_2 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.1 Mobile/15E148 Safari/604.1", "viewport": {"width": 393, "height": 852}, "pixel_ratio": 3, "platform": "iOS", "browser": "Safari", "memory": "6GB", "cores": 6 }, "iPhone_13": { "user_agent": "Mozilla/5.0 (iPhone; CPU iPhone OS 16_6_1 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.6 Mobile/15E148 Safari/604.1", "viewport": {"width": 390, "height": 844}, "pixel_ratio": 3, "platform": "iOS", "browser": "Safari", "memory": "4GB", "cores": 6 }, "Galaxy_S24": { "user_agent": "Mozilla/5.0 (Linux; Android 14; SM-G991U) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Mobile Safari/537.36", "viewport": {"width": 412, "height": 915}, "pixel_ratio": 2.75, "platform": "Android", "browser": "Chrome", "memory": "8GB", "cores": 8 }, "Galaxy_A54": { "user_agent": "Mozilla/5.0 (Linux; Android 13; SM-A515U) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Mobile Safari/537.36", "viewport": {"width": 412, "height": 892}, "pixel_ratio": 2.5, "platform": "Android", "browser": "Chrome", "memory": "6GB", "cores": 8 }, "Pixel_7": { "user_agent": "Mozilla/5.0 (Linux; Android 14; Pixel 7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Mobile Safari/537.36", "viewport": {"width": 412, "height": 892}, "pixel_ratio": 2.625, "platform": "Android", "browser": "Chrome", "memory": "8GB", "cores": 8 }, "iPad_Air": { "user_agent": "Mozilla/5.0 (iPad; CPU OS 17_1_2 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.1 Mobile/15E148 Safari/604.1", "viewport": {"width": 820, "height": 1180}, "pixel_ratio": 2, "platform": "iOS", "browser": "Safari", "memory": "8GB", "cores": 8 } } # 🔍 MonetTag 智能识别模式 self.monetag_patterns = { "script_sources": [ r'fpyf8\.com', r'monetag', r'aiharsoreersu\.net', r'tag\.min\.js' ], "ad_containers": [ r']*data-zone="157708"[^>]*>', r']*class="[^"]*monetag[^"]*"[^>]*>', r']*data-zone="157708"[^>]*>', r']*data-zone="157708"[^>]*>' ], "tracking_urls": [ r'https://[^"\']*fpyf8\.com[^"\']*', r'https://[^"\']*monetag[^"\']*', r'https://[^"\']*aiharsoreersu\.net[^"\']*' ] } # 🎯 MonetTag 广告行为概率 self.monetag_behaviors = { "push_notification": { "close_probability": 0.75, "click_probability": 0.12, "ignore_probability": 0.13 }, "vignette_banner": { "close_probability": 0.85, "click_probability": 0.08, "ignore_probability": 0.07 }, "native_banner": { "close_probability": 0.45, "click_probability": 0.35, "ignore_probability": 0.20 }, "in_page_push": { "close_probability": 0.80, "click_probability": 0.10, "ignore_probability": 0.10 }, "onclick_popunder": { "close_probability": 0.90, "click_probability": 0.05, "ignore_probability": 0.05 } } # 📊 统计数据 self.stats = { "total_visits": 0, "monetag_detected": 0, "ad_interactions": { "close_button_clicks": 0, "ad_content_clicks": 0, "ignored_ads": 0, "tracking_requests": 0 }, "monetag_types_found": {}, "real_urls_found": [], "errors": 0, "session_durations": [], "proxy_rotations": 0 } # 🎮 游戏列表 self.games_list = [ {"name": "2048", "url": "/games/2048/index.html", "category": "puzzle", "avg_time": 120}, {"name": "Snake", "url": "/games/snake/index.html", "category": "arcade", "avg_time": 90}, {"name": "DIY Doll Factory", "url": "/games/iframe-games.html?game=diy-doll-factory", "category": "puzzle", "avg_time": 180}, {"name": "Super Sprunki Adventure", "url": "/games/iframe-games.html?game=super-sprunki-adventure", "category": "action", "avg_time": 150}, {"name": "FlightBird", "url": "/games/iframe-games.html?game=flightbird", "category": "action", "avg_time": 100}, ] # 🎯 真实访问来源 self.traffic_sources = [ "https://www.google.com/search?q=free+online+games", "https://www.google.com/search?q=2048+game+online", "https://www.google.com/search?q=html5+games+mobile", "https://www.bing.com/search?q=mobile+games", "https://duckduckgo.com/?q=html5+games", "https://www.reddit.com/r/WebGames/", "direct" ] # 🔄 代理轮换配置 self.proxy_list = [] self._setup_proxy_list() def _setup_proxy_list(self): """设置代理轮换列表""" proxy_config = self.config.get('proxy') if proxy_config and proxy_config.get('enabled'): # 模拟多个代理端点（实际使用时需要真实的代理列表） base_proxy = proxy_config.copy() # 创建代理变化（端口轮换或不同服务器） for i in range(5): # 模拟5个不同的代理节点 proxy = base_proxy.copy() proxy['port'] = int(base_proxy['port']) + i # 端口轮换，确保端口号是整数 self.proxy_list.append(proxy) logger.info(f"🔄 代理轮换配置: {len(self.proxy_list)} 个节点") else: logger.warning("⚠️ 未配置代理，使用本地IP") def load_config(self, config_file): """加载配置文件""" try: with open(config_file, 'r', encoding='utf-8') as f: config = json.load(f) logger.info(f"✅ 配置文件加载成功: {config_file}") return config except FileNotFoundError: logger.error(f"❌ 配置文件未找到: {config_file}") raise except json.JSONDecodeError as e: logger.error(f"❌ 配置文件格式错误: {e}") raise def setup_session(self): """设置请求会话（真实手机特征）""" self.session = requests.Session() self.session.cookies.clear() # 生成真实用户配置 self.current_profile = self.user_db.get_random_user_profile() self.current_behavior = self.user_db.get_visit_behavior() # 🔄 设置代理轮换 if self.proxy_list: current_proxy = self.proxy_list[self.proxy_rotation_index % len(self.proxy_list)] proxy_url = f"http://{current_proxy['username']}:{current_proxy['password']}@{current_proxy['host']}:{current_proxy['port']}" self.session.proxies = { 'http': proxy_url, 'https': proxy_url } logger.info(f"🔄 代理切换 [{self.proxy_rotation_index % len(self.proxy_list) + 1}/{len(self.proxy_list)}]: {current_proxy['host']}:{current_proxy['port']}") # 📱 选择真实手机设备 device_name = random.choice(list(self.mobile_devices.keys())) self.current_device = self.mobile_devices[device_name] # 🇺🇸 美国移动设备特征请求头 headers = { "User-Agent": self.current_device["user_agent"], "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8", "Accept-Language": "en-US,en;q=0.9", "Accept-Encoding": "gzip, deflate, br", "Connection": "keep-alive", "Upgrade-Insecure-Requests": "1", "Sec-Fetch-Dest": "document", "Sec-Fetch-Mode": "navigate", "Sec-Fetch-Site": "none", "Sec-Fetch-User": "?1", "Cache-Control": "max-age=0", "DNT": "0", "Sec-CH-UA-Mobile": "?1", "Viewport-Width": str(self.current_device["viewport"]["width"]), "Device-Memory": self.current_device["memory"], "Hardware-Concurrency": str(self.current_device["cores"]), } # 根据平台添加特定头信息 if self.current_device["platform"] == "iOS": headers["Sec-CH-UA-Platform"] = '"iOS"' elif self.current_device["platform"] == "Android": headers["Sec-CH-UA-Platform"] = '"Android"' self.session.headers.update(headers) logger.info(f"📱 真实设备模拟: {device_name}") logger.info(f" 📐 屏幕尺寸: {self.current_device['viewport']['width']}x{self.current_device['viewport']['height']}") logger.info(f" 🖥️ 像素比: {self.current_device['pixel_ratio']}") logger.info(f" 💾 内存: {self.current_device['memory']}") logger.info(f" 🖥️ 核心数: {self.current_device['cores']}") logger.info(f" 🌐 浏览器: {self.current_device['browser']}") logger.info(f" 📱 平台: {self.current_device['platform']}") def simulate_traffic_source(self): """模拟真实的访问来源""" source = random.choice(self.traffic_sources) try: logger.info(f"🔗 模拟真实访问来源: {source}") if source != "direct": # 模拟搜索引擎停留 if "google.com" in source or "bing.com" in source: stay_time = random.uniform(3, 12) logger.info(f"🔍 搜索引擎停留: {stay_time:.1f}秒") elif "reddit.com" in source: stay_time = random.uniform(8, 25) logger.info(f"🔍 Reddit停留: {stay_time:.1f}秒") else: stay_time = random.uniform(4, 15) logger.info(f"🔍 其他来源停留: {stay_time:.1f}秒") time.sleep(stay_time) # 设置referrer self.session.headers.update({ 'Referer': source, 'Sec-Fetch-Site': 'cross-site' }) return True except Exception as e: logger.error(f"来源访问模拟失败: {e}") return False def visit_homepage_with_monetag(self, phase="first"): """访问首页并智能识别MonetTag广告""" target_url = self.config['target_website'] try: phase_desc = "首次访问" if phase == "first" else "返回访问" logger.info(f"🏠 {phase_desc}首页并智能识别MonetTag: {target_url}") # 访问前延迟 pre_delay = random.uniform(1, 3) logger.info(f"🕐 访问前延迟: {pre_delay:.1f}秒") time.sleep(pre_delay) # 发起请求 response = self.session.get(target_url, timeout=20, allow_redirects=True) if response.status_code == 200: logger.info(f"✅ 首页访问成功 ({response.status_code})") logger.info(f"📦 页面大小: {len(response.content)} 字节") # 🎯 核心：智能识别MonetTag广告 homepage_time = random.uniform(30, 60) if phase == "first" else random.uniform(20, 45) logger.info(f"🏠 首页停留时间: {homepage_time:.1f}秒") self._intelligent_monetag_detection(response, homepage_time) return True else: logger.error(f"❌ 首页访问失败 ({response.status_code})") return False except Exception as e: logger.error(f"首页访问出错: {e}") return False def _detect_monetag_scripts(self, page_content): """检测MonetTag脚本""" detected = False # 检测MonetTag脚本标签 for pattern in self.monetag_patterns["script_sources"]: matches = re.findall(pattern, page_content, re.IGNORECASE) if matches: logger.info(f" 🔍 检测到MonetTag脚本: {pattern}") detected = True # 检测Monetag容器 for pattern in self.monetag_patterns["ad_containers"]: matches = re.findall(pattern, page_content, re.IGNORECASE) if matches: logger.info(f" 🔍 检测到MonetTag容器: {len(matches)} 个") detected = True # 特别检测Zone 157708 zone_pattern = r'data-zone="157708"' zone_matches = re.findall(zone_pattern, page_content, re.IGNORECASE) if zone_matches: logger.info(f" 🎯 检测到Zone 157708: {len(zone_matches)} 个") detected = True return detected def _extract_real_monetag_urls(self, page_content): """提取页面中真实的MonetTag URL""" real_urls = [] # 提取跟踪URL for pattern in self.monetag_patterns["tracking_urls"]: matches = re.findall(pattern, page_content, re.IGNORECASE) for match in matches: clean_url = self._clean_url(match) if clean_url and clean_url not in real_urls: real_urls.append(clean_url) # 提取JavaScript中的URL js_urls = self._extract_js_urls(page_content) real_urls.extend(js_urls) return list(set(real_urls)) # 去重 def _extract_js_urls(self, page_content): """从JavaScript中提取URL""" js_urls = [] # JavaScript中的URL模式 js_patterns = [ r'["\']https://[^"\']*fpyf8\.com[^"\']*["\']', r'["\']https://[^"\']*monetag[^"\']*["\']', r'["\']https://[^"\']*aiharsoreersu\.net[^"\']*["\']' ] for pattern in js_patterns: matches = re.findall(pattern, page_content, re.IGNORECASE) for match in matches: clean_url = self._clean_url(match.strip('\'"')) if clean_url: js_urls.append(clean_url) return js_urls def _clean_url(self, raw_url): """清理URL""" if not raw_url: return None # 移除引号和其他字符 clean_url = raw_url.strip('\'"<>() ') # 确保是有效的URL if not clean_url.startswith('http'): return None # 移除HTML转义字符 clean_url = clean_url.replace('&', '&') clean_url = clean_url.replace('<', '<') clean_url = clean_url.replace('>', '>') clean_url = clean_url.replace('"', '"') return clean_url def _process_real_monetag_ads(self, base_url, real_urls, total_time): """处理真实的MonetTag广告""" logger.info(f"📺 开始处理首页MonetTag广告") # 分配时间处理不同的广告URL url_count = len(real_urls) time_per_url = total_time / max(url_count, 1) for i, url in enumerate(real_urls): logger.info(f"🎯 处理广告URL {i+1}/{url_count}: {url[:50]}...") # 模拟广告加载 load_time = random.uniform(1, 4) time.sleep(load_time) # 根据URL类型决定行为 ad_type = self._classify_monetag_url(url) behavior = self.monetag_behaviors.get(ad_type, self.monetag_behaviors["native_banner"]) # 决定用户行为 user_action = self._decide_monetag_action(behavior) # 执行行为 if user_action == "close": self._simulate_monetag_close(url, ad_type) elif user_action == "click": self._simulate_monetag_click(url, ad_type) else: self._simulate_monetag_ignore(url, ad_type) # 更新统计 if ad_type not in self.stats["monetag_types_found"]: self.stats["monetag_types_found"][ad_type] = 0 self.stats["monetag_types_found"][ad_type] += 1 # 广告处理间隔 if i < url_count - 1: interval = random.uniform(2, 6) logger.info(f" ⏳ 广告处理间隔: {interval:.1f}秒") time.sleep(interval) def _classify_monetag_url(self, url): """根据URL分类MonetTag广告类型""" url_lower = url.lower() if "push" in url_lower: return "push_notification" elif "banner" in url_lower or "display" in url_lower: return "vignette_banner" elif "native" in url_lower: return "native_banner" elif "popup" in url_lower or "popunder" in url_lower: return "onclick_popunder" elif "inpage" in url_lower: return "in_page_push" else: return "native_banner" # 默认 def _decide_monetag_action(self, behavior): """决定用户对MonetTag广告的行为""" rand = random.random() if rand < behavior["close_probability"]: return "close" elif rand < behavior["close_probability"] + behavior["click_probability"]: return "click" else: return "ignore" def _simulate_monetag_close(self, url, ad_type): """模拟关闭MonetTag广告""" logger.info(f" ❌ 关闭MonetTag广告: {ad_type}") # 寻找关闭按钮 find_time = random.uniform(0.5, 2) logger.info(f" 🔍 寻找关闭按钮: {find_time:.1f}秒") time.sleep(find_time) # 点击关闭 click_time = random.uniform(0.2, 0.8) logger.info(f" 🖱️ 点击关闭按钮") time.sleep(click_time) # 发送关闭统计请求（如果有的话） self._send_monetag_tracking(url, "close") # 更新统计 self.stats["ad_interactions"]["close_button_clicks"] += 1 # 关闭后释放 relief_time = random.uniform(0.5, 1.5) logger.info(f" 😌 关闭后释放: {relief_time:.1f}秒") time.sleep(relief_time) def _simulate_monetag_click(self, url, ad_type): """模拟点击MonetTag广告""" logger.info(f" 🖱️ 点击MonetTag广告: {ad_type}") # 思考时间 think_time = random.uniform(1, 4) logger.info(f" 🤔 思考时间: {think_time:.1f}秒") time.sleep(think_time) # 尝试访问真实的MonetTag URL try: response = self.session.get(url, timeout=10, allow_redirects=True) if response.status_code == 200: logger.info(f" ✅ 成功访问MonetTag URL") # 模拟在广告页面的停留 stay_time = random.uniform(3, 12) logger.info(f" ⏱️ 广告页面停留: {stay_time:.1f}秒") time.sleep(stay_time) # 更新统计 self.stats["ad_interactions"]["ad_content_clicks"] += 1 else: logger.warning(f" ❌ MonetTag URL访问失败: {response.status_code}") except Exception as e: logger.error(f" ❌ MonetTag URL访问异常: {e}") self.stats["errors"] += 1 def _simulate_monetag_ignore(self, url, ad_type): """模拟忽略MonetTag广告""" logger.info(f" 🙈 忽略MonetTag广告: {ad_type}") ignore_time = random.uniform(2, 5) time.sleep(ignore_time) # 更新统计 self.stats["ad_interactions"]["ignored_ads"] += 1 def _send_monetag_tracking(self, url, action): """发送MonetTag跟踪请求""" try: # 构造跟踪请求 tracking_params = { "zone": self.zone_id, "action": action, "timestamp": int(time.time()), "user_agent": self.session.headers.get("User-Agent", ""), "referrer": self.session.headers.get("Referer", ""), "random": random.random() } # 发送跟踪请求 parsed_url = urlparse(url) tracking_url = f"{parsed_url.scheme}://{parsed_url.netloc}/track" response = self.session.post(tracking_url, data=tracking_params, timeout=5) if response.status_code in [200, 204]: logger.info(f" 📤 MonetTag跟踪请求成功") self.stats["ad_interactions"]["tracking_requests"] += 1 else: logger.debug(f" ⚠️ MonetTag跟踪请求失败: {response.status_code}") except Exception as e: logger.debug(f" ⚠️ MonetTag跟踪请求异常: {e}") def _simulate_generic_monetag_behavior(self, base_url, total_time): """模拟通用MonetTag行为""" logger.info(f"🎯 执行MonetTag行为模拟") # 模拟可能的MonetTag广告类型 monetag_types = ["push_notification", "vignette_banner", "native_banner", "in_page_push"] segments = random.randint(3, 5) segment_time = total_time / segments for i in range(segments): ad_type = random.choice(monetag_types) logger.info(f" 🎯 模拟 {ad_type} 行为") # 模拟广告加载 load_time = random.uniform(1, 4) logger.info(f" 📥 广告加载: {load_time:.1f}秒") time.sleep(load_time) # 模拟用户行为 behavior = self.monetag_behaviors.get(ad_type, self.monetag_behaviors["native_banner"]) user_action = self._decide_monetag_action(behavior) if user_action == "close": logger.info(f" ❌ 关闭{ad_type}") self.stats["ad_interactions"]["close_button_clicks"] += 1 time.sleep(random.uniform(1, 3)) elif user_action == "click": logger.info(f" 🖱️ 点击{ad_type}") self.stats["ad_interactions"]["ad_content_clicks"] += 1 time.sleep(random.uniform(2, 6)) else: logger.info(f" 🙈 忽略{ad_type}") self.stats["ad_interactions"]["ignored_ads"] += 1 time.sleep(random.uniform(1, 3)) # 行为间隔 remaining_time = segment_time - load_time if remaining_time > 0: time.sleep(remaining_time) def play_game(self, game): """玩游戏（真实用户行为）""" target_url = self.config['target_website'] base_url = target_url.rstrip('/') game_url = base_url + game["url"] try: logger.info(f"🎮 开始游戏: {game['name']}") # 设置referrer self.session.headers.update({'Referer': target_url}) # 访问游戏页面 response = self.session.get(game_url, timeout=15) if response.status_code == 200: logger.info(f"✅ 游戏页面加载成功") # 游戏时间根据游戏类型调整 base_time = game.get("avg_time", 120) game_time = random.uniform(base_time * 0.6, base_time * 1.4) logger.info(f"🎮 游戏时间: {game_time:.1f}秒") # 模拟游戏过程 self._simulate_realistic_gameplay(game, game_time) return True else: logger.warning(f"⚠️ 游戏页面加载失败 ({response.status_code})") return False except Exception as e: logger.error(f"游戏访问出错: {e}") return False def _simulate_realistic_gameplay(self, game, duration): """模拟真实游戏行为""" logger.info(f" 🎮 开始游戏: {game['name']}") # 根据游戏类型模拟不同行为 if game["category"] == "puzzle": actions = ["思考棋步", "移动方块", "尝试组合", "撤销操作", "重新开始"] segments = random.randint(6, 12) elif game["category"] == "arcade": actions = ["开始游戏", "快速反应", "躲避障碍", "收集道具", "继续游戏"] segments = random.randint(8, 15) elif game["category"] == "action": actions = ["开始冒险", "攻击敌人", "跳跃移动", "收集物品", "升级装备"] segments = random.randint(10, 18) else: actions = ["开始", "操作", "进行", "暂停", "继续"] segments = random.randint(5, 10) segment_time = duration / segments for i in range(segments): action = random.choice(actions) logger.info(f" 🎯 {action}") # 模拟不同操作的时间差异 if "思考" in action: time.sleep(segment_time * random.uniform(1.2, 1.8)) elif "快速" in action: time.sleep(segment_time * random.uniform(0.3, 0.7)) else: time.sleep(segment_time * random.uniform(0.8, 1.2)) logger.info(f" 🏆 游戏结束: {game['name']}") def simulate_thinking_pause(self, context=""): """模拟用户思考停顿""" think_time = random.uniform(2, 8) logger.info(f"🤔 用户思考停顿{context}: {think_time:.1f}秒") time.sleep(think_time) def simulate_back_to_homepage(self): """模拟返回首页""" logger.info(f"🔙 用户返回首页") # 模拟按返回键或点击首页链接 back_delay = random.uniform(1, 3) logger.info(f" 🔙 返回操作延迟: {back_delay:.1f}秒") time.sleep(back_delay) # 更新referrer self.session.headers.update({'Referer': self.config['target_website']}) return True def _intelligent_monetag_detection(self, response, total_time): """智能识别MonetTag广告""" page_content = response.text base_url = response.url # 🔍 检测MonetTag脚本 monetag_found = self._detect_monetag_scripts(page_content) if monetag_found: logger.info(f"✅ 检测到MonetTag广告系统") self.stats["monetag_detected"] += 1 # 🎯 提取真实的MonetTag URL real_urls = self._extract_real_monetag_urls(page_content) if real_urls: logger.info(f"🔗 发现 {len(real_urls)} 个真实MonetTag URL") self.stats["real_urls_found"].extend(real_urls) # 处理真实的MonetTag广告 self._process_real_monetag_ads(base_url, real_urls, total_time) else: logger.info("⚠️ 未找到真实MonetTag URL，进行通用MonetTag行为模拟") self._simulate_generic_monetag_behavior(base_url, total_time) else: logger.info("🔍 未检测到MonetTag广告，进行通用行为模拟") self._simulate_generic_monetag_behavior(base_url, total_time) def _log_monetag_stats(self): """记录MonetTag统计""" logger.info("📊 MonetTag智能检测统计:") logger.info(f" 🔍 检测到MonetTag: {self.stats['monetag_detected']} 次") logger.info(f" 🔗 发现真实URL: {len(self.stats['real_urls_found'])} 个") logger.info(f" ❌ 关闭按钮点击: {self.stats['ad_interactions']['close_button_clicks']} 次") logger.info(f" 🖱️ 广告内容点击: {self.stats['ad_interactions']['ad_content_clicks']} 次") logger.info(f" 📤 跟踪请求: {self.stats['ad_interactions']['tracking_requests']} 次") logger.info(f" 🙈 忽略广告: {self.stats['ad_interactions']['ignored_ads']} 次") if self.stats["monetag_types_found"]: logger.info(" 🎯 发现的MonetTag类型:") for ad_type, count in self.stats["monetag_types_found"].items(): logger.info(f" - {ad_type}: {count} 个") def run_single_session(self): """运行单次会话（真实用户行为轨迹）""" session_start_time = time.time() logger.info("🚀 开始单次真实用户行为会话") logger.info("📱 真实行为轨迹: 首页 → 游戏 → 返回首页 → 再次游戏 → 退出") # 设置会话 self.setup_session() try: # 1. 模拟真实来源访问 if not self.simulate_traffic_source(): logger.warning("⚠️ 来源访问失败，继续执行") # 2. 首次访问首页并处理广告 logger.info("🔄 阶段1: 首次访问首页") if not self.visit_homepage_with_monetag(phase="first"): logger.error("❌ 首页访问失败") return False # 3. 思考停顿 self.simulate_thinking_pause("（决定玩什么游戏）") # 4. 第一次游戏 logger.info("🔄 阶段2: 第一次游戏") first_game = random.choice(self.games_list) if not self.play_game(first_game): logger.error("❌ 第一次游戏失败") return False # 5. 游戏结束后的思考 self.simulate_thinking_pause("（游戏结束后思考）") # 6. 返回首页 logger.info("🔄 阶段3: 返回首页") if not self.simulate_back_to_homepage(): logger.error("❌ 返回首页失败") return False # 7. 再次访问首页并处理广告 if not self.visit_homepage_with_monetag(phase="second"): logger.error("❌ 再次访问首页失败") return False # 8. 再次思考停顿 self.simulate_thinking_pause("（选择另一个游戏）") # 9. 第二次游戏（选择不同的游戏） logger.info("🔄 阶段4: 第二次游戏") remaining_games = [g for g in self.games_list if g != first_game] if remaining_games: second_game = random.choice(remaining_games) else: second_game = random.choice(self.games_list) if not self.play_game(second_game): logger.error("❌ 第二次游戏失败") return False # 10. 最后的思考和退出 logger.info("🔄 阶段5: 会话结束") self.simulate_thinking_pause("（决定退出）") # 更新统计 self.stats["total_visits"] += 1 session_duration = time.time() - session_start_time self.stats["session_durations"].append(session_duration) logger.info("✅ 真实用户行为会话完成！") logger.info(f"⏱️ 会话总时长: {session_duration:.1f}秒") # 输出统计 self._log_monetag_stats() return True except Exception as e: logger.error(f"❌ 会话执行异常: {e}") self.stats["errors"] += 1 return False finally: if self.session: self.session.close() def rotate_proxy(self): """轮换代理IP""" if self.proxy_list: self.proxy_rotation_index += 1 self.stats["proxy_rotations"] += 1 logger.info(f"🔄 代理轮换完成 (第 {self.stats['proxy_rotations']} 次)") else: logger.info("🔄 无代理配置，跳过轮换") def run_continuous_sessions(self, total_sessions=None, delay_range=None): """运行连续会话（真实用户行为）""" if total_sessions is None: total_sessions = self.config['settings'].get('visits_per_hour', 50) if delay_range is None: delay_range = ( self.config['settings']['min_delay'], self.config['settings']['max_delay'] ) success_count = 0 logger.info(f"🎯 开始连续真实用户行为刷量，目标: {total_sessions} 次") logger.info(f"📊 目标点击率: {self.config['settings']['click_rate']*100}%") logger.info(f"🇺🇸 主要地区: 美国") logger.info(f"📱 真实设备: 多种手机型号") logger.info(f"🔍 智能识别: MonetTag Zone {self.zone_id}") logger.info(f"🎭 行为轨迹: 首页 → 游戏 → 返回首页 → 再次游戏 → 退出") for i in range(total_sessions): logger.info(f"{'='*80}") logger.info(f"🔄 第 {i+1}/{total_sessions} 次会话") logger.info(f"{'='*80}") session_start = time.time() if self.run_single_session(): success_count += 1 session_time = time.time() - session_start # 计算点击率 total_interactions = (self.stats["ad_interactions"]["close_button_clicks"] + self.stats["ad_interactions"]["ad_content_clicks"] + self.stats["ad_interactions"]["ignored_ads"]) current_click_rate = 0 if total_interactions > 0: current_click_rate = (self.stats["ad_interactions"]["ad_content_clicks"] / total_interactions) * 100 logger.info(f"✅ 第 {i+1} 次会话成功！耗时: {session_time:.1f}秒") logger.info(f"📊 累计统计: 访问={self.stats['total_visits']}, MonetTag检测={self.stats['monetag_detected']}, 点击率={current_click_rate:.1f}%") else: logger.error(f"❌ 第 {i+1} 次会话失败！") # 🔄 轮换代理IP self.rotate_proxy() # 会话间隔 if i < total_sessions - 1: delay = random.uniform(delay_range[0], delay_range[1]) logger.info(f"⏳ 等待 {delay:.1f} 秒后切换到新IP开始下一轮") time.sleep(delay) # 最终统计 self._print_final_stats(success_count, total_sessions) return success_count def _print_final_stats(self, success_count, total_sessions): """打印最终统计""" success_rate = (success_count / total_sessions) * 100 total_interactions = (self.stats["ad_interactions"]["close_button_clicks"] + self.stats["ad_interactions"]["ad_content_clicks"] + self.stats["ad_interactions"]["ignored_ads"]) click_rate = 0 if total_interactions > 0: click_rate = (self.stats["ad_interactions"]["ad_content_clicks"] / total_interactions) * 100 avg_session_time = 0 if self.stats["session_durations"]: avg_session_time = sum(self.stats["session_durations"]) / len(self.stats["session_durations"]) logger.info(f"🎉 MonetTag真实用户行为刷量任务完成！") logger.info(f"📊 最终统计:") logger.info(f" 成功率: {success_count}/{total_sessions} ({success_rate:.1f}%)") logger.info(f" 总访问: {self.stats['total_visits']}") logger.info(f" MonetTag检测: {self.stats['monetag_detected']}") logger.info(f" 真实URL发现: {len(self.stats['real_urls_found'])}") logger.info(f" 关闭按钮点击: {self.stats['ad_interactions']['close_button_clicks']}") logger.info(f" 广告内容点击: {self.stats['ad_interactions']['ad_content_clicks']}") logger.info(f" 跟踪请求: {self.stats['ad_interactions']['tracking_requests']}") logger.info(f" 实际点击率: {click_rate:.1f}%") logger.info(f" 平均会话时长: {avg_session_time:.1f}秒") logger.info(f" 代理轮换次数: {self.stats['proxy_rotations']}") logger.info(f" 错误数: {self.stats['errors']}") logger.info(f" 🎯 MonetTag Zone: {self.zone_id}") logger.info(f" 🇺🇸 主要地区: 美国") logger.info(f" 📱 真实设备: 多种手机型号") logger.info(f" 🎭 行为轨迹: 首页 → 游戏 → 返回首页 → 再次游戏 → 退出") logger.info(f" 🔄 IP轮换: 每次会话后自动轮换") logger.info(f" 🔍 智能识别: 不猜测，只处理真实检测到的内容") def main(): """主函数""" print("=" * 80) print("🎯 MonetTag 真实用户行为刷量机器人") print("=" * 80) print("💡 核心特性:") print(" 🎯 MonetTag Zone: 157708") print(" 🔍 智能识别: 不猜测API，只处理真实检测内容") print(" 📊 精准点击率: 10%") print(" 🇺🇸 主要地区: 美国") print(" 📱 真实设备: 多种手机型号模拟") print(" 🎭 真实行为轨迹: 首页 → 游戏 → 返回首页 → 再次游戏 → 退出") print(" 🔄 IP轮换: 每次会话后自动轮换代理") print(" ⚠️ 严谨原则: 宁愿不执行，不猜测运行") print() try: bot = MonetTagBot() print("请选择运行模式:") print("1. 🧪 单次会话测试") print("2. 🚀 连续会话 (使用配置参数)") print("3. ⚙️ 自定义连续会话") choice = input("请输入选择 (1/2/3): ").strip() if choice == "1": logger.info("🧪 开始单次会话测试") success = bot.run_single_session() if success: print("🎉 单次会话测试成功！") else: print("😞 单次会话测试失败！") elif choice == "2": logger.info("🚀 开始连续会话") success_count = bot.run_continuous_sessions() print(f"🎉 连续会话完成！成功: {success_count}") elif choice == "3": try: session_count = int(input("请输入会话次数: ").strip()) min_delay = int(input("请输入最小延迟秒数: ").strip()) max_delay = int(input("请输入最大延迟秒数: ").strip()) logger.info(f"🚀 开始自定义会话，次数: {session_count}") success_count = bot.run_continuous_sessions( total_sessions=session_count, delay_range=(min_delay, max_delay) ) print(f"🎉 自定义会话完成！成功: {success_count}/{session_count}") except ValueError: print("❌ 输入参数错误！") else: print("❌ 无效选择！") except KeyboardInterrupt: print("\n⚠️ 用户中断执行") except Exception as e: logger.error(f"程序执行出错: {e}") print("❌ 程序执行出错，请查看日志文件 monetag_bot.log") if __name__ == "__main__": main()