tests/test_metadata_dmm.py
Rama master · réplica de lectura del repositorio original.
"""DMM 的 GraphQL 目录套进站点解析器契约：cid 推导与搜索匹配、发售日换算、payload 逐键钉住，以及失败分档。"""
import json
import re
import unittest
from unittest.mock import Mock, patch
from peach.http import HttpResponse
from peach.jav_cover_fetch import Unavailable
from peach.library_processing import LibraryMetadataProvider
from peach.scraping_access import SourcePaused
from peach.sources import DMM, DmmSource, FailureReason, Page, Session, SourceFailure
from peach.sources.dmm import (ACTRESS_WORKS_QUERY, DETAIL_ATTEMPTS, DETAIL_QUERY, SEARCH_LIMIT, SEARCH_QUERY, code_parts,
code_from_cid, guessed_cid, matching_cids, release_date)
ENDPOINT = "https://api.video.dmm.co.jp/graphql"
PAGE_URL = "https://video.dmm.co.jp/av/content/?id=ssis00057"
COVER = "https://awsimgsrc.dmm.co.jp/pics_dig/digital/video/ssis00057/ssis00057pl.jpg"
SAMPLE = "https://awsimgsrc.dmm.co.jp/pics_dig/digital/video/ssis00057/ssis00057-1.jpg"
#: 2026-09-24 实测 `ppvContent(id: "ssis00057")` 的响应，简介截短。
ITEM = {"id": "ssis00057", "title": "パンティーストッキングマニア 三宮つばき", "description": "新人OLの‘つばき’は才色兼備で",
"makerContentId": "SSIS-057", "makerReleasedAt": "2021-05-06T15:00:00Z", "duration": 9069,
"maker": {"id": "3152", "name": "エスワン ナンバーワンスタイル"}, "label": {"id": "3474", "name": "S1 NO.1 STYLE"},
"series": None, "actresses": [{"id": "1062909", "name": "三宮つばき"}], "directors": [{"id": "110733", "name": "肉尊"}],
"genres": [{"id": "6017", "name": "ギリモザ"}, {"id": "4025", "name": "単体作品"}],
"packageImage": {"largeUrl": COVER, "mediumUrl": COVER.replace("pl.jpg", "ps.jpg")},
"sampleImages": [{"imageUrl": SAMPLE}]}
FKOS = {"id": "h_1721fkos00006", "title": "個人撮影6", "makerContentId": "FKOS-006", "makerReleasedAt": "2023-01-19T15:00:00Z",
"duration": 3600, "maker": {"id": "1", "name": "FKOS"}, "actresses": [], "genres": [], "packageImage": None}
#: `LibraryMetadataProvider.query('SSIS-057', 'dmm')` 对这份夹具交出的 dict，逐键钉住。
EXPECTED = {
"id": "SSIS-057", "content_id": "ssis00057", "source_url": PAGE_URL, "title": "パンティーストッキングマニア 三宮つばき",
"maker": "エスワン ナンバーワンスタイル", "label": "S1 NO.1 STYLE", "series": "", "director": "肉尊",
"release_date": "2021-05-07", "runtime": 151.15, "cover_urls": [COVER], "cover_url": COVER,
"actresses": [{"japanese_name": "三宮つばき", "dmm_id": "1062909"}], "genres": ["ギリモザ", "単体作品"],
"description": "新人OLの‘つばき’は才色兼備で", "sample_images": [SAMPLE], "raw": ITEM}
def graphql(contents=None, items=None):
"""假接口：按请求体里的查询与变量答复。`contents` 是搜索结果的 cid 列，`items` 是 cid → 商品或异常。"""
calls = []
items = items or {}
def call(request, timeout, limit):
payload = json.loads(request.body)
variables = payload["variables"]
calls.append((request.method, request.url, request.headers.get("Referer"), request.headers.get("Content-Type"),
limit, payload["query"], variables))
if payload["query"] == SEARCH_QUERY:
body = {"data": {"legacySearchPPV": {"result": {"contents": [{"id": cid} for cid in contents or []]}}}}
else:
item = items.get(variables["id"])
if isinstance(item, Exception):
raise item
body = {"data": {"ppvContent": item}}
return HttpResponse(200, {}, json.dumps(body, ensure_ascii=False).encode(), request.url)
call.calls = calls
return call
def asked(transport):
"""这趟发了哪几条查询：`('detail', cid)` 或 `('search', 关键词)`。"""
return [("search", variables["queryWord"]) if query == SEARCH_QUERY else ("detail", variables["id"])
for _, _, _, _, _, query, variables in transport.calls]
class CodeMappingTests(unittest.TestCase):
def test_dmm_product_identity_keeps_the_release_suffix_and_uses_the_dmm_namespace(self):
for product, expected in {"1dandy00899": "DANDY-899", "118abw00220": "ABW-220",
"h_086iqqq00026": "IQQQ-026", "1dandy00851c": "DANDY-851C",
"48midv00012ai": "MIDV-012AI"}.items():
self.assertEqual(code_from_cid(product), expected)
self.assertEqual(code_from_cid("opaque_cid"), "")
def test_sod_bonus_edition_maps_to_the_regular_release(self):
# 特典版（官方番号 SDJS-304-V）是通常版正片加特典映像，番号归到通常版。
for product, expected in {"1sdjs00304v": "SDJS-304", "1start00644v": "START-644"}.items():
self.assertEqual(code_from_cid(product), expected)
def test_code_parts_keep_the_numeric_prefix_and_the_digits_as_written(self):
self.assertEqual(code_parts("SSIS-057"), ("", "SSIS", "057"))
self.assertEqual(code_parts("ssis057"), ("", "SSIS", "057"))
self.assertEqual(code_parts("300MIUM-1239"), ("300", "MIUM", "1239"))
self.assertEqual(code_parts("MIDV-012ai"), ("", "MIDV", "012AI"))
for code in ("", None, "FC2-PPV-1812235", "092415_001", "n0780"):
with self.subTest(code=code):
self.assertIsNone(code_parts(code))
def test_the_guess_is_lowercase_letters_and_five_digits(self):
self.assertEqual(guessed_cid("SSIS-057"), "ssis00057")
self.assertEqual(guessed_cid("START-640"), "start00640")
self.assertEqual(guessed_cid("300MIUM-1239"), "300mium01239")
self.assertEqual(guessed_cid("DANDY-851C"), "dandy00851c")
self.assertEqual(guessed_cid("092415_001"), "")
def test_matching_cids_compare_the_letters_and_the_number_not_the_substring(self):
"""搜 `SSIS 057` 前几条是 `ssis00570`；搜 `ERK 116` 会带出 `gerk116`。厂牌数字前缀与 `h_` 前缀都容得下。"""
self.assertEqual(matching_cids("SSIS-057", ["ssis00570", "ssis00579", "ssis00057", "1ssis00057"]),
["ssis00057", "1ssis00057"])
self.assertEqual(matching_cids("ERK-116", ["h_1838erkr01116", "gerk116", "erk116"]), ["erk116"])
self.assertEqual(matching_cids("FKOS-006", ["h_1721fkos00006"]), ["h_1721fkos00006"])
self.assertEqual(matching_cids("FNS-061", ["1fns00061"]), ["1fns00061"])
def test_suffixed_variants_sort_after_the_plain_cid(self):
"""`MIDV 012` 同时命中原版与 AI 重制版，原版先核。"""
self.assertEqual(matching_cids("MIDV-012", ["48midv00012ai", "midv00127", "midv00012", "48midv00012"]),
["midv00012", "48midv00012", "48midv00012ai"])
self.assertEqual(matching_cids("DANDY-851C", ["1dandy00851", "1dandy00851c", "1dandy00851ai"]), ["1dandy00851c"])
def test_a_numeric_prefix_in_the_code_must_lead_the_cid(self):
self.assertEqual(matching_cids("300MIUM-1239", ["300mium01239", "mium01239", "259mium01239"]), ["300mium01239"])
self.assertEqual(matching_cids("092415_001", ["anything"]), [])
def test_release_date_is_the_japanese_calendar_day(self):
"""`makerReleasedAt` 以 UTC 写日本时间零点：SSIS-057 答 05-06T15:00Z，账本与 r18.dev 都是 05-07。"""
self.assertEqual(release_date("2021-05-06T15:00:00Z"), "2021-05-07")
self.assertEqual(release_date("2026-10-21T15:00:00+00:00"), "2026-10-22")
self.assertEqual(release_date("2021-05-07"), "2021-05-07")
self.assertEqual(release_date("2021-05-07T00:00:00"), "2021-05-07")
self.assertEqual(release_date("not a date at all"), "not a date")
self.assertEqual(release_date(None), "")
class DmmSourceTests(unittest.TestCase):
def test_the_config_and_the_endpoint(self):
site = DmmSource()
self.assertIs(site.config, DMM)
self.assertEqual((DMM.name, DMM.label, DMM.provider, DMM.stage, DMM.interval, DMM.cookie, DMM.page_limit),
("dmm", "DMM / FANZA", "dmm-graphql", "official", 2.0, False, 1024 * 1024))
self.assertEqual(site.endpoint(), ENDPOINT)
self.assertEqual(site.content_url("ssis00057"), PAGE_URL)
self.assertEqual(site.content_url("h_1721fkos00006"), PAGE_URL.replace("ssis00057", "h_1721fkos00006"))
def test_parse_reads_the_content_json_without_any_transport(self):
record = DmmSource().parse(Page(PAGE_URL, json.dumps(ITEM, ensure_ascii=False)), "SSIS-057")
self.assertEqual((record.source, record.provenance, record.code, record.source_url),
("dmm", "dmm-graphql", "SSIS-057", PAGE_URL))
self.assertEqual((record.title, record.studio, record.label, record.series, record.director, record.release_date,
record.runtime, record.cover_urls),
(ITEM["title"], "エスワン ナンバーワンスタイル", "S1 NO.1 STYLE", "", "肉尊", "2021-05-07", 151.15,
(COVER,)))
self.assertEqual(record.performers, ({"japanese_name": "三宮つばき", "dmm_id": "1062909"},))
self.assertEqual(record.tags, ("ギリモザ", "単体作品"))
self.assertEqual((record.extra["content_id"], record.extra["sample_images"], record.extra["raw"]),
("ssis00057", [SAMPLE], ITEM))
self.assertEqual(record.payload(), EXPECTED)
def test_parse_tolerates_missing_fields_and_null_objects(self):
"""DMM 对 `series`、`packageImage` 给 null；站上没给的字段留空，不是 `None`。"""
record = DmmSource().parse(Page(PAGE_URL, json.dumps({"id": "ssis00057", "title": "x", "series": None,
"packageImage": None, "duration": 0})), "SSIS-057")
self.assertEqual((record.series, record.director, record.release_date, record.runtime, record.cover_urls,
record.tags, record.performers), ("", "", "", None, (), (), ()))
self.assertEqual(record.payload()["cover_url"], "")
def test_query_hits_the_guessed_cid_in_one_post(self):
transport = graphql(items={"ssis00057": ITEM})
record = DmmSource().query("SSIS-057", session=Session(transport))
self.assertEqual(record.payload(), EXPECTED)
self.assertEqual(asked(transport), [("detail", "ssis00057")])
method, url, referer, content_type, limit, query, variables = transport.calls[0]
self.assertEqual((method, url, referer, content_type, limit, query, variables),
("POST", ENDPOINT, "https://api.video.dmm.co.jp/", "application/json", 1024 * 1024,
DETAIL_QUERY, {"id": "ssis00057"}))
def test_query_falls_back_to_the_search_when_the_guess_is_not_on_dmm(self):
"""`FKOS-006` 的 cid 带厂牌数字前缀 `h_1721`，猜不出来，只有搜索答得出。"""
transport = graphql(contents=["h_1721fkos00006"], items={"h_1721fkos00006": FKOS})
record = DmmSource().query("FKOS-006", session=Session(transport))
self.assertEqual((record.code, record.extra["content_id"], record.release_date, record.cover_urls),
("FKOS-006", "h_1721fkos00006", "2023-01-20", ()))
self.assertEqual(asked(transport), [("detail", "fkos00006"), ("search", "FKOS 006"), ("detail", "h_1721fkos00006")])
self.assertEqual(transport.calls[1][6], {"limit": SEARCH_LIMIT, "sort": "RELEASE_DATE", "queryWord": "FKOS 006"})
def test_a_detail_that_is_another_film_is_not_trusted(self):
"""接口对一个 cid 答回来的是别的商品（id 与 `makerContentId` 都对不上番号）：按没有处理，接着核下一条。"""
other = dict(ITEM, id="ssis00570", makerContentId="SSIS-570")
transport = graphql(contents=["1ssis00057", "2ssis00057"],
items={"1ssis00057": other, "2ssis00057": dict(ITEM, id="2ssis00057")})
record = DmmSource().query("SSIS-057", session=Session(transport))
self.assertEqual(record.extra["content_id"], "2ssis00057")
self.assertEqual(asked(transport), [("detail", "ssis00057"), ("search", "SSIS 057"),
("detail", "1ssis00057"), ("detail", "2ssis00057")])
def test_search_hits_are_capped_and_the_guess_is_not_asked_twice(self):
contents = ["midv00012", "48midv00012", "48midv00012ai"]
transport = graphql(contents=contents, items={})
with self.assertRaises(SourceFailure) as caught:
DmmSource().query("MIDV-012", session=Session(transport))
self.assertEqual((caught.exception.reason, str(caught.exception)), (FailureReason.NOT_FOUND, "DMM 没有这个番号"))
self.assertEqual(asked(transport), [("detail", "midv00012"), ("search", "MIDV 012"),
("detail", "48midv00012"), ("detail", "48midv00012ai")])
self.assertEqual(len(asked(transport)) - 2, DETAIL_ATTEMPTS)
def test_the_provider_payload_equals_the_snapshot_shape_field_for_field(self):
"""`LibraryMetadataProvider.query(code, 'dmm')` 走的就是这一站；现有用例按 `peach.jav_cover_fetch._fetch` 打桩，桩仍然生效。"""
def pages(transport, url, **kwargs):
payload = json.loads(kwargs["body"])
self.assertEqual((kwargs["method"], payload["query"], payload["variables"]), ("POST", DETAIL_QUERY, {"id": "ssis00057"}))
return json.dumps({"data": {"ppvContent": ITEM}}, ensure_ascii=False)
provider = LibraryMetadataProvider.__new__(LibraryMetadataProvider)
provider.transport = Mock()
with patch("peach.jav_cover_fetch._fetch", side_effect=pages) as fetch:
payload = provider.query("SSIS-057", "dmm")
self.assertEqual(payload, EXPECTED)
self.assertEqual(fetch.call_args.args[1], ENDPOINT)
self.assertEqual(fetch.call_args.kwargs["extra_headers"], {"Content-Type": "application/json", "Accept": "application/json"})
def test_failures_come_through_the_shared_reason_table(self):
with self.assertRaises(SourceFailure) as caught:
DmmSource().query("FC2-PPV-1812235", session=Session(graphql()))
self.assertEqual((caught.exception.reason, str(caught.exception)), (FailureReason.NOT_FOUND, "认不出可以问 DMM 的番号"))
with self.assertRaises(SourceFailure) as caught:
DmmSource().parse(Page(PAGE_URL, json.dumps(dict(ITEM, makerContentId="SSIS-570", id="ssis00570"))), "SSIS-057")
self.assertEqual((caught.exception.reason, caught.exception.kind, str(caught.exception)),
(FailureReason.PARSE_ERROR, "unavailable", "来源返回的番号不匹配"))
for body in (b"<html>maintenance</html>", b"[1, 2]", b'{"title": "no id"}'):
with self.subTest(body=body), self.assertRaises(SourceFailure) as caught:
DmmSource().parse(Page(PAGE_URL, body), "SSIS-057")
self.assertEqual((caught.exception.reason, str(caught.exception)),
(FailureReason.PARSE_ERROR, "DMM 返回的不是作品 JSON"))
def test_a_refused_query_and_a_non_json_reply_are_parse_errors(self):
"""字段名过期时接口回 `errors` 且没有 `data`，措辞带上站方那句话。"""
refused = lambda request, timeout, limit: HttpResponse(
200, {}, b'{"errors": [{"message": "Cannot query field \\"largeUrl\\""}]}', request.url)
with self.assertRaises(SourceFailure) as caught:
DmmSource().query("SSIS-057", session=Session(refused))
self.assertEqual((caught.exception.reason, str(caught.exception)),
(FailureReason.PARSE_ERROR, 'DMM 拒绝了这条查询：Cannot query field "largeUrl"'))
cloudflare = lambda request, timeout, limit: HttpResponse(200, {}, b"<html>Just a moment</html>", request.url)
with self.assertRaises(SourceFailure) as caught:
DmmSource().query("SSIS-057", session=Session(cloudflare))
self.assertEqual((caught.exception.reason, str(caught.exception)), (FailureReason.PARSE_ERROR, "DMM 返回的不是作品 JSON"))
def test_transport_tiers_keep_their_semantics(self):
with self.assertRaises(SourceFailure) as caught:
DmmSource().query("SSIS-057", session=Session(graphql(items={"ssis00057": Unavailable("HTTP 503")})))
self.assertEqual((caught.exception.reason, caught.exception.status_code), (FailureReason.SERVER_ERROR, 503))
with self.assertRaises(SourceFailure) as caught:
DmmSource().query("SSIS-057", session=Session(graphql(items={"ssis00057": Unavailable("HTTP 422")})))
self.assertEqual((caught.exception.reason, caught.exception.status_code), (FailureReason.NETWORK, 422))
with self.assertRaises(SourcePaused):
DmmSource().query("SSIS-057", session=Session(graphql(items={"ssis00057": SourcePaused("来源正在冷却")})))
def listed(cid, *, status=None, date="2026-10-30T00:00:00+09:00", cast=("森日向子",), maker="ムーディーズ"):
"""片单里的一行，形状照 2026-11-04 `legacySearchPPV(filter: {actressIds})` 的实测响应写。"""
return {"id": cid, "title": f"{cid} の作品", "deliveryStartAt": date, "releaseStatus": status,
"actresses": [{"id": "1061586" if name == "森日向子" else "9", "name": name} for name in cast],
"maker": {"id": "1", "name": maker}}
def catalogue(pages, maker_codes):
"""假接口：片单按 offset 答 `pages` 里的那一页，别名批量按 cid 答 `maker_codes`。"""
calls = []
def call(request, timeout, limit):
payload = json.loads(request.body)
if payload["query"] == ACTRESS_WORKS_QUERY:
variables = payload["variables"]
calls.append(("list", variables["offset"], variables["filter"]))
rows, has_next, total = pages[variables["offset"]]
body = {"data": {"legacySearchPPV": {"result": {
"pageInfo": {"totalCount": total, "hasNext": has_next}, "contents": rows}}}}
else:
cids = re.findall(r'ppvContent\(id: "([^"]+)"\)', payload["query"])
calls.append(("codes", cids))
body = {"data": {f"a{index}": ({"id": cid, "makerContentId": maker_codes[cid]}
if cid in maker_codes else None)
for index, cid in enumerate(cids)}}
return HttpResponse(200, {}, json.dumps(body, ensure_ascii=False).encode(), request.url)
call.calls = calls
return call
class ActressWorksTests(unittest.TestCase):
"""按女优编号列片单：分页、预购、官方番号与翻页止点（ADR-0135）。"""
def test_pages_until_the_catalogue_says_there_is_no_more(self):
first = [listed("mizd00570", status="PRE_ORDER", date="2026-11-27T00:00:00+09:00"),
*[listed(f"mide{index:05d}") for index in range(99)]]
second = [listed("h_068mxdlp00343", date="2026-10-01T10:00:40+09:00")]
codes = {"mizd00570": "MIZD-570", "h_068mxdlp00343": "MXDLP-0343"}
transport = catalogue({0: (first, True, 101), 100: (second, False, 101)}, codes)
works = DmmSource().actress_works("1061586", session=Session(transport))
self.assertEqual([call[:2] for call in transport.calls if call[0] == "list"], [("list", 0), ("list", 100)])
self.assertEqual(transport.calls[0][2], {"actressIds": {"ids": [{"id": "1061586"}]}})
self.assertEqual((works.name, works.total, len(works.works)), ("森日向子", 101, 101))
head, tail = works.works[0], works.works[-1]
self.assertEqual((head.cid, head.maker_code, head.pre_order, head.release_date),
("mizd00570", "MIZD-570", True, "2026-11-27"))
self.assertEqual((tail.maker_code, tail.pre_order, tail.release_date, tail.studio),
("MXDLP-0343", False, "2026-10-01", "ムーディーズ"))
# 接口没答官方番号的那几部留空，不拿 cid 去推。
self.assertEqual(works.works[1].maker_code, "")
def test_a_page_with_a_seen_work_stops_paging_and_skips_its_codes(self):
first = [listed("mizd00570", status="PRE_ORDER"), listed("mide00001")]
transport = catalogue({0: (first, True, 500)}, {"mizd00570": "MIZD-570", "mide00001": "MIDE-001"})
works = DmmSource().actress_works("1061586", session=Session(transport), known={"mide00001"})
self.assertEqual(transport.calls, [("list", 0, {"actressIds": {"ids": [{"id": "1061586"}]}}),
("codes", ["mizd00570"])])
self.assertEqual([work.maker_code for work in works.works], ["MIZD-570", ""])
def test_the_first_poll_stops_at_the_limit(self):
page = [listed(f"mide{index:05d}") for index in range(100)]
transport = catalogue({0: (page, True, 2066), 100: (page, True, 2066)}, {})
works = DmmSource().actress_works("1008785", session=Session(transport), limit=150)
self.assertEqual(len(works.works), 150)
self.assertEqual([call[1] for call in transport.calls if call[0] == "list"], [0, 100])
def test_an_empty_catalogue_is_zero_works_not_a_failure(self):
works = DmmSource().actress_works("1051912", session=Session(catalogue({0: ([], False, 0)}, {})))
self.assertEqual((works.total, works.works), (0, ()))
def test_failures_are_classified(self):
silent = Mock()
with self.assertRaises(SourceFailure) as caught:
DmmSource().actress_works("森日向子", session=Session(silent))
self.assertEqual(caught.exception.reason, FailureReason.NOT_FOUND)
silent.assert_not_called()
broken = lambda request, timeout, limit: HttpResponse(
200, {}, json.dumps({"data": {"legacySearchPPV": None}}).encode(), request.url)
with self.assertRaises(SourceFailure) as caught:
DmmSource().actress_works("1061586", session=Session(broken))
self.assertEqual(caught.exception.reason, FailureReason.PARSE_ERROR)
def test_rate_limits_cool_the_whole_source(self):
class Limited:
def __init__(self):
self.held = []
def __call__(self, request, timeout, limit):
return HttpResponse(429, {}, b"", request.url)
def hold(self, source, action):
self.held.append((source, action))
transport, site = Limited(), DmmSource()
session = Session(transport)
with self.assertRaises(SourceFailure) as caught, site.holding(session):
site.actress_works("1061586", session=session)
self.assertEqual(caught.exception.reason, FailureReason.RATE_LIMITED)
self.assertEqual(transport.held, [("dmm", "rate_limited")])
if __name__ == "__main__":
unittest.main()