"""补女优别名后继：两站证据确定的艺名直接登记，不确定的不写，写下的能整批撤回（ADR-0055）。
全程临时账本；两站的取页层是真的（缓存、冷却、撞墙判定都走一遍），只把最底下的传输换成
按地址回页面的替身，不联网。页面是 minnano-av 699633 与 av_neme「雲母そら」「有本紗世」的
真实页面裁剪而成，只留解析会读到的那几块：夹具内联在这里而不是另开目录，新目录映射不到
任何测试域，会让每次改动都跑全量。
"""
import contextlib
import importlib.util
import io
import json
import tempfile
import time
import unittest
from pathlib import Path
from types import SimpleNamespace
from unittest import mock
from urllib.parse import quote
from peach import minnano_av
from peach import performer_alias_followup as alias
from peach.entities import normalize_entity_name
from peach.followups import Attempts, attempts_root
from peach.http import HttpResponse
from peach.repository import LedgerDatabase
from peach.review_csv import read_rows
from peach.scraping_access import paused_until
from peach.sources.base import Page
from peach.sources.seesaa import person_profile, renamed_to, search_results, split_names
from support.ledger import fresh_ledger
ROOT = Path(__file__).resolve().parents[1]
STAMP = "2026-09-24T00:00:00.000Z"
PROFILE_URL = f"{minnano_av.SITE}actress699633.html"
MINNANO_PROFILE = """
雲母そらきららそら / Kirara Sora
雲母そら （きららそら / Kirara Sora） 別名 未来ちゃん(FC2) （みらいちゃん / ）
別名 いちかちゃん(FC2) （いちかちゃん / ）
別名 雫つむぎ(FC2) （しずくつむぎ / SizukuTsumugi）
別名 神山ももか （かみやまももか / KamiyamaMomoka）
別名 朝霧いのり(着エロ) （あさぎりいのり / Asagiri Inori）
別名 神山ももか(天然むすめ) （かみやまももか / Kamiyama Momoka）
別名 雲母そら （きららそら / Kirara Sora）
別名 美雲そら【旧名】 （みくもそら / Mikumo Sora）
所属事務所 Bambi Promotion
"""
MINNANO_SEARCH = """
名前
（雲母そら）
（雲母そら）
"""
MINNANO_EMPTY = """
"""
def wiki_url(title: str) -> str:
return alias.AV_NEME_ROOT + "d/" + quote(title.encode("euc_jp"))
def wiki_search_url(key: str) -> str:
return alias.AV_NEME_ROOT + "search?keywords=" + quote(key.encode("euc_jp"))
def wiki_page(section: str, rest: str = "") -> str:
return ('')
def wiki_section(title: str, body: str, level: int = 1) -> str:
return (f'')
AV_NEME_KIRARA = wiki_page(
wiki_section("プロフィール", """
名前(女優名)：雲母そら（きららそら）
旧名義&別名：美雲そら（みくもそら）・朝霧いのり（あさぎりいのり）・雫つむぎ・神山ももか
生年月日：
"""),
# 作品小节里的名义是那一部片给她起的，不是她的艺名，不读。
wiki_section("smuc170| 素人ムクムク-夢中-",
"別名：みく 18歳 本屋の店員ナンバー
347 名前(別名)
有本紗世（ありもとさよ）／有本沙耶／有本紗也／元木小夜（もときさよ）
")))
def wiki_search(*hits) -> str:
"""检索结果页。每条是页名，或 (页名, 摘要)。"""
blocks = []
for hit in hits:
title, text = hit if isinstance(hit, tuple) else (hit, "…")
blocks.append(f'')
return f'{"".join(blocks)}
'
#: `叶芽ゆきな` 在 av_neme 的真实检索结果：五页厂牌页与月份页排在她的人物页 `桜美ゆきな` 前面。
LISTING_TEXT = "&size(18){''名前(女優名)''：[[ 叶芽ゆきな ]]} 仮名：[[ゆきな]]"
YUKINA_SEARCH = (
("ガチ素人", LISTING_TEXT), ("Girl’s Blue", LISTING_TEXT), ("2024年11月", LISTING_TEXT),
("ぐちょぐちょ素人娘", LISTING_TEXT), ("2020年2月", "完璧な男 叶芽ゆきな"),
("桜美ゆきな", "ゆきな（さくらみゆきな） 旧名義&別名：叶芽ゆきな（かなめゆきな）・夢原まみ・尾上さら"
" 生年月日：1999年1月8日"),
("叶芽ゆきな", "女優名が【叶芽ゆきな】から【桜美ゆきな】へ変更になりました。"),
)
AV_NEME_SAKURAMI = wiki_page(wiki_section("プロフィール", """
名前(女優名)：桜美ゆきな（さくらみゆきな）
旧名義&別名：叶芽ゆきな（かなめゆきな）・夢原まみ・尾上さら
生年月日：1999年1月8日
"""))
AV_NEME_LISTING = wiki_page(wiki_section("gsiro018| ガチ素人", "名前(女優名)：叶芽ゆきな\n名前(女優名)：初川みなみ（はつかわみなみ）\n生年月日：\n '
)).encode("euc_jp"))
summary = self.run_followup(minami)
self.assertEqual(summary["outcome"], "没有新写法")
self.assertTrue(summary["sites"][alias.AV_NEME].startswith("命中 " + wiki_url("初川みなみ")))
self.assertNotIn(wiki_search_url("初川みなみ"), self.av_neme.calls)
def test_her_profile_page_is_found_behind_five_archive_pages(self):
"""站上没有叫她名字的页；检索结果前五页是厂牌页与月份页，只读前三页就永远读不到她那一页。"""
yukina = self.entity("叶芽ゆきな")
self.av_neme.pages[wiki_search_url("叶芽ゆきな")] = (
200, wiki_search(*YUKINA_SEARCH[:6]).encode("euc_jp"))
self.av_neme.pages[wiki_url("桜美ゆきな")] = (200, AV_NEME_SAKURAMI.encode("euc_jp"))
for title, _text in YUKINA_SEARCH[:5]:
self.av_neme.pages[wiki_url(title)] = (200, AV_NEME_LISTING.encode("euc_jp"))
summary = self.run_followup(yukina)
self.assertEqual(summary["outcome"], "登记 3 个别名")
self.assertEqual(set(self.aliases(yukina)), {"桜美ゆきな", "夢原まみ", "尾上さら"})
self.assertIn(wiki_url("叶芽ゆきな"), self.av_neme.calls)
self.assertNotIn(wiki_url("2020年2月"), self.av_neme.calls)
def test_a_page_that_does_not_list_her_is_not_used(self):
stranger = self.entity("佐々木ゆうか")
self.minnano.pages[minnano_av.search_url("佐々木ゆうか")] = (
200, MINNANO_PROFILE.encode("utf-8"))
summary = self.run_followup(stranger)
self.assertEqual(summary["outcome"], "两站都没对上她")
self.assertEqual(self.aliases(stranger), {})
def test_a_known_minnano_id_is_read_without_searching(self):
momoka = self.entity("神山ももか")
with self.database.write_transaction(notify=False) as connection:
connection.execute(
"INSERT INTO entity_external_ref(entity_id,provider,external_kind,external_id)"
" VALUES(?,?,'performer','699633')", (momoka, alias.MINNANO))
self.run_followup(momoka)
self.assertNotIn(minnano_av.search_url("神山ももか"), self.minnano.calls)
self.assertIn("雲母そら", self.aliases(momoka))
def test_a_renamed_or_missing_entity_is_not_written(self):
momoka = self.entity("神山ももか")
with self.database.write_transaction(notify=False) as connection:
stale = alias.land(connection, momoka, "別人", alias.MINNANO, "page", ["雲母そら"], "b")
gone = alias.land(connection, 999, "神山ももか", alias.MINNANO, "page", ["雲母そら"], "b")
self.assertEqual([row["action"] for row in stale + gone], [alias.STALE, alias.GONE])
self.assertEqual(self.aliases(momoka), {})
class Fc2cmadbTests(Case):
NICKNAME = "たぬき顔サラサラ黒髪ロング"
def nicknamed(self, *listed: dict) -> int:
"""账本里只有卖家那句称呼的一位，挂着一部 FC2 作品。"""
her = self.entity(self.NICKNAME)
self.work(1, her, code="FC2-PPV-1234567")
if listed:
self.fc2cmadb.listed["1234567"] = list(listed)
return her
def test_the_site_name_behind_a_nickname_is_registered_and_searched_the_same_round(self):
her = self.nicknamed()
with self.database.read_connection() as connection:
self.assertTrue(alias.has_entry(connection, her))
summary = self.run_followup(her)
names = self.aliases(her)
batch = f"{alias.SOURCE}@7"
self.assertEqual(names["神山ももか"], batch)
self.assertEqual(names["雲母そら"], batch)
self.assertNotIn("ちっぱいリクルーター", names)
self.assertEqual(summary["sites"][alias.FC2CMADB],
"命中 " + alias.FC2CMADB_ACTRESS.format(id=9186))
self.assertTrue(summary["sites"][alias.MINNANO].startswith("命中 " + PROFILE_URL))
self.assertEqual([partial for _url, partial in self.fc2cmadb.calls], [False, True])
def test_a_cast_list_of_someone_else_writes_nothing(self):
her = self.nicknamed({"id": 77, "name": "佐々木ゆうか", "alias_name": "別の人"})
summary = self.run_followup(her)
self.assertEqual(self.aliases(her), {})
self.assertEqual(summary["outcome"], "三站都没对上她")
self.assertIn("FC2-PPV-1234567 的女优栏是 佐々木ゆうか", summary["sites"][alias.FC2CMADB])
def test_two_people_on_the_cast_list_claiming_her_is_ambiguous(self):
her = self.nicknamed({"id": 1, "name": "神山ももか", "alias_name": self.NICKNAME},
{"id": 2, "name": "雲母そら", "alias_name": self.NICKNAME})
summary = self.run_followup(her)
self.assertEqual(self.aliases(her), {})
self.assertIn("不止一位", summary["sites"][alias.FC2CMADB])
def test_a_site_name_another_entity_uses_is_left_for_review(self):
other = self.entity("神山ももか")
her = self.nicknamed()
self.run_followup(her)
self.assertNotIn("神山ももか", self.aliases(her))
taken = [row for row in self.review() if row["site"] == alias.FC2CMADB]
self.assertEqual([row["action"] for row in taken], [alias.TAKEN])
self.assertIn(f"实体 {other}", taken[0]["detail"])
def test_a_refusal_pauses_the_site(self):
her = self.nicknamed()
self.fc2cmadb.status = 403
summary = self.run_followup(her)
self.assertEqual(summary["outcome"], "未取得")
self.assertTrue(paused_until(self.cooldown, alias.FC2CMADB))
asked = len(self.fc2cmadb.calls)
self.run_followup(her, run_id=8)
self.assertEqual(len(self.fc2cmadb.calls), asked)
def test_a_performer_without_fc2_works_is_not_asked(self):
momoka = self.entity("神山ももか")
self.work(1, momoka, code="ABP-968")
summary = self.run_followup(momoka)
self.assertEqual(self.fc2cmadb.calls, [])
self.assertNotIn(alias.FC2CMADB, summary["sites"])
class RepeatTests(Case):
def test_a_second_run_writes_nothing_and_asks_nobody(self):
momoka = self.entity("神山ももか")
self.run_followup(momoka, run_id=7)
first = dict(self.aliases(momoka))
requests = len(self.minnano.calls) + len(self.av_neme.calls)
summary = self.run_followup(momoka, run_id=8)
self.assertEqual(summary["outcome"], "没有新写法")
self.assertEqual(self.aliases(momoka), first)
self.assertEqual(len(self.minnano.calls) + len(self.av_neme.calls), requests)
def test_stock_does_not_dispatch_her_again_until_her_names_change(self):
momoka = self.entity("神山ももか")
self.work(1, momoka)
attempts = Attempts(attempts_root(self.generated))
def planned():
with self.database.read_connection() as connection:
return [item.key for item in alias.stock(connection, attempts, limit=5)]
self.assertEqual(planned(), [alias.followup_key(momoka)])
self.run_followup(momoka)
self.assertEqual(planned(), [])
with self.database.write_transaction(notify=False) as connection:
connection.execute("INSERT INTO entity_alias(entity_id,alias,normalized_alias,source)"
" VALUES(?,'新しい名前','新しい名前','manual')", (momoka,))
self.assertEqual(planned(), [alias.followup_key(momoka)])
def test_stock_reads_every_performer_in_a_fixed_number_of_queries_and_judges_like_one_by_one(self):
"""存量规划每分钟跑一次：查询条数不随女优人数涨，判据与逐人查的 `has_entry`、`fingerprint` 一致。"""
momoka = self.entity("神山ももか", "雲母そら")
tried = self.entity("伊吹彩", "月島舞香")
fc2 = self.entity("Ab")
unreachable = self.entity("Cd")
self.work(1, momoka)
self.work(2, tried, code="ABP-968")
self.work(3, fc2, code="FC2-PPV-1234567")
self.work(4, unreachable)
with self.database.write_transaction(notify=False) as connection:
connection.execute("INSERT INTO entity_alias(entity_id,alias,normalized_alias,source)"
" VALUES(?,'深田えいみ','深田えいみ',?)", (tried, f"{alias.SOURCE}@3"))
connection.execute("INSERT INTO entity_external_ref(entity_id,provider,external_kind,external_id)"
" VALUES(?,?,'performer','387589')", (tried, alias.MINNANO))
attempts = Attempts(attempts_root(self.generated))
def planned(connection, limit=10):
statements = []
connection.set_trace_callback(statements.append)
try:
keys = [item.key for item in alias.stock(connection, attempts, limit=limit)]
finally:
connection.set_trace_callback(None)
return sorted(keys), len(statements)
with self.database.read_connection() as connection:
attempts.record(alias.followup_key(tried), alias.fingerprint(connection, tried), "写入")
expected = sorted(alias.followup_key(entity_id) for entity_id in (momoka, tried, fc2, unreachable)
if alias.has_entry(connection, entity_id)
and not attempts.settled(alias.followup_key(entity_id),
alias.fingerprint(connection, entity_id)))
keys, queries = planned(connection)
self.assertEqual(keys, expected)
self.assertEqual(keys, sorted([alias.followup_key(momoka), alias.followup_key(fc2)]))
for index in range(20):
self.work(100 + index, self.entity(f"追加女優{index:02d}", f"別名{index:02d}"))
with self.database.read_connection() as connection:
keys, more = planned(connection, limit=100)
self.assertEqual(len(keys), 22)
self.assertEqual(more, queries, "多出来的女优不该多出查询")
def test_stock_puts_performers_without_an_avatar_first(self):
"""作品多的那位已经有头像，缺头像的那位作品少也排在前面：别名补进来，
补头像后继才多一个名字去图库里找。
"""
busy = self.entity("神山ももか")
quiet = self.entity("伊吹彩")
self.work(1, busy)
self.work(2, busy)
self.work(3, quiet)
avatars = self.generated / "avatars"
avatars.mkdir(parents=True, exist_ok=True)
(avatars / f"performer-{busy}.img").write_bytes(b"installed")
attempts = Attempts(attempts_root(self.generated))
with self.database.read_connection() as connection:
ordered = [item.key for item in alias.stock(connection, attempts, limit=5,
avatar_root=avatars)]
plain = [item.key for item in alias.stock(connection, attempts, limit=5)]
self.assertEqual(ordered, [alias.followup_key(quiet), alias.followup_key(busy)])
self.assertEqual(plain, [alias.followup_key(busy), alias.followup_key(quiet)])
def test_a_new_entry_rule_dispatches_everyone_tried_again(self):
"""入口判据换了版本，名字链没变的也再问一次：新入口能到的页，旧入口没读到。"""
momoka = self.entity("神山ももか")
self.work(1, momoka)
attempts = Attempts(attempts_root(self.generated))
def planned():
with self.database.read_connection() as connection:
return [item.key for item in alias.stock(connection, attempts, limit=5)]
self.run_followup(momoka)
self.assertEqual(planned(), [])
with mock.patch.object(alias, "ENTRY_RULE", alias.ENTRY_RULE + 1):
self.assertEqual(planned(), [alias.followup_key(momoka)])
def test_a_rate_limited_site_is_paused_and_not_asked_again(self):
momoka = self.entity("神山ももか")
self.minnano.pages[minnano_av.search_url("神山ももか")] = (429, b"")
self.av_neme.pages[wiki_search_url("神山ももか")] = (429, b"")
summary = self.run_followup(momoka)
self.assertEqual(summary["outcome"], "未取得")
self.assertIn("429", summary["sites"][alias.MINNANO])
self.assertIn("429", summary["sites"][alias.AV_NEME])
self.assertTrue(paused_until(self.cooldown, alias.MINNANO))
self.assertTrue(paused_until(self.cooldown, alias.AV_NEME))
requests = len(self.minnano.calls) + len(self.av_neme.calls)
self.assertEqual(self.run_followup(momoka, run_id=8)["outcome"], "未取得")
self.assertEqual(len(self.minnano.calls) + len(self.av_neme.calls), requests)
self.assertEqual(self.aliases(momoka), {})
def test_an_unfetched_conclusion_is_dispatched_again_after_a_day(self):
"""站在冷却时的「未取得」说的是那一天，不是她；名字链没变也要再试，其余结论照旧不再派。"""
momoka = self.entity("神山ももか")
self.work(1, momoka)
settled = self.entity("有本紗世")
self.work(2, settled)
self.assertEqual(self.run_followup(settled)["outcome"], "两站都没对上她")
self.minnano.pages[minnano_av.search_url("神山ももか")] = (429, b"")
self.av_neme.pages[wiki_url("神山ももか")] = (429, b"")
self.assertEqual(self.run_followup(momoka)["outcome"], "未取得")
now = time.time()
def planned(clock):
attempts = Attempts(attempts_root(self.generated), clock=lambda: clock)
with self.database.read_connection() as connection:
return [item.key for item in alias.stock(connection, attempts, limit=5)]
self.assertEqual(planned(now), [])
self.assertEqual(planned(now + alias.RETRY_UNFETCHED + 1), [alias.followup_key(momoka)])
self.assertEqual(planned(now + 365 * 86400), [alias.followup_key(momoka)])
def test_a_challenge_page_is_not_cached_as_her_profile(self):
momoka = self.entity("神山ももか")
self.minnano.pages[minnano_av.search_url("神山ももか")] = (
200, b"Just a moment... ")
summary = self.run_followup(momoka)
self.assertIn("未取得", summary["sites"][alias.MINNANO])
self.assertFalse(list((self.generated / "provider-cache" / "minnano").glob("*.json")))
def test_the_cloudflare_beacon_on_a_normal_page_is_not_a_challenge(self):
"""Cloudflare 往正常页末尾插一段 `challenge-platform` 检测脚本；页照常取，来源不进冷却。"""
beacon = (b"")
for url, (status, body) in list(self.minnano.pages.items()):
self.minnano.pages[url] = (status, body + beacon)
momoka = self.entity("神山ももか")
summary = self.run_followup(momoka)
self.assertTrue(summary["sites"][alias.MINNANO].startswith("命中 " + PROFILE_URL))
self.assertFalse(list(self.cooldown.glob("scraping-minnano-av.cooldown.json")))
def test_a_refused_page_is_fetched_in_the_local_browser_and_the_site_stays_there(self):
"""默认传输是 `SourceTransport`：HTTP 客户端被拦时同一页由本机浏览器取，取到就缓存、不冷却，
这一站之后固定走浏览器；补女优资料后继与链接、名册脚本走的是同一个入口。"""
from peach import browser_transport
from peach.scraping_access import fixed_to_browser
class Browser(browser_transport.BrowserTransport):
def __init__(self):
self.calls = []
def __call__(self, request, timeout, max_bytes):
self.calls.append(request.url)
return HttpResponse(200, {}, b"profile ", PROFILE_URL)
url = minnano_av.search_url("神山ももか")
browser = Browser()
pages = alias.MinnanoPages(self.generated / "provider-cache" / "minnano", self.cooldown, limiter=NoWait())
pages.transport.transports[alias.MINNANO] = lambda request, *_args: HttpResponse(403, {}, b"", request.url)
with mock.patch("peach.browser_transport.shared", return_value=browser):
self.assertEqual(pages.get(url), (PROFILE_URL, "profile "))
pages.close()
self.assertEqual(browser.calls, [url])
self.assertTrue(fixed_to_browser(self.cooldown, alias.MINNANO))
self.assertFalse(paused_until(self.cooldown, alias.MINNANO))
def load_revert():
spec = importlib.util.spec_from_file_location(
"revert_auto_landing_under_test", ROOT / "scripts" / "revert_auto_landing.py")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
class RevertTests(Case):
def test_a_batch_is_reverted_and_not_landed_again(self):
momoka = self.entity("神山ももか", "かみやまももか本人")
self.work(1, momoka)
self.run_followup(momoka, run_id=7)
self.assertEqual(len(self.aliases(momoka)), 5)
revert = load_revert()
base = ["--db", str(self.db), "--logo-root", str(self.root / "logos"),
"--source", alias.SOURCE]
with contextlib.redirect_stdout(io.StringIO()) as printed:
self.assertEqual(revert.main([*base, "--batch", f"{alias.SOURCE}@8"]), 0)
self.assertIn("'别名': 0", printed.getvalue())
with contextlib.redirect_stdout(io.StringIO()) as printed:
self.assertEqual(revert.main(base), 0)
self.assertIn("'别名': 4", printed.getvalue())
self.assertEqual(len(self.aliases(momoka)), 5)
with contextlib.redirect_stdout(io.StringIO()):
self.assertEqual(revert.main(
[*base, "--apply", "--backup", str(self.root / "backup.db")]), 0)
self.assertEqual(self.aliases(momoka), {"かみやまももか本人": "manual"})
# 撤回之后她的名字链与撤回前、这一批写入前一样，存量补派不会原样再落一遍。
with self.database.read_connection() as connection:
self.assertEqual(alias.stock(connection, Attempts(attempts_root(self.generated)),
limit=5), [])
class AllocationTests(Case):
def test_stock_aliases_get_their_share_of_the_round(self):
from peach import library_processing, task_runs
people = [self.entity(f"白石まり{index:02d}") for index in range(10)]
for index, entity_id in enumerate(people, start=1):
self.work(index, entity_id)
config = SimpleNamespace(directory=lambda _name: self.generated)
with mock.patch.object(task_runs, "MAX_FOLLOWUPS", 8), \
mock.patch.object(alias, "STOCK_SHARE", 3):
found = library_processing._entity_followups(self.database, config, max(people))
kinds = [item["task_key"] for item in found]
self.assertEqual(len(kinds), 8)
self.assertEqual(kinds.count(alias.TASK_KEY), 3)
def test_new_performers_get_an_alias_followup_ahead_of_stock(self):
from peach import library_processing
old = self.entity("古川ゆうな")
self.work(1, old)
new = self.entity("新田あおい")
config = SimpleNamespace(directory=lambda _name: self.generated)
found = library_processing._entity_followups(self.database, config, old)
keys = [item["key"] for item in found if item["task_key"] == alias.TASK_KEY]
self.assertEqual(keys, [alias.followup_key(new), alias.followup_key(old)])
if __name__ == "__main__":
unittest.main()
コメント欄の誰か