#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 增强版广告刷量机器人 (真实用户行为版) 专门用于与多种广告类型进行真实交互 包括模拟点击关闭按钮的真实用户行为 """ import requests import time import random import json import os import logging from urllib.parse import urlparse, urljoin import re from real_user_database import RealUserDatabase # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('enhanced_ad_bot.log', encoding='utf-8'), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) class EnhancedAdBot: def __init__(self, config_file='config.json'): """ 初始化增强版广告机器人 """ self.config = self.load_config(config_file) self.session = None self.user_db = RealUserDatabase() self.current_profile = None self.current_behavior = None # 🎯 广告类型和真实行为概率 self.ad_behaviors = { "popup": { "close_probability": 0.75, # 75%概率关闭弹窗 "click_probability": 0.15, # 15%概率点击广告 "ignore_probability": 0.10, # 10%概率忽略 "stay_time": (2, 8), # 停留时间范围 "annoyance_level": 0.8 # 用户讨厌程度 }, "interstitial": { "close_probability": 0.60, # 60%概率关闭插页 "click_probability": 0.25, # 25%概率点击广告 "wait_probability": 0.15, # 15%概率等待倒计时 "stay_time": (5, 15), # 停留时间范围 "annoyance_level": 0.6 # 用户讨厌程度 }, "banner": { "close_probability": 0.85, # 85%概率关闭横幅 "click_probability": 0.10, # 10%概率点击广告 "ignore_probability": 0.05, # 5%概率忽略 "stay_time": (1, 4), # 停留时间范围 "annoyance_level": 0.9 # 用户讨厌程度 }, "native": { "close_probability": 0.45, # 45%概率关闭原生广告 "click_probability": 0.35, # 35%概率点击广告 "read_probability": 0.20, # 20%概率仔细阅读 "stay_time": (3, 12), # 停留时间范围 "annoyance_level": 0.3 # 用户讨厌程度较低 }, "side": { "close_probability": 0.70, # 70%概率关闭侧边广告 "click_probability": 0.20, # 20%概率点击广告 "ignore_probability": 0.10, # 10%概率忽略 "stay_time": (1, 5), # 停留时间范围 "annoyance_level": 0.7 # 用户讨厌程度 }, "bottom": { "close_probability": 0.80, # 80%概率关闭底部广告 "click_probability": 0.12, # 12%概率点击广告 "ignore_probability": 0.08, # 8%概率忽略 "stay_time": (1, 3), # 停留时间范围 "annoyance_level": 0.8 # 用户讨厌程度 } } # 🎮 游戏列表 self.games_list = [ {"name": "2048", "url": "/games/2048/index.html", "category": "puzzle"}, {"name": "Snake", "url": "/games/snake/index.html", "category": "arcade"}, {"name": "DIY Doll Factory", "url": "/games/iframe-games.html?game=diy-doll-factory", "category": "puzzle"}, {"name": "Super Sprunki Adventure", "url": "/games/iframe-games.html?game=super-sprunki-adventure", "category": "action"}, {"name": "FlightBird", "url": "/games/iframe-games.html?game=flightbird", "category": "action"}, ] # 🎯 真实的访问来源 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.google.com/search?q=在线小游戏", "https://www.google.com/search?q=免费网页游戏", "https://www.baidu.com/s?wd=2048数字游戏", "https://www.reddit.com/r/WebGames/", "https://github.com/topics/html5-game", "https://news.ycombinator.com/", ] # 📊 访问统计 self.visit_stats = { "total_visits": 0, "ad_interactions": { "popup": {"shows": 0, "closes": 0, "clicks": 0}, "interstitial": {"shows": 0, "closes": 0, "clicks": 0}, "banner": {"shows": 0, "closes": 0, "clicks": 0}, "native": {"shows": 0, "closes": 0, "clicks": 0}, "side": {"shows": 0, "closes": 0, "clicks": 0}, "bottom": {"shows": 0, "closes": 0, "clicks": 0} }, "total_ad_time": 0, "close_button_clicks": 0, "ad_content_clicks": 0 } 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() # 设置代理 proxy_config = self.config.get('proxy') if proxy_config: proxy_url = f"http://{proxy_config['username']}:{proxy_config['password']}@{proxy_config['host']}:{proxy_config['port']}" self.session.proxies = { 'http': proxy_url, 'https': proxy_url } logger.info(f"已配置代理: {proxy_config['host']}:{proxy_config['port']}") # 🎯 真实用户代理 - 优先移动端 if random.random() < 0.85: # 85%概率使用移动端 mobile_agents = [ "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", "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", "Mozilla/5.0 (Linux; Android 14; SM-G991B) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Mobile Safari/537.36", "Mozilla/5.0 (Linux; Android 13; SM-A515F) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Mobile Safari/537.36", "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", ] user_agent = random.choice(mobile_agents) is_mobile = True else: user_agent = self.current_profile["user_agent"] is_mobile = "Mobile" in user_agent # 广告友好的请求头 headers = { "User-Agent": user_agent, "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8", "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", "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", # 不启用DNT，对广告友好 } if is_mobile: headers.update({ "Sec-CH-UA-Mobile": "?1", "Viewport-Width": str(random.choice([375, 414, 390, 393, 412])), }) self.session.headers.update(headers) logger.info(f"🎭 增强版广告机器人配置:") logger.info(f" 📱 移动端: {is_mobile}") logger.info(f" 🎯 访问模式: {self.current_behavior['pattern_type']}") logger.info(f" 😤 用户讨厌广告概率: 75%") def simulate_traffic_source(self): """模拟真实的访问来源""" source = random.choice(self.traffic_sources) try: logger.info(f"🔗 模拟从来源访问: {source}") # 根据来源类型设置不同的行为 if "google.com" in source or "baidu.com" in source: stay_time = random.uniform(2, 8) logger.info(f"🔍 搜索引擎停留: {stay_time:.1f}秒") elif "reddit.com" in source: stay_time = random.uniform(8, 20) logger.info(f"🔍 Reddit停留: {stay_time:.1f}秒") elif "github.com" in source: stay_time = random.uniform(5, 15) logger.info(f"🔍 GitHub停留: {stay_time:.1f}秒") else: stay_time = random.uniform(3, 12) logger.info(f"🔍 其他来源停留: {stay_time:.1f}秒") # 模拟来源网站行为 self._simulate_source_behavior(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 _simulate_source_behavior(self, total_time): """模拟在来源网站的行为""" actions = ["浏览搜索结果", "阅读页面内容", "查看相关链接", "滚动浏览页面"] segments = random.randint(2, 4) segment_time = total_time / segments for i in range(segments): if i > 0: time.sleep(random.uniform(0.5, 2)) action = random.choice(actions) logger.info(f" 🔍 {action}") time.sleep(segment_time * random.uniform(0.7, 1.3)) def visit_homepage_with_enhanced_ads(self): """访问增强版首页并进行真实广告交互""" main_site = self.config['targets']['main_site'] try: logger.info(f"🏠 访问增强版首页: {main_site}") # 访问前延迟 pre_delay = random.uniform(2, 5) logger.info(f"🕐 访问前延迟: {pre_delay:.1f}秒") time.sleep(pre_delay) # 发起请求 response = self.session.get(main_site, timeout=20, allow_redirects=True) if response.status_code == 200: logger.info(f"✅ 首页访问成功 ({response.status_code})") logger.info(f"📦 页面大小: {len(response.content)} 字节") # 🎯 核心：模拟真实的多广告交互 self._simulate_enhanced_ad_interactions(response) return True else: logger.error(f"❌ 首页访问失败 ({response.status_code})") return False except Exception as e: logger.error(f"首页访问出错: {e}") return False def _simulate_enhanced_ad_interactions(self, response): """模拟增强版广告交互 - 真实用户行为""" base_url = response.url # 首页总停留时间 homepage_stay_time = random.uniform(25, 60) # 25-60秒 logger.info(f"🏠 首页总停留时间: {homepage_stay_time:.1f}秒") # 阶段性广告交互 ad_interaction_phases = [ {"name": "页面加载阶段", "duration": 0.15, "ads": ["side", "bottom"]}, {"name": "内容浏览阶段", "duration": 0.25, "ads": ["banner", "native"]}, {"name": "弹窗广告阶段", "duration": 0.30, "ads": ["popup", "interstitial"]}, {"name": "深度浏览阶段", "duration": 0.30, "ads": ["all"]} ] for phase in ad_interaction_phases: phase_time = homepage_stay_time * phase["duration"] logger.info(f"🎯 {phase['name']}: {phase_time:.1f}秒") # 在每个阶段处理对应的广告 if phase["ads"] == ["all"]: # 处理所有剩余广告 self._handle_all_remaining_ads(base_url, phase_time) else: # 处理特定广告类型 for ad_type in phase["ads"]: self._handle_specific_ad_type(base_url, ad_type, phase_time / len(phase["ads"])) # 📊 输出广告交互统计 self._log_ad_interaction_stats() def _handle_specific_ad_type(self, base_url, ad_type, allocated_time): """处理特定类型的广告""" if ad_type not in self.ad_behaviors: return behavior = self.ad_behaviors[ad_type] logger.info(f" 📺 处理{ad_type}广告 (分配时间: {allocated_time:.1f}秒)") # 记录广告显示 self.visit_stats["ad_interactions"][ad_type]["shows"] += 1 # 模拟广告显示延迟 display_delay = random.uniform(0.5, 2.5) logger.info(f" 🔄 广告显示延迟: {display_delay:.1f}秒") time.sleep(display_delay) # 模拟用户看到广告的反应时间 reaction_time = random.uniform(1, 3) logger.info(f" 👀 用户反应时间: {reaction_time:.1f}秒") time.sleep(reaction_time) # 根据广告类型和用户心理决定行为 user_action = self._decide_user_action(ad_type, behavior) # 执行用户行为 self._execute_user_action(base_url, ad_type, user_action, allocated_time - display_delay - reaction_time) def _decide_user_action(self, ad_type, behavior): """基于用户心理决定对广告的行为""" # 考虑用户的讨厌程度 annoyance_factor = behavior["annoyance_level"] # 考虑当前时间段（用户在不同时间对广告的容忍度不同） time_factor = self._get_time_tolerance_factor() # 综合讨厌程度 final_annoyance = min(annoyance_factor * time_factor, 1.0) # 随机数决定行为 rand = random.random() if rand < behavior["close_probability"] * final_annoyance: return "close" elif rand < behavior["close_probability"] * final_annoyance + behavior["click_probability"]: return "click" elif hasattr(behavior, "read_probability") and rand < behavior["close_probability"] * final_annoyance + behavior["click_probability"] + behavior.get("read_probability", 0): return "read" elif hasattr(behavior, "wait_probability") and rand < behavior["close_probability"] * final_annoyance + behavior["click_probability"] + behavior.get("wait_probability", 0): return "wait" else: return "ignore" def _get_time_tolerance_factor(self): """获取时间容忍度因子""" current_hour = time.localtime().tm_hour if 9 <= current_hour <= 18: # 工作时间 return 1.2 # 更讨厌广告 elif 19 <= current_hour <= 23: # 休闲时间 return 0.8 # 相对容忍 else: # 深夜 return 1.0 # 普通 def _execute_user_action(self, base_url, ad_type, action, remaining_time): """执行用户对广告的行为""" logger.info(f" 🎯 用户行为: {action}") if action == "close": self._simulate_close_ad(base_url, ad_type, remaining_time) elif action == "click": self._simulate_click_ad(base_url, ad_type, remaining_time) elif action == "read": self._simulate_read_ad(base_url, ad_type, remaining_time) elif action == "wait": self._simulate_wait_ad(base_url, ad_type, remaining_time) else: # ignore self._simulate_ignore_ad(base_url, ad_type, remaining_time) def _simulate_close_ad(self, base_url, ad_type, time_budget): """模拟点击关闭广告按钮 - 真实用户行为""" logger.info(f" ❌ 关闭{ad_type}广告") # 模拟寻找关闭按钮的时间 find_close_time = random.uniform(0.5, 2.5) logger.info(f" 🔍 寻找关闭按钮: {find_close_time:.1f}秒") time.sleep(find_close_time) # 模拟点击关闭按钮 click_delay = random.uniform(0.2, 0.8) logger.info(f" 🖱️ 点击关闭按钮") time.sleep(click_delay) # 模拟关闭按钮点击请求 self._send_close_ad_request(base_url, ad_type) # 记录统计 self.visit_stats["ad_interactions"][ad_type]["closes"] += 1 self.visit_stats["close_button_clicks"] += 1 # 关闭后的心理释放时间 relief_time = random.uniform(0.5, 1.5) logger.info(f" 😌 关闭后释放: {relief_time:.1f}秒") time.sleep(relief_time) def _simulate_click_ad(self, base_url, ad_type, time_budget): """模拟点击广告内容""" logger.info(f" 🖱️ 点击{ad_type}广告内容") # 模拟思考时间 think_time = random.uniform(2, 5) logger.info(f" 🤔 思考时间: {think_time:.1f}秒") time.sleep(think_time) # 模拟点击广告 click_delay = random.uniform(0.3, 1.0) time.sleep(click_delay) # 模拟广告点击请求 self._send_ad_click_request(base_url, ad_type) # 记录统计 self.visit_stats["ad_interactions"][ad_type]["clicks"] += 1 self.visit_stats["ad_content_clicks"] += 1 # 点击后可能的停留时间 if random.random() < 0.7: # 70%概率会快速返回 return_time = random.uniform(1, 3) logger.info(f" 🔙 快速返回: {return_time:.1f}秒") time.sleep(return_time) def _simulate_read_ad(self, base_url, ad_type, time_budget): """模拟仔细阅读广告""" logger.info(f" 📖 仔细阅读{ad_type}广告") # 模拟阅读时间 read_time = random.uniform(5, 12) logger.info(f" 📚 阅读时间: {read_time:.1f}秒") time.sleep(read_time) # 阅读后可能的行为 if random.random() < 0.3: # 30%概率阅读后点击 self._simulate_click_ad(base_url, ad_type, time_budget - read_time) else: # 阅读后离开 leave_time = random.uniform(1, 2) time.sleep(leave_time) def _simulate_wait_ad(self, base_url, ad_type, time_budget): """模拟等待广告倒计时""" logger.info(f" ⏳ 等待{ad_type}广告倒计时") # 模拟等待倒计时时间 wait_time = random.uniform(5, 8) logger.info(f" ⏰ 等待倒计时: {wait_time:.1f}秒") time.sleep(wait_time) # 倒计时结束后的行为 if random.random() < 0.6: # 60%概率倒计时结束后关闭 self._simulate_close_ad(base_url, ad_type, time_budget - wait_time) def _simulate_ignore_ad(self, base_url, ad_type, time_budget): """模拟忽略广告""" logger.info(f" 🙈 忽略{ad_type}广告") # 模拟忽略时间 ignore_time = min(time_budget, random.uniform(2, 6)) logger.info(f" 😐 忽略时间: {ignore_time:.1f}秒") time.sleep(ignore_time) def _send_close_ad_request(self, base_url, ad_type): """发送关闭广告的请求""" try: # 模拟关闭广告的API请求 close_endpoints = [ f"/api/ads/{ad_type}/close", f"/ads/close?type={ad_type}", f"/tracking/ad_close/{ad_type}", f"/analytics/ad_interaction?action=close&type={ad_type}" ] endpoint = random.choice(close_endpoints) parsed_url = urlparse(base_url) close_url = f"{parsed_url.scheme}://{parsed_url.netloc}{endpoint}" headers = self.session.headers.copy() headers.update({ "X-Requested-With": "XMLHttpRequest", "Content-Type": "application/json", "Referer": base_url, }) response = self.session.post(close_url, headers=headers, json={"ad_type": ad_type, "action": "close"}, timeout=5) logger.info(f" 📤 关闭广告请求: {endpoint} ({response.status_code})") except Exception as e: logger.debug(f" ⚠️ 关闭广告请求失败: {e}") def _send_ad_click_request(self, base_url, ad_type): """发送广告点击请求""" try: # 模拟广告点击的API请求 click_endpoints = [ f"/api/ads/{ad_type}/click", f"/ads/click?type={ad_type}", f"/tracking/ad_click/{ad_type}", f"/analytics/ad_interaction?action=click&type={ad_type}" ] endpoint = random.choice(click_endpoints) parsed_url = urlparse(base_url) click_url = f"{parsed_url.scheme}://{parsed_url.netloc}{endpoint}" headers = self.session.headers.copy() headers.update({ "X-Requested-With": "XMLHttpRequest", "Content-Type": "application/json", "Referer": base_url, }) response = self.session.post(click_url, headers=headers, json={"ad_type": ad_type, "action": "click"}, timeout=5) logger.info(f" 📤 广告点击请求: {endpoint} ({response.status_code})") except Exception as e: logger.debug(f" ⚠️ 广告点击请求失败: {e}") def _handle_all_remaining_ads(self, base_url, phase_time): """处理所有剩余的广告""" logger.info(f" 📺 处理所有剩余广告") # 为每种广告类型分配时间 ad_types = list(self.ad_behaviors.keys()) time_per_ad = phase_time / len(ad_types) for ad_type in ad_types: if random.random() < 0.8: # 80%概率处理该广告 self._handle_specific_ad_type(base_url, ad_type, time_per_ad) def _log_ad_interaction_stats(self): """记录广告交互统计""" logger.info("📊 广告交互统计:") total_shows = 0 total_closes = 0 total_clicks = 0 for ad_type, stats in self.visit_stats["ad_interactions"].items(): if stats["shows"] > 0: close_rate = (stats["closes"] / stats["shows"]) * 100 click_rate = (stats["clicks"] / stats["shows"]) * 100 logger.info(f" {ad_type}: 显示{stats['shows']}次, 关闭{stats['closes']}次({close_rate:.1f}%), 点击{stats['clicks']}次({click_rate:.1f}%)") total_shows += stats["shows"] total_closes += stats["closes"] total_clicks += stats["clicks"] if total_shows > 0: overall_close_rate = (total_closes / total_shows) * 100 overall_click_rate = (total_clicks / total_shows) * 100 logger.info(f" 总计: 显示{total_shows}次, 关闭{total_closes}次({overall_close_rate:.1f}%), 点击{total_clicks}次({overall_click_rate:.1f}%)") logger.info(f" 关闭按钮点击: {self.visit_stats['close_button_clicks']}次") logger.info(f" 广告内容点击: {self.visit_stats['ad_content_clicks']}次") def browse_games_with_ads(self): """浏览游戏页面并处理广告""" main_site = self.config['targets']['main_site'] base_url = main_site.rstrip('/') games_to_browse = random.randint(2, 4) selected_games = random.sample(self.games_list, min(games_to_browse, len(self.games_list))) logger.info(f"🎮 浏览 {games_to_browse} 个游戏页面") for i, game in enumerate(selected_games): game_url = base_url + game["url"] try: logger.info(f"🎯 游戏 {i+1}/{games_to_browse}: {game['name']}") # 设置referrer if i == 0: self.session.headers.update({'Referer': main_site}) else: prev_game_url = base_url + selected_games[i-1]["url"] self.session.headers.update({'Referer': prev_game_url}) # 访问游戏页面 response = self.session.get(game_url, timeout=15) if response.status_code == 200: logger.info(f"✅ 游戏页面访问成功") # 模拟游戏页面的广告交互 self._simulate_game_page_ads(response, game) else: logger.warning(f"⚠️ 游戏页面访问失败 ({response.status_code})") # 页面间延迟 if i < len(selected_games) - 1: delay = random.uniform(3, 8) logger.info(f" ⏳ 页面切换延迟: {delay:.1f}秒") time.sleep(delay) except Exception as e: logger.error(f"游戏页面访问出错: {e}") continue return True def _simulate_game_page_ads(self, response, game): """模拟游戏页面的广告交互""" game_stay_time = random.uniform(30, 90) logger.info(f" 🎮 游戏页面停留: {game_stay_time:.1f}秒") # 分配时间给不同的活动 game_time = game_stay_time * 0.7 # 70%时间玩游戏 ad_time = game_stay_time * 0.3 # 30%时间处理广告 logger.info(f" 🎯 游戏时间: {game_time:.1f}秒") logger.info(f" 💰 广告时间: {ad_time:.1f}秒") # 模拟游戏前广告 if random.random() < 0.8: # 80%概率有游戏前广告 pre_ad_time = random.uniform(5, 15) logger.info(f" 📺 游戏前广告: {pre_ad_time:.1f}秒") # 真实用户行为：大多数人讨厌游戏前广告 if random.random() < 0.7: # 70%概率跳过或关闭 logger.info(f" ⏭️ 用户厌烦，跳过广告") time.sleep(random.uniform(2, 5)) else: logger.info(f" ✅ 用户耐心等待广告") time.sleep(pre_ad_time) # 模拟游戏过程 self._simulate_game_play(game, game_time) # 模拟游戏中广告 if random.random() < 0.6: # 60%概率有游戏中广告 mid_ad_time = random.uniform(3, 10) logger.info(f" 📺 游戏中广告: {mid_ad_time:.1f}秒") # 游戏中广告更容易被关闭 if random.random() < 0.8: # 80%概率关闭 logger.info(f" ❌ 关闭游戏中广告") time.sleep(random.uniform(1, 3)) else: time.sleep(mid_ad_time) # 模拟游戏结束广告 if random.random() < 0.9: # 90%概率有结束广告 end_ad_time = random.uniform(5, 20) logger.info(f" 📺 游戏结束广告: {end_ad_time:.1f}秒") # 结束广告的处理较为复杂 if random.random() < 0.5: # 50%概率立即关闭 logger.info(f" ❌ 立即关闭结束广告") time.sleep(random.uniform(1, 2)) elif random.random() < 0.3: # 30%概率点击 logger.info(f" 🖱️ 点击结束广告") time.sleep(random.uniform(2, 5)) else: # 20%概率看完 logger.info(f" ✅ 看完结束广告") time.sleep(end_ad_time) def _simulate_game_play(self, game, duration): """模拟游戏过程""" logger.info(f" 🎮 开始游戏: {game['name']}") # 根据游戏类型模拟不同行为 if game["category"] == "puzzle": actions = ["思考", "移动", "尝试", "撤销"] segments = random.randint(4, 8) elif game["category"] == "arcade": actions = ["开始", "快速操作", "躲避", "收集"] segments = random.randint(6, 12) else: actions = ["开始", "操作", "进行", "暂停"] segments = random.randint(3, 6) segment_time = duration / segments for i in range(segments): action = random.choice(actions) logger.info(f" 🎯 {action}") time.sleep(segment_time * random.uniform(0.8, 1.2)) def run_enhanced_ad_visit(self): """执行增强版广告访问流程""" logger.info("🚀 开始增强版广告访问流程") # 设置会话 self.setup_session() try: # 1. 模拟来源访问 if not self.simulate_traffic_source(): logger.warning("⚠️ 来源访问失败，继续执行") # 2. 访问增强版首页 if not self.visit_homepage_with_enhanced_ads(): logger.error("❌ 首页广告访问失败") return False # 3. 浏览游戏页面 if not self.browse_games_with_ads(): logger.error("❌ 游戏页面访问失败") return False # 更新统计 self.visit_stats["total_visits"] += 1 logger.info("✅ 增强版广告访问流程完成！") return True except Exception as e: logger.error(f"❌ 访问流程出错: {e}") return False finally: if self.session: self.session.close() def run_continuous_enhanced_visits(self, total_visits=None, delay_range=None): """连续执行增强版广告访问""" if total_visits is None: total_visits = self.config['settings']['default_visits'] if delay_range is None: delay_range = ( self.config['settings']['min_delay'], self.config['settings']['max_delay'] ) success_count = 0 logger.info(f"🎯 开始连续增强版广告访问，目标: {total_visits} 次") for i in range(total_visits): logger.info(f"{'='*70}") logger.info(f"🔄 第 {i+1}/{total_visits} 次增强版广告访问") logger.info(f"{'='*70}") visit_start_time = time.time() if self.run_enhanced_ad_visit(): success_count += 1 visit_time = time.time() - visit_start_time logger.info(f"✅ 第 {i+1} 次访问成功！耗时: {visit_time:.1f}秒") # 显示累计统计 total_ad_shows = sum(stats["shows"] for stats in self.visit_stats["ad_interactions"].values()) logger.info(f"📊 累计: 成功{success_count}, 广告显示{total_ad_shows}, 关闭{self.visit_stats['close_button_clicks']}, 点击{self.visit_stats['ad_content_clicks']}") else: logger.error(f"❌ 第 {i+1} 次访问失败！") # 访问间隔 if i < total_visits - 1: delay = random.uniform(delay_range[0], delay_range[1]) logger.info(f"⏳ 等待 {delay:.1f} 秒") time.sleep(delay) # 最终统计 self._print_final_stats(success_count, total_visits) return success_count def _print_final_stats(self, success_count, total_visits): """打印最终统计信息""" success_rate = (success_count / total_visits) * 100 total_ad_shows = sum(stats["shows"] for stats in self.visit_stats["ad_interactions"].values()) total_ad_closes = sum(stats["closes"] for stats in self.visit_stats["ad_interactions"].values()) total_ad_clicks = sum(stats["clicks"] for stats in self.visit_stats["ad_interactions"].values()) logger.info(f"🎉 增强版广告访问任务完成！") logger.info(f"📊 访问统计:") logger.info(f" 成功率: {success_count}/{total_visits} ({success_rate:.1f}%)") logger.info(f" 广告显示总数: {total_ad_shows}") logger.info(f" 关闭按钮点击: {self.visit_stats['close_button_clicks']} ({(self.visit_stats['close_button_clicks']/total_ad_shows*100):.1f}%)") logger.info(f" 广告内容点击: {self.visit_stats['ad_content_clicks']} ({(self.visit_stats['ad_content_clicks']/total_ad_shows*100):.1f}%)") logger.info(f"💡 真实用户行为: 75%的用户讨厌广告并选择关闭") def main(): """主函数""" config_file = 'config.json' if not os.path.exists(config_file): print(f"❌ 配置文件 {config_file} 不存在！") return try: bot = EnhancedAdBot(config_file) print("=" * 80) print("🎯 增强版广告刷量机器人 (真实用户行为版)") print("=" * 80) print("💡 核心特性:") print(" 🎭 真实用户行为模拟 - 75%概率讨厌广告") print(" ❌ 智能关闭广告按钮 - 模拟真实用户习惯") print(" 📱 移动端优先 - 85%概率使用移动设备") print(" 🎯 6种广告类型 - 弹窗、插页、横幅、原生、侧边、底部") print(" 📊 详细交互统计 - 关闭率、点击率等") print(" ⚠️ 请确保仅用于测试自己的网站！") print() print(f"🎯 目标网站: {bot.config['targets']['main_site']}") print(f"📺 广告类型: {len(bot.ad_behaviors)} 种") print(f"😤 用户讨厌广告: 75%概率") print() print("请选择运行模式:") print("1. 💎 单次增强版广告访问测试") print("2. 🚀 连续增强版广告访问 (使用配置参数)") print("3. ⚙️ 自定义连续增强版广告访问") choice = input("请输入选择 (1/2/3): ").strip() if choice == "1": logger.info("🎬 开始单次增强版广告访问测试") success = bot.run_enhanced_ad_visit() if success: print("🎉 单次增强版广告访问测试成功！") else: print("😞 单次增强版广告访问测试失败！") elif choice == "2": logger.info("🎬 开始连续增强版广告访问") success_count = bot.run_continuous_enhanced_visits() total = bot.config['settings']['default_visits'] print(f"🎉 连续增强版广告访问完成！成功: {success_count}/{total}") elif choice == "3": try: visit_count = int(input("请输入访问次数: ").strip()) min_delay = int(input("请输入最小延迟秒数: ").strip()) max_delay = int(input("请输入最大延迟秒数: ").strip()) logger.info(f"🎬 开始自定义增强版广告访问，次数: {visit_count}") success_count = bot.run_continuous_enhanced_visits( total_visits=visit_count, delay_range=(min_delay, max_delay) ) print(f"🎉 自定义增强版广告访问完成！成功: {success_count}/{visit_count}") except ValueError: print("❌ 输入参数错误！") else: print("❌ 无效选择！") except KeyboardInterrupt: print("\n⚠️ 用户中断执行") except Exception as e: logger.error(f"程序执行出错: {e}") print("❌ 程序执行出错，请查看日志文件 enhanced_ad_bot.log") if __name__ == "__main__": main()