diff --git a/sau_cli.py b/sau_cli.py index 73b2ab4..0a12602 100644 --- a/sau_cli.py +++ b/sau_cli.py @@ -48,6 +48,11 @@ from uploader.weibo_uploader.main import ( weibo_setup, cookie_auth as weibo_cookie_auth, ) +from uploader.hupu_uploader.main import ( + HuPuVideo, + hupu_setup, + cookie_auth as hupu_cookie_auth, +) from uploader.xiaohongshu_uploader.main import ( XIAOHONGSHU_PUBLISH_STRATEGY_IMMEDIATE, XIAOHONGSHU_PUBLISH_STRATEGY_SCHEDULED, @@ -225,6 +230,18 @@ class WeiboVideoUploadRequest: headless: bool = True +@dataclass(slots=True) +class HupuVideoUploadRequest: + account_name: str + video_file: Path + title: str + description: str + tags: list[str] + thumbnail_file: Path | None = None + debug: bool = True + headless: bool = True + + @dataclass(slots=True) class YouTubeVideoUploadRequest: account_name: str @@ -714,6 +731,40 @@ async def upload_weibo_video(request: WeiboVideoUploadRequest) -> Path: return account_file +async def login_hupu_account(account_name: str, headless: bool = False, qrcode_callback=None) -> dict: + account_file = resolve_account_file("hupu", account_name) + return await hupu_setup(str(account_file), handle=True, return_detail=True, headless=headless, qrcode_callback=qrcode_callback) + + +async def check_hupu_account(account_name: str) -> bool: + account_file = resolve_account_file("hupu", account_name) + if not account_file.exists(): + return False + return await hupu_cookie_auth(str(account_file)) + + +async def upload_hupu_video(request: HupuVideoUploadRequest) -> Path: + account_file = resolve_account_file("hupu", request.account_name) + is_ready = await hupu_setup(str(account_file), handle=False) + if not is_ready: + raise RuntimeError( + f"Hupu cookie is missing or expired: {account_file}. Run `sau hupu login --account {request.account_name}` first." + ) + + app = HuPuVideo( + title=request.title, + file_path=str(request.video_file), + tags=request.tags, + account_file=str(account_file), + desc=request.description, + thumbnail_path=str(request.thumbnail_file) if request.thumbnail_file else None, + debug=request.debug, + headless=request.headless, + ) + await app.main() + return account_file + + def existing_file_path(value: str) -> Path: path = Path(value) if not path.is_file(): @@ -923,6 +974,25 @@ def build_parser() -> argparse.ArgumentParser: weibo_upload_video_parser.add_argument("--collection", default=None, help="Optional collection name") add_runtime_flags(weibo_upload_video_parser) + # ── Hupu ── + hupu_parser = platform_parsers.add_parser("hupu", help="Hupu (虎扑) operations") + hupu_actions = hupu_parser.add_subparsers(dest="action", required=True) + + for action_name in ("login", "check"): + action_parser = hupu_actions.add_parser(action_name, help=f"Hupu {action_name}") + action_parser.add_argument("--account", required=True, help="Hupu user-defined account_name") + if action_name == "login": + add_runtime_flags(action_parser) + + hupu_upload_video_parser = hupu_actions.add_parser("upload-video", help="Upload one video to Hupu") + hupu_upload_video_parser.add_argument("--account", required=True, help="Hupu user-defined account_name") + hupu_upload_video_parser.add_argument("--file", required=True, type=existing_file_path, help="Video file path") + hupu_upload_video_parser.add_argument("--title", required=True, help="Video title (4-40 chars)") + hupu_upload_video_parser.add_argument("--desc", default="", help="Optional video description") + hupu_upload_video_parser.add_argument("--tags", default="", help="Comma-separated tags, such as tag1,tag2") + hupu_upload_video_parser.add_argument("--thumbnail", type=existing_file_path, help="Optional cover image path") + add_runtime_flags(hupu_upload_video_parser) + youtube_parser = platform_parsers.add_parser("youtube", help="YouTube operations") youtube_actions = youtube_parser.add_subparsers(dest="action", required=True) @@ -1276,6 +1346,36 @@ async def dispatch(args: argparse.Namespace) -> int: raise RuntimeError(f"Unsupported Weibo action: {args.action}") + if args.platform == "hupu": + if args.action == "login": + result = await login_hupu_account(args.account, headless=args.headless) + if not result["success"]: + raise RuntimeError(result["message"]) + print(f"Hupu login flow completed: {result['account_file']}") + return 0 + + if args.action == "check": + is_valid = await check_hupu_account(args.account) + print("valid" if is_valid else "invalid") + return 0 if is_valid else 1 + + if args.action == "upload-video": + request = HupuVideoUploadRequest( + account_name=args.account, + video_file=args.file, + title=args.title, + description=args.desc, + tags=parse_tags(args.tags), + thumbnail_file=args.thumbnail, + debug=args.debug, + headless=args.headless, + ) + await upload_hupu_video(request) + print(f"Hupu video upload submitted: {request.video_file}") + return 0 + + raise RuntimeError(f"Unsupported Hupu action: {args.action}") + if args.platform == "youtube": if args.action == "login": result = await login_youtube_account(args.account, headless=args.headless) diff --git a/uploader/hupu_uploader/__init__.py b/uploader/hupu_uploader/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/uploader/hupu_uploader/main.py b/uploader/hupu_uploader/main.py new file mode 100644 index 0000000..1efdaa9 --- /dev/null +++ b/uploader/hupu_uploader/main.py @@ -0,0 +1,523 @@ +# -*- coding: utf-8 -*- +"""虎扑视频上传 + 手动登录保存 cookie。 + +功能： + - hupu_cookie_gen: 打开浏览器让用户手动登录（QQ/手机号等），保存 storage_state + - cookie_auth: 验证 cookie 是否有效 + - hupu_setup: 统一入口（检查/触发登录） + - HuPuVideo: 视频上传类 + +基于 playwright codegen 录制脚本改写。 +发布页：https://bbs.hupu.com/newpost?tabkey=2（视频发布标签页） +专区：固定选择「步行街 → 步行街主干道」 + +注意：虎扑有 headless 检测，需配合反检测参数才能正常操作。 +""" +from __future__ import annotations + +import asyncio +import inspect +import os +import re +import time +from pathlib import Path + +from playwright.async_api import BrowserContext, Page, Playwright, async_playwright + +from conf import BASE_DIR, LOCAL_CHROME_HEADLESS, LOCAL_CHROME_PATH +from uploader.base_video import BaseVideoUploader +from utils.log import hupu_logger + + +HUPU_HOME_URL = "https://www.hupu.com/" +HUPU_LOGIN_URL = "https://passport.hupu.com/v2/login?pcPhone=1&jumpurl=https://www.hupu.com&from=https://www.hupu.com" +HUPU_PUBLISH_URL = "https://bbs.hupu.com/newpost?tabkey=2" + +# 虎扑发布成功后跳转的帖子 URL 模式 +HUPU_POST_URL_PATTERN = re.compile(r"bbs\.hupu\.com/\d+\.html") + +# 反检测 UA +_CHROME_UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/127.0.0.0 Safari/537.36" + +# 反 webdriver 检测脚本 +_STEALTH_SCRIPT = """ +Object.defineProperty(navigator, 'webdriver', { get: () => undefined }); +""" + + +def _msg(emoji: str, text: str) -> str: + return f"{emoji} {text}" + + +def _build_login_result(success: bool, status: str, message: str, account_file: str, current_url: str = "") -> dict: + return { + "success": success, + "status": status, + "message": message, + "account_file": str(account_file), + "current_url": current_url, + } + + +async def _emit_qrcode_callback(qrcode_callback, payload: dict): + if not qrcode_callback: + return + callback_result = qrcode_callback(payload) + if inspect.isawaitable(callback_result): + await callback_result + + +def _build_launch_kwargs(headless: bool) -> dict: + launch_kwargs = { + "headless": headless, + "args": ["--disable-blink-features=AutomationControlled"], + } + if LOCAL_CHROME_PATH: + launch_kwargs["executable_path"] = LOCAL_CHROME_PATH + return launch_kwargs + + +def _resolve_account_file(account_file: str | Path) -> str: + path = Path(account_file).expanduser() + if path.is_absolute(): + return str(path) + if len(path.parts) == 1: + return str((Path(BASE_DIR) / "cookies" / path).resolve()) + return str(path.resolve()) + + +async def _create_stealth_context(browser, account_file: str | None = None) -> BrowserContext: + """创建带反检测的 context。""" + kwargs = { + "user_agent": _CHROME_UA, + "viewport": {"width": 1920, "height": 1080}, + } + if account_file and os.path.exists(account_file): + kwargs["storage_state"] = account_file + context = await browser.new_context(**kwargs) + return context + + +async def _new_stealth_page(context: BrowserContext) -> Page: + """创建带反检测 init script 的 page。""" + page = await context.new_page() + await page.add_init_script(_STEALTH_SCRIPT) + return page + + +async def hupu_cookie_gen(account_file, qrcode_callback=None, poll_interval: int = 3, max_checks: int = 120, headless: bool = False): + """QQ 扫码登录虎扑，保存 cookie。 + + 流程：打开虎扑登录页 → 点击 QQ 登录 → 截取 QQ 二维码 → 等待扫码完成 → 保存 storage_state。 + 支持 headless 模式（终端显示二维码）。 + """ + account_file = _resolve_account_file(account_file) + Path(account_file).parent.mkdir(parents=True, exist_ok=True) + result = _build_login_result(False, "failed", "虎扑登录失败", account_file) + + async with async_playwright() as playwright: + browser = await playwright.chromium.launch(**_build_launch_kwargs(headless=headless)) + context = await _create_stealth_context(browser) + try: + page = await _new_stealth_page(context) + await page.goto(HUPU_LOGIN_URL, timeout=60000, wait_until="load") + await page.wait_for_timeout(3000) + + # 点击 QQ 登录按钮 + qq_btn = page.get_by_role("button", name="qq QQ登录") + if not await qq_btn.count(): + hupu_logger.error(_msg("😢", "未找到 QQ 登录按钮")) + result = _build_login_result(False, "failed", "未找到 QQ 登录按钮", account_file, page.url) + return result + + await qq_btn.click() + await page.wait_for_timeout(5000) + hupu_logger.info(_msg("🏃", "已跳转到 QQ 登录页")) + + # 从 QQ iframe 中获取二维码 + qrcode_info = await _grab_qq_qrcode(page, context, account_file) + + if qrcode_info: + await _emit_qrcode_callback(qrcode_callback, qrcode_info) + hupu_logger.info(_msg("🧍", "请用 QQ 手机版扫码登录")) + else: + hupu_logger.warning(_msg("⚠️", "未能获取 QQ 二维码，请在浏览器中手动扫码")) + + # 轮询等待登录完成 + for _ in range(max_checks): + current_url = page.url + # QQ 授权成功后会跳回虎扑首页 + if "www.hupu.com" in current_url and "passport" not in current_url and "graph.qq.com" not in current_url: + hupu_logger.info(_msg("🥳", f"登录成功，跳转到: {current_url}")) + result = _build_login_result(True, "success", "虎扑 QQ 扫码登录成功", account_file, current_url) + break + # 检查是否实际跳转到虎扑 passport 回调页（不是 redirect_uri 参数中包含） + if current_url.startswith("https://passport.hupu.com/pc/qqcallback"): + await page.wait_for_timeout(5000) + current_url = page.url + hupu_logger.info(_msg("🥳", f"QQ 回调成功，当前: {current_url}")) + result = _build_login_result(True, "success", "虎扑 QQ 扫码登录成功", account_file, current_url) + break + # 检查 cookies 中是否出现 u（核心登录态） + cookies = await context.cookies() + if any(c.get("name") == "u" and c.get("value") for c in cookies): + hupu_logger.info(_msg("🥳", f"登录成功（检测到 u cookie），当前: {current_url}")) + result = _build_login_result(True, "success", "虎扑 QQ 扫码登录成功", account_file, current_url) + break + await page.wait_for_timeout(poll_interval * 1000) + else: + result = _build_login_result(False, "timeout", "等待 QQ 扫码登录超时", account_file, page.url) + + if result["success"]: + await asyncio.sleep(2) + # 确保跳转到首页加载完 cookie + if "www.hupu.com" not in page.url: + await page.goto(HUPU_HOME_URL, timeout=30000, wait_until="domcontentloaded") + await page.wait_for_timeout(3000) + await context.storage_state(path=account_file) + hupu_logger.success(_msg("🥳", f"cookie 已保存: {account_file}")) + except Exception as exc: + result = _build_login_result(False, "failed", str(exc), account_file, current_url=page.url if "page" in locals() else "") + finally: + # 清理临时二维码文件 + qr_path = Path(account_file).parent / f"{Path(account_file).stem}_qq_qrcode.png" + if qr_path.exists(): + qr_path.unlink() + if not result["success"]: + hupu_logger.error(_msg("😢", f"登录失败: {result['message']}")) + await context.close() + await browser.close() + return result + + +async def _grab_qq_qrcode(page: Page, context: BrowserContext, account_file: str) -> dict | None: + """从 QQ 登录 iframe 中获取二维码。""" + from utils.login_qrcode import decode_qrcode_from_path, print_terminal_qrcode + + # 等待 QQ iframe 加载（最多 15 秒） + qq_frame = None + for _ in range(5): + for frame in page.frames: + if "xui.ptlogin2.qq.com" in frame.url or "ptlogin2.qq.com" in frame.url: + qq_frame = frame + break + if qq_frame: + break + await asyncio.sleep(3) + + if not qq_frame: + return None + + await asyncio.sleep(3) + + # 获取二维码图片（id="qrlogin_img"） + qr_selectors = ["#qrlogin_img", 'img[src*="ptqrshow"]', 'img[id*="qr"]'] + for sel in qr_selectors: + qr_loc = qq_frame.locator(sel).first + if await qr_loc.count(): + src = await qr_loc.get_attribute("src") + if src and src.startswith("http"): + # 下载二维码图片 + try: + resp = await context.request.get(src) + qr_path = Path(account_file).parent / f"{Path(account_file).stem}_qq_qrcode.png" + qr_path.parent.mkdir(parents=True, exist_ok=True) + qr_path.write_bytes(await resp.body()) + + # 尝试解码并在终端显示 + qrcode_content = decode_qrcode_from_path(qr_path) + if qrcode_content: + print_terminal_qrcode(qrcode_content, qr_path, "QQ手机版") + else: + hupu_logger.warning(_msg("😵", f"终端无法显示二维码，请打开 {qr_path} 扫码")) + + return {"image_path": str(qr_path), "image_data_url": ""} + except Exception as exc: + hupu_logger.warning(_msg("⚠️", f"下载 QQ 二维码失败: {exc}")) + continue + + return None + + +async def cookie_auth(account_file): + """验证虎扑 cookie 是否有效。访问发布页，检测是否能正常加载。""" + account_file = _resolve_account_file(account_file) + async with async_playwright() as playwright: + browser = await playwright.chromium.launch(**_build_launch_kwargs(headless=True)) + try: + context = await _create_stealth_context(browser, account_file) + page = await _new_stealth_page(context) + await page.goto(HUPU_PUBLISH_URL, timeout=60000, wait_until="domcontentloaded") + await page.wait_for_timeout(5000) + + # 检查是否被跳转到登录页 + if "passport" in page.url or "login" in page.url: + hupu_logger.info(_msg("🥹", "cookie 已失效（跳转到登录页）")) + return False + + # 检查发布页是否出现「上传视频」按钮（登录后才有） + upload_btn = page.get_by_role("button", name="上传视频") + if await upload_btn.count(): + hupu_logger.success(_msg("🥳", "cookie 有效")) + return True + + # 兜底：检查 cookie 中是否有 u 字段 + cookies = await context.cookies() + if any(c.get("name") == "u" and c.get("value") for c in cookies): + hupu_logger.success(_msg("🥳", "cookie 有效（u cookie 存在）")) + return True + + hupu_logger.info(_msg("🥹", "cookie 已失效（未检测到登录态）")) + return False + except Exception as exc: + hupu_logger.warning(_msg("😵", f"cookie 校验出错，按失效处理: {exc}")) + return False + finally: + await browser.close() + + +async def hupu_setup(account_file, handle=False, return_detail=False, qrcode_callback=None, headless: bool = False): + """统一入口：检查 cookie → 如无效且 handle=True 则触发手动登录。""" + account_file = _resolve_account_file(account_file) + if not os.path.exists(account_file) or not await cookie_auth(account_file): + if not handle: + result = _build_login_result(False, "cookie_invalid", "cookie 文件不存在或已失效", account_file) + return result if return_detail else False + hupu_logger.info(_msg("🥹", "cookie 文件不存在或已失效，打开浏览器请手动登录")) + result = await hupu_cookie_gen(account_file, qrcode_callback=qrcode_callback, headless=headless) + return result if return_detail else result["success"] + + result = _build_login_result(True, "cookie_valid", "cookie 有效", account_file) + return result if return_detail else True + + +class HuPuVideo(BaseVideoUploader): + """虎扑视频上传。 + + 流程：直接跳转发布页 → 上传视频文件 → 填标题 → 填简介 → + 上传封面 → 选专区（步行街主干道）→ 选原创/二创 → 选 AI 声明 → + 点击「确定发布」→ 等待跳转到帖子页面。 + """ + + def __init__( + self, + title, + file_path, + tags, + account_file, + publish_date=0, + desc: str | None = None, + thumbnail_path: str | None = None, + debug: bool = True, + headless: bool = LOCAL_CHROME_HEADLESS, + ): + self.title = title + self.file_path = file_path + self.tags = tags or [] + self.account_file = _resolve_account_file(account_file) + self.publish_date = publish_date + self.desc = desc or "" + self.thumbnail_path = thumbnail_path + self.debug = debug + self.headless = headless + self.local_executable_path = LOCAL_CHROME_PATH + self.max_title_length = 40 + self.min_title_length = 4 + + async def validate_upload_args(self): + if not os.path.exists(self.account_file): + raise RuntimeError(f"cookie文件不存在，请先完成虎扑登录: {self.account_file}") + if not await cookie_auth(self.account_file): + raise RuntimeError(f"cookie文件已失效，请先完成虎扑登录: {self.account_file}") + if not self.title or not str(self.title).strip(): + raise ValueError("视频标题不能为空") + if len(self.title) < self.min_title_length: + raise ValueError(f"视频标题至少{self.min_title_length}个字") + self.file_path = str(self.validate_video_file(self.file_path)) + if self.thumbnail_path: + self.thumbnail_path = str(self.validate_image_file(self.thumbnail_path)) + + async def upload(self, playwright: Playwright) -> None: + hupu_logger.info(_msg("🧍", "先检查 cookie 和视频文件")) + await self.validate_upload_args() + hupu_logger.info(_msg("🥳", "上传前检查通过")) + + browser = await playwright.chromium.launch(**_build_launch_kwargs(headless=self.headless)) + context = await _create_stealth_context(browser, self.account_file) + + try: + page = await _new_stealth_page(context) + # 直接跳转到视频发布页（绕过首页点击） + await page.goto(HUPU_PUBLISH_URL, timeout=60000, wait_until="load") + await page.wait_for_timeout(5000) + hupu_logger.info(_msg("🏃", f"开始上传视频: {self.title}")) + + # 1) 上传视频文件 + await self._upload_video_file(page) + + # 2) 填写标题 + await self._fill_title(page) + + # 3) 填写简介 + await self._fill_description(page) + + # 4) 上传封面（如有） + if self.thumbnail_path: + await self._upload_thumbnail(page) + + # 5) 选择专区（步行街 → 步行街主干道） + await self._select_zone(page) + + # 6) 选择原创/二创 + AI 声明 + await self._check_declarations(page) + + # 7) 点击发布 + await self._submit_publish(page) + + # 保存 cookie + await context.storage_state(path=self.account_file) + hupu_logger.success(_msg("🥳", "cookie 更新完毕")) + finally: + await context.close() + await browser.close() + + async def _upload_video_file(self, page: Page) -> None: + """点击「上传视频」按钮并设置文件。""" + upload_btn = page.get_by_role("button", name="上传视频") + await upload_btn.wait_for(state="visible", timeout=15000) + + # 通过 file chooser 设置文件 + async with page.expect_file_chooser(timeout=10000) as fc_info: + await upload_btn.click() + file_chooser = await fc_info.value + await file_chooser.set_files(self.file_path) + hupu_logger.info(_msg("🏃", f"已选择视频文件: {self.file_path}")) + + # 等待视频上传就绪（标题输入框出现即可填写） + title_field = page.get_by_placeholder("请输入标题（最少4个字，最多40个字）") + await title_field.wait_for(state="visible", timeout=300000) + hupu_logger.info(_msg("🥳", "视频已就绪")) + + async def _fill_title(self, page: Page) -> None: + """填写标题（4-40字）。""" + title_field = page.get_by_placeholder("请输入标题（最少4个字，最多40个字）") + await title_field.wait_for(state="visible", timeout=15000) + title = self.title[:self.max_title_length] + await title_field.click() + await title_field.fill(title) + hupu_logger.info(_msg("🏷️", f"标题已填写: {title}")) + + async def _fill_description(self, page: Page) -> None: + """填写简介。""" + desc_field = page.get_by_placeholder("请输入简介") + if not await desc_field.count(): + hupu_logger.warning(_msg("⚠️", "未找到简介输入框")) + return + + # 组装描述：正文 + 标签 + content = self.desc + if self.tags: + tag_str = " ".join(f"#{t}#" for t in self.tags) + content = f"{content}\n{tag_str}" if content else tag_str + + if content: + await desc_field.click() + await desc_field.fill(content) + hupu_logger.info(_msg("📝", f"简介已填写（{len(content)}字）")) + + async def _upload_thumbnail(self, page: Page) -> None: + """上传封面：点击「更换封面」→ 设置文件。""" + try: + cover_span = page.locator("span").filter(has_text="更换封面") + await cover_span.wait_for(state="visible", timeout=10000) + + # 通过 file chooser 设置封面图片 + async with page.expect_file_chooser(timeout=10000) as fc_info: + await cover_span.click() + file_chooser = await fc_info.value + await file_chooser.set_files(self.thumbnail_path) + hupu_logger.info(_msg("🏃", f"已选择封面图片: {self.thumbnail_path}")) + await page.wait_for_timeout(3000) + hupu_logger.success(_msg("🖼️", "封面已上传")) + except Exception as exc: + hupu_logger.warning(_msg("⚠️", f"封面上传失败: {exc}，继续发布（使用默认封面）")) + + async def _select_zone(self, page: Page) -> None: + """选择专区：步行街 → 步行街主干道。""" + try: + # 录制脚本：page.get_by_label("发视频").get_by_text("添加专区").click() + add_zone_btn = page.get_by_label("发视频").get_by_text("添加专区") + await add_zone_btn.wait_for(state="visible", timeout=10000) + await add_zone_btn.click() + await page.wait_for_timeout(1500) + + # 选择「步行街」分类 + zone_dialog = page.get_by_label("添加专区") + step_street = zone_dialog.locator("div").filter(has_text=re.compile(r"^步行街$")) + await step_street.click() + await page.wait_for_timeout(1000) + + # 选择「步行街主干道」子分类 + main_road = page.get_by_text("步行街主干道") + await main_road.click() + await page.wait_for_timeout(500) + + # 点击确定 + confirm_btn = page.get_by_role("button", name="确 定") + await confirm_btn.click() + await page.wait_for_timeout(1000) + hupu_logger.info(_msg("🏷️", "已选择专区：步行街主干道")) + except Exception as exc: + hupu_logger.warning(_msg("⚠️", f"选择专区失败: {exc}")) + + async def _check_declarations(self, page: Page) -> None: + """选择原创/二创声明 + AI 声明。""" + try: + # 1) 点击「原创/二创」按钮 + declaration_btn = page.get_by_role("button", name="原创/二创") + if await declaration_btn.count(): + await declaration_btn.click(timeout=5000) + await page.wait_for_timeout(1000) + hupu_logger.info(_msg("🏷️", "已点击「原创/二创」")) + except Exception as exc: + hupu_logger.warning(_msg("⚠️", f"点击原创/二创失败: {exc}")) + + try: + # 2) 选择「含AI生成内容」 + combobox = page.get_by_role("combobox") + if await combobox.count(): + await combobox.click(timeout=5000) + await page.wait_for_timeout(1000) + + ai_option = page.get_by_text("含AI生成内容") + if await ai_option.count(): + await ai_option.click(timeout=5000) + await page.wait_for_timeout(500) + hupu_logger.info(_msg("🏷️", "已选择「含AI生成内容」")) + except Exception as exc: + hupu_logger.warning(_msg("⚠️", f"选择 AI 声明失败: {exc}")) + + async def _submit_publish(self, page: Page) -> None: + """点击「确定发布」并等待跳转到帖子页面。""" + # 录制脚本：page.get_by_label("发视频").get_by_text("确定发布").click() + publish_btn = page.get_by_label("发视频").get_by_text("确定发布") + await publish_btn.wait_for(state="visible", timeout=15000) + await publish_btn.click() + hupu_logger.info(_msg("🏃", "已点击「确定发布」")) + + # 等待跳转到帖子详情页（URL 匹配 bbs.hupu.com/{数字}.html） + start = time.monotonic() + while time.monotonic() - start < 60: + current_url = page.url + if HUPU_POST_URL_PATTERN.search(current_url): + hupu_logger.success(_msg("🥳", f"视频发布成功: {current_url}")) + return + + await page.wait_for_timeout(2000) + + # 超时 - 可能已成功但检测不到 + hupu_logger.warning(_msg("⚠️", f"发布后 60s 未检测到帖子页面跳转，当前 URL: {page.url}")) + + async def main(self): + async with async_playwright() as playwright: + await self.upload(playwright) diff --git a/utils/log.py b/utils/log.py index 6008443..031f569 100644 --- a/utils/log.py +++ b/utils/log.py @@ -60,3 +60,4 @@ xiaohongshu_logger = create_logger('xiaohongshu', 'logs/xiaohongshu.log') youtube_logger = create_logger('youtube', 'logs/youtube.log') alipay_logger = create_logger('alipay', 'logs/alipay.log') weibo_logger = create_logger('weibo', 'logs/weibo.log') +hupu_logger = create_logger('hupu', 'logs/hupu.log')