#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 网站刷量广告机器人 (专业版) 专门用于刷量和广告互动 针对Fun Games Hub网站优化 """ 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('ad_bot.log', encoding='utf-8'), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) class WebsiteAdBot: 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.games_list = [ {"name": "2048", "url": "/games/2048/index.html", "category": "puzzle"}, {"name": "Snake", "url": "/games/snake/index.html", "category": "arcade"}, {"name": "DIY Doll", "url": "/games/diy-doll/index.html", "category": "puzzle"}, {"name": "Sprunki", "url": "/games/sprunki/index.html", "category": "arcade"}, {"name": "FlightBird", "url": "/games/flightbird/index.html", "category": "action"}, {"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"}, ] # 🎯 游戏分类列表 self.game_categories = [ {"name": "Puzzle", "games": 4}, {"name": "Arcade", "games": 3}, {"name": "Action", "games": 2}, {"name": "Strategy", "games": 1}, {"name": "All", "games": 10} ] # 🎯 真实的访问来源（针对游戏网站） self.game_traffic_sources = [ # 游戏相关搜索 "https://www.google.com/search?q=free+online+games", "https://www.google.com/search?q=2048+game+play+online", "https://www.google.com/search?q=snake+game+browser", "https://www.google.com/search?q=puzzle+games+online+free", "https://www.google.com/search?q=fun+games+hub", "https://www.google.com/search?q=html5+games+no+download", "https://www.google.com/search?q=browser+games+2024", # 游戏社区来源 "https://www.reddit.com/r/WebGames/", "https://www.reddit.com/r/incremental_games/", "https://news.ycombinator.com/", "https://www.producthunt.com/", "https://github.com/topics/game", "https://github.com/topics/html5-game", # 中文游戏搜索 "https://www.google.com/search?q=在线小游戏", "https://www.google.com/search?q=免费网页游戏", "https://www.google.com/search?q=2048数字游戏", "https://www.baidu.com/s?wd=网页小游戏", "https://www.baidu.com/s?wd=在线游戏平台", ] 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.8: # 80%概率使用移动端（广告收益更高） 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,image/apng,*/*;q=0.8", "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6", "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，对广告友好 "Pragma": "no-cache", } # 移动端特殊头部 if is_mobile: headers.update({ "Sec-CH-UA-Mobile": "?1", "Viewport-Width": str(random.choice([375, 414, 390, 393, 412])), "Device-Memory": str(random.choice([4, 6, 8])), }) self.session.headers.update(headers) logger.info(f"🎭 广告机器人配置:") logger.info(f" 📱 User-Agent: {user_agent}") logger.info(f" 📱 移动端模式: {is_mobile}") logger.info(f" 🎯 访问模式: {self.current_behavior['pattern_type']}") # 获取当前IP current_ip = self.get_current_ip() if current_ip: print(f"🌍 当前IP: {current_ip}") print(f"🎯 广告刷量模式: 移动端={is_mobile}, 模式={self.current_behavior['pattern_type']}") def get_current_ip(self): """获取当前IP地址""" try: response = self.session.get('https://httpbin.org/ip', timeout=10) if response.status_code == 200: data = response.json() return data.get('origin', 'Unknown') except: pass return None def simulate_traffic_source(self): """模拟真实的游戏网站访问来源""" source = random.choice(self.game_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) # Reddit停留较长 logger.info(f"🔍 Reddit社区停留: {stay_time:.1f}秒") elif "github.com" in source: stay_time = random.uniform(5, 15) # GitHub中等停留 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_ads(self): """访问首页并重点关注广告""" main_site = self.config['targets']['main_site'] try: logger.info(f"🏠 访问首页并关注广告: {main_site}") # 人工延迟 pre_delay = random.uniform(3, 7) 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_homepage_ad_interaction(response) return True else: logger.error(f"❌ 首页访问失败 ({response.status_code})") return False except Exception as e: logger.error(f"首页访问出错: {e}") return False def _simulate_homepage_ad_interaction(self, response): """模拟首页广告交互""" base_url = response.url # 🎯 首页深度停留 - 广告需要更长的展示时间 homepage_stay_time = random.uniform(15, 45) # 15-45秒深度停留 logger.info(f"🏠 首页深度停留时间: {homepage_stay_time:.1f}秒") # 分阶段模拟首页行为 stages = [ {"name": "初始加载", "duration": 0.2, "actions": ["页面加载", "渲染广告"]}, {"name": "浏览游戏", "duration": 0.3, "actions": ["查看热门游戏", "浏览游戏分类"]}, {"name": "广告互动", "duration": 0.3, "actions": ["查看广告", "广告展示", "可能点击"]}, {"name": "深度探索", "duration": 0.2, "actions": ["滚动页面", "查看更多游戏"]} ] for stage in stages: stage_time = homepage_stay_time * stage["duration"] logger.info(f"🎯 {stage['name']}阶段: {stage_time:.1f}秒") # 执行该阶段的行为 action_count = len(stage["actions"]) action_time = stage_time / action_count for action in stage["actions"]: logger.info(f" 📱 {action}") # 特殊处理广告相关行为 if "广告" in action: self._simulate_ad_behavior(base_url, action_time) else: time.sleep(action_time * random.uniform(0.8, 1.2)) # 动作间随机间隔 if random.random() < 0.7: time.sleep(random.uniform(0.5, 2)) # 🎯 额外的广告专项行为 if random.random() < 0.8: # 80%概率进行广告专项行为 self._simulate_dedicated_ad_behavior(base_url) def _simulate_ad_behavior(self, base_url, duration): """模拟广告行为""" logger.info(f" 💰 广告行为开始，持续 {duration:.1f}秒") # 模拟广告加载时间 ad_load_time = random.uniform(1, 3) logger.info(f" 📺 广告加载: {ad_load_time:.1f}秒") time.sleep(ad_load_time) # 模拟广告展示时间 ad_display_time = duration - ad_load_time if ad_display_time > 0: logger.info(f" 👀 广告展示: {ad_display_time:.1f}秒") time.sleep(ad_display_time) # 模拟可能的广告点击 if random.random() < 0.15: # 15%概率点击广告 self._simulate_ad_click(base_url) def _simulate_ad_click(self, base_url): """模拟广告点击""" logger.info(f" 🖱️ 模拟广告点击") # 模拟点击延迟 click_delay = random.uniform(0.5, 2) time.sleep(click_delay) # 模拟广告相关请求 ad_endpoints = [ "/ads/click", "/advertising/track", "/ad/impression", "/analytics/ad", "/track/click" ] for endpoint in random.sample(ad_endpoints, random.randint(1, 2)): try: parsed_url = urlparse(base_url) ad_url = f"{parsed_url.scheme}://{parsed_url.netloc}{endpoint}" ad_headers = self.session.headers.copy() ad_headers.update({ "X-Requested-With": "XMLHttpRequest", "Content-Type": "application/json", "Referer": base_url, "Sec-Fetch-Dest": "empty", "Sec-Fetch-Mode": "cors", "Sec-Fetch-Site": "same-origin", }) response = self.session.get(ad_url, headers=ad_headers, timeout=5) logger.info(f" 📊 广告请求: {endpoint} ({response.status_code})") time.sleep(random.uniform(0.5, 1.5)) except Exception as e: logger.debug(f" ⚠️ 广告请求失败 {endpoint}: {e}") def _simulate_dedicated_ad_behavior(self, base_url): """专项广告行为模拟""" logger.info(f"💰 开始专项广告行为") # 模拟滚动到广告位置 scroll_time = random.uniform(2, 5) logger.info(f" 📜 滚动到广告区域: {scroll_time:.1f}秒") time.sleep(scroll_time) # 模拟在广告区域停留 ad_focus_time = random.uniform(8, 20) logger.info(f" 👁️ 广告区域停留: {ad_focus_time:.1f}秒") time.sleep(ad_focus_time) # 模拟广告交互 if random.random() < 0.25: # 25%概率进行广告交互 logger.info(f" 🔄 广告交互行为") self._simulate_ad_interaction(base_url) def _simulate_ad_interaction(self, base_url): """模拟广告交互""" interactions = [ "悬停在广告上", "查看广告详情", "广告展开", "关闭广告", "点击广告" ] selected_interactions = random.sample(interactions, random.randint(1, 3)) for interaction in selected_interactions: logger.info(f" 🎯 {interaction}") if interaction == "点击广告": self._simulate_ad_click(base_url) else: time.sleep(random.uniform(1, 4)) def browse_games_with_ads(self): """浏览游戏页面并关注广告""" main_site = self.config['targets']['main_site'] base_url = main_site.rstrip('/') # 随机选择要浏览的游戏数量 games_to_browse = random.randint(3, 6) 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']} ({game['category']})") # 设置referrer if i == 0: self.session.headers.update({ 'Referer': main_site, 'Sec-Fetch-Site': 'same-origin' }) else: prev_game_url = base_url + selected_games[i-1]["url"] self.session.headers.update({ 'Referer': prev_game_url, 'Sec-Fetch-Site': 'same-origin' }) # 访问游戏页面 response = self.session.get(game_url, timeout=15, allow_redirects=True) if response.status_code == 200: logger.info(f"✅ 游戏页面访问成功") # 模拟游戏页面的广告行为 self._simulate_game_page_ad_behavior(response, game) else: logger.warning(f"⚠️ 游戏页面访问失败 ({response.status_code})") # 页面间切换延迟 if i < len(selected_games) - 1: switch_delay = random.uniform(2, 6) logger.info(f" ⚡ 切换延迟: {switch_delay:.1f}秒") time.sleep(switch_delay) except Exception as e: logger.error(f"游戏页面访问出错: {e}") continue return True def _simulate_game_page_ad_behavior(self, response, game): """模拟游戏页面的广告行为""" # 游戏页面停留时间（包含广告时间） game_stay_time = random.uniform(20, 60) logger.info(f" 🎮 游戏页面停留: {game_stay_time:.1f}秒") # 分配时间 game_time = game_stay_time * 0.6 # 60%时间玩游戏 ad_time = game_stay_time * 0.4 # 40%时间看广告 logger.info(f" 🎯 游戏时间: {game_time:.1f}秒") logger.info(f" 💰 广告时间: {ad_time:.1f}秒") # 模拟游戏加载 load_time = random.uniform(2, 5) logger.info(f" 🔄 游戏加载: {load_time:.1f}秒") time.sleep(load_time) # 模拟游戏前广告 if random.random() < 0.7: # 70%概率有游戏前广告 pre_ad_time = random.uniform(5, 15) logger.info(f" 📺 游戏前广告: {pre_ad_time:.1f}秒") time.sleep(pre_ad_time) # 模拟跳过广告或看完广告 if random.random() < 0.3: # 30%概率跳过 logger.info(f" ⏭️ 跳过广告") time.sleep(random.uniform(1, 2)) else: logger.info(f" ✅ 看完广告") # 模拟游戏过程 self._simulate_game_play(game, game_time) # 模拟游戏中广告 if random.random() < 0.5: # 50%概率有游戏中广告 mid_ad_time = random.uniform(3, 10) logger.info(f" 📺 游戏中广告: {mid_ad_time:.1f}秒") time.sleep(mid_ad_time) # 模拟游戏结束广告 if random.random() < 0.8: # 80%概率有结束广告 end_ad_time = random.uniform(5, 15) logger.info(f" 📺 游戏结束广告: {end_ad_time:.1f}秒") time.sleep(end_ad_time) def _simulate_game_play(self, game, duration): """模拟游戏过程""" logger.info(f" 🎮 开始游戏: {game['name']}") # 根据游戏类型模拟不同行为 if game["category"] == "puzzle": self._simulate_puzzle_game(duration) elif game["category"] == "arcade": self._simulate_arcade_game(duration) elif game["category"] == "action": self._simulate_action_game(duration) else: self._simulate_general_game(duration) def _simulate_puzzle_game(self, duration): """模拟益智游戏""" 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.7, 1.3)) def _simulate_arcade_game(self, duration): """模拟街机游戏""" actions = ["开始游戏", "快速操作", "躲避", "收集", "游戏结束"] segments = random.randint(4, 8) segment_time = duration / segments for i in range(segments): action = random.choice(actions) logger.info(f" 🕹️ {action}") time.sleep(segment_time * random.uniform(0.6, 1.4)) def _simulate_action_game(self, duration): """模拟动作游戏""" actions = ["移动", "攻击", "跳跃", "技能", "闯关"] segments = random.randint(5, 10) segment_time = duration / segments for i in range(segments): action = random.choice(actions) logger.info(f" ⚔️ {action}") time.sleep(segment_time * random.uniform(0.5, 1.2)) def _simulate_general_game(self, duration): """模拟一般游戏""" 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_ad_focused_visit(self): """执行以广告为重点的访问流程""" logger.info("🚀 开始广告重点访问流程") # 设置会话 self.setup_session() try: # 1. 模拟来源访问 if not self.simulate_traffic_source(): logger.warning("⚠️ 来源访问失败，继续执行") # 2. 访问首页并重点关注广告 if not self.visit_homepage_with_ads(): logger.error("❌ 首页广告访问失败") return False # 3. 浏览游戏页面并关注广告 if not self.browse_games_with_ads(): logger.error("❌ 游戏页面广告访问失败") return False logger.info("✅ 广告重点访问流程完成！") return True except Exception as e: logger.error(f"❌ 访问流程出错: {e}") return False finally: if self.session: self.session.close() def run_continuous_ad_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 total_ad_time = 0 logger.info(f"🎯 开始连续广告访问，目标: {total_visits} 次") for i in range(total_visits): logger.info(f"{'='*60}") logger.info(f"🔄 第 {i+1}/{total_visits} 次广告访问") logger.info(f"{'='*60}") visit_start_time = time.time() if self.run_ad_focused_visit(): success_count += 1 visit_time = time.time() - visit_start_time total_ad_time += visit_time logger.info(f"✅ 第 {i+1} 次访问成功！耗时: {visit_time:.1f}秒") logger.info(f"📊 累计成功: {success_count}, 总广告时间: {total_ad_time:.1f}秒") 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) success_rate = (success_count / total_visits) * 100 avg_ad_time = total_ad_time / success_count if success_count > 0 else 0 logger.info(f"🎉 广告访问任务完成！") logger.info(f"📊 成功率: {success_count}/{total_visits} ({success_rate:.1f}%)") logger.info(f"💰 总广告时间: {total_ad_time:.1f}秒") logger.info(f"📈 平均每次广告时间: {avg_ad_time:.1f}秒") return success_count def main(): """主函数""" config_file = 'config.json' if not os.path.exists(config_file): print(f"❌ 配置文件 {config_file} 不存在！") return try: bot = WebsiteAdBot(config_file) print("=" * 60) print("💰 网站广告刷量机器人 (专业版)") print("=" * 60) print("🎯 专为Fun Games Hub游戏网站优化") print("💰 重点关注广告展示和交互") print("📱 移动端优先（广告收益更高）") print("⚠️ 请确保仅用于测试自己的网站！") print() print(f"🎯 目标网站: {bot.config['targets']['main_site']}") print(f"🎮 游戏类型: {len(bot.games_list)} 个游戏") print(f"📱 移动端概率: 80%") print() print("请选择运行模式:") print("1. 💎 单次广告重点访问测试") print("2. 🚀 连续广告访问 (使用配置参数)") print("3. ⚙️ 自定义连续广告访问") choice = input("请输入选择 (1/2/3): ").strip() if choice == "1": logger.info("🎬 开始单次广告重点访问测试") success = bot.run_ad_focused_visit() if success: print("🎉 单次广告访问测试成功！") else: print("😞 单次广告访问测试失败！") elif choice == "2": logger.info("🎬 开始连续广告访问") success_count = bot.run_continuous_ad_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_ad_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("❌ 程序执行出错，请查看日志文件 ad_bot.log") if __name__ == "__main__": main()