321 lines
17 KiB
Python
321 lines
17 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from biliup_next.core.models import Task, utc_now_iso
|
|
from biliup_next.core.errors import ModuleError
|
|
from biliup_next.modules.comment.providers.bilibili_top_comment import BilibiliTopCommentProvider
|
|
|
|
|
|
class _FakeBilibiliApi:
|
|
def __init__(self) -> None:
|
|
self.reply_calls: list[dict[str, object]] = []
|
|
|
|
def load_cookies(self, path: Path) -> dict[str, str]:
|
|
return {"bili_jct": "csrf-token"}
|
|
|
|
def build_session(self, *, cookies: dict[str, str], referer: str, origin: str | None = None) -> object:
|
|
return object()
|
|
|
|
def get_video_view(self, session, bvid: str, *, error_code: str, error_message: str) -> dict[str, object]:
|
|
return {"aid": 123}
|
|
|
|
def add_reply(self, session, *, csrf: str, aid: int, content: str, error_message: str) -> dict[str, object]:
|
|
self.reply_calls.append({"aid": aid, "content": content, "error_message": error_message})
|
|
raise ModuleError(
|
|
code="COMMENT_POST_FAILED",
|
|
message=f"{error_message}: 当前页面评论功能已关闭",
|
|
retryable=True,
|
|
)
|
|
|
|
def top_reply(self, session, *, csrf: str, aid: int, rpid: int, error_message: str) -> None:
|
|
raise AssertionError("top_reply should not be called when comment is disabled")
|
|
|
|
|
|
class BilibiliTopCommentProviderTests(unittest.TestCase):
|
|
def test_split_comment_aggregates_session_parts_on_anchor_task(self) -> None:
|
|
api = _FakeBilibiliApi()
|
|
provider = BilibiliTopCommentProvider(bilibili_api=api)
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
root = Path(tmpdir)
|
|
task_dir_1 = root / "task-1"
|
|
task_dir_2 = root / "task-2"
|
|
task_dir_1.mkdir(parents=True, exist_ok=True)
|
|
task_dir_2.mkdir(parents=True, exist_ok=True)
|
|
task = Task("task-1", "local_file", str(task_dir_1 / "source-1.mp4"), "task-1", "published", utc_now_iso(), utc_now_iso())
|
|
(task_dir_1 / "songs.txt").write_text("00:00:00 Song A — Artist A\n", encoding="utf-8")
|
|
(task_dir_1 / "songs.json").write_text(json.dumps({"songs": [{"title": "Song A", "artist": "Artist A"}]}), encoding="utf-8")
|
|
(task_dir_1 / "bvid.txt").write_text("BV1SPLIT111", encoding="utf-8")
|
|
(task_dir_2 / "songs.txt").write_text("00:00:00 Song B — Artist B\n", encoding="utf-8")
|
|
(task_dir_2 / "songs.json").write_text(json.dumps({"songs": [{"title": "Song B", "artist": "Artist B"}]}), encoding="utf-8")
|
|
cookies_file = root / "cookies.json"
|
|
cookies_file.write_text("{}", encoding="utf-8")
|
|
|
|
class _Repo:
|
|
def get_task_context(self, task_id): # noqa: ANN001
|
|
mapping = {
|
|
"task-1": type("Ctx", (), {"task_id": "task-1", "session_key": "session-1", "segment_started_at": "2026-04-04T09:23:00+08:00", "source_title": "part-1"})(),
|
|
"task-2": type("Ctx", (), {"task_id": "task-2", "session_key": "session-1", "segment_started_at": "2026-04-04T09:25:00+08:00", "source_title": "part-2"})(),
|
|
}
|
|
return mapping[task_id]
|
|
|
|
def list_task_contexts_by_session_key(self, session_key): # noqa: ANN001
|
|
return [self.get_task_context("task-1"), self.get_task_context("task-2")]
|
|
|
|
def get_task(self, task_id): # noqa: ANN001
|
|
mapping = {
|
|
"task-1": Task("task-1", "local_file", str(task_dir_1 / "source-1.mp4"), "task-1", "published", utc_now_iso(), utc_now_iso()),
|
|
"task-2": Task("task-2", "local_file", str(task_dir_2 / "source-2.mp4"), "task-2", "published", utc_now_iso(), utc_now_iso()),
|
|
}
|
|
return mapping[task_id]
|
|
|
|
result = provider.comment(
|
|
task,
|
|
{
|
|
"session_dir": str(root),
|
|
"cookies_file": str(cookies_file),
|
|
"post_split_comment": True,
|
|
"post_full_video_timeline_comment": False,
|
|
"__repo": _Repo(),
|
|
},
|
|
)
|
|
|
|
self.assertEqual(result["status"], "ok")
|
|
self.assertEqual(result["split"]["status"], "skipped")
|
|
self.assertEqual(result["split"]["reason"], "comment_disabled")
|
|
self.assertEqual(len(api.reply_calls), 1)
|
|
self.assertIn("P1:\n1. Song A — Artist A", api.reply_calls[0]["content"])
|
|
self.assertIn("P2:\n1. Song B — Artist B", api.reply_calls[0]["content"])
|
|
|
|
def test_split_comment_skips_on_non_anchor_task(self) -> None:
|
|
api = _FakeBilibiliApi()
|
|
provider = BilibiliTopCommentProvider(bilibili_api=api)
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
root = Path(tmpdir)
|
|
task_dir = root / "task-2"
|
|
task_dir.mkdir(parents=True, exist_ok=True)
|
|
task = Task("task-2", "local_file", str(task_dir / "source-2.mp4"), "task-2", "published", utc_now_iso(), utc_now_iso())
|
|
(task_dir / "songs.txt").write_text("00:00:00 Song B — Artist B\n", encoding="utf-8")
|
|
(task_dir / "songs.json").write_text(json.dumps({"songs": [{"title": "Song B", "artist": "Artist B"}]}), encoding="utf-8")
|
|
(task_dir / "bvid.txt").write_text("BV1SPLIT222", encoding="utf-8")
|
|
cookies_file = root / "cookies.json"
|
|
cookies_file.write_text("{}", encoding="utf-8")
|
|
|
|
class _Repo:
|
|
def get_task_context(self, task_id): # noqa: ANN001
|
|
mapping = {
|
|
"task-1": type("Ctx", (), {"task_id": "task-1", "session_key": "session-1", "segment_started_at": "2026-04-04T09:23:00+08:00", "source_title": "part-1"})(),
|
|
"task-2": type("Ctx", (), {"task_id": "task-2", "session_key": "session-1", "segment_started_at": "2026-04-04T09:25:00+08:00", "source_title": "part-2"})(),
|
|
}
|
|
return mapping[task_id]
|
|
|
|
def list_task_contexts_by_session_key(self, session_key): # noqa: ANN001
|
|
return [self.get_task_context("task-1"), self.get_task_context("task-2")]
|
|
|
|
def get_task(self, task_id): # noqa: ANN001
|
|
mapping = {
|
|
"task-1": Task("task-1", "local_file", str(root / "task-1" / "source-1.mp4"), "task-1", "published", utc_now_iso(), utc_now_iso()),
|
|
"task-2": Task("task-2", "local_file", str(task_dir / "source-2.mp4"), "task-2", "published", utc_now_iso(), utc_now_iso()),
|
|
}
|
|
return mapping[task_id]
|
|
|
|
result = provider.comment(
|
|
task,
|
|
{
|
|
"session_dir": str(root),
|
|
"cookies_file": str(cookies_file),
|
|
"post_split_comment": True,
|
|
"post_full_video_timeline_comment": False,
|
|
"__repo": _Repo(),
|
|
},
|
|
)
|
|
|
|
self.assertEqual(result["status"], "ok")
|
|
self.assertEqual(result["split"]["status"], "skipped")
|
|
self.assertEqual(result["split"]["reason"], "session_split_comment_owned_by_anchor")
|
|
self.assertEqual(api.reply_calls, [])
|
|
|
|
def test_comment_skips_when_page_comment_is_disabled(self) -> None:
|
|
provider = BilibiliTopCommentProvider(bilibili_api=_FakeBilibiliApi())
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
root = Path(tmpdir)
|
|
work_dir = root / "task-1"
|
|
work_dir.mkdir(parents=True, exist_ok=True)
|
|
task = Task(
|
|
id="task-1",
|
|
source_type="local_file",
|
|
source_path=str(work_dir / "source.mp4"),
|
|
title="task-1",
|
|
status="published",
|
|
created_at=utc_now_iso(),
|
|
updated_at=utc_now_iso(),
|
|
)
|
|
(work_dir / "songs.txt").write_text("00:00:00 Test Song - Tester\n", encoding="utf-8")
|
|
(work_dir / "songs.json").write_text(json.dumps({"songs": [{"title": "Test Song", "artist": "Tester"}]}), encoding="utf-8")
|
|
(work_dir / "bvid.txt").write_text("BV1COMMENT123", encoding="utf-8")
|
|
cookies_file = root / "cookies.json"
|
|
cookies_file.write_text("{}", encoding="utf-8")
|
|
|
|
result = provider.comment(
|
|
task,
|
|
{
|
|
"session_dir": str(root),
|
|
"cookies_file": str(cookies_file),
|
|
"post_split_comment": True,
|
|
"post_full_video_timeline_comment": False,
|
|
},
|
|
)
|
|
|
|
self.assertEqual(result["status"], "ok")
|
|
self.assertEqual(result["split"]["status"], "skipped")
|
|
self.assertEqual(result["split"]["reason"], "comment_disabled")
|
|
self.assertTrue((work_dir / "comment_split_done.flag").exists())
|
|
self.assertTrue((work_dir / "comment_full_done.flag").exists())
|
|
self.assertTrue((work_dir / "comment_done.flag").exists())
|
|
|
|
def test_comment_uses_source_path_parent_when_task_title_differs(self) -> None:
|
|
provider = BilibiliTopCommentProvider(bilibili_api=_FakeBilibiliApi())
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
root = Path(tmpdir)
|
|
work_dir = root / "task-id-dir"
|
|
work_dir.mkdir(parents=True, exist_ok=True)
|
|
task = Task(
|
|
id="task-id",
|
|
source_type="bilibili_url",
|
|
source_path=str(work_dir / "task-id.mp4"),
|
|
title="display-title",
|
|
status="published",
|
|
created_at=utc_now_iso(),
|
|
updated_at=utc_now_iso(),
|
|
)
|
|
(work_dir / "songs.txt").write_text("00:00:00 Test Song - Tester\n", encoding="utf-8")
|
|
(work_dir / "songs.json").write_text(json.dumps({"songs": [{"title": "Test Song", "artist": "Tester"}]}), encoding="utf-8")
|
|
(work_dir / "bvid.txt").write_text("BV1COMMENT123", encoding="utf-8")
|
|
cookies_file = root / "cookies.json"
|
|
cookies_file.write_text("{}", encoding="utf-8")
|
|
|
|
result = provider.comment(
|
|
task,
|
|
{
|
|
"session_dir": str(root),
|
|
"cookies_file": str(cookies_file),
|
|
"post_split_comment": True,
|
|
"post_full_video_timeline_comment": False,
|
|
},
|
|
)
|
|
|
|
self.assertEqual(result["status"], "ok")
|
|
self.assertEqual(result["split"]["status"], "skipped")
|
|
self.assertEqual(result["split"]["reason"], "comment_disabled")
|
|
self.assertTrue((work_dir / "comment_done.flag").exists())
|
|
|
|
def test_full_comment_aggregates_session_parts_on_anchor_task(self) -> None:
|
|
api = _FakeBilibiliApi()
|
|
provider = BilibiliTopCommentProvider(bilibili_api=api)
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
root = Path(tmpdir)
|
|
task_dir_1 = root / "task-1"
|
|
task_dir_2 = root / "task-2"
|
|
task_dir_1.mkdir(parents=True, exist_ok=True)
|
|
task_dir_2.mkdir(parents=True, exist_ok=True)
|
|
task = Task("task-1", "local_file", str(task_dir_1 / "source-1.mp4"), "task-1", "published", utc_now_iso(), utc_now_iso())
|
|
(task_dir_1 / "songs.txt").write_text("00:00:01 Song A\n00:02:00 Song B\n", encoding="utf-8")
|
|
(task_dir_1 / "songs.json").write_text(json.dumps({"songs": [{"title": "Song A"}]}), encoding="utf-8")
|
|
(task_dir_1 / "bvid.txt").write_text("BV1SPLIT111", encoding="utf-8")
|
|
(task_dir_1 / "full_video_bvid.txt").write_text("BV1FULL111", encoding="utf-8")
|
|
(task_dir_2 / "songs.txt").write_text("00:00:03 Song C\n", encoding="utf-8")
|
|
cookies_file = root / "cookies.json"
|
|
cookies_file.write_text("{}", encoding="utf-8")
|
|
|
|
class _Repo:
|
|
def get_task_context(self, task_id): # noqa: ANN001
|
|
mapping = {
|
|
"task-1": type("Ctx", (), {"task_id": "task-1", "session_key": "session-1", "segment_started_at": "2026-04-04T09:23:00+08:00", "source_title": "part-1"})(),
|
|
"task-2": type("Ctx", (), {"task_id": "task-2", "session_key": "session-1", "segment_started_at": "2026-04-04T09:25:00+08:00", "source_title": "part-2"})(),
|
|
}
|
|
return mapping[task_id]
|
|
|
|
def list_task_contexts_by_session_key(self, session_key): # noqa: ANN001
|
|
return [self.get_task_context("task-1"), self.get_task_context("task-2")]
|
|
|
|
def get_task(self, task_id): # noqa: ANN001
|
|
mapping = {
|
|
"task-1": Task("task-1", "local_file", str(task_dir_1 / "source-1.mp4"), "task-1", "published", utc_now_iso(), utc_now_iso()),
|
|
"task-2": Task("task-2", "local_file", str(task_dir_2 / "source-2.mp4"), "task-2", "published", utc_now_iso(), utc_now_iso()),
|
|
}
|
|
return mapping[task_id]
|
|
|
|
result = provider.comment(
|
|
task,
|
|
{
|
|
"session_dir": str(root),
|
|
"cookies_file": str(cookies_file),
|
|
"post_split_comment": False,
|
|
"post_full_video_timeline_comment": True,
|
|
"__repo": _Repo(),
|
|
},
|
|
)
|
|
|
|
self.assertEqual(result["status"], "ok")
|
|
self.assertEqual(result["full"]["status"], "skipped")
|
|
self.assertEqual(result["full"]["reason"], "comment_disabled")
|
|
self.assertEqual(len(api.reply_calls), 1)
|
|
self.assertIn("P1:\n00:00:01 Song A\n00:02:00 Song B", api.reply_calls[0]["content"])
|
|
self.assertIn("P2:\n00:00:03 Song C", api.reply_calls[0]["content"])
|
|
|
|
def test_full_comment_skips_on_non_anchor_task(self) -> None:
|
|
api = _FakeBilibiliApi()
|
|
provider = BilibiliTopCommentProvider(bilibili_api=api)
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
root = Path(tmpdir)
|
|
task_dir = root / "task-2"
|
|
task_dir.mkdir(parents=True, exist_ok=True)
|
|
task = Task("task-2", "local_file", str(task_dir / "source-2.mp4"), "task-2", "published", utc_now_iso(), utc_now_iso())
|
|
(task_dir / "songs.txt").write_text("00:00:03 Song C\n", encoding="utf-8")
|
|
(task_dir / "songs.json").write_text(json.dumps({"songs": [{"title": "Song C"}]}), encoding="utf-8")
|
|
(task_dir / "bvid.txt").write_text("BV1SPLIT222", encoding="utf-8")
|
|
(task_dir / "full_video_bvid.txt").write_text("BV1FULL111", encoding="utf-8")
|
|
cookies_file = root / "cookies.json"
|
|
cookies_file.write_text("{}", encoding="utf-8")
|
|
|
|
class _Repo:
|
|
def get_task_context(self, task_id): # noqa: ANN001
|
|
mapping = {
|
|
"task-1": type("Ctx", (), {"task_id": "task-1", "session_key": "session-1", "segment_started_at": "2026-04-04T09:23:00+08:00", "source_title": "part-1"})(),
|
|
"task-2": type("Ctx", (), {"task_id": "task-2", "session_key": "session-1", "segment_started_at": "2026-04-04T09:25:00+08:00", "source_title": "part-2"})(),
|
|
}
|
|
return mapping[task_id]
|
|
|
|
def list_task_contexts_by_session_key(self, session_key): # noqa: ANN001
|
|
return [self.get_task_context("task-1"), self.get_task_context("task-2")]
|
|
|
|
def get_task(self, task_id): # noqa: ANN001
|
|
mapping = {
|
|
"task-1": Task("task-1", "local_file", str(root / "task-1" / "source-1.mp4"), "task-1", "published", utc_now_iso(), utc_now_iso()),
|
|
"task-2": Task("task-2", "local_file", str(task_dir / "source-2.mp4"), "task-2", "published", utc_now_iso(), utc_now_iso()),
|
|
}
|
|
return mapping[task_id]
|
|
|
|
result = provider.comment(
|
|
task,
|
|
{
|
|
"session_dir": str(root),
|
|
"cookies_file": str(cookies_file),
|
|
"post_split_comment": False,
|
|
"post_full_video_timeline_comment": True,
|
|
"__repo": _Repo(),
|
|
},
|
|
)
|
|
|
|
self.assertEqual(result["status"], "ok")
|
|
self.assertEqual(result["full"]["status"], "skipped")
|
|
self.assertEqual(result["full"]["reason"], "session_full_comment_owned_by_anchor")
|
|
self.assertEqual(api.reply_calls, [])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|