tests/test_fc2_metadata.py
Rama master · réplica de lectura del repositorio original.
# -*- coding: utf-8 -*-
"""fc2cmadb 评论解析。样本取自 3312576 等页面上的真实评论。"""
from __future__ import annotations
import json
import sqlite3
import tempfile
import unittest
from pathlib import Path
from peach.sources.fc2cmadb import collection_parts, page_comments, parse_equivalences, parse_performers
from scripts.fetch_fc2_metadata import (
backfill,
harvest,
harvest_rows,
high_resolution_cover_url,
metadata_candidate_rows,
summarise,
translated_tags,
)
#: 站上评论是 CRLF 换行，样本照原样保留。
CRLF = "\r\n"
# 演员标记：全角空格分隔，一行可有多人。
PERFORMER_COMMENT = """2664942 ひかり
2355314 真夏
2802445 真夏
1955351 みなみ"""
MULTI_PERFORMER_COMMENT = """1934545 ゆう
2589532 未歩なな
2724256 未歩なな 皐月
*
1909413 すず"""
# 等价标记：断言拆成两行写，`=` 起头的行属于上一行。
SPLIT_EQUIV_COMMENT = """3312576-1
= 3090722-3
3312576-4
= 2471432 = 3090722-1 = 4605413"""
# 另一种写法：单行、下划线分片，末尾用 `bad:` 列对不上的分片。
INLINE_EQUIV_COMMENT = """3090722=3312576_1++
2471432=3312576_4
1912367=1909413=1952605=3312576_10
bad:
#2
#3312576-3=9"""
class ParsePerformersTest(unittest.TestCase):
def test_maps_video_id_to_names(self):
found = parse_performers(PERFORMER_COMMENT)
self.assertEqual(found["2355314"], ["真夏"])
self.assertEqual(found["1955351"], ["みなみ"])
def test_keeps_every_name_on_a_co_starred_line(self):
found = parse_performers(MULTI_PERFORMER_COMMENT)
self.assertEqual(found["2724256"], ["未歩なな", "皐月"])
def test_ignores_decorations_and_equivalence_lines(self):
found = parse_performers(MULTI_PERFORMER_COMMENT)
self.assertNotIn("*", found)
# 等价行里的数字不能被当成演员名读进来。
self.assertEqual(parse_performers(INLINE_EQUIV_COMMENT), {})
class ParseEquivalencesTest(unittest.TestCase):
def test_joins_a_continuation_line_onto_its_subject(self):
groups = parse_equivalences(SPLIT_EQUIV_COMMENT)
self.assertIn([("3312576", "1"), ("3090722", "3")], groups)
def test_reads_underscore_parts_and_long_chains(self):
groups = parse_equivalences(INLINE_EQUIV_COMMENT)
self.assertIn([("2471432", ""), ("3312576", "4")], groups)
self.assertIn(
[("1912367", ""), ("1909413", ""), ("1952605", ""), ("3312576", "10")],
groups,
)
def test_stops_at_the_bad_section(self):
"""`bad:` 之后是「对不上」的记录，读成等价会把错误关系写进候选。"""
groups = parse_equivalences(INLINE_EQUIV_COMMENT)
self.assertNotIn([("3312576", "3"), ("9", "")], groups)
for group in groups:
self.assertNotIn("9", [token[0] for token in group])
class CollectionTest(unittest.TestCase):
def test_counts_distinct_parts_of_this_id(self):
groups = parse_equivalences(SPLIT_EQUIV_COMMENT + "\n" + INLINE_EQUIV_COMMENT)
parts = collection_parts("3312576", groups)
self.assertEqual(parts["1"], "3090722")
self.assertEqual(parts["4"], "2471432")
self.assertGreaterEqual(len(parts), 3)
def test_a_standalone_work_is_not_a_collection(self):
groups = parse_equivalences("2471432 = 4605413")
self.assertEqual(collection_parts("2471432", groups), {})
class SummariseTest(unittest.TestCase):
def _props(self, comments, **article):
base = {"title": "作品", "image_url": "https://example.invalid/c.png",
"writer": {"name": "千華繚乱", "slug": "hyakka"}, "tags": []}
base.update(article)
return {"article": base, "comments": {"data": comments}}
def test_withholds_the_cover_from_a_collection(self):
"""合集封面套给每个分片，会让 21 段不同内容显示同一张图。"""
props = self._props([{"id": 1, "body": SPLIT_EQUIV_COMMENT},
{"id": 2, "body": INLINE_EQUIV_COMMENT}])
row = summarise("3312576", props)
self.assertEqual(row["is_collection"], "1")
self.assertEqual(row["cover_url"], "")
self.assertIn("封面不下发", row["note"])
def test_keeps_the_cover_for_a_single_work(self):
row = summarise("2355314", self._props([{"id": 1, "body": PERFORMER_COMMENT}]))
self.assertEqual(row["is_collection"], "")
self.assertEqual(row["cover_url"], "https://example.invalid/c.png")
def test_upgrades_fc2_listing_thumbnail_to_the_measured_1200px_rendition(self):
url = "https://contents-thumbnail2.fc2.com/w276/storage.example/cover.jpg"
self.assertEqual(high_resolution_cover_url(url),
"https://contents-thumbnail2.fc2.com/w1200/storage.example/cover.jpg")
def test_ranks_performers_by_how_many_comments_name_them(self):
props = self._props([{"id": 1, "body": "2355314 真夏"},
{"id": 2, "body": "2355314 真夏"},
{"id": 3, "body": "2355314 ひかり"}])
row = summarise("2355314", props)
self.assertEqual(row["performers"], "真夏 ひかり")
self.assertEqual(row["performer_votes"], "真夏:2 ひかり:1")
class MetadataCandidateTest(unittest.TestCase):
def test_translates_only_reviewed_fc2_tags_and_deduplicates_synonyms(self):
self.assertEqual(
translated_tags("素人 フェラ フェラチオ 口内発射 未知"),
["素人", "口交", "口爆"],
)
def test_article_title_tags_and_cover_evidence_enter_the_unified_review_shape(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
database = root / "ledger.db"
connection = sqlite3.connect(database)
connection.executescript("""
CREATE TABLE asset(
id INTEGER PRIMARY KEY,medium TEXT,code TEXT,size INTEGER,
catalog_title TEXT,disposal TEXT
);
CREATE TABLE asset_tag(asset_id INTEGER,tag TEXT);
CREATE TABLE genre_decision(
source_genre TEXT PRIMARY KEY,raw_genre TEXT NOT NULL,
peach_tag TEXT,decided_at TEXT NOT NULL
);
INSERT INTO asset VALUES(1,'video','FC2-PPV-3701252',1500000000,NULL,NULL);
""")
connection.commit(); connection.close()
source = root / "fc2-candidate-log.csv"
rows = metadata_candidate_rows([{
"code": "FC2 3701252", "video_id": "3701252", "result": "取得",
"title": "作品标题", "tags": "素人 フェラ 口内発射",
"duration": "46:44", "cover_url": "https://example.invalid/cover.jpg",
}], database, raw_snapshot=source, fetched_at="2026-08-19T00:00:00Z")
self.assertEqual({row["field"] for row in rows}, {"title", "tags"})
title = next(row for row in rows if row["field"] == "title")
candidate = json.loads(title["candidates_json"])[0]
self.assertEqual(candidate["provider"], "fc2cmadb")
self.assertEqual(candidate["source"], "fc2")
self.assertTrue(candidate["official"])
self.assertEqual(candidate["catalog_evidence"]["cover_url"]["value"],
"https://example.invalid/cover.jpg")
tag_row = next(row for row in rows if row["field"] == "tags")
tags = json.loads(tag_row["candidates_json"])[0]
self.assertEqual(tags["value"], ["素人", "口交", "口爆"])
class HarvestTest(unittest.TestCase):
"""评论是用户多年攒的人工经验，本地对不上的标记同样要留。"""
def _props(self, comments):
return {"article": {"title": "合集", "tags": []},
"comments": {"data": comments}}
def test_keeps_markings_for_video_ids_we_do_not_own(self):
collected = {}
harvest("3312576", self._props([{"id": 1, "body": MULTI_PERFORMER_COMMENT}]),
collected)
rows = {row["video_id"]: row for row in harvest_rows(collected, owned={"3312576"})}
# 库里没有 2724256，但它的演员标记必须留下来。
self.assertEqual(rows["2724256"]["performers"], "未歩なな 皐月")
self.assertEqual(rows["2724256"]["owned"], "")
self.assertEqual(rows["2724256"]["seen_on"], "3312576")
def test_records_which_page_a_marking_came_from(self):
collected = {}
harvest("3312576", self._props([{"id": 1, "body": PERFORMER_COMMENT}]), collected)
harvest("2355314", self._props([{"id": 2, "body": "2664942 ひかり"}]), collected)
rows = {row["video_id"]: row for row in harvest_rows(collected, owned={"2355314"})}
self.assertEqual(rows["2664942"]["seen_on"], "2355314 3312576")
self.assertEqual(rows["2664942"]["performer_votes"], "ひかり:2")
def test_equivalence_members_are_kept_both_ways(self):
collected = {}
harvest("3312576", self._props([{"id": 1, "body": SPLIT_EQUIV_COMMENT}]), collected)
rows = {row["video_id"]: row for row in harvest_rows(collected, owned=set())}
self.assertIn("3312576", rows["3090722"]["equivalents"].split())
self.assertIn("4605413", rows["2471432"]["equivalents"].split())
class RealWorldShapesTest(unittest.TestCase):
"""线上抓到的真实写法，照着记忆重画的正则读不出它们。"""
def test_reads_a_full_width_equals_sign(self):
"""日文输入法打出的是 `＝`，只认半角等号会把这类断言整条丢掉。"""
groups = parse_equivalences("＝2407240", subject="4143616")
self.assertEqual(groups, [[("4143616", ""), ("2407240", "")]])
def test_an_equals_line_without_a_subject_is_dropped(self):
self.assertEqual(parse_equivalences("＝2407240"), [])
def test_reads_a_name_followed_by_work_links(self):
"""一行名字 + 若干作品链接，语义是同一个人的作品集。"""
body = CRLF.join(["剛毛マキちゃん",
"https://fc2ppvdb.com/articles/3701252",
"https://fc2ppvdb.com/articles/4078398"])
found = parse_performers(body)
self.assertEqual(found["3701252"], ["剛毛マキちゃん"])
self.assertEqual(found["4078398"], ["剛毛マキちゃん"])
def test_link_listing_without_a_single_name_is_not_guessed(self):
"""说明文字超过一行就说不准哪行是名字，宁可不认。"""
body = CRLF.join(["同一人？", "たぶん",
"https://fc2ppvdb.com/articles/3701252"])
self.assertEqual(parse_performers(body), {})
class BackfillTest(unittest.TestCase):
def test_takes_performers_marked_on_another_works_page(self):
"""`4176112` 页上那条链接列表同时认领了 `3701252`，候选行得吃到。"""
collected = {}
body = CRLF.join(["剛毛マキちゃん",
"https://fc2ppvdb.com/articles/3701252"])
harvest("4176112", {"comments": {"data": [{"id": 1, "body": body}]}}, collected)
rows = backfill([{"video_id": "3701252", "performers": "",
"performer_votes": "", "equivalents": ""}], collected)
self.assertEqual(rows[0]["performers"], "剛毛マキちゃん")
def test_keeps_equivalents_found_on_the_works_own_page(self):
collected = {}
harvest("4143616", {"comments": {"data": [{"id": 1, "body": "＝2407240"}]}},
collected)
rows = backfill([{"video_id": "4143616", "performers": "",
"performer_votes": "", "equivalents": "9999999"}], collected)
self.assertEqual(rows[0]["equivalents"], "2407240 9999999")
class PageCommentsTest(unittest.TestCase):
def test_deduplicates_the_two_overlapping_comment_lists(self):
"""`comments.data` 与 `article.comments` 重叠，相加会让一条评论投两票。"""
props = {"comments": {"data": [{"id": 7, "body": "2355314 真夏"}]},
"article": {"comments": [{"id": 7, "body": "2355314 真夏"},
{"id": 8, "body": "2355314 ひかり"}]}}
self.assertEqual(len(page_comments(props)), 2)
self.assertEqual(summarise("2355314", props)["performer_votes"],
"ひかり:1 真夏:1")
if __name__ == "__main__":
unittest.main()