From 8fccaea6890d40f6c4b740237dcdfb5af86a56b2 Mon Sep 17 00:00:00 2001 From: 27942 Date: Fri, 27 Feb 2026 04:01:33 +0800 Subject: [PATCH] hahza --- MIGU_README.md | 99 ++++++++++++++++++++++++++ api_migu.py | 135 +++++++++++++++++++++++++++++++++++ migu_miguSM_dp.py | 163 +++++++++++++++++++++++++++++++++++++++++++ requirements.txt | 1 + run_migu_cli.py | 93 ++++++++++++++++++++++++ tgebrowser_client.py | 159 +++++++++++++++++++++++++++++++++++++++++ 6 files changed, 650 insertions(+) create mode 100644 MIGU_README.md create mode 100644 api_migu.py create mode 100644 migu_miguSM_dp.py create mode 100644 run_migu_cli.py create mode 100644 tgebrowser_client.py diff --git a/MIGU_README.md b/MIGU_README.md new file mode 100644 index 0000000..4edc497 --- /dev/null +++ b/MIGU_README.md @@ -0,0 +1,99 @@ +# 咪咕短剧 flows.cdyylkj.com/miguSM/home 自动化说明 + +使用 **DrissionPage** + **TgeBrowser** 实现自动化：输入手机号、点击发送验证码、填写验证码。 + +## 前置条件 + +1. **安装 TgeBrowser**：从 [https://tgebrowser.com/zh/download](https://tgebrowser.com/zh/download) 下载并安装 +2. **启动 TgeBrowser 客户端**，确保本地 API 服务运行（默认端口 50326） +3. **获取 API Key**：TgeBrowser 客户端 → API → 生成新密钥 +4. **设置环境变量**： + ```powershell + $env:TGEBROWSER_API_KEY = "你的API密钥" + ``` + +## 流程说明 + +- **步骤一**：传入电话号码 → 新建一个 TgeBrowser 浏览器 → 打开 miguSM 页面 → 输入手机号 → 点击发送验证码 → 返回成功 +- **步骤二**：传入验证码 → 在对应会话中填写验证码并提交 + +每个电话号码对应一个新建的浏览器实例。 + +## 使用方式 + +### 方式一：API 服务 + +```bash +# 安装依赖 +pip install -r requirements.txt + +# 启动 API（端口 8001） +python api_migu.py +``` + +**接口一：提交手机号** + +```http +POST http://127.0.0.1:8001/api/submit_phone +Content-Type: application/json + +{ + "phone": "13800138000", + "url": "https://flows.cdyylkj.com/miguSM/home" +} +``` + +响应示例： +```json +{ + "success": true, + "session_id": "uuid-xxx", + "message": "输入电话号码成功" +} +``` + +**接口二：提交验证码** + +```http +POST http://127.0.0.1:8001/api/submit_code +Content-Type: application/json + +{ + "session_id": "上一步返回的 session_id", + "code": "123456" +} +``` + +### 方式二：命令行 + +**推荐：交互模式（一步完成）** + +```bash +python run_migu_cli.py --phone 13800138000 -i +# 输入手机号、点击发送验证码后，会提示「请输入短信验证码」，输入收到的验证码即可 +``` + +**分步执行** + +```bash +# 步骤一 +python run_migu_cli.py --phone 13800138000 + +# 步骤二（仅在同一 shell 内、步骤一之后执行，因为依赖内存中的 page） +python run_migu_cli.py --code 123456 +``` + +> 跨进程/跨请求的会话请使用 API 模式（`api_migu.py`）。 + +## 选择器说明 + +若目标页面结构变化导致找不到输入框或按钮，可修改 `migu_miguSM_dp.py` 中的选择器列表（`_find_first` 所用选择器）。当前已适配常见 H5 表单结构。 + +## 文件说明 + +| 文件 | 说明 | +|------|------| +| `tgebrowser_client.py` | TgeBrowser REST API 客户端（创建/启动/停止浏览器） | +| `migu_miguSM_dp.py` | DrissionPage 自动化逻辑（输入手机号、发送验证码、填写验证码） | +| `api_migu.py` | FastAPI 接口（submit_phone / submit_code） | +| `run_migu_cli.py` | 命令行入口 | diff --git a/api_migu.py b/api_migu.py new file mode 100644 index 0000000..88c2dd3 --- /dev/null +++ b/api_migu.py @@ -0,0 +1,135 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +咪咕短剧 flows.cdyylkj.com/miguSM/home 接口： +- POST /api/submit_phone 传入电话号码 → 新建 TgeBrowser → 输入手机号、点击发送验证码 → 返回成功 +- POST /api/submit_code 传入验证码 → 在对应会话中填写验证码并提交 + +每个电话号码对应一个新建的 TgeBrowser 浏览器会话。 +""" +from __future__ import annotations + +import threading +import uuid +from typing import Any, Optional + +from fastapi import FastAPI, HTTPException +from pydantic import BaseModel, Field + +from migu_miguSM_dp import ( + MIGU_HOME_URL, + connect_dp_to_tgebrowser, + connect_dp_to_ws, + input_code_and_submit, + input_phone_and_send_code, +) +from tgebrowser_client import TgeBrowserClient + +app = FastAPI(title="咪咕短剧 miguSM 自动化 API", description="TgeBrowser + DrissionPage 手机号与验证码提交流程") + +# 会话存储：session_id -> {page, env_id, client} +_sessions: dict[str, dict[str, Any]] = {} +_sessions_lock = threading.Lock() + + +class SubmitPhoneRequest(BaseModel): + phone: str = Field(..., description="手机号码") + url: str = Field(MIGU_HOME_URL, description="目标页面 URL") + + +class SubmitPhoneResponse(BaseModel): + success: bool = True + session_id: str = Field(..., description="会话 ID，submit_code 时需传入") + message: str = "输入电话号码成功" + + +class SubmitCodeRequest(BaseModel): + session_id: str = Field(..., description="submit_phone 返回的会话 ID") + code: str = Field(..., description="短信验证码") + + +class SubmitCodeResponse(BaseModel): + success: bool = True + message: str = "" + + +@app.post("/api/submit_phone", response_model=SubmitPhoneResponse) +def api_submit_phone(req: SubmitPhoneRequest): + """ + 步骤一：传入电话号码。 + 新建 TgeBrowser 浏览器 → 打开 miguSM 页面 → 输入手机号 → 点击发送验证码。 + 返回 session_id，用于后续 submit_code。 + """ + try: + client = TgeBrowserClient() + # 每次输入电话号码新建一个浏览器 + start_data = client.create_and_start( + browser_name=f"miguSM_{req.phone[-4:]}", + start_page_url=req.url, + ) + port = start_data.get("port") + ws = start_data.get("ws") + env_id = start_data.get("envId") + + # 优先用端口连接（DrissionPage 对端口支持更好） + if port: + page = connect_dp_to_tgebrowser(port) + elif ws: + page = connect_dp_to_ws(ws) + else: + raise HTTPException(status_code=500, detail="TgeBrowser 未返回 port 或 ws") + + input_phone_and_send_code(page, req.phone, url=req.url) + + session_id = str(uuid.uuid4()) + with _sessions_lock: + _sessions[session_id] = { + "page": page, + "env_id": env_id, + "client": client, + } + + return SubmitPhoneResponse(success=True, session_id=session_id, message="输入电话号码成功") + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + + +@app.post("/api/submit_code", response_model=SubmitCodeResponse) +def api_submit_code(req: SubmitCodeRequest): + """ + 步骤二：传入验证码。 + 在 submit_phone 创建的会话中填写验证码并提交。 + """ + with _sessions_lock: + sess = _sessions.get(req.session_id) + if not sess: + raise HTTPException(status_code=404, detail=f"会话不存在或已过期: {req.session_id}") + + page = sess["page"] + try: + result = input_code_and_submit(page, req.code) + return SubmitCodeResponse(success=True, message=result.get("message", "验证码已填写")) + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + + +@app.delete("/api/session/{session_id}") +def close_session(session_id: str): + """关闭并清理指定会话（可选，用于释放浏览器）。""" + with _sessions_lock: + sess = _sessions.pop(session_id, None) + if not sess: + return {"success": False, "message": "会话不存在"} + try: + client = sess.get("client") + env_id = sess.get("env_id") + if client and env_id is not None: + client.stop_browser(env_id=env_id) + except Exception: + pass + return {"success": True, "message": "会话已关闭"} + + +if __name__ == "__main__": + import uvicorn + uvicorn.run(app, host="0.0.0.0", port=8001) diff --git a/migu_miguSM_dp.py b/migu_miguSM_dp.py new file mode 100644 index 0000000..d9b485e --- /dev/null +++ b/migu_miguSM_dp.py @@ -0,0 +1,163 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +咪咕短剧超级会员 flows.cdyylkj.com/miguSM/home 自动化（DrissionPage + TgeBrowser）： +1) 新建 TgeBrowser 浏览器 +2) 打开页面，输入手机号，点击发送验证码 → 返回成功 +3) 接收验证码并填写、提交 +""" +from __future__ import annotations + +import time +from typing import Any, Optional + +from DrissionPage import ChromiumOptions, ChromiumPage + + +MIGU_HOME_URL = "https://flows.cdyylkj.com/miguSM/home" + + +def _find_first(page, selectors: list[str], timeout: float = 8): + """尝试多个选择器，返回第一个找到的元素。""" + for sel in selectors: + try: + ele = page.ele(sel, timeout=timeout) + if ele: + return ele + except Exception: + continue + return None + + +def _click_safe(ele) -> None: + try: + ele.click() + except Exception: + try: + ele.click(by_js=True) + except Exception: + ele.run_js("this.click()") + + +def input_phone_and_send_code( + page: ChromiumPage, + phone: str, + url: str = MIGU_HOME_URL, + wait_after_open: float = 1.0, +) -> dict: + """ + 步骤一：输入手机号、点击发送验证码。 + 返回 {"success": True, "message": "输入电话号码成功"} 或抛出异常。 + """ + page.get(url) + time.sleep(wait_after_open) + + # 手机号输入框（按常见 H5 表单结构） + phone_input = _find_first(page, [ + 'x://input[@placeholder*="手机"]', + 'x://input[@placeholder*="电话"]', + 'x://input[@placeholder*="号码"]', + 'x://input[@type="tel"]', + 'x://input[@type="number"]', + "css:input[type='tel']", + "css:input[placeholder*='手机']", + "css:input[placeholder*='电话']", + "css:input.inp-txt", + "css:.phone-input input", + "css:input.phone", + ], timeout=10) + if not phone_input: + raise RuntimeError("未找到手机号输入框，请根据页面调整选择器") + + phone_input.input(phone, clear=True) + time.sleep(0.3) + + # 可选：勾选协议（若页面有） + agree = _find_first(page, [ + "css:input[type='checkbox']", + "css:.agree-checkbox", + 'x://input[@type="checkbox"]', + 'x://i[contains(@class,"checkbox")]', + ], timeout=2) + if agree: + try: + _click_safe(agree) + time.sleep(0.2) + except Exception: + pass + + # 发送验证码按钮 + send_btn = _find_first(page, [ + 'x://button[contains(text(),"获取验证码")]', + 'x://span[contains(text(),"获取验证码")]', + 'x://*[contains(text(),"获取验证码")]', + 'x://button[contains(text(),"发送验证码")]', + 'x://*[contains(text(),"发送验证码")]', + "css:button.btn-code", + "css:.send-code", + "css:.get-code", + "css:button[class*='code']", + "css:.verify-btn", + ], timeout=8) + if not send_btn: + raise RuntimeError("未找到「获取验证码」按钮，请根据页面调整选择器") + + _click_safe(send_btn) + time.sleep(0.5) + + return {"success": True, "message": "输入电话号码成功"} + + +def input_code_and_submit( + page: ChromiumPage, + code: str, +) -> dict: + """ + 步骤二：填写验证码并提交（若存在提交按钮）。 + 返回 {"success": True, "message": "验证码已填写"}。 + """ + code_input = _find_first(page, [ + 'x://input[@placeholder*="验证码"]', + 'x://input[@placeholder*="短信"]', + 'x://input[@placeholder*="验证"]', + "css:input[placeholder*='验证码']", + "css:input[placeholder*='短信']", + "css:input.code-input", + "css:input.verify-input", + "css:input[type='text']", + "css:input.inp-txt", + ], timeout=8) + if not code_input: + raise RuntimeError("未找到验证码输入框") + + code_input.input(code, clear=True) + time.sleep(0.2) + + # 若有确认/提交按钮则点击 + submit_btn = _find_first(page, [ + 'x://button[contains(text(),"确认")]', + 'x://button[contains(text(),"提交")]', + 'x://*[contains(text(),"确认")]', + 'x://*[contains(text(),"登录")]', + 'x://*[contains(text(),"绑定")]', + "css:button.btn-primary", + "css:button.btn-buy", + "css:.submit-btn", + "css:.confirm-btn", + "css:img.btn-buy", + ], timeout=5) + if submit_btn: + _click_safe(submit_btn) + + return {"success": True, "message": "验证码已填写"} + + +def connect_dp_to_tgebrowser(port: int) -> ChromiumPage: + """通过调试端口连接 TgeBrowser 已启动的浏览器。""" + co = ChromiumOptions().set_local_port(port=port) + return ChromiumPage(addr_or_opts=co) + + +def connect_dp_to_ws(ws_url: str) -> ChromiumPage: + """通过 WebSocket 地址连接 TgeBrowser。""" + return ChromiumPage(addr_or_opts=ws_url) diff --git a/requirements.txt b/requirements.txt index c9b73bf..0a20ebf 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,5 +1,6 @@ # 项目依赖 DrissionPage>=4.0.0 +requests>=2.28.0 Pillow>=10.0.0 numpy>=1.24.0 fastapi>=0.100.0 diff --git a/run_migu_cli.py b/run_migu_cli.py new file mode 100644 index 0000000..94e9bb5 --- /dev/null +++ b/run_migu_cli.py @@ -0,0 +1,93 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +咪咕 miguSM 命令行工具（TgeBrowser + DrissionPage）： + 1. 传入手机号 → 新建浏览器、输入手机号、点击发送验证码 + 2. 传入验证码 → 填写并提交 + +用法: + set TGEBROWSER_API_KEY=你的API密钥 + python run_migu_cli.py --phone 13800138000 + python run_migu_cli.py --session-id <上一步返回的session_id> --code 123456 +""" +from __future__ import annotations + +import argparse +import os +import sys + +# 确保 TgeBrowser 客户端和 migu 模块可导入 +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +from migu_miguSM_dp import ( + MIGU_HOME_URL, + connect_dp_to_tgebrowser, + connect_dp_to_ws, + input_code_and_submit, + input_phone_and_send_code, +) +from tgebrowser_client import TgeBrowserClient + + +def cmd_submit_phone(phone: str, url: str) -> str: + """步骤一：输入手机号并发送验证码，返回 session_id（此处为 env_id 的字符串形式，用于后续填写验证码）。""" + client = TgeBrowserClient() + start_data = client.create_and_start( + browser_name=f"miguSM_{phone[-4:]}", + start_page_url=url, + ) + port = start_data.get("port") + ws = start_data.get("ws") + env_id = start_data.get("envId") + + if port: + page = connect_dp_to_tgebrowser(port) + elif ws: + page = connect_dp_to_ws(ws) + else: + raise RuntimeError("TgeBrowser 未返回 port 或 ws") + + input_phone_and_send_code(page, phone, url=url) + print("输入电话号码成功，请查收短信验证码") + # 保存到全局，供 submit_code 使用（CLI 模式只支持单次流程） + cmd_submit_phone._page = page # type: ignore + cmd_submit_phone._client = client # type: ignore + cmd_submit_phone._env_id = env_id # type: ignore + return str(env_id or "ok") + + +def cmd_submit_code(code: str) -> None: + """步骤二：填写验证码并提交。""" + page = getattr(cmd_submit_phone, "_page", None) + if not page: + raise RuntimeError("请先执行步骤一：python run_migu_cli.py --phone 13800138000") + result = input_code_and_submit(page, code) + print(result.get("message", "验证码已填写")) + + +def main(): + parser = argparse.ArgumentParser(description="咪咕 miguSM 自动化 CLI") + parser.add_argument("--phone", help="手机号码（步骤一）") + parser.add_argument("--code", help="短信验证码（步骤二）") + parser.add_argument("--interactive", "-i", action="store_true", help="步骤一后等待输入验证码（同一进程完成两步）") + parser.add_argument("--url", default=MIGU_HOME_URL, help=f"目标页面，默认 {MIGU_HOME_URL}") + args = parser.parse_args() + + if args.phone: + cmd_submit_phone(args.phone, args.url) + if args.interactive: + code = input("请输入短信验证码: ").strip() + if code: + cmd_submit_code(code) + elif args.code: + cmd_submit_code(args.code) + else: + parser.print_help() + print("\n示例:") + print(" python run_migu_cli.py --phone 13800138000 -i # 输入手机号后交互输入验证码") + print(" python run_migu_cli.py --phone 13800138000 # 仅步骤一") + print(" python run_migu_cli.py --code 123456 # 仅步骤二（需在 API 模式或同一进程的步骤一之后）") + + +if __name__ == "__main__": + main() diff --git a/tgebrowser_client.py b/tgebrowser_client.py new file mode 100644 index 0000000..a29d8ae --- /dev/null +++ b/tgebrowser_client.py @@ -0,0 +1,159 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +TgeBrowser API 客户端：用于通过 REST API 新建/启动浏览器环境，获取调试端口。 +文档: https://docs.tgebrowser.com/api.html +""" +from __future__ import annotations + +import os +from typing import Any, Optional + +import requests + +# TgeBrowser 本地 API 默认端口 +DEFAULT_API_BASE = "http://127.0.0.1:50326" +# API Key（写死，也可通过环境变量 TGEBROWSER_API_KEY 覆盖） +DEFAULT_API_KEY = "asp_ad11de1727c77a5d514256b3209eec52130c6e5b4f9ccd92" +ENV_API_KEY = "TGEBROWSER_API_KEY" + + +def get_api_key() -> str: + """优先从环境变量获取 API Key，未配置时使用默认值。""" + key = os.environ.get(ENV_API_KEY) + if key and key.strip(): + return key.strip() + return DEFAULT_API_KEY + + +class TgeBrowserClient: + """TgeBrowser 本地 API 客户端。""" + + def __init__(self, base_url: str = DEFAULT_API_BASE, api_key: Optional[str] = None): + self.base_url = base_url.rstrip("/") + self.api_key = api_key or get_api_key() + self._headers = { + "Authorization": f"Bearer {self.api_key}", + "Content-Type": "application/json", + } + + def _post(self, path: str, json: Optional[dict] = None) -> dict: + resp = requests.post( + f"{self.base_url}{path}", + headers=self._headers, + json=json or {}, + timeout=30, + ) + resp.raise_for_status() + data = resp.json() + if not data.get("success", False): + msg = data.get("message", "未知错误") + raise RuntimeError(f"TgeBrowser API 失败: {msg}") + return data + + def _get(self, path: str) -> dict: + resp = requests.get( + f"{self.base_url}{path}", + headers=self._headers, + timeout=30, + ) + resp.raise_for_status() + data = resp.json() + if not data.get("success", False): + msg = data.get("message", "未知错误") + raise RuntimeError(f"TgeBrowser API 失败: {msg}") + return data + + def status(self) -> dict: + """检查 API 服务状态。""" + return self._get("/api/status") + + def create_browser( + self, + browser_name: str = "miguSM_automation", + start_page_url: Optional[str] = None, + **kwargs: Any, + ) -> dict: + """ + 创建新的浏览器环境。 + 返回 data.envId 用于后续 start。 + """ + # 使用 TgeBrowser 文档要求的完整结构，避免 400 + start_page_value = [start_page_url] if start_page_url else [] + start_page_mode = "custom" if start_page_value else "none" + + # 指定手机端指纹，随机生成手机 UA + mobile_fingerprint = { + "os": "Android", + "platformVersion": 12, + } + payload = { + "browserName": browser_name, + "proxy": {"protocol": "direct"}, + "fingerprint": mobile_fingerprint, + "startInfo": { + "startPage": {"mode": start_page_mode, "value": start_page_value}, + "otherConfig": {"openConfigPage": False, "checkPage": False}, + }, + **kwargs, + } + data = self._post("/api/browser/create", json=payload) + return data.get("data", {}) + + def start_browser( + self, + env_id: Optional[int] = None, + user_index: Optional[int] = None, + port: Optional[int] = None, + **kwargs: Any, + ) -> dict: + """ + 启动浏览器环境。 + 返回 data 含 ws、port、http 等，用于 DrissionPage 连接。 + """ + payload: dict[str, Any] = {} + if env_id is not None: + payload["envId"] = env_id + elif user_index is not None: + payload["userIndex"] = user_index + else: + raise ValueError("必须指定 envId 或 userIndex") + + if port is not None: + payload["port"] = port + payload.update(kwargs) + + data = self._post("/api/browser/start", json=payload) + return data.get("data", {}) + + def stop_browser(self, env_id: Optional[int] = None, user_index: Optional[int] = None) -> dict: + """停止浏览器环境。""" + payload: dict[str, Any] = {} + if env_id is not None: + payload["envId"] = env_id + elif user_index is not None: + payload["userIndex"] = user_index + else: + raise ValueError("必须指定 envId 或 userIndex") + return self._post("/api/browser/stop", json=payload) + + def create_and_start( + self, + browser_name: str = "miguSM_automation", + start_page_url: Optional[str] = None, + ) -> dict: + """ + 创建并启动新浏览器，每次调用都会新建一个环境。 + 返回含 ws、port、envId 等，供 DrissionPage 连接。 + """ + create_data = self.create_browser( + browser_name=browser_name, + start_page_url=start_page_url, + ) + env_id = create_data.get("envId") + if env_id is None: + raise RuntimeError("创建浏览器失败：未返回 envId") + + start_data = self.start_browser(env_id=env_id) + start_data["envId"] = env_id + return start_data