diff --git a/backend/quick_test_proxy.py b/backend/quick_test_proxy.py new file mode 100644 index 0000000..8be2e77 --- /dev/null +++ b/backend/quick_test_proxy.py @@ -0,0 +1,73 @@ +""" +快速测试代理是否生效 +""" +import asyncio +from playwright.async_api import async_playwright + + +async def quick_test(): + """快速测试代理""" + # 配置代理 + proxy = { + "server": "http://210.51.27.194:50001", + "username": "hb6su3", + "password": "acv2ciow" + } + + print("\n" + "="*60) + print(f"测试代理: {proxy['server']}") + print("="*60) + + playwright = await async_playwright().start() + + # 启动浏览器（带代理） + browser = await playwright.chromium.launch( + headless=False, + proxy=proxy + ) + + context = await browser.new_context() + page = await context.new_page() + + # 监听所有请求，查看是否通过代理 + def on_request(request): + print(f"\n[请求] {request.method} {request.url}") + # 打印请求头中的代理信息 + headers = request.headers + if 'proxy-authorization' in headers: + print(f" [代理认证] 包含代理认证头") + + page.on("request", on_request) + + print("\n[步骤1] 访问IP查询网站...") + try: + await page.goto('https://httpbin.org/ip', timeout=30000) + ip_info = await page.locator('body').inner_text() + print(f"\n当前IP信息:\n{ip_info}") + except Exception as e: + print(f"访问失败: {e}") + + print("\n[步骤2] 访问小红书创作者中心...") + try: + await page.goto('https://creator.xiaohongshu.com/login', timeout=60000) + print(f"页面URL: {page.url}") + + if 'captcha' in page.url: + print("⚠️ 触发验证页面") + else: + print("✅ 正常访问登录页") + + await asyncio.sleep(3) + except Exception as e: + print(f"访问失败: {e}") + + await browser.close() + await playwright.stop() + + print("\n" + "="*60) + print("测试完成") + print("="*60) + + +if __name__ == "__main__": + asyncio.run(quick_test()) diff --git a/backend/test_proxy_connection.py b/backend/test_proxy_connection.py new file mode 100644 index 0000000..01d9a0e --- /dev/null +++ b/backend/test_proxy_connection.py @@ -0,0 +1,156 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- + +""" +测试代理IP连接 +使用requests验证代理是否可用 +""" + +import requests +from damai_proxy_config import get_all_enabled_proxies +import time + + +def test_proxy_with_requests(proxy_config: dict, test_url: str = "http://httpbin.org/ip"): + """ + 使用requests测试代理连接 + + Args: + proxy_config: 代理配置 + test_url: 测试URL + """ + print(f"\n{'='*60}") + print(f"🔄 测试代理: {proxy_config['name']}") + print(f" 服务器: {proxy_config['server']}") + print(f" 用户名: {proxy_config['username']}") + print(f"{'='*60}\n") + + # 格式化代理URL + server = proxy_config['server'].replace('http://', '').replace('https://', '') + username = proxy_config['username'] + password = proxy_config['password'] + proxy_url = f"http://{username}:{password}@{server}" + + proxies = { + "http": proxy_url, + "https": proxy_url + } + + print(f"📍 代理URL: {proxy_url}") + + # 测试HTTP + print(f"\n✅ 测试HTTP...") + try: + start_time = time.time() + response = requests.get("http://httpbin.org/ip", proxies=proxies, timeout=10) + elapsed_time = time.time() - start_time + + if response.status_code == 200: + print(f" ✅ HTTP连接成功 - 响应时间: {elapsed_time:.2f}秒") + try: + result = response.json() + print(f" 🌍 出IP: {result.get('origin', 'N/A')}") + except: + pass + else: + print(f" ❌ HTTP失败 - 状态码: {response.status_code}") + return False + except Exception as e: + print(f" ❌ HTTP错误: {str(e)}") + return False + + # 测试HTTPS + print(f"\n🔒 测试HTTPS...") + try: + start_time = time.time() + response = requests.get("https://httpbin.org/ip", proxies=proxies, timeout=10) + elapsed_time = time.time() - start_time + + if response.status_code == 200: + print(f" ✅ HTTPS连接成功 - 响应时间: {elapsed_time:.2f}秒") + try: + result = response.json() + print(f" 🌍 出IP: {result.get('origin', 'N/A')}") + except: + pass + return True + else: + print(f" ❌ HTTPS失败 - 状态码: {response.status_code}") + return False + except Exception as e: + print(f" ❌ HTTPS错误: {str(e)}") + print(f" ⚠️ 该代理可能不支持HTTPS隧道(CONNECT方法)") + return False + + +def test_all_proxies(): + """测试所有代理""" + proxies = get_all_enabled_proxies() + + print(f"\n{'='*60}") + print(f" 代理连接测试") + print(f" 共有 {len(proxies)} 个代理需要测试") + print(f"{'='*60}") + + results = [] + + for i, proxy_config in enumerate(proxies, 1): + print(f"\n[{i}/{len(proxies)}]") + success = test_proxy_with_requests(proxy_config) + results.append({ + 'name': proxy_config['name'], + 'server': proxy_config['server'], + 'success': success + }) + + if i < len(proxies): + print("\n⏳ 等待2秒...") + time.sleep(2) + + # 输出汇总 + print(f"\n\n{'='*60}") + print(" 测试汇总") + print(f"{'='*60}\n") + + success_count = sum(1 for r in results if r['success']) + fail_count = len(results) - success_count + + print(f"✅ 成功: {success_count}/{len(results)}") + print(f"❌ 失败: {fail_count}/{len(results)}") + print(f"\n详细结果:") + + for result in results: + status = "✅ 可用" if result['success'] else "❌ 不可用" + print(f" {status} - {result['name']} ({result['server']})") + + return results + + +if __name__ == '__main__': + print("\n" + "="*60) + print(" 小红书代理IP连接测试") + print(" 使用requests库测试代理连通性") + print("="*60) + + results = test_all_proxies() + + # 给出建议 + available_proxies = [r for r in results if r['success']] + + if available_proxies: + print(f"\n\n💡 建议:") + print(f" 以下代理可用，可以在damai_proxy_config.py中保留:") + for proxy in available_proxies: + print(f" ✅ {proxy['name']}") + + unavailable_proxies = [r for r in results if not r['success']] + if unavailable_proxies: + print(f"\n 以下代理不可用，建议在damai_proxy_config.py中设置enabled=False:") + for proxy in unavailable_proxies: + print(f" ❌ {proxy['name']}") + else: + print(f"\n\n⚠️ 警告: 所有代理均不可用!") + print(f" 请检查:") + print(f" 1. 代理服务器是否在线") + print(f" 2. 用户名和密码是否正确") + print(f" 3. 本机网络是否正常") diff --git a/backend/test_proxy_playwright.py b/backend/test_proxy_playwright.py new file mode 100644 index 0000000..f66d3b7 --- /dev/null +++ b/backend/test_proxy_playwright.py @@ -0,0 +1,106 @@ +""" +测试Playwright使用代理访问HTTP和HTTPS网站 +""" +import asyncio +import warnings +import sys +from playwright.async_api import async_playwright +from damai_proxy_config import get_proxy_1, format_proxy_for_playwright + +# 抑制Windows下asyncio的资源警告 +warnings.filterwarnings('ignore', category=ResourceWarning) + +async def test_proxy_with_playwright(): + """使用Playwright测试代理""" + + # 获取代理配置 + proxy_config = get_proxy_1() + proxy = format_proxy_for_playwright(proxy_config) + + print("\n" + "="*60) + print("Playwright代理测试") + print("="*60) + print(f"\n代理配置:") + print(f" 服务器: {proxy['server']}") + print(f" 用户名: {proxy['username']}") + + playwright = None + browser = None + + try: + playwright = await async_playwright().start() + + print(f"\n启动浏览器...") + browser = await playwright.chromium.launch( + headless=True, + proxy=proxy + ) + + print(f"浏览器启动成功\n") + + context = await browser.new_context() + page = await context.new_page() + + # 测试1: HTTP网站 + print("-"*60) + print("测试1: 访问HTTP网站 (http://httpbin.org/ip)") + print("-"*60) + try: + response = await page.goto("http://httpbin.org/ip", timeout=15000, wait_until="domcontentloaded") + print(f"成功 - 状态码: {response.status}") + content = await page.content() + if 'origin' in content: + print("页面包含IP信息\n") + except Exception as e: + print(f"失败: {str(e)}\n") + + # 测试2: HTTPS网站 + print("-"*60) + print("测试2: 访问HTTPS网站 (https://httpbin.org/ip)") + print("-"*60) + try: + response = await page.goto("https://httpbin.org/ip", timeout=15000, wait_until="domcontentloaded") + print(f"成功 - 状态码: {response.status}\n") + except Exception as e: + print(f"失败: {str(e)}\n") + + # 测试3: 小红书HTTP + print("-"*60) + print("测试3: 访问小红书HTTP (http://xiaohongshu.com)") + print("-"*60) + try: + response = await page.goto("http://xiaohongshu.com", timeout=15000, wait_until="domcontentloaded") + print(f"成功 - 状态码: {response.status}") + print(f"最终URL: {page.url}\n") + except Exception as e: + print(f"失败: {str(e)}\n") + + # 测试4: 小红书HTTPS + print("-"*60) + print("测试4: 访问小红书HTTPS (https://xiaohongshu.com)") + print("-"*60) + try: + response = await page.goto("https://xiaohongshu.com", timeout=15000, wait_until="domcontentloaded") + print(f"成功 - 状态码: {response.status}\n") + except Exception as e: + print(f"失败: {str(e)}\n") + + await context.close() + + except Exception as e: + print(f"\n测试过程出错: {str(e)}") + import traceback + traceback.print_exc() + finally: + if browser: + await browser.close() + if playwright: + await playwright.stop() + + print("="*60) + print("测试完成") + print("="*60 + "\n") + + +if __name__ == "__main__": + asyncio.run(test_proxy_with_playwright()) diff --git a/backend/test_proxy_verification.py b/backend/test_proxy_verification.py new file mode 100644 index 0000000..7988781 --- /dev/null +++ b/backend/test_proxy_verification.py @@ -0,0 +1,172 @@ +""" +测试代理是否生效 +验证Playwright是否真的通过代理IP访问网站 +""" +import asyncio +import sys +from playwright.async_api import async_playwright +from damai_proxy_config import get_random_proxy, format_proxy_for_playwright + + +async def test_proxy(): + """测试代理IP是否生效""" + print("=" * 60, file=sys.stderr) + print("开始测试代理IP是否生效", file=sys.stderr) + print("=" * 60, file=sys.stderr) + + # 1. 获取本机真实IP（不使用代理） + print("\n[步骤1] 获取本机真实IP（不使用代理）", file=sys.stderr) + playwright = await async_playwright().start() + browser = await playwright.chromium.launch(headless=False) + context = await browser.new_context() + page = await context.new_page() + + # 访问IP查询网站 + await page.goto('https://httpbin.org/ip', timeout=30000) + real_ip_text = await page.locator('body').inner_text() + print(f"本机IP信息:\n{real_ip_text}", file=sys.stderr) + + await browser.close() + await playwright.stop() + + # 2. 使用代理IP访问 + print("\n[步骤2] 使用代理IP访问", file=sys.stderr) + + # 获取随机代理 + proxy_config = get_random_proxy() + proxy = format_proxy_for_playwright(proxy_config) + print(f"使用代理: {proxy_config.get('name', '未知')} - {proxy['server']}", file=sys.stderr) + + playwright = await async_playwright().start() + + # 启动浏览器时配置代理 + launch_kwargs = { + "headless": False, + "proxy": proxy, + "args": [ + '--disable-blink-features=AutomationControlled', + ] + } + + browser = await playwright.chromium.launch(**launch_kwargs) + context = await browser.new_context() + + # 注入反检测脚本 + await context.add_init_script(""" + Object.defineProperty(navigator, 'webdriver', { + get: () => undefined + }); + """) + + page = await context.new_page() + + # 访问IP查询网站 + print("正在访问 httpbin.org/ip ...", file=sys.stderr) + await page.goto('https://httpbin.org/ip', timeout=30000) + proxy_ip_text = await page.locator('body').inner_text() + print(f"代理IP信息:\n{proxy_ip_text}", file=sys.stderr) + + # 3. 访问多个IP查询网站进行交叉验证 + print("\n[步骤3] 交叉验证 - 访问其他IP查询网站", file=sys.stderr) + + test_urls = [ + 'https://api.ipify.org?format=json', + 'https://ifconfig.me/all.json', + ] + + for url in test_urls: + try: + print(f"\n访问: {url}", file=sys.stderr) + await page.goto(url, timeout=30000) + content = await page.locator('body').inner_text() + print(f"返回内容:\n{content}", file=sys.stderr) + except Exception as e: + print(f"访问失败: {str(e)}", file=sys.stderr) + + # 4. 测试访问小红书 + print("\n[步骤4] 测试访问小红书创作者中心", file=sys.stderr) + try: + await page.goto('https://creator.xiaohongshu.com/login', timeout=60000) + print(f"小红书页面URL: {page.url}", file=sys.stderr) + print(f"页面标题: {await page.title()}", file=sys.stderr) + + # 检查是否被重定向到验证页面 + if 'captcha' in page.url: + print("⚠️ 警告: 页面被重定向到验证页面，代理IP可能已被标记", file=sys.stderr) + else: + print("✅ 成功访问小红书登录页", file=sys.stderr) + + # 等待5秒观察页面 + print("\n等待5秒，您可以观察浏览器窗口...", file=sys.stderr) + await asyncio.sleep(5) + + except Exception as e: + print(f"访问小红书失败: {str(e)}", file=sys.stderr) + + await browser.close() + await playwright.stop() + + print("\n" + "=" * 60, file=sys.stderr) + print("测试完成", file=sys.stderr) + print("=" * 60, file=sys.stderr) + print("\n对比上述IP信息:", file=sys.stderr) + print("- 如果本机IP和代理IP不同，说明代理生效", file=sys.stderr) + print("- 如果IP相同，说明代理未生效", file=sys.stderr) + + +async def test_all_proxies(): + """测试所有启用的代理""" + from damai_proxy_config import get_all_enabled_proxies + + enabled_proxies = get_all_enabled_proxies() + print(f"\n找到 {len(enabled_proxies)} 个已启用的代理", file=sys.stderr) + + for i, proxy_config in enumerate(enabled_proxies, 1): + print(f"\n{'=' * 60}", file=sys.stderr) + print(f"测试代理 {i}/{len(enabled_proxies)}: {proxy_config.get('name', '未知')}", file=sys.stderr) + print(f"{'=' * 60}", file=sys.stderr) + + try: + proxy = format_proxy_for_playwright(proxy_config) + + playwright = await async_playwright().start() + browser = await playwright.chromium.launch( + headless=True, + proxy=proxy, + args=['--disable-blink-features=AutomationControlled'] + ) + context = await browser.new_context() + page = await context.new_page() + + # 访问IP查询网站 + await page.goto('https://httpbin.org/ip', timeout=30000) + ip_text = await page.locator('body').inner_text() + print(f"代理IP: {ip_text}", file=sys.stderr) + + # 测试访问小红书 + await page.goto('https://creator.xiaohongshu.com/login', timeout=60000) + + if 'captcha' in page.url: + print(f"❌ {proxy_config['name']} - 触发验证页面", file=sys.stderr) + else: + print(f"✅ {proxy_config['name']} - 可正常访问", file=sys.stderr) + + await browser.close() + await playwright.stop() + + except Exception as e: + print(f"❌ {proxy_config['name']} - 测试失败: {str(e)}", file=sys.stderr) + + # 间隔2秒 + await asyncio.sleep(2) + + +if __name__ == "__main__": + import sys + + if len(sys.argv) > 1 and sys.argv[1] == "all": + # 测试所有代理 + asyncio.run(test_all_proxies()) + else: + # 测试单个随机代理（详细模式） + asyncio.run(test_proxy()) diff --git a/backend/test_xhs_proxy.py b/backend/test_xhs_proxy.py new file mode 100644 index 0000000..af23608 --- /dev/null +++ b/backend/test_xhs_proxy.py @@ -0,0 +1,143 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- + +""" +测试小红书代理配置 +验证代理IP是否可用 +""" + +import asyncio +from xhs_login import XHSLoginService +from damai_proxy_config import get_all_enabled_proxies, format_proxy_for_playwright + + +async def test_proxy_access(): + """测试代理访问小红书""" + proxies = get_all_enabled_proxies() + + print(f"\n📋 共有 {len(proxies)} 个可用代理\n") + + for i, proxy_config in enumerate(proxies): + print(f"\n{'='*60}") + print(f"🔄 测试代理 {i+1}/{len(proxies)}: {proxy_config['name']}") + print(f" 服务器: {proxy_config['server']}") + print(f" 用户名: {proxy_config['username']}") + print(f"{'='*60}\n") + + try: + # 格式化代理配置 + proxy_dict = format_proxy_for_playwright(proxy_config) + print(f"🌐 代理配置: {proxy_dict}\n") + + # 创建登录服务实例，使用指定代理 + login_service = XHSLoginService( + use_pool=False, # 测试时不使用浏览器池 + headless=False, # 有头模式方便观察 + ) + + # 初始化浏览器，传入代理 + print("🚀 正在启动浏览器...") + await login_service.init_browser( + proxy=proxy_dict, # 使用代理字典 + use_random_proxy=False # 禁用自动代理，使用我们指定的 + ) + + print("✅ 浏览器启动成功\n") + + # 访问小红书首页 + print("🌍 正在访问小红书首页...") + await login_service.page.goto('https://www.xiaohongshu.com', timeout=30000) + + # 检查当前URL + current_url = login_service.page.url + print(f"📍 当前URL: {current_url}\n") + + # 判断是否被风控 + if '/website-login/captcha' in current_url or 'verifyUuid=' in current_url: + print("❌ 该代理被风控，需要验证\n") + else: + print("✅ 代理正常，成功访问小红书\n") + + # 等待5秒观察页面 + print("⏳ 等待5秒观察页面...") + await asyncio.sleep(5) + + # 关闭浏览器 + await login_service.close() + print("✅ 浏览器已关闭\n") + + except Exception as e: + print(f"❌ 测试失败: {str(e)}\n") + try: + await login_service.close() + except: + pass + + +async def test_random_proxy(): + """测试随机代理""" + print(f"\n{'='*60}") + print("🎲 测试随机代理功能") + print(f"{'='*60}\n") + + try: + # 创建登录服务实例 + login_service = XHSLoginService( + use_pool=False, + headless=False, + ) + + # 初始化浏览器，自动选择随机代理 + print("🚀 正在启动浏览器（自动选择随机代理）...") + await login_service.init_browser(use_random_proxy=True) + + print("✅ 浏览器启动成功\n") + + # 访问小红书首页 + print("🌍 正在访问小红书首页...") + await login_service.page.goto('https://www.xiaohongshu.com', timeout=30000) + + # 检查当前URL + current_url = login_service.page.url + print(f"📍 当前URL: {current_url}\n") + + # 判断是否被风控 + if '/website-login/captcha' in current_url or 'verifyUuid=' in current_url: + print("❌ 该代理被风控，需要验证\n") + else: + print("✅ 代理正常，成功访问小红书\n") + + # 等待10秒观察页面 + print("⏳ 等待10秒观察页面...") + await asyncio.sleep(10) + + # 关闭浏览器 + await login_service.close() + print("✅ 浏览器已关闭\n") + + except Exception as e: + print(f"❌ 测试失败: {str(e)}\n") + try: + await login_service.close() + except: + pass + + +if __name__ == '__main__': + print("\n" + "="*60) + print(" 小红书代理配置测试") + print("="*60) + + # 选择测试模式 + print("\n请选择测试模式:") + print("1. 测试所有代理") + print("2. 测试随机代理功能") + + choice = input("\n请输入选项(1 or 2): ").strip() + + if choice == '1': + asyncio.run(test_proxy_access()) + elif choice == '2': + asyncio.run(test_random_proxy()) + else: + print("无效选项")