1189 lines
48 KiB
Python
1189 lines
48 KiB
Python
#!/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() |