oadmin/shualiangv1
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

first commit

Browse Source
This commit is contained in:
huangzhenpc
2025-07-18 13:52:41 +08:00
parent 67a78b0c56
commit bb133586ab
9 changed files with 3390 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

880
enhanced_ad_bot.py Normal file
View File

@@ -0,0 +1,880 @@
#!/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()