Files
shualiangv1/universal_ad_bot.py
huangzhenpc bb133586ab first commit
2025-07-18 13:52:41 +08:00

1189 lines
48 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/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'<div[^>]*class="[^"]*ad[^"]*"[^>]*>',
r'<div[^>]*id="[^"]*ad[^"]*"[^>]*>',
r'<ins[^>]*class="[^"]*adsbygoogle[^"]*"[^>]*>',
r'<iframe[^>]*src="[^"]*ads[^"]*"[^>]*>',
r'<div[^>]*class="[^"]*banner[^"]*"[^>]*>',
r'<div[^>]*class="[^"]*popup[^"]*"[^>]*>',
r'<div[^>]*class="[^"]*modal[^"]*"[^>]*>',
r'<div[^>]*class="[^"]*overlay[^"]*"[^>]*>',
r'<div[^>]*class="[^"]*interstitial[^"]*"[^>]*>',
r'<div[^>]*class="[^"]*native[^"]*"[^>]*>',
r'<div[^>]*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()