tests/test_fc2_cast_review.py
Rama master · réplica de lectura del repositorio original.
# -*- coding: utf-8 -*-
"""FC2 演员复核清单：账本已有的演员与 fc2cmadb 对不上的番号列出来，只出清单不写库。"""
from __future__ import annotations
import json
import sqlite3
import tempfile
import unittest
from pathlib import Path
from peach.jav_cover_fetch import NotFound
from peach.review_csv import read_candidates, read_rows
from scripts.fc2_cast_review import build_parser, run
from support.ledger import fresh_ledger
class _Mirror:
"""按番号回 fc2cmadb 那一页；没登记的番号是 404，登记成异常的原样抛。"""
def __init__(self, pages):
self.pages = pages
self.asked = []
def site(self, source, code):
self.asked.append((source, code))
page = self.pages.get(code)
if page is None:
raise NotFound("HTTP 404")
if isinstance(page, Exception):
raise page
return page
def _page(code, *names):
return {"id": code, "source_url": f"https://fc2cmadb.com/articles/{code.rsplit('-', 1)[-1]}",
"actresses": list(names)}
class Fc2CastReviewTests(unittest.TestCase):
def setUp(self):
tmp = tempfile.TemporaryDirectory()
self.addCleanup(tmp.cleanup)
self.root = Path(tmp.name).resolve()
self.db = fresh_ledger(self.root)
self.snapshots = self.root / "library-metadata"
self.snapshots.mkdir()
self.output = self.root / "generated" / "fc2-cast-candidates.csv"
con = sqlite3.connect(self.db)
try:
con.executemany(
"INSERT INTO asset(id,location,path,name,medium,size,first_seen,code)"
" VALUES(?,'local',?,?,'video',100,'2026-01-01',?)",
[(1, r"R:\media\a.mp4", "a.mp4", "FC2-PPV-2629971"),
(2, r"R:\media\b.mp4", "b.mp4", "FC2-PPV-1449453"),
(3, r"R:\media\c.mp4", "c.mp4", "FC2-PPV-3518061"),
(4, r"R:\media\d.mp4", "d.mp4", "FC2-PPV-2851534")])
con.executemany(
"INSERT INTO entity(id,kind,canonical_name,normalized_name,metadata_json,"
"created_at,updated_at) VALUES(?,'performer',?,?,'{}','t','t')",
[(11, "安娜", "安娜"), (12, "Chisa", "chisa"), (13, "梨奈", "梨奈"), (14, "みお", "みお")])
con.execute("INSERT INTO entity_alias(entity_id,alias,normalized_alias,source)"
" VALUES(13,'りな','りな','test')")
con.executemany(
"INSERT INTO asset_entity(asset_id,entity_id,role,source,confidence)"
" VALUES(?,?,'performer',?,1.0)",
[(1, 11, "javinizer:javdb:performer"), (2, 12, "javinizer:javdb:performer"),
(3, 13, "javinizer:javdb:performer"), (4, 14, "user")])
con.commit()
finally:
con.close()
def _snapshot(self, code, *names):
(self.snapshots / f"{code}-fc2cmadb.json").write_text(
json.dumps(_page(code, *names), ensure_ascii=False), encoding="utf-8")
def _run(self, *extra, provider=None):
self.pauses = []
args = build_parser().parse_args(
["--db", str(self.db), "--snapshots", str(self.snapshots),
"--output", str(self.output), "--misses", str(self.root / "misses.json"), *extra])
return run(args, provider=provider, sleep=self.pauses.append)
def test_a_different_name_on_the_mirror_becomes_one_review_row(self):
self._snapshot("FC2-PPV-2629971", "あんな")
self._snapshot("FC2-PPV-3518061", "りな")
result = self._run("--offline")
rows = read_rows(self.output)
self.assertEqual([row["item_key"] for row in rows], ["FC2-PPV-2629971:performers:fc2cmadb"])
self.assertEqual(rows[0]["current_value"], "安娜")
candidate = json.loads(rows[0]["candidates_json"])[0]
self.assertEqual((candidate["source"], candidate["display_value"]), ("fc2cmadb", "あんな"))
self.assertEqual(result["review"], 1)
# 别名登记过的同一个人不算分歧；没存快照的离线时算「站上没有」。
self.assertEqual((result["agreed_or_empty"], result["not_on_mirror"]), (1, 1))
def test_a_code_with_a_cast_the_user_set_is_left_alone(self):
self._snapshot("FC2-PPV-2851534", "梨奈")
self._run("--offline")
self.assertEqual(read_rows(self.output), [])
def test_the_review_page_reads_the_list_as_metadata_fields(self):
self._snapshot("FC2-PPV-1449453", "大村阿美香")
self._run("--offline")
rows, _label, skipped = read_candidates("metadata_fields", self.output.parent)
self.assertEqual(skipped, 0)
self.assertEqual([(row["item_key"], row["field"]) for row in rows],
[("FC2-PPV-1449453:performers:fc2cmadb", "performers")])
def test_a_live_answer_is_kept_as_a_snapshot_and_not_asked_twice(self):
mirror = _Mirror({"FC2-PPV-1449453": _page("FC2-PPV-1449453", "大村阿美香")})
first = self._run(provider=mirror)
self.assertEqual((first["review"], first["not_on_mirror"]), (1, 2))
self.assertTrue((self.snapshots / "FC2-PPV-1449453-fc2cmadb.json").is_file())
asked = len(mirror.asked)
self._run(provider=mirror)
self.assertNotIn(("fc2cmadb", "FC2-PPV-1449453"), mirror.asked[asked:])
def test_a_code_the_mirror_does_not_have_is_not_asked_again(self):
mirror = _Mirror({})
self._run(provider=mirror)
self.assertEqual(len(mirror.asked), 3)
result = self._run(provider=mirror)
self.assertEqual(len(mirror.asked), 3)
self.assertEqual(result["not_on_mirror"], 3)
def test_live_questions_are_spaced_but_snapshots_are_not(self):
self._snapshot("FC2-PPV-1449453", "大村阿美香")
self._run("--interval", "7", provider=_Mirror({}))
# 三个番号里一个有快照，剩下两次联网之间停一次。
self.assertEqual(self.pauses, [7.0])
def test_a_snapshot_without_the_cast_column_is_asked_once_more(self):
"""女优栏要单独再问一跳，没问过那一跳的快照是空的（`2629971` 站上是 `あんな`）。"""
self._snapshot("FC2-PPV-2629971")
self._snapshot("FC2-PPV-3518061")
mirror = _Mirror({"FC2-PPV-2629971": _page("FC2-PPV-2629971", "あんな"),
"FC2-PPV-3518061": _page("FC2-PPV-3518061")})
self._run(provider=mirror)
self.assertEqual([row["code"] for row in read_rows(self.output)], ["FC2-PPV-2629971"])
asked = len(mirror.asked)
self._run(provider=mirror)
# 重问过仍空的一周内不再问；有了女优栏的读快照。
self.assertNotIn("FC2-PPV-3518061", [code for _, code in mirror.asked[asked:]])
self.assertNotIn("FC2-PPV-2629971", [code for _, code in mirror.asked[asked:]])
def test_a_failure_other_than_absence_stops_and_keeps_what_was_found(self):
mirror = _Mirror({"FC2-PPV-1449453": _page("FC2-PPV-1449453", "大村阿美香"),
"FC2-PPV-2629971": TimeoutError("read timed out")})
result = self._run(provider=mirror)
self.assertTrue(result["stopped"].startswith("FC2-PPV-2629971"))
self.assertEqual([row["code"] for row in read_rows(self.output)], ["FC2-PPV-1449453"])
if __name__ == "__main__":
unittest.main()