572 lines
21 KiB
Python
572 lines
21 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('traffic_bot_realistic.log', encoding='utf-8'),
|
||
logging.StreamHandler()
|
||
]
|
||
)
|
||
logger = logging.getLogger(__name__)
|
||
|
||
class WebTrafficBotRealistic:
|
||
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
|
||
|
||
# 真实的GitHub来源页面（更丰富的列表）
|
||
self.github_referrers = [
|
||
"https://github.com/trending",
|
||
"https://github.com/trending/javascript",
|
||
"https://github.com/trending/typescript",
|
||
"https://github.com/topics/javascript",
|
||
"https://github.com/topics/game",
|
||
"https://github.com/topics/html5",
|
||
"https://github.com/topics/2048",
|
||
"https://github.com/search?q=2048+game",
|
||
"https://github.com/search?q=html5+games",
|
||
"https://github.com/collections/javascript-game-engines",
|
||
"https://github.com/explore",
|
||
"https://github.com/",
|
||
]
|
||
|
||
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.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']}")
|
||
|
||
# 使用真实用户数据库生成的头部
|
||
realistic_headers = self.user_db.get_realistic_headers(self.current_profile)
|
||
self.session.headers.update(realistic_headers)
|
||
|
||
# 记录用户特征
|
||
logger.info(f"用户配置:")
|
||
logger.info(f" 操作系统: {self.current_profile['operating_system']}")
|
||
logger.info(f" 屏幕分辨率: {self.current_profile['screen_resolution']}")
|
||
logger.info(f" 语言: {self.current_profile['language']}")
|
||
logger.info(f" 时区: {self.current_profile['timezone']}")
|
||
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"👤 用户配置: {self.current_profile['operating_system']} | {self.current_profile['screen_resolution']} | {self.current_behavior['pattern_type']}")
|
||
|
||
def get_current_ip(self):
|
||
"""获取当前IP地址"""
|
||
ip_services = [
|
||
'https://httpbin.org/ip',
|
||
'https://api.ipify.org?format=json',
|
||
'https://ipinfo.io/json',
|
||
]
|
||
|
||
for service in ip_services:
|
||
try:
|
||
logger.info(f"正在获取IP地址: {service}")
|
||
response = self.session.get(service, timeout=10)
|
||
|
||
if response.status_code == 200:
|
||
if 'json' in service or 'httpbin' in service or 'ipinfo' in service:
|
||
try:
|
||
data = response.json()
|
||
if 'origin' in data:
|
||
ip = data['origin']
|
||
elif 'ip' in data:
|
||
ip = data['ip']
|
||
else:
|
||
ip = str(data)
|
||
except:
|
||
ip = response.text.strip()
|
||
else:
|
||
ip = response.text.strip()
|
||
|
||
logger.info(f"✅ 当前IP地址: {ip}")
|
||
return ip
|
||
|
||
except Exception as e:
|
||
logger.warning(f"从 {service} 获取IP失败: {e}")
|
||
continue
|
||
|
||
logger.error("❌ 无法获取当前IP地址")
|
||
return None
|
||
|
||
def simulate_realistic_github_visit(self):
|
||
"""模拟真实的GitHub访问过程"""
|
||
github_page = random.choice(self.github_referrers)
|
||
|
||
try:
|
||
logger.info(f"🔍 模拟从GitHub访问: {github_page}")
|
||
|
||
# 更新headers以模拟GitHub访问
|
||
github_headers = self.user_db.get_realistic_headers(
|
||
self.current_profile,
|
||
referrer=None
|
||
)
|
||
|
||
# 模拟在GitHub的真实浏览行为
|
||
github_stay_time = self.user_db.get_realistic_timing(
|
||
random.uniform(5, 15),
|
||
self.current_behavior
|
||
)
|
||
|
||
logger.info(f"📚 在GitHub页面停留 {github_stay_time:.1f} 秒")
|
||
|
||
# 模拟GitHub上的行为
|
||
self._simulate_github_browsing(github_stay_time)
|
||
|
||
# 设置referrer用于后续访问
|
||
self.session.headers.update({
|
||
'Referer': github_page,
|
||
'Sec-Fetch-Site': 'cross-site'
|
||
})
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error(f"GitHub访问模拟失败: {e}")
|
||
return False
|
||
|
||
def _simulate_github_browsing(self, total_time):
|
||
"""模拟在GitHub上的浏览行为"""
|
||
actions = [
|
||
"查看项目描述",
|
||
"阅读README",
|
||
"查看代码文件",
|
||
"检查更新时间",
|
||
"查看星标数量",
|
||
"浏览提交历史"
|
||
]
|
||
|
||
segments = random.randint(3, 6)
|
||
segment_time = total_time / segments
|
||
|
||
for i in range(segments):
|
||
if i > 0:
|
||
delay = self.user_db.simulate_human_delays("reading")
|
||
time.sleep(delay)
|
||
|
||
action = random.choice(actions)
|
||
logger.info(f" GitHub行为: {action}")
|
||
|
||
# 模拟不同行为的停留时间
|
||
if "阅读" in action:
|
||
time.sleep(segment_time * 0.4)
|
||
elif "查看" in action:
|
||
time.sleep(segment_time * 0.3)
|
||
else:
|
||
time.sleep(segment_time * 0.2)
|
||
|
||
def visit_main_site_realistic(self):
|
||
"""真实模拟访问主网站"""
|
||
main_site = self.config['targets']['main_site']
|
||
|
||
try:
|
||
logger.info(f"🏠 访问主网站: {main_site}")
|
||
|
||
# 发起请求
|
||
response = self.make_realistic_request(main_site)
|
||
if not response:
|
||
return False
|
||
|
||
# 模拟真实的页面浏览行为
|
||
self._simulate_realistic_browsing(response, is_main_page=True)
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error(f"主网站访问失败: {e}")
|
||
return False
|
||
|
||
def visit_game_page_realistic(self):
|
||
"""真实模拟访问游戏页面"""
|
||
game_page = self.config['targets']['game_page']
|
||
main_site = self.config['targets']['main_site']
|
||
|
||
try:
|
||
logger.info(f"🎮 访问游戏页面: {game_page}")
|
||
|
||
# 更新referrer为主站
|
||
self.session.headers.update({
|
||
'Referer': main_site,
|
||
'Sec-Fetch-Site': 'same-origin'
|
||
})
|
||
|
||
# 发起请求
|
||
response = self.make_realistic_request(game_page)
|
||
if not response:
|
||
return False
|
||
|
||
# 模拟真实的游戏页面行为
|
||
self._simulate_realistic_gaming(response)
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error(f"游戏页面访问失败: {e}")
|
||
return False
|
||
|
||
def make_realistic_request(self, url, timeout=10):
|
||
"""发起真实的HTTP请求"""
|
||
try:
|
||
# 添加随机的人为延迟
|
||
pre_request_delay = self.user_db.simulate_human_delays("thinking")
|
||
time.sleep(pre_request_delay)
|
||
|
||
response = self.session.get(url, timeout=timeout, allow_redirects=True)
|
||
|
||
# 记录详细信息
|
||
logger.info(f"📡 访问 {url}")
|
||
logger.info(f" 状态码: {response.status_code}")
|
||
logger.info(f" 响应大小: {len(response.content)} 字节")
|
||
logger.info(f" 响应时间: {response.elapsed.total_seconds():.2f}秒")
|
||
|
||
if response.headers.get('content-type'):
|
||
logger.info(f" 内容类型: {response.headers.get('content-type')}")
|
||
|
||
response.raise_for_status()
|
||
return response
|
||
|
||
except requests.exceptions.RequestException as e:
|
||
logger.error(f"请求失败 {url}: {e}")
|
||
return None
|
||
|
||
def _simulate_realistic_browsing(self, response, is_main_page=False):
|
||
"""模拟真实的页面浏览行为"""
|
||
content = response.text
|
||
|
||
# 估算页面内容长度和阅读时间
|
||
text_length = len(re.sub(r'<[^>]+>', '', content))
|
||
reading_time = text_length / self.current_behavior['reading_speed'] * 60 # 转换为秒
|
||
|
||
logger.info(f"📖 页面内容长度: {text_length} 字符")
|
||
logger.info(f"📖 预估阅读时间: {reading_time:.1f} 秒")
|
||
|
||
# 获取真实的停留时间
|
||
if is_main_page:
|
||
base_time = random.uniform(*self.config['settings']['main_site_stay_time'])
|
||
else:
|
||
base_time = reading_time
|
||
|
||
stay_time = self.user_db.get_realistic_timing(base_time, self.current_behavior)
|
||
|
||
logger.info(f"⏱️ 实际停留时间: {stay_time:.1f} 秒")
|
||
|
||
# 模拟分段浏览
|
||
self._simulate_browsing_segments(stay_time, content)
|
||
|
||
def _simulate_browsing_segments(self, total_time, content):
|
||
"""模拟分段浏览行为"""
|
||
# 查找页面中的链接
|
||
links = re.findall(r'href=["\'](.*?)["\']', content)
|
||
internal_links = [link for link in links if not link.startswith('http') or self.config['targets']['main_site'] in link]
|
||
|
||
segments = random.randint(3, 8)
|
||
segment_time = total_time / segments
|
||
|
||
browsing_actions = [
|
||
"阅读页面内容",
|
||
"查看导航菜单",
|
||
"滚动浏览",
|
||
"查看页脚信息",
|
||
"检查页面链接",
|
||
"观察页面布局"
|
||
]
|
||
|
||
for i in range(segments):
|
||
action = random.choice(browsing_actions)
|
||
logger.info(f" 浏览行为: {action}")
|
||
|
||
# 根据行为类型调整时间
|
||
if action == "滚动浏览":
|
||
self._simulate_scrolling_behavior(segment_time)
|
||
elif action == "检查页面链接" and internal_links:
|
||
self._simulate_link_hovering(internal_links)
|
||
time.sleep(segment_time * 0.8)
|
||
else:
|
||
# 添加人为的不规律性
|
||
actual_segment_time = segment_time * random.uniform(0.7, 1.3)
|
||
time.sleep(actual_segment_time)
|
||
|
||
def _simulate_scrolling_behavior(self, duration):
|
||
"""模拟真实的滚动行为"""
|
||
scroll_sessions = random.randint(2, 5)
|
||
session_time = duration / scroll_sessions
|
||
|
||
for session in range(scroll_sessions):
|
||
logger.info(f" 滚动会话 {session + 1}")
|
||
|
||
# 模拟快速滚动
|
||
quick_scrolls = random.randint(2, 4)
|
||
for _ in range(quick_scrolls):
|
||
time.sleep(random.uniform(0.2, 0.8))
|
||
|
||
# 模拟停顿阅读
|
||
if random.random() < 0.7: # 70%概率停顿阅读
|
||
pause_time = random.uniform(1, 4)
|
||
logger.info(f" 停顿阅读 {pause_time:.1f}秒")
|
||
time.sleep(pause_time)
|
||
|
||
def _simulate_link_hovering(self, links):
|
||
"""模拟鼠标悬停在链接上"""
|
||
hover_count = min(random.randint(1, 3), len(links))
|
||
sample_links = random.sample(links, hover_count)
|
||
|
||
for link in sample_links:
|
||
logger.info(f" 查看链接: {link[:50]}...")
|
||
time.sleep(random.uniform(0.5, 2))
|
||
|
||
def _simulate_realistic_gaming(self, response):
|
||
"""模拟真实的游戏行为"""
|
||
logger.info("🎲 开始模拟2048游戏")
|
||
|
||
# 游戏前的准备时间
|
||
prep_time = self.user_db.simulate_human_delays("thinking")
|
||
logger.info(f"🤔 游戏准备时间: {prep_time:.1f}秒")
|
||
time.sleep(prep_time)
|
||
|
||
# 获取游戏停留时间
|
||
base_time = random.uniform(*self.config['settings']['game_page_stay_time'])
|
||
game_time = self.user_db.get_realistic_timing(base_time, self.current_behavior)
|
||
|
||
logger.info(f"🎮 游戏总时长: {game_time:.1f}秒")
|
||
|
||
# 模拟游戏过程
|
||
self._simulate_game_sessions(game_time)
|
||
|
||
def _simulate_game_sessions(self, total_time):
|
||
"""模拟游戏会话"""
|
||
# 将游戏时间分成多个会话
|
||
sessions = random.randint(2, 5)
|
||
|
||
for session in range(sessions):
|
||
session_time = total_time / sessions * random.uniform(0.8, 1.2)
|
||
logger.info(f"🎯 游戏会话 {session + 1}, 时长: {session_time:.1f}秒")
|
||
|
||
self._simulate_single_game_session(session_time)
|
||
|
||
# 会话间休息
|
||
if session < sessions - 1:
|
||
break_time = random.uniform(2, 8)
|
||
logger.info(f"⏸️ 会话间休息: {break_time:.1f}秒")
|
||
time.sleep(break_time)
|
||
|
||
def _simulate_single_game_session(self, session_time):
|
||
"""模拟单个游戏会话"""
|
||
game_moves = ["⬆️上移", "⬇️下移", "⬅️左移", "➡️右移"]
|
||
|
||
start_time = time.time()
|
||
move_count = 0
|
||
|
||
while time.time() - start_time < session_time:
|
||
# 随机选择移动方向
|
||
move = random.choice(game_moves)
|
||
move_count += 1
|
||
|
||
logger.info(f" 第{move_count}步: {move}")
|
||
|
||
# 模拟不同难度的思考时间
|
||
if move_count % 10 == 0: # 每10步深度思考
|
||
think_time = random.uniform(3, 8)
|
||
logger.info(f" 深度思考: {think_time:.1f}秒")
|
||
time.sleep(think_time)
|
||
elif random.random() < 0.3: # 30%概率短暂思考
|
||
think_time = random.uniform(0.5, 2)
|
||
time.sleep(think_time)
|
||
else: # 快速移动
|
||
time.sleep(random.uniform(0.3, 1))
|
||
|
||
# 模拟偶尔的错误操作
|
||
if random.random() < 0.05: # 5%概率
|
||
logger.info(" 误操作，快速纠正")
|
||
time.sleep(0.2)
|
||
logger.info(f" 纠正: {random.choice(game_moves)}")
|
||
time.sleep(random.uniform(0.3, 0.8))
|
||
|
||
def run_single_visit(self):
|
||
"""执行一次完整的真实访问流程"""
|
||
logger.info("🚀 开始执行真实用户访问流程")
|
||
|
||
# 设置会话
|
||
self.setup_session()
|
||
|
||
try:
|
||
# 1. 模拟真实的GitHub访问
|
||
if not self.simulate_realistic_github_visit():
|
||
logger.warning("GitHub访问模拟失败，继续执行")
|
||
|
||
# 2. 访问主网站
|
||
if not self.visit_main_site_realistic():
|
||
logger.error("主网站访问失败")
|
||
return False
|
||
|
||
# 3. 访问游戏页面
|
||
if not self.visit_game_page_realistic():
|
||
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(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
|
||
|
||
for i in range(total_visits):
|
||
logger.info(f"🔄 执行第 {i+1}/{total_visits} 次真实访问")
|
||
|
||
if self.run_single_visit():
|
||
success_count += 1
|
||
|
||
# 智能延迟（根据时间模式调整）
|
||
if i < total_visits - 1:
|
||
base_delay = random.uniform(delay_range[0], delay_range[1])
|
||
behavior = self.user_db.get_visit_behavior()
|
||
|
||
# 根据访问模式调整延迟
|
||
if behavior['pattern_type'] == "工作时间":
|
||
delay = base_delay * 0.8 # 工作时间间隔较短
|
||
elif behavior['pattern_type'] == "深夜":
|
||
delay = base_delay * 1.5 # 深夜间隔较长
|
||
else:
|
||
delay = base_delay
|
||
|
||
logger.info(f"⏳ 智能延迟 {delay:.1f} 秒 (模式: {behavior['pattern_type']})")
|
||
time.sleep(delay)
|
||
|
||
logger.info(f"🎉 真实访问完成，成功: {success_count}/{total_visits}")
|
||
return success_count
|
||
|
||
def main():
|
||
"""主函数"""
|
||
config_file = 'config.json'
|
||
|
||
if not os.path.exists(config_file):
|
||
print(f"❌ 配置文件 {config_file} 不存在！")
|
||
return
|
||
|
||
try:
|
||
bot = WebTrafficBotRealistic(config_file)
|
||
|
||
print("=== 网站流量模拟脚本 (真实用户行为版) ===")
|
||
print("🎭 使用真实用户数据库，模拟最真实的访问轨迹")
|
||
print("⚠️ 请确保仅用于测试自己的网站！")
|
||
print("目标网站:", bot.config['targets']['main_site'])
|
||
print("游戏页面:", bot.config['targets']['game_page'])
|
||
print()
|
||
|
||
print("请选择运行模式:")
|
||
print("1. 单次真实访问测试")
|
||
print("2. 连续真实访问模式 (使用配置文件设置)")
|
||
print("3. 连续真实访问模式 (自定义参数)")
|
||
|
||
choice = input("请输入选择 (1/2/3): ").strip()
|
||
|
||
if choice == "1":
|
||
logger.info("开始单次真实访问测试")
|
||
success = bot.run_single_visit()
|
||
if success:
|
||
print("✅ 单次真实访问测试成功！")
|
||
else:
|
||
print("❌ 单次真实访问测试失败！")
|
||
|
||
elif choice == "2":
|
||
logger.info("开始连续真实访问（配置文件模式）")
|
||
success_count = bot.run_continuous()
|
||
print(f"✅ 连续真实访问完成！成功: {success_count}/{bot.config['settings']['default_visits']}")
|
||
|
||
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(
|
||
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("❌ 程序执行出错，请检查日志文件 traffic_bot_realistic.log")
|
||
|
||
if __name__ == "__main__":
|
||
main() |