awaioi/EveryPublish
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(douyin): require explicit self-declaration

Browse Source
This commit is contained in:
lmr
2026-07-18 02:07:55 +08:00
parent d32280c98b
commit e89d3a875a
4 changed files with 154 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sau_cli.py
+7
View File
@@ -63,6 +63,7 @@ class DouyinVideoUploadRequest:
thumbnail_portrait_file: Path | None = None
product_link: str = ""
product_title: str = ""
declaration: str | None = None
publish_strategy: str = DOUYIN_PUBLISH_STRATEGY_IMMEDIATE
debug: bool = True
headless: bool = True
@@ -354,6 +355,7 @@ async def upload_video(request: DouyinVideoUploadRequest) -> Path:
) if request.thumbnail_portrait_file or request.thumbnail_file else None,
productLink=request.product_link,
productTitle=request.product_title,
declaration=request.declaration,
publish_strategy=request.publish_strategy,
debug=request.debug,
headless=request.headless,
@@ -600,6 +602,10 @@ def build_parser() -> argparse.ArgumentParser:
upload_video_parser.add_argument("--thumbnail-portrait", type=existing_file_path, help="Optional 3:4 portrait thumbnail path")
upload_video_parser.add_argument("--product-link", default="", help="Optional product link")
upload_video_parser.add_argument("--product-title", default="", help="Optional product title")
upload_video_parser.add_argument(
"--declaration",
help="Exact Douyin self-declaration option text; omitted means do not set one",
)
add_runtime_flags(upload_video_parser)
upload_note_parser = douyin_actions.add_parser("upload-note", help="Upload one note to Douyin")
@@ -762,6 +768,7 @@ async def dispatch(args: argparse.Namespace) -> int:
thumbnail_portrait_file=args.thumbnail_portrait,
product_link=args.product_link,
product_title=args.product_title,
declaration=args.declaration,
publish_strategy=publish_strategy,
debug=args.debug,
headless=args.headless,
tests/test_douyin_declaration.py
+98
View File
@@ -0,0 +1,98 @@
# Language: 中文
import asyncio
import unittest
from unittest.mock import AsyncMock, MagicMock, patch
from uploader.douyin_uploader import main as douyin_main
from uploader.douyin_uploader.main import DouYinVideo
class DouyinDeclarationTests(unittest.TestCase):
def test_upload_only_sets_explicit_declaration(self):
video = DouYinVideo(
"标题", "/tmp/demo.mp4", [], 0, "/tmp/cookie.json",
declaration="已确认声明原文",
)
self.assertEqual(video.declaration, "已确认声明原文")
def test_missing_declaration_does_not_fall_back_to_personal_opinion(self):
video = DouYinVideo("标题", "/tmp/demo.mp4", [], 0, "/tmp/cookie.json")
self.assertIsNone(video.declaration)
def test_legacy_positional_runtime_flags_keep_their_meaning(self):
video = DouYinVideo(
"标题", "/tmp/demo.mp4", [], 0, "/tmp/cookie.json",
None, "", "", None, "",
"scheduled", False, False,
)
self.assertEqual(video.publish_strategy, "scheduled")
self.assertFalse(video.debug)
self.assertFalse(video.headless)
self.assertIsNone(video.declaration)
def test_apply_declaration_skips_when_unspecified(self):
video = DouYinVideo("标题", "/tmp/demo.mp4", [], 0, "/tmp/cookie.json")
video.set_self_declaration = AsyncMock()
asyncio.run(video.apply_self_declaration(object()))
video.set_self_declaration.assert_not_awaited()
def test_apply_declaration_uses_exact_explicit_text(self):
page = object()
video = DouYinVideo(
"标题", "/tmp/demo.mp4", [], 0, "/tmp/cookie.json",
declaration="已确认声明原文",
)
video.set_self_declaration = AsyncMock(return_value=True)
asyncio.run(video.apply_self_declaration(page))
video.set_self_declaration.assert_awaited_once_with(page, "已确认声明原文")
def test_explicit_declaration_failure_blocks_publish(self):
video = DouYinVideo(
"标题", "/tmp/demo.mp4", [], 0, "/tmp/cookie.json",
declaration="已确认声明原文",
)
video.set_self_declaration = AsyncMock(return_value=False)
with self.assertRaisesRegex(RuntimeError, "自主声明"):
asyncio.run(video.apply_self_declaration(object()))
def test_declaration_failure_closes_browser_resources(self):
video = DouYinVideo(
"标题", "/tmp/demo.mp4", [], 0, "/tmp/cookie.json",
declaration="已确认声明原文",
)
video.validate_upload_args = AsyncMock()
video.fill_title_and_description = AsyncMock()
video.set_thumbnail = AsyncMock()
video.apply_self_declaration = AsyncMock(side_effect=RuntimeError("抖音自主声明设置失败"))
locator = MagicMock()
locator.set_input_files = AsyncMock()
locator.count = AsyncMock(return_value=1)
page = MagicMock()
page.goto = AsyncMock()
page.wait_for_url = AsyncMock()
page.wait_for_selector = AsyncMock()
page.locator.return_value = locator
context = MagicMock()
context.new_page = AsyncMock(return_value=page)
context.close = AsyncMock(side_effect=OSError("close failed"))
browser = MagicMock()
browser.new_context = AsyncMock(return_value=context)
browser.close = AsyncMock()
playwright = MagicMock()
playwright.chromium.launch = AsyncMock(return_value=browser)
with (
patch.object(douyin_main, "set_init_script", AsyncMock(return_value=context)),
patch.object(douyin_main.asyncio, "sleep", AsyncMock()),
self.assertRaisesRegex(RuntimeError, "自主声明"),
):
asyncio.run(video.upload(playwright))
context.close.assert_awaited_once()
browser.close.assert_awaited_once()
if __name__ == "__main__":
unittest.main()
tests/test_sau_browser_cli.py
+24
View File
@@ -68,6 +68,28 @@ class BrowserCliParserTests(unittest.TestCase):
self.assertEqual(args.thumbnail_landscape, landscape_path)
self.assertEqual(args.thumbnail_portrait, portrait_path)
def test_douyin_upload_video_accepts_explicit_declaration(self):
with tempfile.TemporaryDirectory() as tmp_dir:
video_path = Path(tmp_dir) / "demo.mp4"
video_path.write_bytes(b"video")
parser = sau_cli.build_parser()
args = parser.parse_args([
"douyin", "upload-video", "--account", "creator",
"--file", str(video_path), "--title", "标题",
"--declaration", "已确认声明原文",
])
self.assertEqual(args.declaration, "已确认声明原文")
def test_douyin_upload_video_has_no_implicit_declaration(self):
with tempfile.TemporaryDirectory() as tmp_dir:
video_path = Path(tmp_dir) / "demo.mp4"
video_path.write_bytes(b"video")
args = sau_cli.build_parser().parse_args([
"douyin", "upload-video", "--account", "creator",
"--file", str(video_path), "--title", "标题",
])
self.assertIsNone(args.declaration)
def test_tencent_upload_video_accepts_dual_thumbnail_aspects(self):
with tempfile.TemporaryDirectory() as tmp_dir:
video_path = Path(tmp_dir) / "demo.mp4"
@@ -210,6 +232,7 @@ class BrowserCliDispatchTests(unittest.TestCase):
thumbnail_portrait=Path("portrait.png"),
product_link="",
product_title="",
declaration="已确认声明原文",
debug=False,
headless=True,
)
@@ -219,6 +242,7 @@ class BrowserCliDispatchTests(unittest.TestCase):
request = mock_upload.await_args.args[0]
self.assertEqual(request.thumbnail_landscape_file, Path("landscape.png"))
self.assertEqual(request.thumbnail_portrait_file, Path("portrait.png"))
self.assertEqual(request.declaration, "已确认声明原文")
def test_dispatch_tencent_upload_video_uses_dual_thumbnail_request_fields(self):
args = Namespace(
uploader/douyin_uploader/main.py
+25 -8
View File
@@ -420,12 +420,8 @@ class DouYinBaseUploader(BaseVideoUploader):
douyin_logger.error(_msg("😢", f"设置商品链接时出错: {str(e)}"))
return False
async def set_self_declaration(self, page: Page, declaration: str = "内容为个人观点或见解") -> None:
"""抖音「自主声明」为发布必选项：打开声明弹窗 → 选指定类型 → 确定。
入口和弹窗都是异步渲染，等不到就记 warning 跳过、继续发布，绝不因此中断
（与小红书话题、视频号声明原创的容错策略保持一致）。
"""
async def set_self_declaration(self, page: Page, declaration: str) -> bool:
"""按调用方给出的平台原文选择自主声明；失败返回 False。"""
try:
# 发布页底部「自主声明」行，未选时显示占位文案「请选择自主声明」
entry = page.get_by_text("请选择自主声明").first
@@ -446,8 +442,10 @@ class DouYinBaseUploader(BaseVideoUploader):
await dialog.get_by_role("button", name="确定").click(timeout=6000)
await dialog.wait_for(state="hidden", timeout=6000)
douyin_logger.info(_msg("🧾", f"自主声明已选择「{declaration}」"))
return True
except Exception as exc:
douyin_logger.warning(_msg("🧾", f"自主声明设置失败，跳过该步骤继续发布：{exc}"))
douyin_logger.warning(_msg("🧾", f"自主声明设置失败：{exc}"))
return False
async def select_bgm(self, page: Page, bgm_name: str) -> bool:
"""为图文发布选择 BGM：可选增强功能，搜索无结果或异常均跳过不中断发布。"""
@@ -532,6 +530,7 @@ class DouYinVideo(DouYinBaseUploader):
publish_strategy: str = DOUYIN_PUBLISH_STRATEGY_IMMEDIATE,
debug: bool = DEBUG_MODE,
headless: bool = LOCAL_CHROME_HEADLESS,
declaration: str | None = None,
):
super().__init__(
publish_date=publish_date,
@@ -548,6 +547,13 @@ class DouYinVideo(DouYinBaseUploader):
self.productLink = productLink
self.productTitle = productTitle
self.desc = desc or ""
self.declaration = declaration.strip() if declaration and declaration.strip() else None
async def apply_self_declaration(self, page: Page) -> None:
if not self.declaration:
return
if not await self.set_self_declaration(page, self.declaration):
raise RuntimeError(f"自主声明「{self.declaration}」设置失败，拒绝继续发布")
async def _submit_sms_verify_code(self, page: Page, sms_input, code: str, code_file: str) -> bool:
douyin_logger.info(_msg("✍️", f"已获取验证码，准备填入: {code}"))
@@ -731,7 +737,18 @@ class DouYinVideo(DouYinBaseUploader):
await self.set_thumbnail(page)
await self.set_self_declaration(page)
try:
await self.apply_self_declaration(page)
except Exception:
try:
await context.close()
except Exception:
pass
try:
await browser.close()
except Exception:
pass
raise
third_part_element = '[class^="info"] > [class^="first-part"] div div.semi-switch'
if await page.locator(third_part_element).count():