723 lines
29 KiB
Python
723 lines
29 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
网站刷量广告机器人 (专业版)
|
||
专门用于刷量和广告互动
|
||
针对Fun Games Hub网站优化
|
||
"""
|
||
|
||
import requests
|
||
import time
|
||
import random
|
||
import json
|
||
import os
|
||
import logging
|
||
from urllib.parse import urlparse, urljoin
|
||
import re
|
||
from real_user_database import RealUserDatabase
|
||
|
||
# 配置日志
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format='%(asctime)s - %(levelname)s - %(message)s',
|
||
handlers=[
|
||
logging.FileHandler('ad_bot.log', encoding='utf-8'),
|
||
logging.StreamHandler()
|
||
]
|
||
)
|
||
logger = logging.getLogger(__name__)
|
||
|
||
class WebsiteAdBot:
|
||
def __init__(self, config_file='config.json'):
|
||
"""
|
||
初始化广告刷量机器人
|
||
"""
|
||
self.config = self.load_config(config_file)
|
||
self.session = None
|
||
self.user_db = RealUserDatabase()
|
||
self.current_profile = None
|
||
self.current_behavior = None
|
||
|
||
# 🎮 基于网站截图的真实游戏列表
|
||
self.games_list = [
|
||
{"name": "2048", "url": "/games/2048/index.html", "category": "puzzle"},
|
||
{"name": "Snake", "url": "/games/snake/index.html", "category": "arcade"},
|
||
{"name": "DIY Doll", "url": "/games/diy-doll/index.html", "category": "puzzle"},
|
||
{"name": "Sprunki", "url": "/games/sprunki/index.html", "category": "arcade"},
|
||
{"name": "FlightBird", "url": "/games/flightbird/index.html", "category": "action"},
|
||
{"name": "Diy Doll Factory", "url": "/games/iframe-games.html?game=diy-doll-factory", "category": "puzzle"},
|
||
{"name": "Super Sprunki Adventure", "url": "/games/iframe-games.html?game=super-sprunki-adventure", "category": "action"},
|
||
]
|
||
|
||
# 🎯 游戏分类列表
|
||
self.game_categories = [
|
||
{"name": "Puzzle", "games": 4},
|
||
{"name": "Arcade", "games": 3},
|
||
{"name": "Action", "games": 2},
|
||
{"name": "Strategy", "games": 1},
|
||
{"name": "All", "games": 10}
|
||
]
|
||
|
||
# 🎯 真实的访问来源（针对游戏网站）
|
||
self.game_traffic_sources = [
|
||
# 游戏相关搜索
|
||
"https://www.google.com/search?q=free+online+games",
|
||
"https://www.google.com/search?q=2048+game+play+online",
|
||
"https://www.google.com/search?q=snake+game+browser",
|
||
"https://www.google.com/search?q=puzzle+games+online+free",
|
||
"https://www.google.com/search?q=fun+games+hub",
|
||
"https://www.google.com/search?q=html5+games+no+download",
|
||
"https://www.google.com/search?q=browser+games+2024",
|
||
|
||
# 游戏社区来源
|
||
"https://www.reddit.com/r/WebGames/",
|
||
"https://www.reddit.com/r/incremental_games/",
|
||
"https://news.ycombinator.com/",
|
||
"https://www.producthunt.com/",
|
||
"https://github.com/topics/game",
|
||
"https://github.com/topics/html5-game",
|
||
|
||
# 中文游戏搜索
|
||
"https://www.google.com/search?q=在线小游戏",
|
||
"https://www.google.com/search?q=免费网页游戏",
|
||
"https://www.google.com/search?q=2048数字游戏",
|
||
"https://www.baidu.com/s?wd=网页小游戏",
|
||
"https://www.baidu.com/s?wd=在线游戏平台",
|
||
]
|
||
|
||
def load_config(self, config_file):
|
||
"""加载配置文件"""
|
||
try:
|
||
with open(config_file, 'r', encoding='utf-8') as f:
|
||
config = json.load(f)
|
||
logger.info(f"配置文件加载成功: {config_file}")
|
||
return config
|
||
except FileNotFoundError:
|
||
logger.error(f"配置文件未找到: {config_file}")
|
||
raise
|
||
except json.JSONDecodeError as e:
|
||
logger.error(f"配置文件格式错误: {e}")
|
||
raise
|
||
|
||
def setup_session(self):
|
||
"""设置请求会话 - 广告友好版"""
|
||
self.session = requests.Session()
|
||
self.session.cookies.clear()
|
||
|
||
# 生成真实用户配置
|
||
self.current_profile = self.user_db.get_random_user_profile()
|
||
self.current_behavior = self.user_db.get_visit_behavior()
|
||
|
||
# 设置代理
|
||
proxy_config = self.config.get('proxy')
|
||
if proxy_config:
|
||
proxy_url = f"http://{proxy_config['username']}:{proxy_config['password']}@{proxy_config['host']}:{proxy_config['port']}"
|
||
self.session.proxies = {
|
||
'http': proxy_url,
|
||
'https': proxy_url
|
||
}
|
||
logger.info(f"已配置代理: {proxy_config['host']}:{proxy_config['port']}")
|
||
|
||
# 🎯 广告友好的用户代理 - 移动端为主
|
||
if random.random() < 0.8: # 80%概率使用移动端（广告收益更高）
|
||
mobile_agents = [
|
||
"Mozilla/5.0 (iPhone; CPU iPhone OS 17_1_2 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.1 Mobile/15E148 Safari/604.1",
|
||
"Mozilla/5.0 (iPhone; CPU iPhone OS 16_6_1 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.6 Mobile/15E148 Safari/604.1",
|
||
"Mozilla/5.0 (Linux; Android 14; SM-G991B) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Mobile Safari/537.36",
|
||
"Mozilla/5.0 (Linux; Android 13; SM-A515F) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Mobile Safari/537.36",
|
||
"Mozilla/5.0 (iPad; CPU OS 17_1_2 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.1 Mobile/15E148 Safari/604.1",
|
||
]
|
||
user_agent = random.choice(mobile_agents)
|
||
is_mobile = True
|
||
else:
|
||
user_agent = self.current_profile["user_agent"]
|
||
is_mobile = "Mobile" in user_agent
|
||
|
||
# 完整的广告友好头部
|
||
headers = {
|
||
"User-Agent": user_agent,
|
||
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8",
|
||
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6",
|
||
"Accept-Encoding": "gzip, deflate, br",
|
||
"Connection": "keep-alive",
|
||
"Upgrade-Insecure-Requests": "1",
|
||
"Sec-Fetch-Dest": "document",
|
||
"Sec-Fetch-Mode": "navigate",
|
||
"Sec-Fetch-Site": "none",
|
||
"Sec-Fetch-User": "?1",
|
||
"Cache-Control": "max-age=0",
|
||
"DNT": "0", # 不启用DNT，对广告友好
|
||
"Pragma": "no-cache",
|
||
}
|
||
|
||
# 移动端特殊头部
|
||
if is_mobile:
|
||
headers.update({
|
||
"Sec-CH-UA-Mobile": "?1",
|
||
"Viewport-Width": str(random.choice([375, 414, 390, 393, 412])),
|
||
"Device-Memory": str(random.choice([4, 6, 8])),
|
||
})
|
||
|
||
self.session.headers.update(headers)
|
||
|
||
logger.info(f"🎭 广告机器人配置:")
|
||
logger.info(f" 📱 User-Agent: {user_agent}")
|
||
logger.info(f" 📱 移动端模式: {is_mobile}")
|
||
logger.info(f" 🎯 访问模式: {self.current_behavior['pattern_type']}")
|
||
|
||
# 获取当前IP
|
||
current_ip = self.get_current_ip()
|
||
if current_ip:
|
||
print(f"🌍 当前IP: {current_ip}")
|
||
print(f"🎯 广告刷量模式: 移动端={is_mobile}, 模式={self.current_behavior['pattern_type']}")
|
||
|
||
def get_current_ip(self):
|
||
"""获取当前IP地址"""
|
||
try:
|
||
response = self.session.get('https://httpbin.org/ip', timeout=10)
|
||
if response.status_code == 200:
|
||
data = response.json()
|
||
return data.get('origin', 'Unknown')
|
||
except:
|
||
pass
|
||
return None
|
||
|
||
def simulate_traffic_source(self):
|
||
"""模拟真实的游戏网站访问来源"""
|
||
source = random.choice(self.game_traffic_sources)
|
||
|
||
try:
|
||
logger.info(f"🔗 模拟从游戏相关来源访问: {source}")
|
||
|
||
# 根据来源类型设置不同的停留时间
|
||
if "google.com" in source or "baidu.com" in source:
|
||
stay_time = random.uniform(2, 8) # 搜索引擎停留较短
|
||
logger.info(f"🔍 搜索引擎停留: {stay_time:.1f}秒")
|
||
elif "reddit.com" in source:
|
||
stay_time = random.uniform(8, 20) # Reddit停留较长
|
||
logger.info(f"🔍 Reddit社区停留: {stay_time:.1f}秒")
|
||
elif "github.com" in source:
|
||
stay_time = random.uniform(5, 15) # GitHub中等停留
|
||
logger.info(f"🔍 GitHub停留: {stay_time:.1f}秒")
|
||
else:
|
||
stay_time = random.uniform(3, 12) # 其他网站
|
||
logger.info(f"🔍 其他来源停留: {stay_time:.1f}秒")
|
||
|
||
# 模拟来源网站行为
|
||
self._simulate_source_behavior(stay_time)
|
||
|
||
# 设置referrer
|
||
self.session.headers.update({
|
||
'Referer': source,
|
||
'Sec-Fetch-Site': 'cross-site'
|
||
})
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error(f"来源访问模拟失败: {e}")
|
||
return False
|
||
|
||
def _simulate_source_behavior(self, total_time):
|
||
"""模拟在来源网站的行为"""
|
||
actions = [
|
||
"浏览搜索结果",
|
||
"阅读页面内容",
|
||
"查看相关链接",
|
||
"滚动浏览页面",
|
||
"点击相关内容"
|
||
]
|
||
|
||
segments = random.randint(2, 4)
|
||
segment_time = total_time / segments
|
||
|
||
for i in range(segments):
|
||
if i > 0:
|
||
time.sleep(random.uniform(0.5, 2))
|
||
|
||
action = random.choice(actions)
|
||
logger.info(f" 🔍 {action}")
|
||
time.sleep(segment_time * random.uniform(0.7, 1.3))
|
||
|
||
def visit_homepage_with_ads(self):
|
||
"""访问首页并重点关注广告"""
|
||
main_site = self.config['targets']['main_site']
|
||
|
||
try:
|
||
logger.info(f"🏠 访问首页并关注广告: {main_site}")
|
||
|
||
# 人工延迟
|
||
pre_delay = random.uniform(3, 7)
|
||
logger.info(f"🕐 访问前延迟: {pre_delay:.1f}秒")
|
||
time.sleep(pre_delay)
|
||
|
||
# 发起请求
|
||
response = self.session.get(main_site, timeout=20, allow_redirects=True)
|
||
|
||
if response.status_code == 200:
|
||
logger.info(f"✅ 首页访问成功 ({response.status_code})")
|
||
logger.info(f"📦 页面大小: {len(response.content)} 字节")
|
||
|
||
# 🎯 专注于首页停留和广告交互
|
||
self._simulate_homepage_ad_interaction(response)
|
||
|
||
return True
|
||
else:
|
||
logger.error(f"❌ 首页访问失败 ({response.status_code})")
|
||
return False
|
||
|
||
except Exception as e:
|
||
logger.error(f"首页访问出错: {e}")
|
||
return False
|
||
|
||
def _simulate_homepage_ad_interaction(self, response):
|
||
"""模拟首页广告交互"""
|
||
base_url = response.url
|
||
|
||
# 🎯 首页深度停留 - 广告需要更长的展示时间
|
||
homepage_stay_time = random.uniform(15, 45) # 15-45秒深度停留
|
||
logger.info(f"🏠 首页深度停留时间: {homepage_stay_time:.1f}秒")
|
||
|
||
# 分阶段模拟首页行为
|
||
stages = [
|
||
{"name": "初始加载", "duration": 0.2, "actions": ["页面加载", "渲染广告"]},
|
||
{"name": "浏览游戏", "duration": 0.3, "actions": ["查看热门游戏", "浏览游戏分类"]},
|
||
{"name": "广告互动", "duration": 0.3, "actions": ["查看广告", "广告展示", "可能点击"]},
|
||
{"name": "深度探索", "duration": 0.2, "actions": ["滚动页面", "查看更多游戏"]}
|
||
]
|
||
|
||
for stage in stages:
|
||
stage_time = homepage_stay_time * stage["duration"]
|
||
logger.info(f"🎯 {stage['name']}阶段: {stage_time:.1f}秒")
|
||
|
||
# 执行该阶段的行为
|
||
action_count = len(stage["actions"])
|
||
action_time = stage_time / action_count
|
||
|
||
for action in stage["actions"]:
|
||
logger.info(f" 📱 {action}")
|
||
|
||
# 特殊处理广告相关行为
|
||
if "广告" in action:
|
||
self._simulate_ad_behavior(base_url, action_time)
|
||
else:
|
||
time.sleep(action_time * random.uniform(0.8, 1.2))
|
||
|
||
# 动作间随机间隔
|
||
if random.random() < 0.7:
|
||
time.sleep(random.uniform(0.5, 2))
|
||
|
||
# 🎯 额外的广告专项行为
|
||
if random.random() < 0.8: # 80%概率进行广告专项行为
|
||
self._simulate_dedicated_ad_behavior(base_url)
|
||
|
||
def _simulate_ad_behavior(self, base_url, duration):
|
||
"""模拟广告行为"""
|
||
logger.info(f" 💰 广告行为开始，持续 {duration:.1f}秒")
|
||
|
||
# 模拟广告加载时间
|
||
ad_load_time = random.uniform(1, 3)
|
||
logger.info(f" 📺 广告加载: {ad_load_time:.1f}秒")
|
||
time.sleep(ad_load_time)
|
||
|
||
# 模拟广告展示时间
|
||
ad_display_time = duration - ad_load_time
|
||
if ad_display_time > 0:
|
||
logger.info(f" 👀 广告展示: {ad_display_time:.1f}秒")
|
||
time.sleep(ad_display_time)
|
||
|
||
# 模拟可能的广告点击
|
||
if random.random() < 0.15: # 15%概率点击广告
|
||
self._simulate_ad_click(base_url)
|
||
|
||
def _simulate_ad_click(self, base_url):
|
||
"""模拟广告点击"""
|
||
logger.info(f" 🖱️ 模拟广告点击")
|
||
|
||
# 模拟点击延迟
|
||
click_delay = random.uniform(0.5, 2)
|
||
time.sleep(click_delay)
|
||
|
||
# 模拟广告相关请求
|
||
ad_endpoints = [
|
||
"/ads/click",
|
||
"/advertising/track",
|
||
"/ad/impression",
|
||
"/analytics/ad",
|
||
"/track/click"
|
||
]
|
||
|
||
for endpoint in random.sample(ad_endpoints, random.randint(1, 2)):
|
||
try:
|
||
parsed_url = urlparse(base_url)
|
||
ad_url = f"{parsed_url.scheme}://{parsed_url.netloc}{endpoint}"
|
||
|
||
ad_headers = self.session.headers.copy()
|
||
ad_headers.update({
|
||
"X-Requested-With": "XMLHttpRequest",
|
||
"Content-Type": "application/json",
|
||
"Referer": base_url,
|
||
"Sec-Fetch-Dest": "empty",
|
||
"Sec-Fetch-Mode": "cors",
|
||
"Sec-Fetch-Site": "same-origin",
|
||
})
|
||
|
||
response = self.session.get(ad_url, headers=ad_headers, timeout=5)
|
||
logger.info(f" 📊 广告请求: {endpoint} ({response.status_code})")
|
||
|
||
time.sleep(random.uniform(0.5, 1.5))
|
||
|
||
except Exception as e:
|
||
logger.debug(f" ⚠️ 广告请求失败 {endpoint}: {e}")
|
||
|
||
def _simulate_dedicated_ad_behavior(self, base_url):
|
||
"""专项广告行为模拟"""
|
||
logger.info(f"💰 开始专项广告行为")
|
||
|
||
# 模拟滚动到广告位置
|
||
scroll_time = random.uniform(2, 5)
|
||
logger.info(f" 📜 滚动到广告区域: {scroll_time:.1f}秒")
|
||
time.sleep(scroll_time)
|
||
|
||
# 模拟在广告区域停留
|
||
ad_focus_time = random.uniform(8, 20)
|
||
logger.info(f" 👁️ 广告区域停留: {ad_focus_time:.1f}秒")
|
||
time.sleep(ad_focus_time)
|
||
|
||
# 模拟广告交互
|
||
if random.random() < 0.25: # 25%概率进行广告交互
|
||
logger.info(f" 🔄 广告交互行为")
|
||
self._simulate_ad_interaction(base_url)
|
||
|
||
def _simulate_ad_interaction(self, base_url):
|
||
"""模拟广告交互"""
|
||
interactions = [
|
||
"悬停在广告上",
|
||
"查看广告详情",
|
||
"广告展开",
|
||
"关闭广告",
|
||
"点击广告"
|
||
]
|
||
|
||
selected_interactions = random.sample(interactions, random.randint(1, 3))
|
||
|
||
for interaction in selected_interactions:
|
||
logger.info(f" 🎯 {interaction}")
|
||
|
||
if interaction == "点击广告":
|
||
self._simulate_ad_click(base_url)
|
||
else:
|
||
time.sleep(random.uniform(1, 4))
|
||
|
||
def browse_games_with_ads(self):
|
||
"""浏览游戏页面并关注广告"""
|
||
main_site = self.config['targets']['main_site']
|
||
base_url = main_site.rstrip('/')
|
||
|
||
# 随机选择要浏览的游戏数量
|
||
games_to_browse = random.randint(3, 6)
|
||
selected_games = random.sample(self.games_list, min(games_to_browse, len(self.games_list)))
|
||
|
||
logger.info(f"🎮 开始浏览 {games_to_browse} 个游戏页面")
|
||
|
||
for i, game in enumerate(selected_games):
|
||
game_url = base_url + game["url"]
|
||
|
||
try:
|
||
logger.info(f"🎯 游戏 {i+1}/{games_to_browse}: {game['name']} ({game['category']})")
|
||
|
||
# 设置referrer
|
||
if i == 0:
|
||
self.session.headers.update({
|
||
'Referer': main_site,
|
||
'Sec-Fetch-Site': 'same-origin'
|
||
})
|
||
else:
|
||
prev_game_url = base_url + selected_games[i-1]["url"]
|
||
self.session.headers.update({
|
||
'Referer': prev_game_url,
|
||
'Sec-Fetch-Site': 'same-origin'
|
||
})
|
||
|
||
# 访问游戏页面
|
||
response = self.session.get(game_url, timeout=15, allow_redirects=True)
|
||
|
||
if response.status_code == 200:
|
||
logger.info(f"✅ 游戏页面访问成功")
|
||
|
||
# 模拟游戏页面的广告行为
|
||
self._simulate_game_page_ad_behavior(response, game)
|
||
|
||
else:
|
||
logger.warning(f"⚠️ 游戏页面访问失败 ({response.status_code})")
|
||
|
||
# 页面间切换延迟
|
||
if i < len(selected_games) - 1:
|
||
switch_delay = random.uniform(2, 6)
|
||
logger.info(f" ⚡ 切换延迟: {switch_delay:.1f}秒")
|
||
time.sleep(switch_delay)
|
||
|
||
except Exception as e:
|
||
logger.error(f"游戏页面访问出错: {e}")
|
||
continue
|
||
|
||
return True
|
||
|
||
def _simulate_game_page_ad_behavior(self, response, game):
|
||
"""模拟游戏页面的广告行为"""
|
||
# 游戏页面停留时间（包含广告时间）
|
||
game_stay_time = random.uniform(20, 60)
|
||
logger.info(f" 🎮 游戏页面停留: {game_stay_time:.1f}秒")
|
||
|
||
# 分配时间
|
||
game_time = game_stay_time * 0.6 # 60%时间玩游戏
|
||
ad_time = game_stay_time * 0.4 # 40%时间看广告
|
||
|
||
logger.info(f" 🎯 游戏时间: {game_time:.1f}秒")
|
||
logger.info(f" 💰 广告时间: {ad_time:.1f}秒")
|
||
|
||
# 模拟游戏加载
|
||
load_time = random.uniform(2, 5)
|
||
logger.info(f" 🔄 游戏加载: {load_time:.1f}秒")
|
||
time.sleep(load_time)
|
||
|
||
# 模拟游戏前广告
|
||
if random.random() < 0.7: # 70%概率有游戏前广告
|
||
pre_ad_time = random.uniform(5, 15)
|
||
logger.info(f" 📺 游戏前广告: {pre_ad_time:.1f}秒")
|
||
time.sleep(pre_ad_time)
|
||
|
||
# 模拟跳过广告或看完广告
|
||
if random.random() < 0.3: # 30%概率跳过
|
||
logger.info(f" ⏭️ 跳过广告")
|
||
time.sleep(random.uniform(1, 2))
|
||
else:
|
||
logger.info(f" ✅ 看完广告")
|
||
|
||
# 模拟游戏过程
|
||
self._simulate_game_play(game, game_time)
|
||
|
||
# 模拟游戏中广告
|
||
if random.random() < 0.5: # 50%概率有游戏中广告
|
||
mid_ad_time = random.uniform(3, 10)
|
||
logger.info(f" 📺 游戏中广告: {mid_ad_time:.1f}秒")
|
||
time.sleep(mid_ad_time)
|
||
|
||
# 模拟游戏结束广告
|
||
if random.random() < 0.8: # 80%概率有结束广告
|
||
end_ad_time = random.uniform(5, 15)
|
||
logger.info(f" 📺 游戏结束广告: {end_ad_time:.1f}秒")
|
||
time.sleep(end_ad_time)
|
||
|
||
def _simulate_game_play(self, game, duration):
|
||
"""模拟游戏过程"""
|
||
logger.info(f" 🎮 开始游戏: {game['name']}")
|
||
|
||
# 根据游戏类型模拟不同行为
|
||
if game["category"] == "puzzle":
|
||
self._simulate_puzzle_game(duration)
|
||
elif game["category"] == "arcade":
|
||
self._simulate_arcade_game(duration)
|
||
elif game["category"] == "action":
|
||
self._simulate_action_game(duration)
|
||
else:
|
||
self._simulate_general_game(duration)
|
||
|
||
def _simulate_puzzle_game(self, duration):
|
||
"""模拟益智游戏"""
|
||
actions = ["思考", "移动", "尝试", "撤销", "重新开始"]
|
||
segments = random.randint(3, 6)
|
||
segment_time = duration / segments
|
||
|
||
for i in range(segments):
|
||
action = random.choice(actions)
|
||
logger.info(f" 🧩 {action}")
|
||
time.sleep(segment_time * random.uniform(0.7, 1.3))
|
||
|
||
def _simulate_arcade_game(self, duration):
|
||
"""模拟街机游戏"""
|
||
actions = ["开始游戏", "快速操作", "躲避", "收集", "游戏结束"]
|
||
segments = random.randint(4, 8)
|
||
segment_time = duration / segments
|
||
|
||
for i in range(segments):
|
||
action = random.choice(actions)
|
||
logger.info(f" 🕹️ {action}")
|
||
time.sleep(segment_time * random.uniform(0.6, 1.4))
|
||
|
||
def _simulate_action_game(self, duration):
|
||
"""模拟动作游戏"""
|
||
actions = ["移动", "攻击", "跳跃", "技能", "闯关"]
|
||
segments = random.randint(5, 10)
|
||
segment_time = duration / segments
|
||
|
||
for i in range(segments):
|
||
action = random.choice(actions)
|
||
logger.info(f" ⚔️ {action}")
|
||
time.sleep(segment_time * random.uniform(0.5, 1.2))
|
||
|
||
def _simulate_general_game(self, duration):
|
||
"""模拟一般游戏"""
|
||
actions = ["开始", "操作", "进行", "暂停", "继续"]
|
||
segments = random.randint(3, 6)
|
||
segment_time = duration / segments
|
||
|
||
for i in range(segments):
|
||
action = random.choice(actions)
|
||
logger.info(f" 🎯 {action}")
|
||
time.sleep(segment_time * random.uniform(0.8, 1.2))
|
||
|
||
def run_ad_focused_visit(self):
|
||
"""执行以广告为重点的访问流程"""
|
||
logger.info("🚀 开始广告重点访问流程")
|
||
|
||
# 设置会话
|
||
self.setup_session()
|
||
|
||
try:
|
||
# 1. 模拟来源访问
|
||
if not self.simulate_traffic_source():
|
||
logger.warning("⚠️ 来源访问失败，继续执行")
|
||
|
||
# 2. 访问首页并重点关注广告
|
||
if not self.visit_homepage_with_ads():
|
||
logger.error("❌ 首页广告访问失败")
|
||
return False
|
||
|
||
# 3. 浏览游戏页面并关注广告
|
||
if not self.browse_games_with_ads():
|
||
logger.error("❌ 游戏页面广告访问失败")
|
||
return False
|
||
|
||
logger.info("✅ 广告重点访问流程完成！")
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error(f"❌ 访问流程出错: {e}")
|
||
return False
|
||
|
||
finally:
|
||
if self.session:
|
||
self.session.close()
|
||
|
||
def run_continuous_ad_visits(self, total_visits=None, delay_range=None):
|
||
"""连续执行广告重点访问"""
|
||
if total_visits is None:
|
||
total_visits = self.config['settings']['default_visits']
|
||
|
||
if delay_range is None:
|
||
delay_range = (
|
||
self.config['settings']['min_delay'],
|
||
self.config['settings']['max_delay']
|
||
)
|
||
|
||
success_count = 0
|
||
total_ad_time = 0
|
||
|
||
logger.info(f"🎯 开始连续广告访问，目标: {total_visits} 次")
|
||
|
||
for i in range(total_visits):
|
||
logger.info(f"{'='*60}")
|
||
logger.info(f"🔄 第 {i+1}/{total_visits} 次广告访问")
|
||
logger.info(f"{'='*60}")
|
||
|
||
visit_start_time = time.time()
|
||
|
||
if self.run_ad_focused_visit():
|
||
success_count += 1
|
||
visit_time = time.time() - visit_start_time
|
||
total_ad_time += visit_time
|
||
|
||
logger.info(f"✅ 第 {i+1} 次访问成功！耗时: {visit_time:.1f}秒")
|
||
logger.info(f"📊 累计成功: {success_count}, 总广告时间: {total_ad_time:.1f}秒")
|
||
else:
|
||
logger.error(f"❌ 第 {i+1} 次访问失败！")
|
||
|
||
# 访问间隔
|
||
if i < total_visits - 1:
|
||
delay = random.uniform(delay_range[0], delay_range[1])
|
||
logger.info(f"⏳ 等待 {delay:.1f} 秒")
|
||
time.sleep(delay)
|
||
|
||
success_rate = (success_count / total_visits) * 100
|
||
avg_ad_time = total_ad_time / success_count if success_count > 0 else 0
|
||
|
||
logger.info(f"🎉 广告访问任务完成！")
|
||
logger.info(f"📊 成功率: {success_count}/{total_visits} ({success_rate:.1f}%)")
|
||
logger.info(f"💰 总广告时间: {total_ad_time:.1f}秒")
|
||
logger.info(f"📈 平均每次广告时间: {avg_ad_time:.1f}秒")
|
||
|
||
return success_count
|
||
|
||
def main():
|
||
"""主函数"""
|
||
config_file = 'config.json'
|
||
|
||
if not os.path.exists(config_file):
|
||
print(f"❌ 配置文件 {config_file} 不存在！")
|
||
return
|
||
|
||
try:
|
||
bot = WebsiteAdBot(config_file)
|
||
|
||
print("=" * 60)
|
||
print("💰 网站广告刷量机器人 (专业版)")
|
||
print("=" * 60)
|
||
print("🎯 专为Fun Games Hub游戏网站优化")
|
||
print("💰 重点关注广告展示和交互")
|
||
print("📱 移动端优先（广告收益更高）")
|
||
print("⚠️ 请确保仅用于测试自己的网站！")
|
||
print()
|
||
print(f"🎯 目标网站: {bot.config['targets']['main_site']}")
|
||
print(f"🎮 游戏类型: {len(bot.games_list)} 个游戏")
|
||
print(f"📱 移动端概率: 80%")
|
||
print()
|
||
|
||
print("请选择运行模式:")
|
||
print("1. 💎 单次广告重点访问测试")
|
||
print("2. 🚀 连续广告访问 (使用配置参数)")
|
||
print("3. ⚙️ 自定义连续广告访问")
|
||
|
||
choice = input("请输入选择 (1/2/3): ").strip()
|
||
|
||
if choice == "1":
|
||
logger.info("🎬 开始单次广告重点访问测试")
|
||
success = bot.run_ad_focused_visit()
|
||
if success:
|
||
print("🎉 单次广告访问测试成功！")
|
||
else:
|
||
print("😞 单次广告访问测试失败！")
|
||
|
||
elif choice == "2":
|
||
logger.info("🎬 开始连续广告访问")
|
||
success_count = bot.run_continuous_ad_visits()
|
||
total = bot.config['settings']['default_visits']
|
||
print(f"🎉 连续广告访问完成！成功: {success_count}/{total}")
|
||
|
||
elif choice == "3":
|
||
try:
|
||
visit_count = int(input("请输入访问次数: ").strip())
|
||
min_delay = int(input("请输入最小延迟秒数: ").strip())
|
||
max_delay = int(input("请输入最大延迟秒数: ").strip())
|
||
|
||
logger.info(f"🎬 开始自定义广告访问，次数: {visit_count}")
|
||
success_count = bot.run_continuous_ad_visits(
|
||
total_visits=visit_count,
|
||
delay_range=(min_delay, max_delay)
|
||
)
|
||
|
||
print(f"🎉 自定义广告访问完成！成功: {success_count}/{visit_count}")
|
||
|
||
except ValueError:
|
||
print("❌ 输入参数错误！")
|
||
else:
|
||
print("❌ 无效选择！")
|
||
|
||
except KeyboardInterrupt:
|
||
print("\n⚠️ 用户中断执行")
|
||
except Exception as e:
|
||
logger.error(f"程序执行出错: {e}")
|
||
print("❌ 程序执行出错，请查看日志文件 ad_bot.log")
|
||
|
||
if __name__ == "__main__":
|
||
main() |