diff --git a/examples/get_baijiahao_cookie_headless.py b/examples/get_baijiahao_cookie_headless.py new file mode 100644 index 0000000..e914d09 --- /dev/null +++ b/examples/get_baijiahao_cookie_headless.py @@ -0,0 +1,111 @@ +# -*- coding: utf-8 -*- +"""百家号(headless)扫码登录 → 保存 cookie。 + +说明： + - 无头浏览器打开百家号登录页并提取登录二维码，放大截图后保存为 png， + 并用终端 ASCII 二维码展示；你手机/百度APP扫码登录。 + - 扫码成功后自动把登录态写入 cookies/baijiahao_uploader/account.json。 + +用法： + python examples/get_baijiahao_cookie_headless.py +""" +from pathlib import Path + +from playwright.async_api import async_playwright +from conf import BASE_DIR +from utils.login_qrcode import ( + build_login_qrcode_path, + print_terminal_qrcode, +) +from uploader.baijiahao_uploader.main import cookie_auth, baijiahao_logger + + +QR_SELECTOR = 'img[src^="https://passport.baidu.com/v2/api/qrcode"]' +LOGIN_BTN_TEXT = "登录" +LOGIN_URL = "https://baijiahao.baidu.com/builder/theme/bjh/login" + + +async def _grab_qr(page, qrcode_path: Path) -> str: + qr = page.locator(QR_SELECTOR).first + await qr.wait_for(state="attached", timeout=60000) + src = await qr.get_attribute("src") + if src and src.startswith("https://"): + resp = await page.context.request.get(src) + qrcode_path.parent.mkdir(parents=True, exist_ok=True) + qrcode_path.write_bytes(await resp.body()) + else: + await qr.screenshot(path=str(qrcode_path)) + + qrcode_content = "" + try: + import cv2 + img = cv2.imread(str(qrcode_path)) + if img is not None: + h, w = img.shape[:2] + up = cv2.resize(img, (w * 3, h * 3), interpolation=cv2.INTER_CUBIC) + qrcode_content = cv2.QRCodeDetector().detectAndDecode(up)[0] or "" + except Exception: + pass + return qrcode_content or "" + + +async def _wait_login(page, max_checks: int = 120, interval: int = 3) -> bool: + import asyncio + async def _logged_in(): + if "login" in page.url: + return False + ctx = page.context + cookies = await ctx.cookies() + if any(c.get("name") in ("BDUSS", "STOKEN") for c in cookies): + return True + return False + + for _ in range(max_checks): + if await _logged_in(): + return True + await asyncio.sleep(interval) + return False + + +async def main(): + account_file = Path(BASE_DIR / "cookies" / "baijiahao_uploader" / "account.json") + account_file.parent.mkdir(parents=True, exist_ok=True) + + import os + if os.path.exists(account_file) and await cookie_auth(str(account_file)): + baijiahao_logger.success("[+] cookie 已有效，无需重新登录") + return + + qrcode_path = build_login_qrcode_path(str(account_file)) + async with async_playwright() as playwright: + browser = await playwright.chromium.launch( + headless=True, args=["--no-sandbox", "--disable-blink-features=AutomationControlled"] + ) + context = await browser.new_context() + page = await context.new_page() + await page.goto(LOGIN_URL, timeout=60000, wait_until="domcontentloaded") + await page.wait_for_timeout(4000) + # 进入登录弹窗 + await page.get_by_text(LOGIN_BTN_TEXT, exact=True).first.click(timeout=10000) + await page.wait_for_timeout(4000) + + qrcode_content = await _grab_qr(page, qrcode_path) + baijiahao_logger.info(f"🖼️ 二维码已保存: {qrcode_path}") + if qrcode_content: + print_terminal_qrcode(qrcode_content, qrcode_path, "百度APP/手机百度") + else: + print(f"未能解码二维码，请直接打开文件扫码:\n {qrcode_path}") + + if await _wait_login(page): + baijiahao_logger.success("[+] 扫码登录成功，正在保存 cookie...") + await context.storage_state(path=str(account_file)) + baijiahao_logger.success(f"[+] cookie 已保存: {account_file}") + else: + baijiahao_logger.error("[-] 等待扫码超时（约 6 分钟），未完成登录。") + await browser.close() + + +if __name__ == "__main__": + import asyncio + + asyncio.run(main()) \ No newline at end of file diff --git a/sau_cli.py b/sau_cli.py index a896718..e0dafa8 100644 --- a/sau_cli.py +++ b/sau_cli.py @@ -9,6 +9,11 @@ from pathlib import Path from typing import Iterable, Sequence from conf import BASE_DIR +from uploader.baijiahao_uploader.main import ( + BaiJiaHaoVideo, + baijiahao_setup, + cookie_auth as baijiahao_cookie_auth, +) from uploader.bilibili_uploader.runtime import run_biliup_command from uploader.douyin_uploader.main import ( DOUYIN_PUBLISH_STRATEGY_IMMEDIATE, @@ -168,6 +173,19 @@ class TencentVideoUploadRequest: headless: bool = True +@dataclass(slots=True) +class BaijiahaoVideoUploadRequest: + account_name: str + video_file: Path + title: str + description: str + tags: list[str] + thumbnail_file: Path | None = None + collection_name: str | None = None + debug: bool = True + headless: bool = True + + @dataclass(slots=True) class YouTubeVideoUploadRequest: account_name: str @@ -549,6 +567,41 @@ async def upload_tencent_video(request: TencentVideoUploadRequest) -> Path: return account_file +async def login_baijiahao_account(account_name: str, headless: bool = True, qrcode_callback=None) -> dict: + account_file = resolve_account_file("baijiahao", account_name) + return await baijiahao_setup(str(account_file), handle=True, return_detail=True, headless=headless, qrcode_callback=qrcode_callback) + + +async def check_baijiahao_account(account_name: str) -> bool: + account_file = resolve_account_file("baijiahao", account_name) + if not account_file.exists(): + return False + return await baijiahao_cookie_auth(str(account_file)) + + +async def upload_baijiahao_video(request: BaijiahaoVideoUploadRequest) -> Path: + account_file = resolve_account_file("baijiahao", request.account_name) + is_ready = await baijiahao_setup(str(account_file), handle=False) + if not is_ready: + raise RuntimeError( + f"Baijiahao cookie is missing or expired: {account_file}. Run `sau baijiahao login --account {request.account_name}` first." + ) + + app = BaiJiaHaoVideo( + 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, + collection_name=request.collection_name, + 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(): @@ -736,6 +789,26 @@ def build_parser() -> argparse.ArgumentParser: youtube_upload_video_parser.add_argument( "--visibility", default="public", choices=["public", "unlisted", "private"], help="Video visibility") add_runtime_flags(youtube_upload_video_parser) + + baijiahao_parser = platform_parsers.add_parser("baijiahao", help="Baidu Baijiahao operations") + baijiahao_actions = baijiahao_parser.add_subparsers(dest="action", required=True) + + for action_name in ("login", "check"): + action_parser = baijiahao_actions.add_parser(action_name, help=f"Baijiahao {action_name}") + action_parser.add_argument("--account", required=True, help="Baijiahao user-defined account_name") + if action_name == "login": + add_runtime_flags(action_parser) + + baijiahao_upload_video_parser = baijiahao_actions.add_parser("upload-video", help="Upload one video to Baijiahao") + baijiahao_upload_video_parser.add_argument("--account", required=True, help="Baijiahao user-defined account_name") + baijiahao_upload_video_parser.add_argument("--file", required=True, type=existing_file_path, help="Video file path") + baijiahao_upload_video_parser.add_argument("--title", required=True, help="Video title") + baijiahao_upload_video_parser.add_argument("--desc", default="", help="Optional video description") + baijiahao_upload_video_parser.add_argument("--tags", default="", help="Comma-separated tags, such as tag1,tag2") + baijiahao_upload_video_parser.add_argument("--thumbnail", type=existing_file_path, help="Optional cover image path") + baijiahao_upload_video_parser.add_argument("--collection", default=None, help="Optional collection name") + add_runtime_flags(baijiahao_upload_video_parser) + return parser @@ -1016,6 +1089,37 @@ async def dispatch(args: argparse.Namespace) -> int: raise RuntimeError(f"Unsupported YouTube action: {args.action}") + if args.platform == "baijiahao": + if args.action == "login": + result = await login_baijiahao_account(args.account, headless=args.headless) + if not result["success"]: + raise RuntimeError(result["message"]) + print(f"Baijiahao login flow completed: {result['account_file']}") + return 0 + + if args.action == "check": + is_valid = await check_baijiahao_account(args.account) + print("valid" if is_valid else "invalid") + return 0 if is_valid else 1 + + if args.action == "upload-video": + request = BaijiahaoVideoUploadRequest( + account_name=args.account, + video_file=args.file, + title=args.title, + description=args.desc, + tags=parse_tags(args.tags), + thumbnail_file=args.thumbnail, + collection_name=args.collection, + debug=args.debug, + headless=args.headless, + ) + await upload_baijiahao_video(request) + print(f"Baijiahao video upload submitted: {request.video_file}") + return 0 + + raise RuntimeError(f"Unsupported Baijiahao action: {args.action}") + raise RuntimeError(f"Unsupported platform: {args.platform}") diff --git a/uploader/baijiahao_uploader/main.py b/uploader/baijiahao_uploader/main.py index 2839afd..2d679bd 100644 --- a/uploader/baijiahao_uploader/main.py +++ b/uploader/baijiahao_uploader/main.py @@ -1,506 +1,597 @@ # -*- coding: utf-8 -*- -import random -from datetime import datetime +"""百家号（百度百家号）视频上传 + 扫码登录。 -from playwright.async_api import Playwright, async_playwright, Page +功能： + - baijiahao_cookie_gen: headless 扫码登录（百度 passport 二维码） + - cookie_auth: 验证 cookie 是否有效 + - baijiahao_setup: 统一入口（检查/触发登录） + - BaiJiaHaoVideo: 视频上传类 +""" +from __future__ import annotations + +import asyncio +import inspect +import json as _json import os import time -import asyncio +from pathlib import Path -from conf import LOCAL_CHROME_PATH, LOCAL_CHROME_HEADLESS -from utils.base_social_media import set_init_script +from playwright.async_api import Page, Playwright, TimeoutError as PWTimeoutError, async_playwright + +from conf import BASE_DIR, LOCAL_CHROME_HEADLESS, LOCAL_CHROME_PATH +from uploader.base_video import BaseVideoUploader from utils.log import baijiahao_logger -from utils.network import async_retry +from utils.login_qrcode import build_login_qrcode_path, decode_qrcode_from_path, print_terminal_qrcode, remove_qrcode_file -async def baijiahao_cookie_gen(account_file): +BAIJIAHAO_LOGIN_URL = "https://baijiahao.baidu.com/builder/theme/bjh/login" +BAIJIAHAO_HOME_URL = "https://baijiahao.baidu.com/builder/rc/home" +BAIJIAHAO_PUBLISH_URL = "https://baijiahao.baidu.com/builder/rc/edit?type=videoV2" +# 发布成功后跳转到的 URL 前缀 +BAIJIAHAO_SUCCESS_URL_PREFIX = "https://baijiahao.baidu.com/builder/rc/clue" + +# 百度 passport 二维码图片选择器 +QR_SELECTOR = 'img[src^="https://passport.baidu.com/v2/api/qrcode"]' + + +def _msg(emoji: str, text: str) -> str: + return f"{emoji} {text}" + + +def _build_login_result(success: bool, status: str, message: str, account_file: str, qrcode: dict | None = None, current_url: str = "") -> dict: + return { + "success": success, + "status": status, + "message": message, + "account_file": str(account_file), + "qrcode": qrcode, + "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} + 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" / "baijiahao_uploader" / path).resolve()) + return str(path.resolve()) + + +async def _grab_qr(page: Page, account_file: str) -> dict: + """截取百度 passport 扫码登录二维码。 + + 百家号登录页点「登录」后弹出百度统一登录框，其中二维码是 img[src] 指向 + passport.baidu.com 的图片 URL，可以直接下载或截图。 + """ + qr = page.locator(QR_SELECTOR).first + await qr.wait_for(state="attached", timeout=60000) + + qrcode_path = build_login_qrcode_path(account_file) + qrcode_path.parent.mkdir(parents=True, exist_ok=True) + + # 优先直接下载高清图片 URL + src = await qr.get_attribute("src") + if src and src.startswith("https://"): + try: + resp = await page.context.request.get(src) + qrcode_path.write_bytes(await resp.body()) + except Exception: + await qr.screenshot(path=str(qrcode_path)) + else: + await qr.screenshot(path=str(qrcode_path)) + + qrcode_content = decode_qrcode_from_path(qrcode_path) + baijiahao_logger.info(_msg("🖼️", f"二维码已保存到: {qrcode_path}")) + if qrcode_content: + print_terminal_qrcode(qrcode_content, qrcode_path, "百度APP/手机百度") + else: + baijiahao_logger.warning(_msg("😵", f"终端没法完整显示二维码，请打开 {qrcode_path} 扫码")) + return {"image_path": str(qrcode_path), "image_data_url": ""} + + +async def _is_login_completed(page: Page) -> bool: + """判断百度登录是否完成：URL 离开 login 页 或 出现 BDUSS cookie。""" + if "login" in page.url.lower(): + # 还在登录页，检查 cookies + cookies = await page.context.cookies() + if any(c.get("name") in ("BDUSS", "STOKEN") for c in cookies): + return True + return False + # 跳走了说明登录成功 + return True + + +async def baijiahao_cookie_gen(account_file, qrcode_callback=None, poll_interval: int = 3, max_checks: int = 120, headless: bool = LOCAL_CHROME_HEADLESS): + """无头/有头扫码登录百家号，保存 cookie。 + + 流程：打开登录页 → 点「登录」按钮弹出百度 passport 登录框 → 截取二维码 → 等待扫码完成 → 保存 storage_state。 + 返回标准 login result dict。 + """ + account_file = _resolve_account_file(account_file) + Path(account_file).parent.mkdir(parents=True, exist_ok=True) + qrcode_path = None + result = _build_login_result(False, "failed", "百家号登录失败", account_file) + async with async_playwright() as playwright: - options = { - 'args': [ - '--lang en-GB' - ], - 'headless': LOCAL_CHROME_HEADLESS, # Set headless option here - } - # Make sure to run headed. - browser = await playwright.chromium.launch(**options) - # Setup context however you like. - context = await browser.new_context() # Pass any options - context = await set_init_script(context) - # Pause the page, and start recording manually. - page = await context.new_page() - await page.goto("https://baijiahao.baidu.com/builder/theme/bjh/login") - await page.pause() - # 点击调试器的继续，保存cookie - await context.storage_state(path=account_file) - baijiahao_logger.success("cookie saved") + browser = await playwright.chromium.launch(**_build_launch_kwargs(headless=headless)) + context = await browser.new_context() + try: + page = await context.new_page() + await page.goto(BAIJIAHAO_LOGIN_URL, timeout=60000, wait_until="domcontentloaded") + await page.wait_for_timeout(4000) + + # 点击「登录」按钮触发百度 passport 弹窗 + login_btn = page.get_by_text("登录", exact=True).first + try: + await login_btn.click(timeout=10000) + except Exception: + # 有些情况直接就在登录状态 + pass + await page.wait_for_timeout(4000) + + if headless: + baijiahao_logger.info(_msg("🧍", "无头登录中：二维码已存为图片，请用百度APP扫码")) + else: + baijiahao_logger.info(_msg("🧍", "请在打开的浏览器中扫码登录百家号")) + + # 截取二维码 + qrcode_info = await _grab_qr(page, account_file) + qrcode_path = Path(qrcode_info["image_path"]) if qrcode_info.get("image_path") else None + await _emit_qrcode_callback(qrcode_callback, qrcode_info) + + baijiahao_logger.info(_msg("🧍", "请扫码，正在耐心等待登录完成")) + + # 轮询等待登录完成 + for _ in range(max_checks): + if await _is_login_completed(page): + baijiahao_logger.info(_msg("🥳", f"扫码成功，当前页面: {page.url}")) + result = _build_login_result(True, "success", "百家号扫码登录成功", account_file, qrcode_info, page.url) + break + await page.wait_for_timeout(poll_interval * 1000) + else: + result = _build_login_result(False, "timeout", "等待百家号扫码登录超时", account_file, qrcode_info, page.url) + + if result["success"]: + await asyncio.sleep(2) + await context.storage_state(path=account_file) + baijiahao_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: + if remove_qrcode_file(qrcode_path): + baijiahao_logger.info(_msg("🧹", f"临时二维码文件已清理: {qrcode_path}")) + if not result["success"]: + baijiahao_logger.error(_msg("😢", f"登录失败: {result['message']}")) + await context.close() + await browser.close() + return result 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(headless=LOCAL_CHROME_HEADLESS) - context = await browser.new_context(storage_state=account_file) - context = await set_init_script(context) - # 创建一个新的页面 - page = await context.new_page() - # 访问指定的 URL - await page.goto("https://baijiahao.baidu.com/builder/rc/home") - await page.wait_for_timeout(timeout=5000) + browser = await playwright.chromium.launch(**_build_launch_kwargs(headless=True)) + try: + context = await browser.new_context(storage_state=account_file) + page = await context.new_page() + await page.goto(BAIJIAHAO_HOME_URL, timeout=60000, wait_until="domcontentloaded") + await page.wait_for_timeout(5000) - if await page.get_by_text('注册/登录百家号').count(): - baijiahao_logger.error("等待5秒 cookie 失效") + if await page.get_by_text("注册/登录百家号").count(): + baijiahao_logger.info(_msg("🥹", "cookie 已失效")) + return False + else: + baijiahao_logger.success(_msg("🥳", "cookie 有效")) + return True + except Exception as exc: + baijiahao_logger.warning(_msg("😵", f"cookie 校验出错，按失效处理: {exc}")) return False - else: - baijiahao_logger.success("[+] cookie 有效") - return True + finally: + await browser.close() -async def baijiahao_setup(account_file, handle=False): +async def baijiahao_setup(account_file, handle=False, return_detail=False, qrcode_callback=None, headless: bool = LOCAL_CHROME_HEADLESS): + """统一入口：检查 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: - return False - baijiahao_logger.error("cookie文件不存在或已失效，即将自动打开浏览器，请扫码登录，登陆后会自动生成cookie文件") - await baijiahao_cookie_gen(account_file) - return True + result = _build_login_result(False, "cookie_invalid", "cookie 文件不存在或已失效", account_file) + return result if return_detail else False + baijiahao_logger.info(_msg("🥹", "cookie 文件不存在或已失效，自动打开浏览器请扫码登录")) + result = await baijiahao_cookie_gen(account_file, qrcode_callback=qrcode_callback, headless=headless) + return result if return_detail else result["success"] -class BaiJiaHaoVideo(object): - def __init__(self, title, file_path, tags, publish_date: datetime, account_file, proxy_setting=None): - self.title = title # 视频标题 + result = _build_login_result(True, "cookie_valid", "cookie 有效", account_file) + return result if return_detail else True + + +class BaiJiaHaoVideo(BaseVideoUploader): + """百家号视频上传。 + + 流程：打开发布页 → 上传视频文件 → 填标题 → 等待上传/转码完成 → 等封面生成 → 点击发布。 + """ + + def __init__( + self, + title, + file_path, + tags, + account_file, + publish_date=0, + desc: str | None = None, + thumbnail_path: str | None = None, + collection_name: str | None = None, + debug: bool = True, + headless: bool = LOCAL_CHROME_HEADLESS, + ): + self.title = title self.file_path = file_path - self.tags = tags + self.tags = tags or [] + self.account_file = _resolve_account_file(account_file) self.publish_date = publish_date - self.account_file = account_file - self.date_format = '%Y年%m月%d日 %H:%M' + self.desc = desc or "" + self.thumbnail_path = thumbnail_path + self.collection_name = collection_name + self.debug = debug + self.headless = headless self.local_executable_path = LOCAL_CHROME_PATH - self.headless = LOCAL_CHROME_HEADLESS - self.proxy_setting = proxy_setting + self.max_title_length = 30 - async def set_schedule_time(self, page, publish_date): - """ - todo 时间选择，日后在处理 百家号的时间选择不准确，目前是随机 - """ - publish_date_day = f"{publish_date.month}月{publish_date.day}日" if publish_date.day >9 else f"{publish_date.month}月0{publish_date.day}日" - publish_date_hour = f"{publish_date.hour}点" - publish_date_min = f"{publish_date.minute}分" - await page.wait_for_selector('div.select-wrap', timeout=5000) - for _ in range(3): - try: - await page.locator('div.select-wrap').nth(0).click() - await page.wait_for_selector('div.rc-virtual-list div.cheetah-select-item', timeout=5000) - break - except: - await page.locator('div.select-wrap').nth(0).click() - # page.locator(f'div.rc-virtual-list-holder-inner >> text={publish_date_day}').click() - await page.wait_for_timeout(2000) - await page.locator(f'div.rc-virtual-list div.cheetah-select-item >> text={publish_date_day}').click() - await page.wait_for_timeout(2000) - - # 改为随机点击一个 hour - for _ in range(3): - try: - await page.locator('div.select-wrap').nth(1).click() - await page.wait_for_selector('div.rc-virtual-list div.rc-virtual-list-holder-inner:visible', timeout=5000) - break - except: - await page.locator('div.select-wrap').nth(1).click() - await page.wait_for_timeout(2000) - current_choice_hour = await page.locator('div.rc-virtual-list:visible div.cheetah-select-item-option').count() - await page.wait_for_timeout(2000) - await page.locator('div.rc-virtual-list:visible div.cheetah-select-item-option').nth( - random.randint(1, current_choice_hour-3)).click() - # 2024.08.05 current_choice_hour的获取可能有问题，页面有7，这里获取了10，暂时硬编码至6 - - await page.wait_for_timeout(2000) - await page.locator("button >> text=定时发布").click() - - - async def handle_upload_error(self, page): - # 日后实现，目前没遇到 - return - print("视频出错了，重新上传中") + 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 not self.thumbnail_path: + raise ValueError("百家号视频发布必须提供横版封面图（--thumbnail）") + self.file_path = str(self.validate_video_file(self.file_path)) + self.thumbnail_path = str(self.validate_image_file(self.thumbnail_path)) async def upload(self, playwright: Playwright) -> None: - # 使用 Chromium 浏览器启动一个浏览器实例 - browser = await playwright.chromium.launch(headless=self.headless, executable_path=self.local_executable_path, proxy=self.proxy_setting) - # 创建一个浏览器上下文，使用指定的 cookie 文件 - context = await browser.new_context(storage_state=f"{self.account_file}", user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/127.0.4324.150 Safari/537.36') - # context = await set_init_script(context) - await context.grant_permissions(['geolocation']) + baijiahao_logger.info(_msg("🧍", "先检查 cookie 和视频文件")) + await self.validate_upload_args() + baijiahao_logger.info(_msg("🥳", "上传前检查通过")) - # 创建一个新的页面 - page = await context.new_page() - # 访问指定的 URL - await page.goto("https://baijiahao.baidu.com/builder/rc/edit?type=videoV2", timeout=60000) - baijiahao_logger.info(f"正在上传-------{self.title}.mp4") - # 等待页面跳转到指定的 URL，没进入，则自动等待到超时 - baijiahao_logger.info('正在打开主页...') - await page.wait_for_url("https://baijiahao.baidu.com/builder/rc/edit?type=videoV2", timeout=60000) + browser = await playwright.chromium.launch(**_build_launch_kwargs(headless=self.headless)) + context = await browser.new_context(storage_state=self.account_file) + await context.grant_permissions(["geolocation"]) - # 点击 "上传视频" 按钮 - await page.locator("div[class^='video-main-container'] input").set_input_files(self.file_path) + try: + page = await context.new_page() + await page.goto(BAIJIAHAO_PUBLISH_URL, timeout=120000, wait_until="domcontentloaded") + baijiahao_logger.info(_msg("🏃", f"开始上传视频: {self.title}")) - # 等待页面跳转到指定的 URL + # 等待发布页加载 + await page.wait_for_timeout(3000) + + # 1) 上传视频文件 + file_input = page.locator('input[type="file"][accept*="video"], input[type="file"][accept*="mp4"]').first + if not await file_input.count(): + file_input = page.locator("div[class^='video-main-container'] input[type='file']").first + if not await file_input.count(): + file_input = page.locator('input[type="file"]').first + await file_input.wait_for(state="attached", timeout=30000) + await file_input.set_input_files(self.file_path) + baijiahao_logger.info(_msg("🏃", f"已选择视频文件: {self.file_path}")) + + # 2) 等待进入表单页面（contenteditable 标题区出现即表单渲染完毕） + title_editor = page.locator('div[class*="contentEditable"]').first + await title_editor.wait_for(state="visible", timeout=180000) + await page.wait_for_timeout(1000) + + # 3) 填写标题 + await self._fill_title(page) + + # 4) 等待视频上传完成 + await self._wait_upload_complete(page) + + # 5) 上传横版封面（必填） + await self._upload_thumbnail(page) + + # 6) 勾选「含AI生成内容」 + await self._check_ai_declaration(page) + + # 7) 选择合集（如有配置） + await self._apply_collection(page) + + # 8) 点击发布 + await self._submit_publish(page) + + # 保存 cookie + await context.storage_state(path=self.account_file) + baijiahao_logger.success(_msg("🥳", "cookie 更新完毕")) + finally: + await context.close() + await browser.close() + + async def _fill_title(self, page: Page) -> None: + title_field = page.locator('div[class*="contentEditable"]').first + await title_field.wait_for(state="visible", timeout=15000) + title = self.title + # 百家号标题最少9字 + if len(title) <= 8: + title += " 你不知道的" + title = title[: self.max_title_length] + # 清空原有内容（可能自动填了文件名），再输入标题 + await title_field.click() + await page.keyboard.press("Control+a") + await page.keyboard.press("Backspace") + await title_field.fill(title) + baijiahao_logger.info(_msg("🏷️", f"标题已填写: {title}")) + + async def _wait_upload_complete(self, page: Page, timeout: int = 600) -> None: + """等待视频真正上传完成。 + + 百度真实上传进度是一段百分比文字（9%…99%，上传完成后消失，本文件实测约 35s）。 + 旧实现用 'div .cover-overlay:has-text("上传中")' 判断——经实测该元素恒不存在， + 导致选完文件立即误判"上传完毕"（约 4s）。大文件此时其实还在后台上传，随后点 + 发布会被百度以"确保视频已经上传完毕"拒绝（产出 0 作品）。改为跟踪百分比进度： + 出现过进度且进度消失/达 100% 才算真正上传完成。 + """ + import re as _re + start = time.monotonic() + seen_progress = False + gone_count = 0 while True: - # 判断是是否进入视频发布页面，没进入，则自动等待到超时 + if time.monotonic() - start > timeout: + baijiahao_logger.warning(_msg("⚠️", f"等待上传超时（>{timeout}s），继续后续步骤")) + return + + body = "" try: - await page.wait_for_selector("div#formMain:visible") - break - except: - baijiahao_logger.info("正在等待进入视频发布页面...") - await asyncio.sleep(0.1) + body = await page.inner_text("body") + except Exception: + pass - # 填充标题和话题 - # 这里为了避免页面变化，故使用相对位置定位：作品标题父级右侧第一个元素的input子元素 - await asyncio.sleep(1) - baijiahao_logger.info("正在填充标题和话题...") - await self.add_title_tags(page) + if "上传失败" in body: + raise RuntimeError("视频上传失败") - upload_status = await self.uploading_video(page) - if not upload_status: - baijiahao_logger.error(f"发现上传出错了... 文件:{self.file_path}") - raise + m = _re.search(r'(\d{1,3})\s*%', body) + pct = int(m.group(1)) if m else None - # 判断视频封面图是否生成成功 - while True: - baijiahao_logger.info("正在确认封面完成, 准备去点击定时/发布...") - if await page.locator("div.cheetah-spin-container img").count(): - baijiahao_logger.info("封面已完成，点击定时/发布...") - break - else: - baijiahao_logger.info("等待封面生成...") - await asyncio.sleep(3) - - await self.publish_video(page, self.publish_date) - await page.wait_for_timeout(2000) - if await page.locator('div.passMod_dialog-container >> text=百度安全验证:visible').count(): - baijiahao_logger.error("出现验证，退出") - raise Exception("出现验证，退出") - await page.wait_for_url("https://baijiahao.baidu.com/builder/rc/clue**", timeout=5000) - baijiahao_logger.success("视频发布成功") - - await context.storage_state(path=self.account_file) # 保存cookie - baijiahao_logger.info('cookie更新完毕！') - await asyncio.sleep(2) # 这里延迟是为了方便眼睛直观的观看 - # 关闭浏览器上下文和浏览器实例 - await context.close() - await browser.close() - - - @async_retry(timeout=300) # 例如，最多重试3次，超时时间为180秒 - async def uploading_video(self, page): - while True: - upload_failed = await page.locator('div .cover-overlay:has-text("上传失败")').count() - if upload_failed: - baijiahao_logger.error("发现上传出错了...") - # await self.handle_upload_error(page) # 假设这是处理上传错误的函数 - return False - - uploading = await page.locator('div .cover-overlay:has-text("上传中")').count() - if uploading: - baijiahao_logger.info("正在上传视频中...") - await asyncio.sleep(2) # 等待2秒再次检查 + if pct is not None and pct < 100: + seen_progress = True + gone_count = 0 + baijiahao_logger.info(_msg("🏃", f"上传中 {pct}%")) + await asyncio.sleep(2) continue - # 检查上传是否成功 - if not uploading and not upload_failed: - baijiahao_logger.success("视频上传完毕") - return True + if seen_progress: + # 进度百分比已消失/到 100%，连续两次确认后判为上传完成 + gone_count += 1 + if gone_count >= 2: + baijiahao_logger.success(_msg("🥳", "视频上传完毕")) + return + await asyncio.sleep(2) + continue - async def set_schedule_publish(self, page, publish_date): - while True: - schedule_element = page.locator("div.op-btn-outter-content >> text=定时发布").locator("..").locator( - 'button') - try: - await schedule_element.click() - await page.wait_for_selector('div.select-wrap:visible', timeout=3000) - await page.wait_for_timeout(timeout=2000) - baijiahao_logger.info("开始点击发布定时...") - await self.set_schedule_time(page, publish_date) - break - except Exception as e: - baijiahao_logger.error(f"定时发布失败: {e}") - raise # 重新抛出异常，让重试装饰器捕获 + # 一直没出现过进度：小文件可能秒传完成；给 15s 窗口后放行 + if time.monotonic() - start > 15: + baijiahao_logger.success(_msg("🥳", "视频上传完毕")) + return + await asyncio.sleep(2) - @async_retry(timeout=300) # 例如，最多重试3次，超时时间为180秒 - async def publish_video(self, page: Page, publish_date): - if publish_date != 0: - # 定时发布 - await self.set_schedule_publish(page, publish_date) - else: - # 立即发布 - await self.direct_publish(page) + async def _upload_thumbnail(self, page: Page) -> None: + """上传横版封面（必填）。 + + 流程：点击「选择封面」→ 弹窗中点「上传」按钮 → 设置图片文件 → 等待上传完成 → 确认。 + 如果没有提供 thumbnail_path，等待系统自动生成封面即可。 + """ + if not self.thumbnail_path: + # 没有自定义封面，等系统自动生成 + await self._wait_cover_ready(page) + return - async def direct_publish(self, page): try: - publish_button = page.locator("button >> text=发布") - if await publish_button.count(): - await publish_button.click() - except Exception as e: - baijiahao_logger.error(f"直接发布视频失败: {e}") - raise # 重新抛出异常，让重试装饰器捕获 + # 1) 点击「选择封面」入口 + cover_entry = page.locator('[data-testid="select-cover"]').first + if not await cover_entry.count(): + # 备选：通过文本定位 + cover_entry = page.get_by_text("选择封面", exact=True).first + await cover_entry.scroll_into_view_if_needed() + await cover_entry.click(timeout=10000) + baijiahao_logger.info(_msg("🏃", "已点击「选择封面」")) + await page.wait_for_timeout(2000) - async def add_title_tags(self, page): - title_container = page.get_by_placeholder('添加标题获得更多推荐') - if len(self.title) <= 8: - self.title += " 你不知道的" - await title_container.fill(self.title[:30]) + # 2) 弹窗中找「上传」按钮并点击 + # 百家号封面弹窗通常有「上传」tab/按钮 + upload_btn = page.locator('button:has-text("上传"), div:has-text("上传"):not(:has(*)):visible').first + if not await upload_btn.count(): + upload_btn = page.get_by_text("上传", exact=True).first + await upload_btn.click(timeout=8000) + await page.wait_for_timeout(1500) + + # 3) 设置图片文件到 file input + # 弹窗中会出现 input[type=file] + img_input = page.locator('input[type="file"][accept*="image"], input[type="file"][accept*="jpg"], input[type="file"][accept*="png"]').first + if not await img_input.count(): + # 通用 fallback：弹窗内最新出现的 file input + img_input = page.locator('input[type="file"]').last + await img_input.set_input_files(self.thumbnail_path) + baijiahao_logger.info(_msg("🏃", f"已选择封面图片: {self.thumbnail_path}")) + + # 4) 等待并点击确认/完成按钮（如有裁剪弹窗）。 + # 裁剪弹窗渲染有延迟（图片上传+服务端处理），之前用固定 sleep(3s) 后 + # 一次性检查 confirm_btn，弹窗还没渲染出来时会被误判为"无需确认"而跳过点击， + # 导致封面选择实际未提交，但日志仍打「封面已上传」成功——这是本次线上 + # 百家号视频没有封面、日志却显示成功的根因。改为轮询等待（不放大超时时长本身 + # 不算错误：裁剪弹窗本就是可选的，等不到也可能是流程本身没有该弹窗）。 + confirm_btn = page.locator('button:has-text("确定"), button:has-text("完成"), button:has-text("确认")').first + confirmed = False + try: + await confirm_btn.wait_for(state="visible", timeout=15000) + await confirm_btn.click(timeout=8000) + await page.wait_for_timeout(1000) + confirmed = True + except PWTimeoutError: + baijiahao_logger.debug("封面确认按钮未出现，可能本次流程无需裁剪确认") + + if confirmed: + baijiahao_logger.success(_msg("🖼️", "封面已上传")) + else: + # 没有等到确认按钮：不确定封面是否真正生效，不再冒充成功， + # 交给下方 except 分支同一套"等待系统自动封面"兜底逻辑核实/兜底。 + raise RuntimeError("封面确认按钮未出现，无法确认封面是否生效") + except Exception as exc: + baijiahao_logger.warning(_msg("⚠️", f"封面上传失败: {exc}，尝试等待系统自动封面")) + # fallback：等系统自动生成 + await self._wait_cover_ready(page) + + async def _check_ai_declaration(self, page: Page) -> None: + """选择「含AI生成内容」创作声明。 + + 点击「请选择创作声明」input → 弹出 modal 弹窗 → 点选「含AI生成内容」→ 点「确定」。 + """ + try: + # 点击创作声明输入框触发弹窗 + trigger = page.locator('input[placeholder="请选择创作声明"]').first + await trigger.scroll_into_view_if_needed() + await trigger.click(force=True, timeout=8000) + await page.wait_for_timeout(3000) + + # 弹窗内点选「含AI生成内容」 + ai_option = page.locator('.cheetah-modal-wrap :text("含AI生成内容")').first + if not await ai_option.count(): + ai_option = page.locator('text="含AI生成内容"').first + await ai_option.wait_for(state="visible", timeout=10000) + await ai_option.click(timeout=5000) + await page.wait_for_timeout(1000) + + # 点「确定」按钮关闭弹窗（弹窗可能在点选后仍存在） + modal = page.locator('.cheetah-modal-wrap:visible').first + if await modal.count(): + confirm_btn = modal.locator('button:has-text("确定")').first + if await confirm_btn.count() and await confirm_btn.is_visible(): + await confirm_btn.click(timeout=5000) + await page.wait_for_timeout(500) + else: + # 确定按钮不可见，尝试 force click 或按 Escape 关闭 + await page.keyboard.press("Escape") + await page.wait_for_timeout(500) + + baijiahao_logger.success(_msg("🏷️", "已选择「含AI生成内容」")) + except Exception as exc: + # 如果失败，尝试关闭可能残留的弹窗 + try: + await page.keyboard.press("Escape") + await page.wait_for_timeout(500) + except Exception: + pass + baijiahao_logger.warning(_msg("⚠️", f"选择 AI 声明失败: {exc}")) + + async def _apply_collection(self, page: Page) -> None: + """选择合集（cheetah-select 下拉搜索框）。 + + placeholder: "选择同主题的合集，可获得更多播放机会" + 有 collection_name 时点开下拉 → 搜索/选中目标合集；没有则跳过。 + """ + if not self.collection_name: + return + try: + # 定位合集下拉框（通过 placeholder 文案） + select_box = page.locator('.cheetah-select:has(.cheetah-select-selection-placeholder:has-text("选择同主题的合集"))').first + if not await select_box.count(): + select_box = page.locator('.cheetah-select-selection-placeholder:has-text("合集")').locator('xpath=ancestor::div[contains(@class,"cheetah-select")]').first + if not await select_box.count(): + baijiahao_logger.warning(_msg("⚠️", "未找到合集选择器，跳过")) + return + + await select_box.scroll_into_view_if_needed() + await select_box.click(timeout=8000) + await page.wait_for_timeout(1500) + + # 在搜索框中输入合集名（触发搜索过滤） + search_input = select_box.locator('input.cheetah-select-selection-search-input').first + if await search_input.count(): + await search_input.fill(self.collection_name) + await page.wait_for_timeout(1500) + + # 从下拉选项中选中目标合集 + option = page.locator(f'[role="option"]:has-text("{self.collection_name}"), .cheetah-select-item:has-text("{self.collection_name}")').first + if await option.count(): + await option.click(timeout=5000) + await page.wait_for_timeout(500) + baijiahao_logger.success(_msg("🥳", f"已选择合集：{self.collection_name}")) + else: + baijiahao_logger.warning(_msg("⚠️", f"账号中无「{self.collection_name}」合集，跳过")) + await page.keyboard.press("Escape") + except Exception as exc: + baijiahao_logger.warning(_msg("⚠️", f"选择合集失败，跳过: {exc}")) + + async def _wait_cover_ready(self, page: Page, timeout: int = 120) -> None: + """等待百家号自动生成封面图。""" + start = time.monotonic() + while True: + if time.monotonic() - start > timeout: + baijiahao_logger.warning(_msg("⚠️", "等待封面生成超时，继续发布")) + return + if await page.locator("div.cheetah-spin-container img").count(): + baijiahao_logger.info(_msg("🖼️", "封面已生成")) + return + baijiahao_logger.info(_msg("🏃", "等待封面生成...")) + await asyncio.sleep(3) + + async def _submit_publish(self, page: Page) -> None: + """点击发布按钮并确认成功。""" + # 确保没有残留弹窗遮挡 + modal = page.locator('.cheetah-modal-wrap:visible').first + if await modal.count(): + await page.keyboard.press("Escape") + await page.wait_for_timeout(1000) + + # 百家号发布按钮有 data-testid="publish-btn" + publish_btn = page.locator('[data-testid="publish-btn"]').first + if not await publish_btn.count(): + publish_btn = page.locator('button:text-is("发布")').first + if not await publish_btn.count(): + publish_btn = page.locator('button:has-text("发布")').last + await publish_btn.wait_for(state="visible", timeout=15000) + await publish_btn.click(force=True) + baijiahao_logger.info(_msg("🏃", "已点击发布按钮")) + + # 等待跳转或成功提示（最多30s） + start = time.monotonic() + while time.monotonic() - start < 30: + url = page.url + # 发布成功跳转 + if BAIJIAHAO_SUCCESS_URL_PREFIX in url or "/rc/content" in url or "/rc/home" in url: + baijiahao_logger.success(_msg("🥳", "视频发布成功")) + return + # 检查是否出现百度安全验证 + if await page.locator('text="百度安全验证"').count(): + raise RuntimeError("出现百度安全验证，需人工处理") + # 检查是否有错误提示阻止发布 + error_toast = page.locator('.cheetah-message-error, .cheetah-message-warning').first + if await error_toast.count() and await error_toast.is_visible(): + err_text = await error_toast.inner_text() + baijiahao_logger.warning(_msg("⚠️", f"发布提示: {err_text}")) + await page.wait_for_timeout(1000) + + # 超时后再检查一次 + if BAIJIAHAO_SUCCESS_URL_PREFIX in page.url or "/rc/content" in page.url: + baijiahao_logger.success(_msg("🥳", "视频发布成功")) + else: + raise RuntimeError(f"发布后未跳转到成功页面（30s），当前 URL: {page.url}") async def main(self): async with async_playwright() as playwright: await self.upload(playwright) - - - - # 使用 AI成片 功能 - async def ai2video(self, playwright: Playwright) -> None: - # 使用 Chromium 浏览器启动一个浏览器实例 - browser = await playwright.chromium.launch(headless=self.headless, executable_path=self.local_executable_path, proxy=self.proxy_setting) - # 创建一个浏览器上下文，使用指定的 cookie 文件 - context = await browser.new_context( - viewport={"width": 1600, "height": 900}, - storage_state=f"{self.account_file}", - user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/127.0.4324.150 Safari/537.36' - ) - # context = await set_init_script(context) - await context.grant_permissions(['geolocation']) - - # 创建一个新的页面 - page = await context.new_page() - # 访问指定的 URL - await page.goto("https://aigc.baidu.com/make", timeout=60000) - # 等待页面跳转到指定的 URL，没进入，则自动等待到超时 - baijiahao_logger.info('正在打开主页...') - await page.wait_for_url("https://aigc.baidu.com/make", timeout=60000) - - # 点击"全网"标签 - await page.locator('div.rounded-lg.border:has-text("全网")').click() - await asyncio.sleep(1) # 这里延迟是为了方便眼睛直观的观看 - - # 点击 "上传视频" 按钮 - # await page.locator("div[class^='video-main-container'] input").set_input_files(self.file_path) - - # region 操作处 - - # 生成日期时间键名（格式：ai2video_YYYYMMDDHHMM） - now = datetime.now() - datetime_str = now.strftime("%Y%m%d%H%M") - processed_key = "ai2video_processed_titles" - batch_key = f"ai2video_{datetime_str}" - - # 初始化LocalStorage - await page.evaluate(f""" - if (!localStorage.getItem("{processed_key}")) {{ - localStorage.setItem("{processed_key}", JSON.stringify([])); - }} - if (!localStorage.getItem("{batch_key}")) {{ - localStorage.setItem("{batch_key}", JSON.stringify([])); - }} - """) - - # 定位新闻列表容器（转义特殊CSS字符） - container_selector = '.overflow-auto.flex-grow.h-0.saas-scrollbar.mt\-\[-4px\].pl\-\[24px\].pr\-\[10px\].pb\-\[18px\]' - news_items = await page.locator(container_selector).locator('div.py\-\[6px\].group.cursor-pointer').all() - - for item in news_items: - try: - # 获取新闻标题 - title_elem = item.locator('div.flex.text-gray-darker.items-center.relative.pr\-\[56px\] > span') - title = await title_elem.text_content() - if not title: - continue - - # 检查是否已处理过 - is_processed = await page.evaluate( - f"""title => {{ - const processedList = JSON.parse(localStorage.getItem("{processed_key}") || "[]"); - return processedList.includes(title); - }}""", - title - ) - - if is_processed: - print(f"[跳过] {title}") - continue - - # 悬停显示按钮（根据HTML结构，按钮在悬停时显示） - await item.hover() - - # 点击生成文案按钮 - button = item.locator('button:has-text("生成文案")') - await button.click() - print(f"[点击] {title}") - - # 等待30秒 - # await page.wait_for_timeout(30000) - print(f"[等待完成] {title}") - - # 监听"一键成片"按钮 - print(f"[开始监听] 一键成片按钮") - should_exit_while_loop = False # 添加标志变量 - while True: - # 定位"一键成片"按钮 - one_key_button = page.locator("button:has-text('一键成片')") - - # 检查按钮是否存在 - if await one_key_button.count() > 0: - # 检查按钮是否有disabled属性 - is_disabled = await one_key_button.get_attribute("disabled") - - if is_disabled is None: - # 按钮不再被禁用，点击它 - print(f"[发现可点击按钮] 一键成片") - await one_key_button.click() # 先点击一键成片按钮 - - # 等待可能出现的"温馨提示"窗口 - print(f"[检查] 是否出现温馨提示窗口") - await page.wait_for_timeout(2000) # 等待2秒，让窗口有时间显示 - - try: - # 检查是否存在"温馨提示"窗口，设置较短的超时时间 - tip_window = page.locator("div:has-text('温馨提示') >> visible=true") - if await tip_window.count() > 0: - print(f"[发现] 温馨提示窗口") - - # 定位并点击"知道了"按钮，设置较短的超时时间 - know_button = page.locator("button:has-text('知道了')") - if await know_button.count() > 0: - try: - # 设置较短的超时时间进行点击 - await know_button.click(timeout=5000) - print(f"[已点击] 知道了按钮") - except Exception as e: - print(f"[警告] 点击知道了按钮时出错: {str(e)}") - else: - print(f"[警告] 未找到知道了按钮") - else: - print(f"[信息] 未出现温馨提示窗口，继续执行") - except Exception as e: - print(f"[警告] 处理温馨提示窗口时出错: {str(e)}") - # 继续执行，不要因为这个错误中断流程 - - # 记录到LocalStorage前打印日志 - print(f"[开始记录] 准备将标题 '{title}' 记录到LocalStorage") - - # 记录到LocalStorage - await page.evaluate( - f""" - (title, processedKey, batchKey) => {{ - // 更新已处理列表 - const processedList = JSON.parse(localStorage.getItem(processedKey) || "[]"); - if (!processedList.includes(title)) {{ - processedList.push(title); - localStorage.setItem(processedKey, JSON.stringify(processedList)); - }} - - // 更新当前批次记录 - const batchList = JSON.parse(localStorage.getItem(batchKey) || "[]"); - if (!batchList.includes(title)) {{ - batchList.push(title); - localStorage.setItem(batchKey, JSON.stringify(batchList)); - }} - }} - """, - title, processed_key, batch_key - ) - - # 记录完成后打印日志 - print(f"[记录完成] 标题 '{title}' 已成功记录到LocalStorage") - - print(f"[记录完成] {title}") - - # 监听新打开的标签页 - print(f"[监听] 等待新标签页打开") - # 获取当前所有页面 - current_pages = context.pages - current_page_count = len(current_pages) - - # 等待新标签页打开（最多等待10秒） - new_page = None - max_wait_time = 10 # 最大等待时间（秒） - start_time = time.time() - - while time.time() - start_time < max_wait_time: - # 获取最新的页面列表 - pages = context.pages - # 如果页面数量增加，说明新标签页已打开 - if len(pages) > current_page_count: - # 获取最新打开的页面（通常是列表中的最后一个） - new_page = pages[-1] - print(f"[发现] 新标签页已打开") - break - # 短暂等待后再次检查 - await asyncio.sleep(0.5) - - # 如果找到新标签页，获取其标题和URL并保存 - if new_page: - # 等待页面加载完成 - try: - await new_page.wait_for_load_state("domcontentloaded", timeout=5000) - # 获取页面标题和URL - page_title = await new_page.title() - page_url = new_page.url - - print(f"[获取] 标题: {page_title}") - print(f"[获取] URL: {page_url}") - - # 将标题和URL保存到url.txt文件 - with open("url.txt", "a", encoding="utf-8") as f: - f.write(f"{page_title}\n{page_url}\n\n") - - print(f"[保存] 标题和URL已保存到url.txt") - - # 等待5秒后关闭新标签页 - print(f"[等待] 5秒后将关闭新标签页") - await asyncio.sleep(5) - await new_page.close() - print(f"[关闭] 新标签页已关闭") - except Exception as e: - print(f"[错误] 处理新标签页时出错: {str(e)}") - try: - # 尝试关闭页面，即使出错 - await new_page.close() - print(f"[关闭] 新标签页已关闭（出错后）") - except: - pass - else: - print(f"[警告] 未检测到新标签页打开") - - # 跳出整个while循环 - print(f"[操作] 跳出所有循环，不再处理其他新闻") - should_exit_while_loop = True # 设置标志变量 - break # 跳出while循环 - - # 检查是否需要跳出while循环 - if should_exit_while_loop: - break - - # 每秒检查一次按钮状态 - await page.wait_for_timeout(1000) - - # 检查是否需要跳出for循环 - if should_exit_while_loop: - print(f"[操作] 跳出for循环，完全结束处理") - break # 跳出for循环 - except Exception as e: - print(f"处理新闻时出错: {str(e)}") - continue - - - # endregion 操作处 - - print(f"[循环完成] 准备关闭浏览器") - - # 暂停 1000s - await asyncio.sleep(1000) # 这里延迟是为了方便眼睛直观的观看 - - # 退出前保存 storage 信息 - await context.storage_state(path=self.account_file) # 保存cookie - baijiahao_logger.info('cookie更新完毕！') - await asyncio.sleep(2) # 这里延迟是为了方便眼睛直观的观看 - # 关闭浏览器上下文和浏览器实例 - await context.close() - await browser.close() - - - async def mainAi(self): - async with async_playwright() as playwright: - await self.ai2video(playwright)