230 lines
7.4 KiB
Python
230 lines
7.4 KiB
Python
"""
|
|
固定代理IP下小红书登录和发文功能示例
|
|
展示如何在实际应用中使用代理IP进行小红书操作
|
|
"""
|
|
import asyncio
|
|
import json
|
|
import sys
|
|
from xhs_login import XHSLoginService
|
|
from xhs_publish import XHSPublishService
|
|
from damai_proxy_config import get_proxy_config
|
|
|
|
|
|
async def login_with_proxy(phone: str, code: str, proxy_index: int = 0):
|
|
"""
|
|
使用代理进行小红书登录
|
|
|
|
Args:
|
|
phone: 手机号
|
|
code: 验证码
|
|
proxy_index: 代理索引 (0 或 1)
|
|
"""
|
|
print(f"\n{'='*60}")
|
|
print(f"📱 使用代理登录小红书")
|
|
print(f"{'='*60}")
|
|
|
|
# 获取代理配置
|
|
proxy_config = get_proxy_config(proxy_index)
|
|
proxy_server = proxy_config['server'].replace('http://', '')
|
|
proxy_url = f"http://{proxy_config['username']}:{proxy_config['password']}@{proxy_server}"
|
|
|
|
print(f"✅ 使用代理: 代理{proxy_index + 1}")
|
|
print(f" 代理服务器: {proxy_config['server']}")
|
|
|
|
# 创建登录服务
|
|
login_service = XHSLoginService()
|
|
|
|
try:
|
|
# 初始化浏览器（使用代理）
|
|
user_agent = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36'
|
|
await login_service.init_browser(proxy=proxy_url, user_agent=user_agent)
|
|
print("✅ 浏览器初始化成功（已启用代理）")
|
|
|
|
# 执行登录
|
|
result = await login_service.login(phone, code)
|
|
|
|
if result.get('success'):
|
|
print("✅ 登录成功！")
|
|
|
|
# 保存Cookies到文件
|
|
cookies_full = result.get('cookies_full', [])
|
|
if cookies_full:
|
|
with open('cookies_proxy.json', 'w', encoding='utf-8') as f:
|
|
json.dump(cookies_full, f, ensure_ascii=False, indent=2)
|
|
print("✅ 已保存登录Cookies到 cookies_proxy.json")
|
|
|
|
return result
|
|
else:
|
|
print(f"❌ 登录失败: {result.get('error')}")
|
|
return result
|
|
|
|
except Exception as e:
|
|
print(f"❌ 登录过程异常: {str(e)}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return {"success": False, "error": str(e)}
|
|
finally:
|
|
await login_service.close_browser()
|
|
|
|
|
|
async def publish_with_proxy(title: str, content: str, images: list = None, tags: list = None, proxy_index: int = 0, cookies_file: str = 'cookies.json'):
|
|
"""
|
|
使用代理发布小红书笔记
|
|
|
|
Args:
|
|
title: 笔记标题
|
|
content: 笔记内容
|
|
images: 图片路径列表
|
|
tags: 标签列表
|
|
proxy_index: 代理索引 (0 或 1)
|
|
cookies_file: Cookies文件路径
|
|
"""
|
|
print(f"\n{'='*60}")
|
|
print(f"📝 使用代理发布小红书笔记")
|
|
print(f"{'='*60}")
|
|
|
|
# 获取代理配置
|
|
proxy_config = get_proxy_config(proxy_index)
|
|
proxy_server = proxy_config['server'].replace('http://', '')
|
|
proxy_url = f"http://{proxy_config['username']}:{proxy_config['password']}@{proxy_server}"
|
|
|
|
print(f"✅ 使用代理: 代理{proxy_index + 1}")
|
|
print(f" 代理服务器: {proxy_config['server']}")
|
|
|
|
# 读取Cookies
|
|
try:
|
|
with open(cookies_file, 'r', encoding='utf-8') as f:
|
|
cookies = json.load(f)
|
|
print(f"✅ 成功读取Cookies: {len(cookies)} 个")
|
|
except FileNotFoundError:
|
|
print(f"❌ Cookies文件不存在: {cookies_file}")
|
|
return {"success": False, "error": f"Cookies文件不存在: {cookies_file}"}
|
|
except Exception as e:
|
|
print(f"❌ 读取Cookies失败: {str(e)}")
|
|
return {"success": False, "error": str(e)}
|
|
|
|
# 准备发布数据
|
|
images = images or []
|
|
tags = tags or []
|
|
|
|
print(f"📝 发布内容:")
|
|
print(f" 标题: {title}")
|
|
print(f" 内容: {content[:50]}...") # 只显示前50个字符
|
|
print(f" 图片: {len(images)} 张")
|
|
print(f" 标签: {tags}")
|
|
|
|
# 创建发布服务
|
|
try:
|
|
publisher = XHSPublishService(cookies, proxy=proxy_url)
|
|
|
|
# 执行发布
|
|
result = await publisher.publish(
|
|
title=title,
|
|
content=content,
|
|
images=images,
|
|
tags=tags
|
|
)
|
|
|
|
if result.get('success'):
|
|
print("✅ 发布成功！")
|
|
else:
|
|
print(f"❌ 发布失败: {result.get('error')}")
|
|
|
|
return result
|
|
|
|
except Exception as e:
|
|
print(f"❌ 发布过程异常: {str(e)}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return {"success": False, "error": str(e)}
|
|
|
|
|
|
async def test_proxy_functionality():
|
|
"""测试代理功能的完整流程"""
|
|
print("🚀 开始测试代理功能完整流程")
|
|
|
|
# 1. 测试代理连接
|
|
print(f"\n{'-'*40}")
|
|
print("1. 测试代理连接...")
|
|
|
|
for i in range(2):
|
|
proxy_config = get_proxy_config(i)
|
|
proxy_server = proxy_config['server'].replace('http://', '')
|
|
proxy_url = f"http://{proxy_config['username']}:{proxy_config['password']}@{proxy_server}"
|
|
|
|
print(f" 代理{i+1}: {proxy_config['server']} - {proxy_url}")
|
|
|
|
# 2. 演示如何使用代理登录（仅展示，不实际执行）
|
|
print(f"\n{'-'*40}")
|
|
print("2. 代理登录示例（代码演示）...")
|
|
print("""
|
|
# 登录示例代码：
|
|
async def example_login():
|
|
result = await login_with_proxy(
|
|
phone="你的手机号", # 实际手机号
|
|
code="验证码", # 实际验证码
|
|
proxy_index=0 # 使用代理1
|
|
)
|
|
return result
|
|
""")
|
|
|
|
# 3. 演示如何使用代理发布（仅展示，不实际执行）
|
|
print(f"\n{'-'*40}")
|
|
print("3. 代理发布示例（代码演示）...")
|
|
print("""
|
|
# 发布示例代码：
|
|
async def example_publish():
|
|
result = await publish_with_proxy(
|
|
title="测试标题",
|
|
content="测试内容",
|
|
images=["图片路径1", "图片路径2"], # 可选
|
|
tags=["标签1", "标签2"], # 可选
|
|
proxy_index=1, # 使用代理2
|
|
cookies_file="cookies.json" # Cookies文件路径
|
|
)
|
|
return result
|
|
""")
|
|
|
|
# 4. 代理轮换策略
|
|
print(f"\n{'-'*40}")
|
|
print("4. 代理轮换策略...")
|
|
print("""
|
|
# 代理轮换示例：
|
|
class ProxyManager:
|
|
def __init__(self):
|
|
self.current_proxy = 0
|
|
|
|
def get_next_proxy(self):
|
|
proxy_config = get_proxy_config(self.current_proxy)
|
|
self.current_proxy = (self.current_proxy + 1) % 2 # 循环使用两个代理
|
|
return proxy_config
|
|
""")
|
|
|
|
print(f"\n{'-'*40}")
|
|
print("✅ 代理功能演示完成！")
|
|
|
|
|
|
def main():
|
|
"""主函数"""
|
|
print("="*60)
|
|
print("🎯 固定代理IP下小红书登录发文功能示例")
|
|
print("="*60)
|
|
|
|
# 运行测试
|
|
asyncio.run(test_proxy_functionality())
|
|
|
|
print(f"\n{'='*60}")
|
|
print("💡 使用说明:")
|
|
print(" 1. 使用 login_with_proxy() 函数进行带代理的登录")
|
|
print(" 2. 使用 publish_with_proxy() 函数进行带代理的发布")
|
|
print(" 3. 可以轮换使用两个代理IP以提高稳定性")
|
|
print(" 4. 代理配置在 damai_proxy_config.py 中管理")
|
|
print("="*60)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# Windows环境下设置事件循环策略
|
|
if sys.platform == 'win32':
|
|
asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())
|
|
|
|
main() |