From d3a475c9026a444389e613431573bf3e2cbcc890 Mon Sep 17 00:00:00 2001 From: huangzhenpc Date: Fri, 18 Jul 2025 09:45:52 +0800 Subject: [PATCH] =?UTF-8?q?=E6=AD=A3=E5=BC=8F2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 74 ++++-- requirements_protocol.txt | 1 + setup_and_run.bat | 66 +++-- website_traffic_bot_protocol.py | 451 ++++++++++++++++++++++++++++++++ 4 files changed, 548 insertions(+), 44 deletions(-) create mode 100644 requirements_protocol.txt create mode 100644 website_traffic_bot_protocol.py diff --git a/README.md b/README.md index ca8390b..3fa05f1 100644 --- a/README.md +++ b/README.md @@ -10,6 +10,7 @@ - 🤖 模拟真实用户行为（滚动、点击、键盘操作） - 📊 详细的日志记录 - ⚙️ 灵活的配置选项 +- 🚀 **新增：纯协议版本，无需浏览器窗口** ## 文件说明 @@ -17,34 +18,54 @@ |--------|------| | `website_traffic_bot.py` | 基础版本，代理配置硬编码在脚本中 | | `website_traffic_bot_config.py` | 配置文件版本，通过config.json配置 | +| `website_traffic_bot_protocol.py` | **推荐** 纯协议版本，不打开浏览器 | | `config.json` | 配置文件，包含代理、目标网站等设置 | -| `requirements.txt` | Python依赖包列表 | +| `requirements.txt` | 完整版Python依赖包列表 | +| `requirements_protocol.txt` | 纯协议版依赖包列表（仅需requests） | | `setup_and_run.bat` | Windows自动安装和运行工具 | +## 版本对比 + +| 特性 | 基础版 | 配置版 | 纯协议版 ⭐ | +|------|--------|--------|------------| +| 需要Chrome浏览器 | ✅ | ✅ | ❌ | +| 打开浏览器窗口 | ✅ | ✅ | ❌ | +| 资源消耗 | 高 | 高 | 低 | +| 运行速度 | 慢 | 慢 | 快 | +| 隐蔽性 | 低 | 低 | 高 | +| 配置文件支持 | ❌ | ✅ | ✅ | +| 登录检测 | ❌ | ❌ | ✅ | + ## 快速开始 -### Windows用户 +### Windows用户（推荐） 1. 下载所有文件到同一文件夹 2. 双击运行 `setup_and_run.bat` -3. 按提示选择运行模式 +3. 选择 **选项3：纯协议版本（推荐）** -### 手动安装（所有平台） +### 手动安装 -1. 确保已安装 Python 3.7+ -2. 安装依赖： - ```bash - pip install -r requirements.txt - ``` -3. 安装Chrome浏览器 -4. 运行脚本： - ```bash - # 基础版本 - python website_traffic_bot.py - - # 配置文件版本 - python website_traffic_bot_config.py - ``` +#### 纯协议版本（推荐） +```bash +# 仅需安装requests +pip install -r requirements_protocol.txt + +# 运行纯协议版本 +python website_traffic_bot_protocol.py +``` + +#### 完整版本 +```bash +# 安装完整依赖（包括selenium） +pip install -r requirements.txt + +# 运行基础版本 +python website_traffic_bot.py + +# 或运行配置文件版本 +python website_traffic_bot_config.py +``` ## 配置说明 @@ -54,7 +75,7 @@ { "proxy": { "host": "代理服务器地址", - "port": "代理端口", + "port": "代理端口", "username": "代理用户名", "password": "代理密码" }, @@ -96,7 +117,7 @@ 1. **合法使用**：仅用于测试自己的网站 2. **代理设置**：确保代理配置正确且有效 -3. **Chrome浏览器**：确保已安装Chrome浏览器 +3. **登录要求**：如果网站需要登录，纯协议版本会自动检测并提示 4. **网络环境**：确保网络连接稳定 5. **适度使用**：避免对服务器造成过大负担 @@ -104,7 +125,7 @@ ### 常见问题 -1. **WebDriver失败** +1. **WebDriver失败**（仅影响完整版本） - 确保Chrome浏览器已安装 - 检查网络连接 @@ -116,6 +137,17 @@ - 检查目标网站是否可访问 - 确认网络连接稳定 +4. **检测到登录页面** + - 检查网站是否需要登录 + - 考虑使用公开访问的页面 + - 或在配置中添加登录功能 + +### 版本选择建议 + +- **首次使用**：推荐纯协议版本 `website_traffic_bot_protocol.py` +- **需要复杂交互**：使用完整版本 +- **长期运行**：强烈推荐纯协议版本（资源消耗低） + ### 获取帮助 如遇到问题，请检查 `traffic_bot.log` 日志文件中的错误信息。 diff --git a/requirements_protocol.txt b/requirements_protocol.txt new file mode 100644 index 0000000..659c37c --- /dev/null +++ b/requirements_protocol.txt @@ -0,0 +1 @@ +requests>=2.31.0 \ No newline at end of file diff --git a/setup_and_run.bat b/setup_and_run.bat index 738f188..a261093 100644 --- a/setup_and_run.bat +++ b/setup_and_run.bat @@ -29,35 +29,26 @@ if %errorlevel% neq 0 ( echo ✅ pip可用 -:: 安装依赖 -echo. -echo 📦 安装Python依赖包... -pip install -r requirements.txt - -if %errorlevel% neq 0 ( - echo ❌ 依赖安装失败！ - pause - exit /b 1 -) - -echo ✅ 依赖安装成功 - -:: 检查Chrome Driver -echo. -echo 🔍 检查Chrome和ChromeDriver... -echo 如果没有安装Chrome浏览器，请先安装： -echo https://www.google.com/chrome/ - :: 显示脚本选择菜单 echo. echo 📋 请选择要运行的脚本: -echo 1. 基础版本 (website_traffic_bot.py) -echo 2. 配置文件版本 (website_traffic_bot_config.py) -echo 3. 退出 +echo 1. 基础版本 (website_traffic_bot.py) - 需要Chrome +echo 2. 配置文件版本 (website_traffic_bot_config.py) - 需要Chrome +echo 3. 纯协议版本 (website_traffic_bot_protocol.py) - 推荐，无需浏览器 +echo 4. 退出 -set /p choice="请输入选择 (1/2/3): " +set /p choice="请输入选择 (1/2/3/4): " if "%choice%"=="1" ( + echo. + echo 📦 安装完整版依赖包... + pip install -r requirements.txt + if %errorlevel% neq 0 ( + echo ❌ 依赖安装失败！ + pause + exit /b 1 + ) + echo ✅ 依赖安装成功 echo. echo 🚀 运行基础版本... python website_traffic_bot.py @@ -69,9 +60,38 @@ if "%choice%"=="1" ( exit /b 1 ) echo. + echo 📦 安装完整版依赖包... + pip install -r requirements.txt + if %errorlevel% neq 0 ( + echo ❌ 依赖安装失败！ + pause + exit /b 1 + ) + echo ✅ 依赖安装成功 + echo. echo 🚀 运行配置文件版本... python website_traffic_bot_config.py ) else if "%choice%"=="3" ( + if not exist config.json ( + echo ❌ 配置文件 config.json 不存在！ + echo 请先检查配置文件是否存在。 + pause + exit /b 1 + ) + echo. + echo 📦 安装纯协议版依赖包... + pip install -r requirements_protocol.txt + if %errorlevel% neq 0 ( + echo ❌ 依赖安装失败！ + pause + exit /b 1 + ) + echo ✅ 依赖安装成功 + echo. + echo 🚀 运行纯协议版本（推荐）... + echo 💡 此版本不会打开浏览器窗口，更加隐蔽高效 + python website_traffic_bot_protocol.py +) else if "%choice%"=="4" ( echo 👋 再见！ exit /b 0 ) else ( diff --git a/website_traffic_bot_protocol.py b/website_traffic_bot_protocol.py new file mode 100644 index 0000000..3a0767e --- /dev/null +++ b/website_traffic_bot_protocol.py @@ -0,0 +1,451 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +网站流量模拟脚本 (纯协议版本) +用途：测试网站访问流程和性能 +请确保仅用于测试自己的网站！ +不会打开浏览器窗口，完全基于HTTP协议 +""" + +import requests +import time +import random +import json +import os +import logging +from urllib.parse import urlparse, urljoin +import re + +# 配置日志 +logging.basicConfig( + level=logging.INFO, + format='%(asctime)s - %(levelname)s - %(message)s', + handlers=[ + logging.FileHandler('traffic_bot.log', encoding='utf-8'), + logging.StreamHandler() + ] +) +logger = logging.getLogger(__name__) + +class WebTrafficBotProtocol: + def __init__(self, config_file='config.json'): + """ + 初始化纯协议版流量机器人 + + Args: + config_file (str): 配置文件路径 + """ + self.config = self.load_config(config_file) + self.session = None + + # 用户代理列表 + self.user_agents = [ + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36", + "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", + "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", + "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/121.0", + "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:109.0) Gecko/20100101 Firefox/121.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() + + # 设置代理 + 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']}") + + # 设置随机用户代理和其他头部 + user_agent = random.choice(self.user_agents) + self.session.headers.update({ + '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', + }) + + logger.info(f"使用用户代理: {user_agent}") + + def simulate_github_visit(self): + """模拟从GitHub访问""" + github_referrers = self.config.get('github_referrers', []) + if not github_referrers: + return True + + github_page = random.choice(github_referrers) + + try: + logger.info(f"模拟访问GitHub页面: {github_page}") + + # 更新referrer头 + self.session.headers.update({ + 'Referer': github_page, + 'Sec-Fetch-Site': 'cross-site' + }) + + # 模拟在GitHub停留 + stay_time = random.uniform(2, 8) + logger.info(f"在GitHub页面停留 {stay_time:.1f} 秒") + time.sleep(stay_time) + + return True + + except Exception as e: + logger.error(f"GitHub访问模拟失败: {e}") + return False + + def check_login_required(self, response): + """检查是否需要登录""" + if not response: + return False + + content = response.text.lower() + + # 检查常见的登录页面特征 + login_indicators = [ + '登录', 'login', '用户名', 'username', 'password', '密码', + 'signin', 'sign in', '账号', 'account', '验证码' + ] + + login_count = sum(1 for indicator in login_indicators if indicator in content) + + # 如果发现多个登录相关的关键词，可能是登录页面 + if login_count >= 3: + logger.warning("⚠️ 检测到可能的登录页面") + logger.info("建议：") + logger.info("1. 检查网站是否需要登录访问") + logger.info("2. 如需登录，请在配置文件中添加登录凭据") + logger.info("3. 或者更换为公开可访问的页面") + return True + + return False + + def make_request(self, url, timeout=10): + """发起HTTP请求""" + try: + response = self.session.get(url, timeout=timeout, allow_redirects=True) + + # 记录响应信息 + logger.info(f"访问 {url} - 状态码: {response.status_code}") + logger.info(f"响应大小: {len(response.content)} 字节") + + if response.headers.get('content-type'): + logger.info(f"内容类型: {response.headers.get('content-type')}") + + # 检查是否被重定向到登录页面 + if response.url != url and 'login' in response.url.lower(): + logger.warning(f"⚠️ 被重定向到登录页面: {response.url}") + + # 检查页面内容是否为登录页面 + self.check_login_required(response) + + response.raise_for_status() + return response + + except requests.exceptions.RequestException as e: + logger.error(f"请求失败 {url}: {e}") + return None + + def simulate_page_interaction(self, response): + """模拟页面交互（解析页面内容，模拟点击等）""" + if not response: + return + + try: + # 模拟解析页面内容 + content = response.text + + # 查找页面中的链接 + links = re.findall(r'href=["\'](.*?)["\']', content) + if links: + # 随机选择一些链接进行"预加载"模拟 + random_links = random.sample(links[:10], min(3, len(links))) + for link in random_links: + if link.startswith('http') or link.startswith('/'): + if not link.startswith('http'): + link = urljoin(response.url, link) + + # 30%概率模拟点击链接 + if random.random() < 0.3: + logger.info(f"模拟预加载链接: {link}") + # 这里不实际请求，只是记录行为 + time.sleep(random.uniform(0.5, 1.5)) + + # 模拟页面停留时的一些行为 + time.sleep(random.uniform(1, 3)) + + except Exception as e: + logger.warning(f"页面交互模拟出错: {e}") + + def visit_main_site(self): + """访问主网站""" + main_site = self.config['targets']['main_site'] + stay_time_range = self.config['settings']['main_site_stay_time'] + + try: + logger.info(f"访问主网站: {main_site}") + + # 发起请求 + response = self.make_request(main_site) + if not response: + return False + + # 模拟页面交互 + self.simulate_page_interaction(response) + + # 随机停留时间 + stay_time = random.uniform(stay_time_range[0], stay_time_range[1]) + logger.info(f"在主网站停留 {stay_time:.1f} 秒") + time.sleep(stay_time) + + return True + + except Exception as e: + logger.error(f"主网站访问失败: {e}") + return False + + def visit_game_page(self): + """访问游戏页面""" + game_page = self.config['targets']['game_page'] + stay_time_range = self.config['settings']['game_page_stay_time'] + + try: + logger.info(f"访问游戏页面: {game_page}") + + # 更新referrer为主站 + main_site = self.config['targets']['main_site'] + self.session.headers.update({ + 'Referer': main_site, + 'Sec-Fetch-Site': 'same-origin' + }) + + # 发起请求 + response = self.make_request(game_page) + if not response: + return False + + # 模拟游戏页面的特殊交互 + self.simulate_game_interaction(response) + + # 在游戏页面停留更长时间 + stay_time = random.uniform(stay_time_range[0], stay_time_range[1]) + logger.info(f"在游戏页面停留 {stay_time:.1f} 秒") + + # 模拟游戏过程中的多次请求 + self.simulate_gaming_behavior(stay_time) + + return True + + except Exception as e: + logger.error(f"游戏页面访问失败: {e}") + return False + + def simulate_game_interaction(self, response): + """模拟游戏交互""" + try: + logger.info("模拟2048游戏交互") + + # 模拟游戏开始 + time.sleep(random.uniform(2, 5)) + + # 模拟一些游戏动作（通过延迟模拟） + game_actions = ["上移", "下移", "左移", "右移"] + action_count = random.randint(8, 20) + + for i in range(action_count): + action = random.choice(game_actions) + logger.info(f"模拟游戏动作: {action}") + time.sleep(random.uniform(1, 3)) + + # 偶尔模拟长思考时间 + if random.random() < 0.2: + think_time = random.uniform(5, 15) + logger.info(f"模拟思考停顿 {think_time:.1f} 秒") + time.sleep(think_time) + + except Exception as e: + logger.warning(f"游戏交互模拟出错: {e}") + + def simulate_gaming_behavior(self, total_time): + """模拟游戏过程中的行为""" + try: + segments = random.randint(3, 6) # 将总时间分成几段 + segment_time = total_time / segments + + for i in range(segments): + # 每段时间内的行为 + if i > 0: + time.sleep(segment_time) + + # 模拟一些可能的游戏相关请求 + if random.random() < 0.3: + logger.info("模拟游戏状态检查") + + if random.random() < 0.1: + logger.info("模拟分数同步") + + except Exception as e: + logger.warning(f"游戏行为模拟出错: {e}") + + def run_single_visit(self): + """执行一次完整的访问流程""" + logger.info("开始执行访问流程（纯协议模式）") + + # 设置会话 + self.setup_session() + + try: + # 1. 模拟从GitHub访问 + if not self.simulate_github_visit(): + logger.warning("GitHub访问模拟失败，继续执行") + + # 2. 访问主网站 + if not self.visit_main_site(): + logger.error("主网站访问失败") + return False + + # 3. 访问游戏页面 + if not self.visit_game_page(): + 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: # 最后一次不需要延迟 + delay = random.uniform(delay_range[0], delay_range[1]) + logger.info(f"等待 {delay:.1f} 秒后进行下次访问") + 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} 不存在！") + print("请先创建配置文件") + return + + try: + # 创建机器人实例 + bot = WebTrafficBotProtocol(config_file) + + print("=== 网站流量模拟脚本 (纯协议版) ===") + print("🚀 不会打开浏览器窗口，完全基于HTTP协议") + 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.log") + +if __name__ == "__main__": + main() \ No newline at end of file