118 lines
4.5 KiB
Python
118 lines
4.5 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
Flask Web 登录小红书创作者中心
|
||
- 前端输入手机号和验证码
|
||
- 后端用 Playwright 自动登录
|
||
"""
|
||
|
||
from flask import Flask, render_template, request, redirect, url_for, flash, session
|
||
|
||
|
||
import time
|
||
from playwright.sync_api import sync_playwright
|
||
import requests
|
||
|
||
def get_working_proxy(test_url="https://creator.xiaohongshu.com"):
|
||
proxy_url = 'http://api.tianqiip.com/getip?secret=ew9mj7j3yplbk3xb&num=1&type=txt&port=1&time=3&mr=1&sign=5451e454a54b9f1f06222606c418e12f'
|
||
try:
|
||
resp = requests.get(proxy_url, timeout=10)
|
||
proxy_content = resp.content.decode("utf-8").strip()
|
||
proxy_ip, proxy_port = proxy_content.split(":", 1)
|
||
proxy_server = f"http://{proxy_ip}:{proxy_port}"
|
||
proxies = {"http": proxy_server, "https": proxy_server}
|
||
# 检查代理可用性
|
||
try:
|
||
test = requests.get(test_url, proxies=proxies, timeout=8)
|
||
if test.status_code == 200:
|
||
print(f"代理可用: {proxy_server}")
|
||
return proxy_server
|
||
else:
|
||
print(f"代理返回非200: {proxy_server}")
|
||
except Exception as e:
|
||
print(f"代理不可用: {proxy_server}, {e}")
|
||
except Exception as e:
|
||
print("获取代理失败", e)
|
||
return None
|
||
|
||
app = Flask(__name__)
|
||
app.secret_key = 'xhs_secret_key'
|
||
|
||
# 步骤1：输入手机号，自动打开浏览器并填手机号，等待用户手动点击发送验证码
|
||
@app.route('/', methods=['GET'])
|
||
def index():
|
||
return render_template('login.html')
|
||
|
||
@app.route('/send_code', methods=['POST'])
|
||
def send_code():
|
||
phone = request.form.get('phone')
|
||
if not phone:
|
||
flash('请输入手机号')
|
||
return redirect(url_for('index'))
|
||
session['phone'] = phone
|
||
# 获取可用代理
|
||
proxy_server = get_working_proxy()
|
||
if proxy_server:
|
||
print(f"使用代理: {proxy_server}")
|
||
else:
|
||
print("未获取到可用代理，使用本地IP")
|
||
|
||
with sync_playwright() as p:
|
||
launch_args = {"headless": False}
|
||
if proxy_server:
|
||
launch_args["proxy"] = {"server": proxy_server}
|
||
browser = p.chromium.launch(**launch_args)
|
||
context = browser.new_context()
|
||
page = context.new_page()
|
||
page.goto("https://creator.xiaohongshu.com/login?source=&redirectReason=401&lastUrl=%252Fnew%252Fhome")
|
||
page.get_by_role("textbox", name="手机号").click()
|
||
page.get_by_role("textbox", name="手机号").fill(phone)
|
||
page.get_by_text("发送验证码").click()
|
||
# 等待用户收到验证码
|
||
input("验证码已自动发送，请查收手机短信并输入验证码，操作完成后回到命令行按回车...")
|
||
# 保存 context 到磁盘，后续用
|
||
context.storage_state(path="state.json")
|
||
browser.close()
|
||
flash('验证码已发送，请查收手机短信并输入验证码')
|
||
return redirect(url_for('index'))
|
||
|
||
# 步骤2：输入验证码，自动登录
|
||
@app.route('/login', methods=['POST'])
|
||
def login():
|
||
code = request.form.get('code')
|
||
phone = session.get('phone')
|
||
if not code or not phone:
|
||
flash('请先输入手机号并发送验证码')
|
||
return redirect(url_for('index'))
|
||
# 获取可用代理
|
||
proxy_server = get_working_proxy()
|
||
if proxy_server:
|
||
print(f"使用代理: {proxy_server}")
|
||
else:
|
||
print("未获取到可用代理，使用本地IP")
|
||
|
||
with sync_playwright() as p:
|
||
launch_args = {"headless": False}
|
||
if proxy_server:
|
||
launch_args["proxy"] = {"server": proxy_server}
|
||
context = p.chromium.launch_persistent_context(user_data_dir=".", storage_state="state.json", **launch_args)
|
||
page = context.pages[0] if context.pages else context.new_page()
|
||
page.goto("https://creator.xiaohongshu.com/login?source=&redirectReason=401&lastUrl=%252Fnew%252Fhome")
|
||
# 填写验证码并登录
|
||
page.wait_for_selector('input[placeholder*="验证码"]', timeout=15000)
|
||
vcode_box = page.locator('input[placeholder*="验证码"]')
|
||
vcode_box.fill(code)
|
||
page.get_by_role("button", name="登 录").click()
|
||
# 等待跳转
|
||
try:
|
||
page.wait_for_url("**/new/home", timeout=20000)
|
||
flash('登录成功！')
|
||
# 保存 cookies
|
||
context.storage_state(path="cookies.json")
|
||
except Exception as e:
|
||
flash('登录失败，请检查验证码')
|
||
context.close()
|
||
return redirect(url_for('index'))
|
||
|
||
if __name__ == '__main__':
|
||
app.run(debug=True)
|