#!/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('universal_ad_bot.log', encoding='utf-8'), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) class UniversalAdBot: 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_selectors = { # 通用广告容器 "containers": [ "div[class*='ad']", "div[id*='ad']", "div[class*='banner']", "div[class*='popup']", "div[class*='modal']", "div[class*='overlay']", "div[class*='interstitial']", "div[class*='native']", "div[class*='sponsor']", "iframe[src*='ads']", "iframe[src*='doubleclick']", "iframe[src*='googlesyndication']", "ins[class*='adsbygoogle']", "[data-ad-client]", "[data-ad-slot]" ], # 关闭按钮模式 "close_buttons": [ "button[class*='close']", "span[class*='close']", "div[class*='close']", "a[class*='close']", "i[class*='close']", "[aria-label*='close']", "[title*='close']", "[alt*='close']", "button[type='button']", ".close", ".btn-close", ".modal-close", ".popup-close", "span:contains('×')", "span:contains('✕')", "span:contains('Close')" ], # 广告点击区域 "clickable_areas": [ "a[href*='ads']", "a[href*='click']", "a[href*='redirect']", "button[onclick*='ad']", "div[onclick*='ad']", "[data-click]", "img[src*='ad']", "video[src*='ad']", "[data-ad-click]" ] } # 🎯 广告行为概率配置 self.ad_behaviors = { "popup_modal": { "close_probability": 0.80, # 80%概率关闭弹窗 "click_probability": 0.12, # 12%概率点击 "ignore_probability": 0.08, # 8%概率忽略 "annoyance_level": 0.9 # 高讨厌程度 }, "banner_display": { "close_probability": 0.85, # 85%概率关闭横幅 "click_probability": 0.10, # 10%概率点击 "ignore_probability": 0.05, # 5%概率忽略 "annoyance_level": 0.8 # 高讨厌程度 }, "native_content": { "close_probability": 0.45, # 45%概率关闭原生广告 "click_probability": 0.35, # 35%概率点击 "read_probability": 0.20, # 20%概率阅读 "annoyance_level": 0.3 # 低讨厌程度 }, "video_preroll": { "skip_probability": 0.70, # 70%概率跳过视频广告 "watch_probability": 0.20, # 20%概率观看 "click_probability": 0.10, # 10%概率点击 "annoyance_level": 0.7 # 较高讨厌程度 }, "interstitial_fullscreen": { "close_probability": 0.65, # 65%概率关闭全屏广告 "click_probability": 0.20, # 20%概率点击 "wait_probability": 0.15, # 15%概率等待 "annoyance_level": 0.6 # 中等讨厌程度 } } # 🎮 游戏列表 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": { "detected_ads": 0, "close_button_clicks": 0, "ad_content_clicks": 0, "skip_actions": 0, "ignored_ads": 0 }, "ad_types_found": {}, "total_ad_time": 0, "detection_success_rate": 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" 🔍 智能广告识别: 启用") 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_universal_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._intelligent_ad_detection_and_interaction(response) return True else: logger.error(f"❌ 首页访问失败 ({response.status_code})") return False except Exception as e: logger.error(f"首页访问出错: {e}") return False def _intelligent_ad_detection_and_interaction(self, response): """智能识别和处理广告""" base_url = response.url page_content = response.text # 首页总停留时间 homepage_stay_time = random.uniform(25, 60) logger.info(f"🏠 首页总停留时间: {homepage_stay_time:.1f}秒") # 🔍 智能检测页面上的广告 detected_ads = self._detect_ads_in_page(page_content) if detected_ads: logger.info(f"🎯 智能检测到 {len(detected_ads)} 个广告元素") self.visit_stats["ad_interactions"]["detected_ads"] += len(detected_ads) # 分阶段处理检测到的广告 self._process_detected_ads(base_url, detected_ads, homepage_stay_time) else: logger.info("🔍 未检测到明显的广告元素，进行通用广告行为模拟") self._simulate_generic_ad_behavior(base_url, homepage_stay_time) # 📊 输出检测和交互统计 self._log_universal_ad_stats() def _detect_ads_in_page(self, page_content): """在页面内容中检测广告""" detected_ads = [] # 检测广告脚本标签 ad_scripts = [ r'googlesyndication\.com', r'doubleclick\.net', r'googletagmanager\.com', r'facebook\.com/tr', r'monetag', r'adnxs\.com', r'adsystem\.com', r'amazon-adsystem\.com', r'ads\.yahoo\.com', r'media\.net', r'popads\.net', r'propellerads\.com', r'adsterra\.com', r'exoclick\.com', r'clickadu\.com', r'hilltopads\.com', r'theprofitzone\.com', r'revcontent\.com', r'outbrain\.com', r'taboola\.com', r'mgid\.com' ] for script_pattern in ad_scripts: matches = re.findall(script_pattern, page_content, re.IGNORECASE) if matches: ad_type = self._classify_ad_type(script_pattern) detected_ads.append({ "type": ad_type, "source": script_pattern, "count": len(matches) }) logger.info(f" 🔍 检测到 {ad_type} 广告: {len(matches)} 个") # 检测广告HTML元素 ad_elements = [ r']*class="[^"]*ad[^"]*"[^>]*>', r']*id="[^"]*ad[^"]*"[^>]*>', r']*class="[^"]*adsbygoogle[^"]*"[^>]*>', r']*src="[^"]*ads[^"]*"[^>]*>', r']*class="[^"]*banner[^"]*"[^>]*>', r']*class="[^"]*popup[^"]*"[^>]*>', r']*class="[^"]*modal[^"]*"[^>]*>', r']*class="[^"]*overlay[^"]*"[^>]*>', r']*class="[^"]*interstitial[^"]*"[^>]*>', r']*class="[^"]*native[^"]*"[^>]*>', r']*class="[^"]*sponsor[^"]*"[^>]*>', r'data-ad-client="[^"]*"', r'data-ad-slot="[^"]*"' ] for element_pattern in ad_elements: matches = re.findall(element_pattern, page_content, re.IGNORECASE) if matches: ad_type = self._classify_ad_element_type(element_pattern) detected_ads.append({ "type": ad_type, "source": "html_element", "count": len(matches) }) logger.info(f" 🔍 检测到 {ad_type} 元素: {len(matches)} 个") return detected_ads def _classify_ad_type(self, script_pattern): """根据脚本模式分类广告类型""" type_mapping = { "googlesyndication": "Google AdSense", "doubleclick": "Google DoubleClick", "googletagmanager": "Google Tag Manager", "facebook": "Facebook Pixel", "monetag": "MonetTag", "amazon-adsystem": "Amazon Associates", "popads": "PopAds", "propellerads": "PropellerAds", "adsterra": "Adsterra", "exoclick": "ExoClick", "clickadu": "ClickAdu", "hilltopads": "HilltopAds", "revcontent": "RevContent", "outbrain": "Outbrain", "taboola": "Taboola", "mgid": "MGID" } for key, value in type_mapping.items(): if key in script_pattern: return value return "Unknown Ad Network" def _classify_ad_element_type(self, element_pattern): """根据元素模式分类广告类型""" if "banner" in element_pattern: return "banner_display" elif "popup" in element_pattern or "modal" in element_pattern or "overlay" in element_pattern: return "popup_modal" elif "interstitial" in element_pattern: return "interstitial_fullscreen" elif "native" in element_pattern or "sponsor" in element_pattern: return "native_content" elif "adsbygoogle" in element_pattern or "data-ad" in element_pattern: return "google_adsense" else: return "generic_ad" def _process_detected_ads(self, base_url, detected_ads, total_time): """处理检测到的广告""" logger.info(f"📺 开始处理检测到的广告") # 为每个广告分配处理时间 ad_count = len(detected_ads) time_per_ad = total_time / max(ad_count, 1) for i, ad in enumerate(detected_ads): ad_type = ad["type"] ad_count = ad["count"] logger.info(f"🎯 处理广告 {i+1}/{len(detected_ads)}: {ad_type} ({ad_count}个)") # 根据广告类型选择处理策略 if ad_type in ["popup_modal", "interstitial_fullscreen"]: self._handle_popup_interstitial_ads(base_url, ad_type, time_per_ad) elif ad_type in ["banner_display", "google_adsense"]: self._handle_banner_display_ads(base_url, ad_type, time_per_ad) elif ad_type in ["native_content"]: self._handle_native_content_ads(base_url, ad_type, time_per_ad) else: self._handle_generic_ads(base_url, ad_type, time_per_ad) # 更新统计 if ad_type not in self.visit_stats["ad_types_found"]: self.visit_stats["ad_types_found"][ad_type] = 0 self.visit_stats["ad_types_found"][ad_type] += ad_count def _handle_popup_interstitial_ads(self, base_url, ad_type, allocated_time): """处理弹窗和插页广告""" logger.info(f" 📺 处理弹窗/插页广告: {ad_type}") # 模拟广告显示延迟 display_delay = random.uniform(1, 4) logger.info(f" 🔄 广告显示延迟: {display_delay:.1f}秒") time.sleep(display_delay) # 模拟用户反应 reaction_time = random.uniform(0.5, 2) logger.info(f" 👀 用户反应时间: {reaction_time:.1f}秒") time.sleep(reaction_time) # 决定用户行为 behavior = self.ad_behaviors.get("popup_modal", self.ad_behaviors["popup_modal"]) user_action = self._decide_user_action_for_ad(behavior) # 执行行为 if user_action == "close": self._simulate_close_popup(base_url, ad_type) elif user_action == "click": self._simulate_click_popup(base_url, ad_type) else: self._simulate_ignore_popup(base_url, ad_type) def _handle_banner_display_ads(self, base_url, ad_type, allocated_time): """处理横幅显示广告""" logger.info(f" 📺 处理横幅广告: {ad_type}") # 模拟广告加载 load_time = random.uniform(0.5, 2) logger.info(f" 📥 广告加载时间: {load_time:.1f}秒") time.sleep(load_time) # 模拟用户浏览时注意到广告 notice_time = random.uniform(2, 6) logger.info(f" 👁️ 用户注意到广告: {notice_time:.1f}秒") time.sleep(notice_time) # 决定用户行为 behavior = self.ad_behaviors.get("banner_display", self.ad_behaviors["banner_display"]) user_action = self._decide_user_action_for_ad(behavior) # 执行行为 if user_action == "close": self._simulate_close_banner(base_url, ad_type) elif user_action == "click": self._simulate_click_banner(base_url, ad_type) else: self._simulate_ignore_banner(base_url, ad_type) def _handle_native_content_ads(self, base_url, ad_type, allocated_time): """处理原生内容广告""" logger.info(f" 📺 处理原生广告: {ad_type}") # 模拟用户阅读内容时发现广告 read_time = random.uniform(3, 8) logger.info(f" 📖 用户阅读发现广告: {read_time:.1f}秒") time.sleep(read_time) # 决定用户行为 behavior = self.ad_behaviors.get("native_content", self.ad_behaviors["native_content"]) user_action = self._decide_user_action_for_ad(behavior) # 执行行为 if user_action == "close": self._simulate_close_native(base_url, ad_type) elif user_action == "click": self._simulate_click_native(base_url, ad_type) elif user_action == "read": self._simulate_read_native(base_url, ad_type) else: self._simulate_ignore_native(base_url, ad_type) def _handle_generic_ads(self, base_url, ad_type, allocated_time): """处理通用广告""" logger.info(f" 📺 处理通用广告: {ad_type}") # 模拟通用广告行为 generic_time = random.uniform(2, 6) logger.info(f" 🎯 通用广告交互: {generic_time:.1f}秒") time.sleep(generic_time) # 随机决定行为 if random.random() < 0.75: # 75%概率关闭 self._simulate_close_generic(base_url, ad_type) elif random.random() < 0.15: # 15%概率点击 self._simulate_click_generic(base_url, ad_type) else: # 10%概率忽略 self._simulate_ignore_generic(base_url, ad_type) def _decide_user_action_for_ad(self, behavior): """决定用户对广告的行为""" # 考虑时间因素 time_factor = self._get_time_tolerance_factor() # 调整概率 close_prob = behavior["close_probability"] * time_factor click_prob = behavior["click_probability"] # 随机决定 rand = random.random() if rand < close_prob: return "close" elif rand < close_prob + click_prob: return "click" elif "read_probability" in behavior and rand < close_prob + click_prob + behavior["read_probability"]: return "read" 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 _simulate_close_popup(self, base_url, ad_type): """模拟关闭弹窗广告""" logger.info(f" ❌ 关闭弹窗广告") # 寻找关闭按钮 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_ad_interaction_request(base_url, ad_type, "close") # 更新统计 self.visit_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_click_popup(self, base_url, ad_type): """模拟点击弹窗广告""" logger.info(f" 🖱️ 点击弹窗广告") # 思考时间 think_time = random.uniform(1, 3) logger.info(f" 🤔 思考时间: {think_time:.1f}秒") time.sleep(think_time) # 点击广告 click_time = random.uniform(0.3, 1) time.sleep(click_time) # 发送点击请求 self._send_ad_interaction_request(base_url, ad_type, "click") # 更新统计 self.visit_stats["ad_interactions"]["ad_content_clicks"] += 1 # 点击后可能返回 if random.random() < 0.7: return_time = random.uniform(1, 3) logger.info(f" 🔙 快速返回: {return_time:.1f}秒") time.sleep(return_time) def _simulate_ignore_popup(self, base_url, ad_type): """模拟忽略弹窗广告""" logger.info(f" 🙈 忽略弹窗广告") ignore_time = random.uniform(2, 5) logger.info(f" 😐 忽略时间: {ignore_time:.1f}秒") time.sleep(ignore_time) # 更新统计 self.visit_stats["ad_interactions"]["ignored_ads"] += 1 def _simulate_close_banner(self, base_url, ad_type): """模拟关闭横幅广告""" logger.info(f" ❌ 关闭横幅广告") # 寻找关闭按钮 find_time = random.uniform(0.3, 1.5) logger.info(f" 🔍 寻找关闭按钮: {find_time:.1f}秒") time.sleep(find_time) # 点击关闭 click_time = random.uniform(0.2, 0.6) logger.info(f" 🖱️ 点击关闭按钮") time.sleep(click_time) # 发送关闭请求 self._send_ad_interaction_request(base_url, ad_type, "close") # 更新统计 self.visit_stats["ad_interactions"]["close_button_clicks"] += 1 def _simulate_click_banner(self, base_url, ad_type): """模拟点击横幅广告""" logger.info(f" 🖱️ 点击横幅广告") # 点击广告 click_time = random.uniform(0.5, 1.5) time.sleep(click_time) # 发送点击请求 self._send_ad_interaction_request(base_url, ad_type, "click") # 更新统计 self.visit_stats["ad_interactions"]["ad_content_clicks"] += 1 def _simulate_ignore_banner(self, base_url, ad_type): """模拟忽略横幅广告""" logger.info(f" 🙈 忽略横幅广告") ignore_time = random.uniform(1, 3) time.sleep(ignore_time) # 更新统计 self.visit_stats["ad_interactions"]["ignored_ads"] += 1 def _simulate_close_native(self, base_url, ad_type): """模拟关闭原生广告""" logger.info(f" ❌ 关闭原生广告") # 原生广告关闭按钮通常较小 find_time = random.uniform(1, 3) logger.info(f" 🔍 寻找关闭按钮: {find_time:.1f}秒") time.sleep(find_time) # 点击关闭 click_time = random.uniform(0.3, 1) logger.info(f" 🖱️ 点击关闭按钮") time.sleep(click_time) # 发送关闭请求 self._send_ad_interaction_request(base_url, ad_type, "close") # 更新统计 self.visit_stats["ad_interactions"]["close_button_clicks"] += 1 def _simulate_click_native(self, base_url, ad_type): """模拟点击原生广告""" logger.info(f" 🖱️ 点击原生广告") # 原生广告点击前通常会多思考 think_time = random.uniform(3, 6) logger.info(f" 🤔 思考时间: {think_time:.1f}秒") time.sleep(think_time) # 点击广告 click_time = random.uniform(0.5, 1.5) time.sleep(click_time) # 发送点击请求 self._send_ad_interaction_request(base_url, ad_type, "click") # 更新统计 self.visit_stats["ad_interactions"]["ad_content_clicks"] += 1 def _simulate_read_native(self, base_url, ad_type): """模拟阅读原生广告""" logger.info(f" 📖 阅读原生广告") # 阅读时间 read_time = random.uniform(5, 12) logger.info(f" 📚 阅读时间: {read_time:.1f}秒") time.sleep(read_time) # 阅读后可能点击 if random.random() < 0.3: self._simulate_click_native(base_url, ad_type) def _simulate_ignore_native(self, base_url, ad_type): """模拟忽略原生广告""" logger.info(f" 🙈 忽略原生广告") ignore_time = random.uniform(2, 4) time.sleep(ignore_time) # 更新统计 self.visit_stats["ad_interactions"]["ignored_ads"] += 1 def _simulate_close_generic(self, base_url, ad_type): """模拟关闭通用广告""" logger.info(f" ❌ 关闭通用广告") close_time = random.uniform(1, 3) logger.info(f" 🖱️ 关闭操作: {close_time:.1f}秒") time.sleep(close_time) # 发送关闭请求 self._send_ad_interaction_request(base_url, ad_type, "close") # 更新统计 self.visit_stats["ad_interactions"]["close_button_clicks"] += 1 def _simulate_click_generic(self, base_url, ad_type): """模拟点击通用广告""" logger.info(f" 🖱️ 点击通用广告") click_time = random.uniform(1, 2) time.sleep(click_time) # 发送点击请求 self._send_ad_interaction_request(base_url, ad_type, "click") # 更新统计 self.visit_stats["ad_interactions"]["ad_content_clicks"] += 1 def _simulate_ignore_generic(self, base_url, ad_type): """模拟忽略通用广告""" logger.info(f" 🙈 忽略通用广告") ignore_time = random.uniform(2, 4) time.sleep(ignore_time) # 更新统计 self.visit_stats["ad_interactions"]["ignored_ads"] += 1 def _simulate_generic_ad_behavior(self, base_url, total_time): """模拟通用广告行为（当未检测到具体广告时）""" logger.info(f"🎯 执行通用广告行为模拟") # 模拟可能存在但未检测到的广告 possible_ads = ["弹窗广告", "横幅广告", "原生广告", "视频广告"] segments = random.randint(2, 4) segment_time = total_time / segments for i in range(segments): ad_type = random.choice(possible_ads) logger.info(f" 🎯 模拟 {ad_type} 行为") # 模拟广告加载和交互 load_time = random.uniform(1, 3) logger.info(f" 📥 广告加载: {load_time:.1f}秒") time.sleep(load_time) # 模拟用户行为 if random.random() < 0.75: # 75%概率关闭 logger.info(f" ❌ 关闭{ad_type}") self._send_ad_interaction_request(base_url, ad_type, "close") self.visit_stats["ad_interactions"]["close_button_clicks"] += 1 elif random.random() < 0.15: # 15%概率点击 logger.info(f" 🖱️ 点击{ad_type}") self._send_ad_interaction_request(base_url, ad_type, "click") self.visit_stats["ad_interactions"]["ad_content_clicks"] += 1 else: # 10%概率忽略 logger.info(f" 🙈 忽略{ad_type}") self.visit_stats["ad_interactions"]["ignored_ads"] += 1 # 行为间隔 time.sleep(segment_time - load_time) def _send_ad_interaction_request(self, base_url, ad_type, action): """发送广告交互请求""" try: # 模拟广告交互API请求 endpoints = [ f"/api/ads/{action}", f"/ads/{action}", f"/tracking/ad_{action}", f"/analytics/ad_interaction", f"/stats/ad_{action}" ] endpoint = random.choice(endpoints) parsed_url = urlparse(base_url) request_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, }) payload = { "ad_type": ad_type, "action": action, "timestamp": int(time.time()), "user_agent": self.session.headers.get("User-Agent", ""), "page_url": base_url } response = self.session.post(request_url, headers=headers, json=payload, timeout=5) logger.info(f" 📤 广告交互请求: {endpoint} ({response.status_code})") except Exception as e: logger.debug(f" ⚠️ 广告交互请求失败: {e}") def _log_universal_ad_stats(self): """记录通用广告统计""" logger.info("📊 智能广告检测和交互统计:") logger.info(f" 🔍 检测到广告: {self.visit_stats['ad_interactions']['detected_ads']} 个") logger.info(f" ❌ 关闭按钮点击: {self.visit_stats['ad_interactions']['close_button_clicks']} 次") logger.info(f" 🖱️ 广告内容点击: {self.visit_stats['ad_interactions']['ad_content_clicks']} 次") logger.info(f" 🙈 忽略广告: {self.visit_stats['ad_interactions']['ignored_ads']} 次") if self.visit_stats["ad_types_found"]: logger.info(" 🎯 发现的广告类型:") for ad_type, count in self.visit_stats["ad_types_found"].items(): logger.info(f" - {ad_type}: {count} 个") 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._handle_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 _handle_game_page_ads(self, response, game): """处理游戏页面的广告""" game_stay_time = random.uniform(30, 90) logger.info(f" 🎮 游戏页面停留: {game_stay_time:.1f}秒") # 检测游戏页面广告 detected_ads = self._detect_ads_in_page(response.text) if detected_ads: logger.info(f" 🔍 游戏页面检测到 {len(detected_ads)} 个广告") # 模拟游戏前广告 if random.random() < 0.8: # 80%概率有游戏前广告 logger.info(f" 📺 游戏前广告处理") if random.random() < 0.7: # 70%概率跳过 logger.info(f" ⏭️ 跳过游戏前广告") self.visit_stats["ad_interactions"]["close_button_clicks"] += 1 time.sleep(random.uniform(1, 3)) else: logger.info(f" ✅ 观看游戏前广告") time.sleep(random.uniform(5, 15)) # 模拟游戏过程 self._simulate_game_play(game, game_stay_time * 0.6) # 模拟游戏中广告 if random.random() < 0.6: # 60%概率有游戏中广告 logger.info(f" 📺 游戏中广告处理") if random.random() < 0.8: # 80%概率关闭 logger.info(f" ❌ 关闭游戏中广告") self.visit_stats["ad_interactions"]["close_button_clicks"] += 1 time.sleep(random.uniform(1, 2)) else: time.sleep(random.uniform(3, 8)) # 模拟游戏结束广告 if random.random() < 0.9: # 90%概率有结束广告 logger.info(f" 📺 游戏结束广告处理") if random.random() < 0.5: # 50%概率关闭 logger.info(f" ❌ 关闭游戏结束广告") self.visit_stats["ad_interactions"]["close_button_clicks"] += 1 time.sleep(random.uniform(1, 2)) elif random.random() < 0.3: # 30%概率点击 logger.info(f" 🖱️ 点击游戏结束广告") self.visit_stats["ad_interactions"]["ad_content_clicks"] += 1 time.sleep(random.uniform(2, 5)) else: logger.info(f" ✅ 观看游戏结束广告") time.sleep(random.uniform(5, 15)) else: # 没有检测到广告，正常游戏 logger.info(f" 🎮 正常游戏，未检测到广告") self._simulate_game_play(game, game_stay_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_universal_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_universal_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_universal_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_universal_ad_visit(): success_count += 1 visit_time = time.time() - visit_start_time logger.info(f"✅ 第 {i+1} 次访问成功！耗时: {visit_time:.1f}秒") # 显示累计统计 logger.info(f"📊 累计: 成功{success_count}, 检测广告{self.visit_stats['ad_interactions']['detected_ads']}, 关闭{self.visit_stats['ad_interactions']['close_button_clicks']}, 点击{self.visit_stats['ad_interactions']['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_universal_stats(success_count, total_visits) return success_count def _print_final_universal_stats(self, success_count, total_visits): """打印最终统计信息""" success_rate = (success_count / total_visits) * 100 logger.info(f"🎉 通用广告访问任务完成！") logger.info(f"📊 访问统计:") logger.info(f" 成功率: {success_count}/{total_visits} ({success_rate:.1f}%)") logger.info(f" 检测广告总数: {self.visit_stats['ad_interactions']['detected_ads']}") logger.info(f" 关闭按钮点击: {self.visit_stats['ad_interactions']['close_button_clicks']}") logger.info(f" 广告内容点击: {self.visit_stats['ad_interactions']['ad_content_clicks']}") logger.info(f" 忽略广告: {self.visit_stats['ad_interactions']['ignored_ads']}") total_interactions = (self.visit_stats['ad_interactions']['close_button_clicks'] + self.visit_stats['ad_interactions']['ad_content_clicks'] + self.visit_stats['ad_interactions']['ignored_ads']) if total_interactions > 0: close_rate = (self.visit_stats['ad_interactions']['close_button_clicks'] / total_interactions) * 100 click_rate = (self.visit_stats['ad_interactions']['ad_content_clicks'] / total_interactions) * 100 ignore_rate = (self.visit_stats['ad_interactions']['ignored_ads'] / total_interactions) * 100 logger.info(f" 📈 用户行为分析:") logger.info(f" 关闭率: {close_rate:.1f}%") logger.info(f" 点击率: {click_rate:.1f}%") logger.info(f" 忽略率: {ignore_rate:.1f}%") if self.visit_stats["ad_types_found"]: logger.info(f" 🎯 发现的广告类型:") for ad_type, count in self.visit_stats["ad_types_found"].items(): logger.info(f" - {ad_type}: {count} 个") logger.info(f"💡 智能识别: 自动适应各种广告网络和类型") def main(): """主函数""" config_file = 'config.json' if not os.path.exists(config_file): print(f"❌ 配置文件 {config_file} 不存在！") return try: bot = UniversalAdBot(config_file) print("=" * 80) print("🎯 通用广告刷量机器人 (智能识别版)") print("=" * 80) print("💡 核心特性:") print(" 🔍 智能广告识别 - 自动检测各种广告类型") print(" 🎭 真实用户行为 - 75%概率讨厌广告") print(" ❌ 智能关闭按钮 - 自动寻找并点击关闭按钮") print(" 📱 移动端优先 - 85%概率使用移动设备") print(" 🌐 通用适配 - 适用于任何广告网络") print(" 📊 详细统计 - 检测率、关闭率、点击率") print(" ⚠️ 请确保仅用于测试自己的网站！") print() print(f"🎯 目标网站: {bot.config['targets']['main_site']}") print(f"🔍 支持广告网络: Google AdSense, MonetTag, PropellerAds 等") 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_universal_ad_visit() if success: print("🎉 单次通用广告访问测试成功！") else: print("😞 单次通用广告访问测试失败！") elif choice == "2": logger.info("🎬 开始连续通用广告访问") success_count = bot.run_continuous_universal_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_universal_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("❌ 程序执行出错，请查看日志文件 universal_ad_bot.log") if __name__ == "__main__": main()