tests/test_search_suggest.py
Rama master · réplica de lectura del repositorio original.
"""搜索栏补全：补出来的每一项都点得开，口径与 `/api/items` 同源。
这一域最容易出的不是报错，是两侧漂开：补全按一套判据取词、搜索按另一套判据找片，
于是下拉里给出的名字搜出零条，或者搜得到的标签补不出来。所以这里的断言分两层——
一层钉具体行为，一层遍历每个补全项拿它去真的搜一次。
"""
from __future__ import annotations
import sqlite3
import tempfile
import unittest
from pathlib import Path
from peach import web_contract as rm_web
from peach.catalog_rules import normalise_code_key
from peach.web_entity import SUGGEST_GROUPS, suggest_kinds
from support.ledger import fresh_ledger
STAMP = "2026-09-07T00:00:00Z"
class LedgerFixture(unittest.TestCase):
"""一份小账本：几部片挂着一位有别名的女优，另有厂牌、系列、标签和事务所。"""
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
root = Path(self.tmp.name).resolve()
self.db = fresh_ledger(root)
self.con = sqlite3.connect(self.db)
self.con.row_factory = sqlite3.Row
self.addCleanup(self.con.close)
self.con.executemany(
"INSERT INTO asset(id,location,path,name,medium,size,first_seen,code,"
"catalog_title,disposal) VALUES(?,'local',?,?,'video',100,'2026-01-01',?,?,?)",
[(1, r"R:\Media\one.mp4", "one.mp4", None, None, None),
(2, r"R:\Media\two.mp4", "two.mp4", None, None, None),
(3, r"R:\Media\three.mp4", "three.mp4", None, None, None),
(4, r"R:\Media\ABW-123.mp4", "ABW-123.mp4", "ABW-123", "潜入取材の一日", None),
(5, r"R:\Media\ABW-999.mp4", "ABW-999.mp4", "ABW-999", "已删除的一部", "trash"),
# 既无番号也无发行标题，只有一个相机文件名。
(6, r"R:\Media\IMG_2757_682.MOV", "IMG_2757_682.MOV", None, None, None),
# 文件名里有别人的番号，它自己没有番号。
(7, r"R:\Media\old-ABW-recording.mp4", "old-ABW-recording.mp4", None, None, None)],
)
self.con.executemany(
"INSERT INTO entity(id,kind,canonical_name,normalized_name,created_at,updated_at)"
" VALUES(?,?,?,?,'t','t')",
[(11, "performer", "涼森れむ", "涼森れむ"),
(12, "performer", "神雪", "神雪"),
(13, "studio", "ラグジュTV", "ラグジュtv"),
(14, "series", "ドキュメント", "ドキュメント"),
(15, "tag", "足交", "足交"),
# 与女优同名的标签是另一个身份的冒充，标签组不该收它。
(16, "tag", "神雪", "神雪"),
# 时长标签是筛选控件，不是可搜的词。
(17, "tag", "短片-2分内", "短片-2分内"),
(18, "agency", "Capsule Agency", "capsule agency"),
# 名下没有任何作品的人和事务所：账本里有身份，按名字搜是空的。
(19, "performer", "涼森みつ", "涼森みつ"),
(20, "agency", "Capsule Reserve", "capsule reserve"),
# 只出现在回收站那条上的标签。
(21, "tag", "足底", "足底"),
# 三个字的标签落在搜索的 FTS 分支上，两个字的落在 LIKE 分支上。
(22, "tag", "巨乳系", "巨乳系"),
# 同一个字头的一批标签，用来看每组的条数上限。
(23, "tag", "足元", "足元"), (24, "tag", "足指", "足指"),
(25, "tag", "足首", "足首"), (26, "tag", "足裏", "足裏"),
(27, "tag", "足音", "足音"),
# 「MO」是 MOODYZ 的词首、SO MODEL AGENT 里一个词的词首，
# 而在 kemonokai 中间。两个字母的拉丁输入只该命中前两个。
(28, "studio", "MOODYZ", "moodyz"),
(29, "creator", "kemonokai", "kemonokai"),
(30, "agency", "SO MODEL AGENT", "so model agent")],
)
self.con.executemany(
"INSERT INTO entity_alias(entity_id,alias,normalized_alias,source)"
" VALUES(?,?,?,'test')",
[(11, "凉森", "凉森"), (18, "カプセル", "カプセル")],
)
self.con.execute(
"INSERT INTO entity_search_term(entity_id,term,purpose,source,created_at)"
" VALUES(11,'Remu','discovery','test',?)", (STAMP,))
self.con.executemany(
"INSERT INTO asset_entity(asset_id,entity_id,role,source,confidence)"
" VALUES(?,?,?,'test',1.0)",
[(1, 11, "performer"), (2, 11, "performer"), (3, 11, "performer"),
(1, 12, "performer"), (1, 13, "studio"), (2, 13, "studio"),
(2, 14, "series"), (1, 15, "tag"), (2, 15, "tag"), (3, 16, "tag"),
(3, 17, "tag"), (5, 21, "tag"), (1, 22, "tag"), (2, 22, "tag"),
(1, 23, "tag"), (1, 24, "tag"), (1, 25, "tag"), (1, 26, "tag"),
(1, 27, "tag"), (3, 28, "studio"), (3, 29, "creator")],
)
self.con.executemany(
"INSERT INTO entity_membership(member_id,agency_id,source,checked_at)"
" VALUES(?,?,'test',?)",
[(11, 18, STAMP), (12, 18, STAMP), (19, 20, STAMP), (29, 30, STAMP)],
)
self.con.commit()
self.contract = rm_web.WebContract(self.db)
def suggest(self, query, limit=5):
return rm_web.q_suggest(self.contract, query, limit)
def group(self, query, kind, limit=5):
for group in self.suggest(query, limit)["groups"]:
if group["kind"] == kind:
return group["items"]
return []
def values(self, query, kind, limit=5):
return [item["value"] for item in self.group(query, kind, limit)]
def total(self, query):
return rm_web.q_items(self.contract, {"q": query, "limit": "10"})["total"]
class SuggestTests(LedgerFixture):
"""补全按输入给出馆藏里的身份与作品。"""
def test_a_canonical_name_completes_from_its_first_characters(self):
self.assertEqual(self.values("涼森", "performer"), ["涼森れむ"])
def test_the_suggestion_carries_how_many_works_are_behind_it(self):
self.assertEqual(self.group("涼森", "performer")[0]["n"], 3)
def test_an_alias_completes_to_the_canonical_name(self):
"""输入的写法是别名，补出来的必须是这个人在馆藏里的统称。"""
self.assertEqual(self.values("凉森", "performer"), ["涼森れむ"])
def test_a_hit_on_an_alias_says_which_writing_matched(self):
"""不说明凭什么，用户会以为补错了人。"""
self.assertEqual(self.group("凉森", "performer")[0]["matched"], "凉森")
def test_a_hit_on_the_canonical_name_says_nothing_extra(self):
self.assertEqual(self.group("涼森", "performer")[0]["matched"], "")
def test_a_search_term_reaches_the_same_person(self):
self.assertEqual(self.values("Remu", "performer"), ["涼森れむ"])
def test_a_person_with_no_works_never_shows_up(self):
"""`涼森みつ` 的名字也以「涼森」开头，但她名下一部片都没有。"""
self.assertNotIn("涼森みつ", self.values("涼森", "performer"))
def test_studios_and_series_have_their_own_groups(self):
self.assertEqual(self.values("ラグジュ", "studio"), ["ラグジュTV"])
self.assertEqual(self.values("ドキュ", "series"), ["ドキュメント"])
def test_an_agency_counts_the_works_of_its_members(self):
self.assertEqual(self.group("Capsule", "agency")[0]["n"], 3)
def test_an_agency_reachable_only_by_its_former_name_still_completes(self):
self.assertEqual(self.values("カプセル", "agency"), ["Capsule Agency"])
def test_an_agency_whose_roster_has_no_works_never_shows_up(self):
self.assertNotIn("Capsule Reserve", self.values("Capsule", "agency"))
def test_a_tag_completes_like_any_other_identity(self):
self.assertIn("足交", self.values("足交", "tag"))
def test_a_tag_that_is_really_a_performer_name_stays_out(self):
self.assertEqual(self.values("神雪", "tag"), [])
self.assertEqual(self.values("神雪", "performer"), ["神雪"])
def test_a_length_tag_is_a_filter_control_not_a_word(self):
self.assertEqual(self.values("短片", "tag"), [])
def test_a_tag_only_seen_in_the_trash_stays_out(self):
self.assertEqual(self.values("足底", "tag"), [])
def test_a_code_completes_to_a_work_that_can_be_opened(self):
items = self.group("ABW", "asset")
self.assertEqual(items[0]["value"], "ABW-123")
self.assertEqual(items[0]["id"], 4)
def test_a_work_in_the_trash_is_not_offered(self):
self.assertNotIn("ABW-999", self.values("ABW", "asset"))
def test_a_work_completes_by_its_release_title_too(self):
self.assertEqual(self.values("潜入", "asset"), ["ABW-123"])
def test_a_hit_in_the_file_name_ranks_last(self):
"""`old-ABW-recording.mp4` 里有别人的番号。它排在真的叫这个番号的那条之后。"""
self.assertEqual(self.values("ABW", "asset"), ["ABW-123", "old-ABW-recording"])
def test_local_extras_do_not_supply_work_suggestions_or_performer_counts(self):
self.con.executemany(
"INSERT INTO asset(id,location,path,name,medium,size,first_seen,code) "
"VALUES(?,'local',?,?,'video',100,'2026-01-01','ABW-777')",
[(90, r"R:\Media\ABW-777-clip2.mp4", "ABW-777-clip2.mp4"),
(91, r"R:\Media\ABW-777\Extras\Disc 1\making.mp4", "making.mp4")])
self.con.executemany(
"INSERT INTO asset_entity(asset_id,entity_id,role,source,confidence) "
"VALUES(?,11,'performer','test',1.0)", [(90,), (91,)])
self.con.commit()
self.assertEqual(self.values("ABW-777", "asset"), [])
self.assertEqual(self.group("涼森", "performer")[0]["n"], 3)
def test_a_file_name_loses_its_extension(self):
"""`.MOV` 是存储事实，不是这部片叫什么。"""
self.assertEqual(self.values("IMG", "asset"), ["IMG_2757_682"])
def test_an_empty_query_offers_nothing(self):
"""空输入下该出现的是搜索记录和推荐，不是补全。"""
self.assertEqual(self.suggest("")["groups"], [])
def test_groups_come_back_in_the_declared_order(self):
order = [kind for kind, _ in SUGGEST_GROUPS]
kinds = [group["kind"] for group in self.suggest("涼")["groups"]]
self.assertEqual(kinds, sorted(kinds, key=order.index))
def test_each_group_carries_its_own_label(self):
labels = {group["kind"]: group["label"] for group in self.suggest("涼森")["groups"]}
self.assertEqual(labels["performer"], "女优")
def test_a_group_never_exceeds_the_asked_for_size(self):
"""六个标签以「足」开头，一组只给得下五个。"""
self.assertEqual(len(self.group("足", "tag", limit=5)), 5)
self.assertEqual(len(self.group("足", "tag", limit=2)), 2)
def test_a_prefix_hit_outranks_a_hit_in_the_middle(self):
"""「足交」是前缀命中，其余几个是同一个字打头的同辈。"""
self.assertEqual(self.values("足交", "tag")[0], "足交")
def test_every_suggestion_finds_something_when_actually_searched(self):
"""补全与搜索同源的唯一验收方式：拿补出来的词去真的搜一次。
逐条搜是这条测试的成本，也是它的意义——两侧判据漂开时，别的断言都还绿着。
"""
for query in ("涼森", "凉森", "Remu", "ラグジュ", "ドキュ", "Capsule",
"カプセル", "足", "足交", "巨乳", "ABW", "潜入",
"MO", "mon", "IMG"):
for group in self.suggest(query)["groups"]:
for item in group["items"]:
with self.subTest(query=query, kind=group["kind"], value=item["value"]):
self.assertGreater(
self.total(item["value"]), 0,
f"补全给出 {item['value']}，搜它却是空的")
def test_a_suggested_work_opens_the_row_it_points_at(self):
for item in self.group("ABW", "asset"):
self.assertEqual(rm_web.q_item(self.contract, item["id"])["id"], item["id"])
class SuggestCardTests(LedgerFixture):
"""下拉栏画得出每一行：人的脸、事务所与近作，作品的小图与署名，每类的总数。"""
def setUp(self):
super().setUp()
root = Path(self.tmp.name)
covers = root / "covers"
covers.mkdir()
(covers / "ABW-123.jpg").write_bytes(b"\xff\xd8\xff\xd9")
snapshot = root / "two.jpg"
snapshot.write_bytes(b"\xff\xd8\xff\xd9")
# ABW-123 有官方封面，two.mp4 只有抽帧，one/three 两样都没有。
self.con.execute("INSERT INTO asset_entity(asset_id,entity_id,role,source,confidence)"
" VALUES(4,11,'performer','test',1.0)")
self.con.execute("UPDATE asset SET snapshot_path=? WHERE id=2", (str(snapshot),))
self.con.commit()
self.contract = rm_web.WebContract(self.db, cover_root=covers)
def test_a_person_row_names_her_agency(self):
self.assertEqual(self.group("涼森", "performer")[0]["agency"], "Capsule Agency")
def test_a_person_row_brings_only_works_that_have_a_picture(self):
works = self.group("涼森", "performer")[0]["works"]
self.assertEqual({work["id"] for work in works}, {2, 4})
cover = next(work for work in works if work["id"] == 4)
self.assertTrue(cover["has_cover"])
self.assertEqual(cover["code"], "ABW-123")
def test_a_face_that_cannot_be_drawn_is_not_offered(self):
"""没装实体图、代表作也没有抽帧：页面直接画首字母，不先出一张等 404 的图。"""
item = self.group("神雪", "performer")[0]
self.assertFalse(item["has_image"])
self.assertIsNone(item["rep"])
def test_a_suggested_work_carries_its_credit_and_picture(self):
item = self.group("ABW", "asset")[0]
self.assertEqual(item["who"], "涼森れむ")
self.assertTrue(item["card"]["has_cover"])
def test_a_suggested_work_without_a_picture_has_no_card(self):
self.assertIsNone(self.group("IMG", "asset")[0]["card"])
def test_each_group_says_how_many_it_matched_in_all(self):
"""六个标签以「足」开头，一组只列五个，总数仍是六。"""
tags = next(group for group in self.suggest("足")["groups"] if group["kind"] == "tag")
self.assertEqual((len(tags["items"]), tags["total"]), (5, 6))
def test_the_work_total_counts_every_hit(self):
works = next(group for group in self.suggest("ABW", limit=1)["groups"]
if group["kind"] == "asset")
self.assertEqual((len(works["items"]), works["total"]), (1, 2))
def test_asking_for_one_kind_returns_only_that_kind(self):
groups = rm_web.q_suggest(self.contract, "涼", 20, ("performer",))["groups"]
self.assertEqual([group["kind"] for group in groups], ["performer"])
def test_an_unknown_kind_is_dropped(self):
self.assertEqual(suggest_kinds("asset,bogus,performer"), ("performer", "asset"))
def test_an_agency_row_carries_its_site_mark_slot(self):
"""事务所没有标识文件，门面是官网圆标；这份夹具里没有官网，那一格就是空的。"""
self.assertIsNone(self.group("Capsule", "agency")[0]["mark"])
class AgencyGroupTests(LedgerFixture):
"""事务所这一组的排序与计数。
事务所自己不挂作品，名下作品数是跨成员的一次去重计数；排序先看整串前缀命中，
再看作品数，最后按名字。单个字母在真实账本上能命中十几家事务所，这一组的
先后和总数由这里钉住。
"""
def setUp(self):
super().setUp()
self.con.executemany(
"INSERT INTO entity(id,kind,canonical_name,normalized_name,created_at,updated_at)"
" VALUES(?,?,?,?,'t','t')",
[(40, "agency", "Alpha Talent", "alpha talent"), (41, "performer", "星空", "星空"),
(42, "agency", "Blue Agent", "blue agent"), (43, "performer", "月野", "月野"),
# 唯一的成员只在回收站那条上出现过：名下没有可见作品。
(44, "agency", "Amber Empty", "amber empty"), (45, "performer", "雨宮", "雨宮")],
)
self.link(4, 41)
self.link(2, 43)
self.link(5, 45)
self.con.executemany(
"INSERT INTO entity_membership(member_id,agency_id,source,checked_at)"
" VALUES(?,?,'test',?)",
[(41, 40, STAMP), (43, 42, STAMP), (45, 44, STAMP)],
)
self.con.commit()
def link(self, asset_id, entity_id, source="test"):
self.con.execute(
"INSERT INTO asset_entity(asset_id,entity_id,role,source,confidence)"
" VALUES(?,?,'performer',?,1.0)", (asset_id, entity_id, source))
self.con.commit()
def capsule_works(self):
return self.group("Capsule", "agency")[0]["n"]
def test_agencies_rank_by_prefix_hit_then_works_then_name(self):
"""单个字母比词首：`Alpha Talent` 整串以它开头，其余三家是某个词的词首。
前缀命中压过作品更多的 `Capsule Agency`；同为词首命中的按作品数，作品数
相同的按名字。
"""
self.assertEqual(
[(item["value"], item["n"]) for item in self.group("a", "agency")],
[("Alpha Talent", 1), ("Capsule Agency", 3), ("Blue Agent", 1),
("SO MODEL AGENT", 1)])
def test_the_total_counts_every_agency_with_works(self):
"""五家命中「a」。`Amber Empty` 名下没有可见作品，不进列表，也不进总数。"""
agencies = next(group for group in self.suggest("a", limit=1)["groups"]
if group["kind"] == "agency")
self.assertEqual((len(agencies["items"]), agencies["total"]), (1, 4))
def test_the_same_work_registered_by_another_source_counts_once(self):
self.link(1, 11, source="scrape")
self.assertEqual(self.capsule_works(), 3)
def test_a_member_work_in_the_trash_is_not_counted(self):
self.link(5, 12)
self.assertEqual(self.capsule_works(), 3)
def test_a_member_picture_is_not_counted(self):
self.con.execute(
"INSERT INTO asset(id,location,path,name,medium,size,first_seen)"
" VALUES(8,'local',?,'still.jpg','image',100,'2026-01-01')",
(r"R:\Media\still.jpg",))
self.link(8, 12)
self.assertEqual(self.capsule_works(), 3)
class ShortLatinQueryTests(LedgerFixture):
"""两个字母的拉丁输入比词首，不比子串。
子串比出来的多半是别的词中间那两个字母：真实账本上「MO」会捞出 `kemonokai`、
「Pr」会捞出 `chf3_prob4`，而用户在打的是 MOODYZ 和 Prestige。汉字和假名没有
分词空格，那一侧仍按子串，否则「凉森」这种两字输入连自己都补不出来。
"""
def test_a_two_letter_query_completes_a_name_that_starts_with_it(self):
self.assertEqual(self.values("MO", "studio"), ["MOODYZ"])
def test_a_two_letter_query_reaches_the_head_of_an_inner_word(self):
self.assertEqual(self.values("MO", "agency"), ["SO MODEL AGENT"])
def test_a_two_letter_query_skips_the_middle_of_a_word(self):
self.assertEqual(self.values("MO", "creator"), [])
def test_a_third_letter_brings_the_substring_match_back(self):
self.assertEqual(self.values("mon", "creator"), ["kemonokai"])
def test_an_extension_in_a_file_name_is_not_a_word_head(self):
"""`IMG_2757_682.MOV` 里的 `.MOV` 不该让这条片出现在「MO」的补全里。"""
self.assertEqual(self.values("MO", "asset"), [])
def test_two_han_characters_still_match_inside_a_name(self):
self.assertEqual(self.values("乳系", "tag"), ["巨乳系"])
def test_two_kana_still_match_inside_a_name(self):
self.assertEqual(self.values("ュメ", "series"), ["ドキュメント"])
class ShortQueryReachesEveryKindTests(LedgerFixture):
"""搜索的两条分支必须覆盖同样的 kind。
三字以上走 FTS，它索引的 `entities` 聚合作品名下的全部实体、不挑 kind；两字以内
走 LIKE。两边认的 kind 不一样时，同一个标签写三个字搜得到、写两个字搜不到，
用户看到的是搜索时灵时不灵，而补全会照样把那个词补出来。
"""
def test_a_two_character_tag_query_finds_its_works(self):
self.assertEqual(self.total("足交"), 2)
def test_a_three_character_tag_query_finds_its_works(self):
self.assertEqual(self.total("巨乳系"), 2)
def test_both_branches_agree_on_the_same_tag(self):
"""`巨乳系` 三个字走 FTS，`巨乳` 两个字走 LIKE，命中的是同一批作品。"""
self.assertEqual(self.total("巨乳"), self.total("巨乳系"))
def test_a_two_character_series_query_finds_its_works(self):
self.assertEqual(self.total("ドキ"), 1)
def test_a_two_character_performer_query_still_works(self):
self.assertEqual(self.total("凉森"), 3)
class OneWorkOneSlotTests(LedgerFixture):
"""同一部作品在下拉栏里只占一格。
真实账本上一部片常有几条来源挂着同一位女优（`r18` 与 `javinizer` 各记一条），同一个
番号也常存着几个文件。逐行取的话，一个人的四张近作里会有两张是同一张封面，视频组
也会并排摆出三张一样的图。
"""
def setUp(self):
super().setUp()
root = Path(self.tmp.name)
covers = root / "covers"
covers.mkdir()
for code in ("ABW-123", "ABW-124"):
(covers / f"{code}.jpg").write_bytes(b"\xff\xd8\xff\xd9")
frame = root / "ssis-001-b.jpg"
frame.write_bytes(b"\xff\xd8\xff\xd9")
self.con.executemany(
"INSERT INTO asset(id,location,path,name,medium,size,first_seen,code,"
"catalog_title,snapshot_path) VALUES(?,'local',?,?,'video',100,'2026-01-01',?,?,?)",
[# ABW-123 的另外两个文件，一个写法规范，一个是下划线小写。
(8, r"R:\Media\ABW-123-B.mp4", "ABW-123-B.mp4", "ABW-123", "潜入取材の一日", None),
(9, r"R:\Media\abw_123.mp4", "abw_123.mp4", "abw_123", None, None),
(10, r"R:\Media\ABW-124.mp4", "ABW-124.mp4", "ABW-124", "二日目", None),
# SSIS-001 没有封面，两个文件里只有后入库的那个抽过帧。
(11, r"R:\Media\SSIS-001.mp4", "SSIS-001.mp4", "SSIS-001", None, None),
(12, r"R:\Media\SSIS-001-B.mp4", "SSIS-001-B.mp4", "SSIS-001", None, str(frame))],
)
self.con.executemany(
"INSERT INTO asset_entity(asset_id,entity_id,role,source,confidence)"
" VALUES(?,11,'performer',?,1.0)",
[(4, "test"), (8, "test"), (9, "test"), (10, "test"), (10, "r18"),
(11, "test"), (12, "test")],
)
self.con.commit()
self.contract = rm_web.WebContract(self.db, cover_root=covers)
def work_keys(self):
works = self.group("涼森", "performer")[0]["works"]
return [normalise_code_key(work["code"]) for work in works]
def test_a_work_credited_by_two_sources_takes_one_slot(self):
self.assertEqual(self.work_keys().count("ABW-124"), 1)
def test_every_file_of_one_code_takes_one_slot(self):
"""ABW-123 存着三个文件，其中一个番号写成 `abw_123`，画出来都是同一张封面。"""
self.assertEqual(self.work_keys().count("ABW-123"), 1)
def test_a_work_shows_up_through_the_file_that_has_a_frame(self):
"""SSIS-001 没有封面，出面的必须是抽过帧的那个文件，否则整部作品掉出近作。"""
works = self.group("涼森", "performer")[0]["works"]
self.assertEqual([work["id"] for work in works], [12, 10, 4])
self.assertTrue(works[0]["has_thumb"])
def test_the_video_group_lists_each_work_once(self):
self.assertEqual(self.values("ABW", "asset"), ["ABW-124", "ABW-123", "old-ABW-recording"])
def test_the_video_group_opens_the_file_written_the_canonical_way(self):
item = next(item for item in self.group("ABW", "asset") if item["value"] == "ABW-123")
self.assertEqual(item["id"], 4)
def test_the_video_total_counts_works_not_files(self):
works = next(group for group in self.suggest("ABW", limit=1)["groups"]
if group["kind"] == "asset")
self.assertEqual((len(works["items"]), works["total"]), (1, 3))
if __name__ == "__main__":
unittest.main()