tests/test_resource_identification.py
Rama master · réplica de lectura del repositorio original.
"""无番号视频识别清单与候选合并的隔离回归。"""
import json
import sqlite3
import tempfile
import unittest
from pathlib import Path
from peach.resource_identification import (
CANDIDATE_FILENAME,
build_worklist,
ingest_results,
performer_guess,
query_variants,
)
from peach.review_csv import read_rows, write_rows
from support.ledger import fresh_ledger
class QueryVariantTests(unittest.TestCase):
def test_chinese_performer_prefix_stays_verbatim_and_guessable(self):
name = "梓怡-背著老公和合租室友的狂歡 長相乖巧，被大肉棒瘋狂抽插.mp4"
variants = query_variants(name)
self.assertEqual(variants[0], "梓怡-背著老公和合租室友的狂歡 長相乖巧，被大肉棒瘋狂抽插")
self.assertIn("梓怡 背著老公和合租室友的狂歡 長相乖巧，被大肉棒瘋狂抽插", variants)
self.assertEqual(performer_guess(name), "梓怡")
def test_unspaced_english_filename_gains_word_bounds(self):
variants = query_variants("YouGetFootjobStandingFromStarttoFinish9.mp4")
self.assertIn("You Get Footjob Standing From Startto Finish 9", variants)
self.assertEqual(performer_guess("YouGetFootjobStandingFromStarttoFinish9.mp4"), "")
def test_watermarks_and_quality_tails_are_stripped_in_variants(self):
variants = query_variants("www.98t.la@MIAD-573-uncensored-HD.mp4")
self.assertEqual(variants[0], "www.98t.la@MIAD-573-uncensored-HD")
self.assertIn("MIAD-573-uncensored-HD", variants)
self.assertIn("MIAD 573", variants)
def test_jav_code_stems_are_not_performer_guesses(self):
self.assertEqual(performer_guess("259LUXU-1055.mp4"), "")
self.assertEqual(performer_guess("ABW-153.mp4"), "")
class ResourceWorklistTests(unittest.TestCase):
def setUp(self):
temporary = tempfile.TemporaryDirectory()
self.addCleanup(temporary.cleanup)
self.root = Path(temporary.name).resolve()
self.db_path = fresh_ledger(self.root)
def add(self, asset_id, location, path, medium, size=1000, duration=None,
code=None, disposal=None):
connection = sqlite3.connect(self.db_path)
try:
connection.execute(
"INSERT INTO asset(id,location,path,name,medium,size,duration,code,disposal) "
"VALUES(?,?,?,?,?,?,?,?,?)",
(asset_id, location, path, path.rsplit("\\", 1)[-1], medium,
size, duration, code, disposal),
)
connection.commit()
finally:
connection.close()
def test_worklist_pairs_a_cover_and_skips_code_assets(self):
base = r"B:\xxr\0208 (23)"
self.add(1, "115", rf"{base}\梓怡-背著老公.mp4", "video",
985 * 1024**2, 1647)
self.add(2, "115", rf"{base}\梓怡-背著老公.png", "image", 2 * 1024**2)
self.add(3, "115", rf"{base}\ABW-153.mp4", "video",
500 * 1024**2, 3600, code="ABW-153")
connection = sqlite3.connect(self.db_path)
connection.row_factory = sqlite3.Row
try:
rows = build_worklist(connection, locations=("115",), prefix=base)
finally:
connection.close()
self.assertEqual([row["asset_id"] for row in rows], [1])
self.assertEqual(rows[0]["cover_path"], rf"{base}\梓怡-背著老公.png")
self.assertEqual(rows[0]["performer_guess"], "梓怡")
self.assertIn("梓怡", rows[0]["query_variants"])
def test_paired_only_keeps_videos_with_a_sibling_image(self):
self.add(1, "115", r"B:\xxr\a\有图.mp4", "video", 100 * 1024**2, 600)
self.add(2, "115", r"B:\xxr\a\有图.jpg", "image", 1024)
self.add(3, "115", r"B:\xxr\b\无图.mp4", "video", 100 * 1024**2, 600)
connection = sqlite3.connect(self.db_path)
connection.row_factory = sqlite3.Row
try:
rows = build_worklist(connection, locations=("115",), prefix=r"B:\xxr",
paired_only=True)
finally:
connection.close()
self.assertEqual([row["asset_id"] for row in rows], [1])
def test_sparse_only_keeps_assets_without_any_metadata(self):
self.add(1, "115", r"B:\xxr\a\已有标题.mp4", "video", 100 * 1024**2, 600)
self.add(2, "115", r"B:\xxr\b\没有元信息.mp4", "video", 100 * 1024**2, 600)
connection = sqlite3.connect(self.db_path)
connection.row_factory = sqlite3.Row
try:
with connection:
connection.execute(
"UPDATE asset SET catalog_title='已有标题' WHERE id=1")
rows = build_worklist(connection, locations=("115",), prefix=r"B:\xxr",
sparse_only=True)
all_rows = build_worklist(connection, locations=("115",), prefix=r"B:\xxr")
finally:
connection.close()
self.assertEqual([row["asset_id"] for row in rows], [2])
self.assertEqual(rows[0]["metadata_hits"], 0)
self.assertEqual([row["asset_id"] for row in all_rows], [2, 1])
self.assertEqual(all_rows[1]["metadata_hits"], 1)
class ResourceIngestTests(unittest.TestCase):
def setUp(self):
temporary = tempfile.TemporaryDirectory()
self.addCleanup(temporary.cleanup)
self.root = Path(temporary.name).resolve()
self.db_path = fresh_ledger(self.root)
self.candidates = self.root / "generated" / CANDIDATE_FILENAME
self.candidates.parent.mkdir(parents=True)
connection = sqlite3.connect(self.db_path)
try:
connection.execute(
"INSERT INTO asset(id,location,path,name,medium,size,duration) "
"VALUES(1,'115',?,?,'video',?,1647)",
(r"B:\xxr\0208 (23)\梓怡-背著老公.mp4", "梓怡-背著老公.mp4", 985 * 1024**2),
)
connection.execute(
"INSERT INTO asset(id,location,path,name,medium,size,duration,code) "
"VALUES(2,'115',?,?,'video',?,3600,'ABW-153')",
(r"B:\xxr\0208 (23)\ABW-153.mp4", "ABW-153.mp4", 500 * 1024**2),
)
connection.commit()
finally:
connection.close()
def write_existing_local_candidate(self):
write_rows(self.candidates, (
"item_key", "code", "query", "asset_id", "asset_path", "field", "field_label",
"current_value", "candidates_json", "source_count", "source_profile",
"policy_version", "status", "size_gb", "videos", "fetched_at",
), [{
"item_key": "asset:1:title", "code": "", "query": "梓怡-背著老公.mp4",
"asset_id": 1, "asset_path": r"B:\xxr\0208 (23)\梓怡-背著老公.mp4",
"field": "title", "field_label": "标题", "current_value": "",
"candidates_json": json.dumps([{
"candidate_key": "local_nfo:abc", "source": "local_nfo",
"value": "旧标题", "display_value": "旧标题", "confidence": 0.9,
}], ensure_ascii=False),
"source_count": 1, "source_profile": "library", "policy_version": "library-v1",
"status": "candidate", "size_gb": 0.96, "videos": 1, "fetched_at": "",
}], atomic=True)
RESULTS = [
{"asset_id": "1", "field": "title",
"value": "背着老公和合租室友的狂欢", "source_url": "https://av911.tv/video/113896",
"confidence": "0.7", "note": "麻豆传媒作品页"},
{"asset_id": "1", "field": "performers", "value": "梓怡",
"source_url": "https://av911.tv/video/113896", "confidence": "0.7", "note": ""},
{"asset_id": "1", "field": "studio", "value": "麻豆传媒",
"source_url": "https://av911.tv/video/113896", "confidence": "0.6", "note": ""},
{"asset_id": "2", "field": "title", "value": "不会写入",
"source_url": "", "confidence": "0.5", "note": ""},
]
def test_results_merge_with_existing_sources_and_stay_idempotent(self):
self.write_existing_local_candidate()
connection = sqlite3.connect(self.db_path)
connection.row_factory = sqlite3.Row
try:
stats = ingest_results(connection, self.candidates, self.RESULTS)
finally:
connection.close()
self.assertEqual(stats["applied"], 3)
self.assertEqual(stats["skipped"], 1)
groups = {row["item_key"]: row for row in read_rows(self.candidates)}
title_choices = json.loads(groups["asset:1:title"]["candidates_json"])
self.assertEqual({choice["source"] for choice in title_choices},
{"local_nfo", "websearch"})
performer = json.loads(groups["asset:1:performers"]["candidates_json"])[0]
self.assertEqual(performer["value"], [{"name": "梓怡"}])
self.assertEqual(performer["display_value"], "梓怡")
self.assertEqual(groups["asset:1:performers"]["asset_path"],
r"B:\xxr\0208 (23)\梓怡-背著老公.mp4")
connection = sqlite3.connect(self.db_path)
connection.row_factory = sqlite3.Row
try:
ingest_results(connection, self.candidates, self.RESULTS)
finally:
connection.close()
groups = {row["item_key"]: row for row in read_rows(self.candidates)}
self.assertEqual(len(json.loads(groups["asset:1:title"]["candidates_json"])), 2)
def test_invalid_values_are_skipped(self):
connection = sqlite3.connect(self.db_path)
connection.row_factory = sqlite3.Row
try:
stats = ingest_results(connection, self.candidates, [
{"asset_id": "1", "field": "release_date", "value": "2026/01/01"},
{"asset_id": "1", "field": "tags", "value": "有码"},
{"asset_id": "1", "field": "title", "value": " "},
{"asset_id": "404", "field": "title", "value": "不存在"},
])
finally:
connection.close()
self.assertEqual(stats["applied"], 0)
self.assertEqual(stats["skipped"], 4)
if __name__ == "__main__":
unittest.main()