"""广告复核队列的判据回归。 样本全部来自 2026-08-15 的真实误判与用户手工标记的真广告：命中推广词本身 不是证据，要看剥掉推广词后还剩不剩内容。 """ import sqlite3 import tempfile import unittest from contextlib import closing from pathlib import Path from unittest import mock from peach import junk, web_batch from peach.web_contract import ( BUNDLE_DIR_ASSETS, CONTENT_BYTES, PART_MARK, PROMO_CLUSTER_FILES, PROMO_DOMAIN, PROMO_PHRASE, REAL_CODE, WebContract, has_sibling_original, promo_residue, q_ads, q_items, w_batch, ) from support.ledger import fresh_ledger def promo_hit(name: str) -> bool: return bool(PROMO_PHRASE.search(name) or PROMO_DOMAIN.search(name)) class AdJudgementTests(unittest.TestCase): #: 用户手工标记为待删的真广告，名字剥完应当基本不剩内容。 REAL_ADS = ( "点击观看 房间火爆", "国产大片", "女神在线视频www.55h.me", "澳门皇冠赌场~vip0955.com", "tuu26.com", "极欲鬼滅之刃366gm.com", "苍老师强力推荐", "全中文AV在线观看【9se.tv】", "【91狼友之家91home.cc】4", ) #: 真实内容，只是被打了站点水印或剧情里出现了推广同形词。 REAL_CONTENT = ( "1248 - #饭饭吖 NTR剧情_给单男口交深喉，被母狗式爆操，还要微信跟老公汇报战果", "236953.xyz 推特新晋4年绿帽美腿淫妻网黄「一个ren」《榨精美足》" "谁能在这美妙的裸足压榨下保持不败", "No.017 - 서서 펠라_뒤치기【02분 34초】", ) def test_real_ads_have_nothing_left_after_stripping_promo(self): for name in self.REAL_ADS: with self.subTest(name=name): self.assertTrue(promo_hit(name), "应命中推广词或域名") self.assertLess(promo_residue(name), 14, "广告名剥完不应还剩大段实质描述") def test_watermarked_content_keeps_its_description(self): """名字里有域名/推广同形词，但正文是真实描述，必须留下足够残留。""" for name in self.REAL_CONTENT: with self.subTest(name=name): self.assertGreaterEqual( promo_residue(name), 14, "真实内容剥完仍应剩下可观描述，否则会被当成广告") def test_plot_word_is_not_a_contact_method(self): """“微信跟老公汇报战果”是剧情；“加微信”“微信号”才是联系方式。""" self.assertIsNone(PROMO_PHRASE.search("还要微信跟老公汇报战果")) self.assertIsNotNone(PROMO_PHRASE.search("加微信看完整版")) self.assertIsNotNone(PROMO_PHRASE.search("微信号：abc123")) def test_domain_is_detected_when_followed_by_a_word_character(self): """`uuc82.com_2` 用 \\b 收尾会漏：m 与 _ 都是词字符，构不成边界。""" self.assertIsNotNone(PROMO_DOMAIN.search("妹妹在精彩表演 哥哥快来大饱眼福uuc82.com_2")) self.assertIsNotNone(PROMO_DOMAIN.search("星空天使ccc.18my.cc")) self.assertIsNone(PROMO_DOMAIN.search("普通标题 没有站点")) def test_only_real_codes_drive_the_longer_version_rule(self): """RAIKUN325 是被误填进 code 的创作者账号，不能拿来比时长。""" self.assertIsNone(REAL_CODE.match("RAIKUN325")) self.assertIsNotNone(REAL_CODE.match("ABW-153")) self.assertIsNotNone(REAL_CODE.match("259LUXU-1141")) self.assertIsNotNone(REAL_CODE.match("FC2-PPV-1234567")) def test_disposal_marks_mix_two_different_intents(self): """`disposal` 同时承载“这是广告”和“这个我不喜欢”，不能当纯广告标注集。 用户 2026-08-15 澄清：`1726897607_*` 那批 10 个时间戳命名的长录像被标删， 是因为不喜欢该来源，与广告无关。判据不该、也不可能识别出这类。 """ for name in ("1726897607_1726897607_20250831-214903", "1726897607_1726897607_20250823-215124"): with self.subTest(name=name): self.assertFalse(promo_hit(name), "时间戳文件名不含推广特征") def test_part_markers_still_protect_split_volumes(self): for name in ("Movie CD2", "feature part1", "作品 分卷", "title-2", "title (3)"): with self.subTest(name=name): self.assertIsNotNone(PART_MARK.search(name)) def test_game_interstitial_padding_does_not_count_as_content(self): """手游插页的残留全是固定套话，2026-09-03 之前它们把分数压到门槛以下。 `WAAA-415-uncensored-HD` 目录里这批图命中「扫码」，可剥完剩下的是 「免费18禁手游」「安装」——一个内容字都没有，却让残留数越算越高。 """ for name in ("爱姬远征-免费18禁手游-扫码安装", "天下布魔-免费18禁手游-扫码安装", "工口.R18-成人遊戲-免费18禁手游-扫码安装", "工口MH-扫描访问", "QR CODE--扫一扫"): with self.subTest(name=name): self.assertTrue(promo_hit(name), "应命中推广词") self.assertLess(promo_residue(name), 6, "剥完只剩游戏名，不该算作实质描述") def test_filler_words_alone_are_not_evidence(self): """「免费」「手游」「安装」只降低残留，自己不构成命中。 真片名里也有这些词（`免费` 尤其常见），拿它们当命中依据就等于按题材删片。 """ for name in ("免费的午后 上门安装师傅", "手游主播的私拍"): with self.subTest(name=name): self.assertFalse(promo_hit(name)) def test_shot_tail_only_strips_the_thumbnail_suffix(self): """`MIAD573_02_s` 要退回 `MIAD573_02`，不能退到 `MIAD573`：那是另一条正片。""" self.assertTrue(has_sibling_original("MIAD573_02_s", {"miad573_02"})) self.assertTrue(has_sibling_original("BNST-033-cover", {"bnst-033"})) self.assertFalse(has_sibling_original("MIAD573_02_s", {"miad573"})) self.assertFalse(has_sibling_original("别的广告图", {"miad573_02"})) def test_poster_suffix_pairs_with_the_disc_it_belongs_to(self): """`pl`／`ps` 是 JAV 海报的通用写法，`lxvs006pl` 要配上 `LXVS006-BD`。 配对时剥掉的载体尾巴（`-BD`）在索引一侧完成，这里只验证海报尾巴。 """ self.assertTrue(has_sibling_original("lxvs006pl", {"lxvs006"})) self.assertTrue(has_sibling_original("lxvs006ps", {"lxvs006"})) self.assertFalse(has_sibling_original("lxvs006pl", {"cawd241"})) def test_content_threshold_is_far_above_any_promo_image(self): """插页整包加起来不到 10 MB；判据只要不把这类体积算成内容就够。""" self.assertGreaterEqual(CONTENT_BYTES, 512 * 1024**2) class ResourceJunkQueueTests(unittest.TestCase): """物理资源都要经过垃圾判断，类型不能成为免检条件。""" def setUp(self): self.temporary = tempfile.TemporaryDirectory() self.addCleanup(self.temporary.cleanup) self.db_path = fresh_ledger(Path(self.temporary.name).resolve()) self.contract = WebContract(self.db_path) def add(self, asset_id, location, path, medium, size=1000, duration=None, creator=None, disposal=None, code=None): connection = sqlite3.connect(self.db_path) try: connection.execute( "INSERT INTO asset(id,location,path,name,medium,size,duration,creator,disposal,code) " "VALUES(?,?,?,?,?,?,?,?,?,?)", (asset_id, location, path, path.rsplit("\\", 1)[-1], medium, size, duration, creator, disposal, code), ) connection.commit() finally: connection.close() def test_automatic_isolation_requires_a_scope_and_observes_source_and_directory_boundaries(self): for aid, location, folder in ((1, 'local', 'media'), (2, 'local', 'media2'), (3, '115', 'media')): self.add(aid, location, rf'R:\{folder}\点击进入.url', 'other') with self.assertRaises(ValueError): junk.isolate(self.db_path) self.assertEqual(junk.isolate(self.db_path, asset_ids=[])['isolated'], 0) self.assertEqual(junk.isolate(self.db_path, roots=[])['isolated'], 0) checked = junk.isolate(self.db_path, roots=[('local', r'r:\MEDIA')]) self.assertEqual(checked['ids'], [1]) self.assertEqual(junk.isolate(self.db_path, roots=[('local', r'R:\media')])['isolated'], 0) with self.contract.read_connection() as connection: self.assertEqual([tuple(row) for row in connection.execute( 'SELECT id,disposal,disposal_reason FROM asset ORDER BY id')], [(1, 'trash', 'junk'), (2, None, None), (3, None, None)]) def test_automatic_isolation_preserves_user_choices_records_subtitles_and_main_identity(self): for aid in range(1, 8): self.add(aid, 'local', rf'R:\media\{aid}\点击进入.url', 'other') self.add(8, 'local', r'R:\media\123.com-ABC-123.mp4', 'video', 2000, 5) with self.contract.database.write_transaction() as connection: connection.execute("UPDATE asset SET feedback='like' WHERE id=1") connection.execute('UPDATE asset SET play_count=1 WHERE id=2') connection.execute("INSERT INTO watch_queue(profile_id,asset_id,added_at) VALUES('local-default',3,'now')") connection.execute("INSERT INTO asset_subtitle(asset_id,location,path,name,format,pairing,first_seen,last_seen) " "VALUES(4,'local','R:\\media\\sub.srt','sub.srt','srt','exact','now','now')") w_batch(self.contract, {'ids': [5], 'operation': 'dismiss-junk'}) checked = junk.isolate(self.contract.database, asset_ids=list(range(1, 9)), protected_ids=[6]) self.assertEqual(checked['ids'], [7]) w_batch(self.contract, {'ids': [7], 'operation': 'restore'}) w_batch(self.contract, {'ids': [7], 'operation': 'dismiss-junk'}) self.assertEqual(junk.isolate(self.db_path, asset_ids=[7])['isolated'], 0) def test_isolation_rechecks_evidence_after_scoring(self): self.add(1, 'local', r'R:\media\点击进入.url', 'other') original = junk.scored_candidates def score_then_change(*args, **options): result = original(*args, **options) with self.contract.database.write_transaction() as connection: connection.execute("UPDATE asset SET name='notes.txt' WHERE id=1") return result with mock.patch.object(junk, 'scored_candidates', side_effect=score_then_change): self.assertEqual(junk.isolate(self.db_path, asset_ids=[1])['isolated'], 0) def test_unmeasured_and_failed_videos_await_review(self): for aid, duration in enumerate((None, 0, -1), 1): self.add(aid, 'local', rf'R:\media\{aid}\点击进入.mp4', 'video', 1000, duration) result = junk.isolate(self.db_path, asset_ids=[1, 2, 3]) self.assertEqual((result['isolated'], result['review']), (0, 3)) def test_short_attachment_structure_stays_in_review_during_automatic_processing(self): self.release_with_shorts(r'B:\发行物\正片标题') checked = junk.isolate(self.db_path, roots=[('115', r'B:\发行物')]) self.assertEqual((checked['isolated'], checked['review']), (0, 3)) def test_video_image_audio_archive_and_url_shortcut_are_all_judged(self): self.add(1, "115", r"B:\广告\tuu26.com.mp4", "video", 10 * 1024**2, 60) self.add(2, "115", r"B:\番号\CAWD-241\亚博体育hayob9.com.jpg", "image", 28725) self.add(3, "pikpak", r"A:\资源\点击这里.url", "other", 128) self.add(4, "local", r"R:\media\福利群vip0955.com.zip", "archive", 2048) self.add(5, "local", r"R:\media\苍老师强力推荐.mp3", "audio", 4096) # 域名属于作品出处且仍有长内容描述，不是推广垃圾。 self.add( 6, "115", r"B:\创作者\Dakota Doll - [LegalPorno.com] - [2024] - Breakfast Sex.mp4.jpg", "image", 280000, creator="Dakota Doll", ) # online 的 path 本来就是 URL，不是等待清理的物理快捷方式。 self.add(7, "online", "https://example.com/post/1.url", "other", 0) result = q_ads(self.contract, limit=200) self.assertEqual(result["total"], 5) self.assertEqual( {item["medium"] for item in result["items"]}, {"video", "image", "audio", "archive", "other"}, ) by_id = {item["id"]: item for item in result["items"]} self.assertIn("网址快捷方式", by_id[3]["why"]) self.assertEqual(by_id[3]["junk_kind"], "url") self.assertIn("整个名字都是推广语", by_id[2]["why"]) self.assertNotIn(6, by_id) self.assertNotIn(7, by_id) self.assertTrue(all("path" not in item for item in result["items"])) def test_game_download_promo_name_enters_review_without_automatic_trash(self): name = "18+游戏大全(996gg.cc)-七龍珠H版-三國志H版-三國群淫傳等.mp4" self.add(80, "115", "B:\\云下载\\MNGS-071\\" + name, "video", 2001226, 11.378005) self.add(81, "115", "B:\\云下载\\ROYD-358\\" + name, "video", 2001226) self.add(82, "115", "B:\\游戏录像\\" + name, "video", 5 * 1024**2, 11) self.add(83, "115", "B:\\游戏录像\\七龍珠H版游戏实况.mp4", "video", 2001226, 11) self.add(84, "115", "B:\\完整录像\\" + name, "video", 2001226, 3600) candidates = {row["id"]: row for row in q_ads(self.contract)["items"]} self.assertEqual(set(candidates), {80, 81}) for row in candidates.values(): self.assertIn("游戏下载站推广视频", row["why"]) self.assertGreaterEqual(row["score"], web_batch.JUNK_REVIEW_SCORE) self.assertLess(row["score"], web_batch.AUTO_TRASH_SCORE) self.assertEqual(junk.isolate(self.db_path, asset_ids=list(candidates))['isolated'], 0) def release_with_shorts(self, base, first_id=300, names=None, **short_values): self.add(first_id, "115", base + "\\" + base.rsplit("\\", 1)[-1] + ".mp4", "video", 1103584330, 3264.96) for offset, name in enumerate(names or ("附件甲", "附件乙", "附件丙"), 1): values = dict(size=24880490, duration=19.319002) values.update(short_values) self.add(first_id + offset, "115", base + "\\" + name + ".mp4", "video", **values) def test_short_attachment_cluster_is_reviewed_without_promo_words(self): self.release_with_shorts(r"B:\发行物\正片标题", names=( "海贼王3D版 成人欲海狂潮 黄 游", "附件甲", "附件乙", "附件丙", "附件丁", "附件戊")) items = q_ads(self.contract)["items"] self.assertEqual({row["id"] for row in items}, set(range(301, 307))) self.assertTrue(all("6 条极短小附件" in row["why"] for row in items)) self.assertTrue(all(web_batch.JUNK_REVIEW_SCORE <= row["score"] < web_batch.AUTO_TRASH_SCORE for row in items)) self.assertEqual(junk.isolate(self.db_path, asset_ids=[row['id'] for row in items])['isolated'], 0) def test_a_pair_of_shorts_and_standalone_animation_loops_stay_out(self): self.release_with_shorts(r"B:\发行物\正片标题", names=("附件甲", "附件乙")) for asset_id in range(310, 316): self.add(asset_id, "115", rf"B:\画师作品集\动画{asset_id}.mp4", "video", 24880490, 19.319002) self.assertEqual(q_ads(self.contract)["items"], []) def test_short_attachment_identity_and_measurements_must_be_eligible(self): cases = ( ({"duration": None}, None), ({"duration": 0}, None), ({"duration": -1}, None), ({"duration": 31}, None), ({"size": 0}, None), ({"size": 33 * 1024**2}, None), ({"creator": "示例画师"}, None), ({}, ("作品 CD1", "作品 part2", "作品 分卷")), ({}, ("ABW-153", "ABW-154", "ABW-155")), ) for index, (values, names) in enumerate(cases): self.release_with_shorts(rf"B:\发行物\正片{index}", 400 + index * 10, names=names, **values) self.assertEqual(q_ads(self.contract)["items"], []) def test_short_attachments_can_have_inherited_directory_codes(self): self.release_with_shorts(r"B:\发行物\ABW-153", code="ABW-153") self.assertEqual({row["id"] for row in q_ads(self.contract)["items"]}, {301, 302, 303}) def test_mixed_folder_or_inactive_main_cannot_support_attachment_evidence(self): for index, main_values in enumerate(( {"path": r"B:\发行物\正片0\别的标题.mp4"}, {"disposal": "trash"}, {"location": "online"}, {"duration": 1100}, {"size": 40 * 1024**2})): main_id = 500 + index * 10 self.release_with_shorts(rf"B:\发行物\正片{index}", main_id) with closing(sqlite3.connect(self.db_path)) as connection: for column, value in main_values.items(): connection.execute(f"UPDATE asset SET {column}=? WHERE id=?", (value, main_id)) connection.commit() self.assertEqual(q_ads(self.contract)["items"], []) def test_cross_release_spread_uses_measurements_with_different_names(self): for index in range(3): self.release_with_shorts(rf"B:\发行物\正片{index}", 600 + index * 10, names=(f"无推广字样的名字{index}",)) with closing(sqlite3.connect(self.db_path)) as connection: connection.execute("UPDATE asset SET width=1920,height=1080") connection.commit() items = q_ads(self.contract)["items"] self.assertEqual({row["id"] for row in items}, {601, 611, 621}) self.assertTrue(all("3 个发行目录" in row["why"] for row in items)) self.assertEqual(junk.isolate(self.db_path, asset_ids=[row['id'] for row in items])['isolated'], 0) def test_cross_release_spread_requires_full_matching_measurements(self): for index in range(3): self.release_with_shorts(rf"B:\发行物\正片{index}", 600 + index * 10, names=("附件",)) self.assertEqual(q_ads(self.contract)["items"], []) with closing(sqlite3.connect(self.db_path)) as connection: connection.execute("UPDATE asset SET width=1920,height=1080") connection.execute("UPDATE asset SET duration=20 WHERE id=621") connection.commit() self.assertEqual(q_ads(self.contract)["items"], []) def test_dismissed_short_attachment_can_be_reconsidered(self): self.release_with_shorts(r"B:\发行物\正片标题") w_batch(self.contract, {"ids": [301], "operation": "dismiss-junk"}) self.assertEqual({row["id"] for row in q_ads(self.contract)["items"]}, {302, 303}) w_batch(self.contract, {"ids": [301], "operation": "reconsider-junk"}) self.assertEqual({row["id"] for row in q_ads(self.contract)["items"]}, {301, 302, 303}) def test_domain_watermarked_release_volumes_are_not_junk(self): names = ("www.98T.la@FC2-1529680-CD2.mp4", "www.example.com_ABW-153-part1.mp4", "FC2-PPV-1234567-2.mp4", "ABW-153 (3).mp4") for asset_id, name in enumerate(names, 700): self.add(asset_id, "115", "B:\\发行物\\" + name, "video", 10 * 1024**2, 400) self.assertEqual(q_ads(self.contract)["items"], []) self.assertEqual(junk.isolate(self.db_path, asset_ids=range(700, 704))['isolated'], 0) def test_numbered_source_watermarks_only_enter_review(self): for asset_id, name in enumerate(("【来源站example.com】1", "[example.com]84 (2)"), 710): self.add(asset_id, "115", "B:\\短片\\" + name + ".mp4", "video", 5 * 1024**2, 10) items = q_ads(self.contract)["items"] self.assertEqual({row["id"] for row in items}, {710, 711}) self.assertTrue(all(row["score"] < web_batch.AUTO_TRASH_SCORE for row in items)) self.assertEqual(junk.isolate(self.db_path, asset_ids=[row['id'] for row in items])['isolated'], 0) def test_non_video_trash_items_remain_visible_and_recoverable(self): self.add(10, "115", r"B:\番号\广告图.jpg", "image", disposal="trash") self.add(11, "115", r"B:\番号\备用网址.url", "other", disposal="trash") self.add(12, "115", r"B:\番号\正片.mp4", "video", 1024, 60) result = q_items(self.contract, { "state": "trash", "limit": "60", "thumb": "1", "exclude_vertical": "1", "dur_min": "3600", "jav": "1", }) self.assertEqual(result["total"], 2) self.assertEqual( {(item["id"], item["medium"]) for item in result["items"]}, {(10, "image"), (11, "other")}, ) def test_nfo_and_large_same_code_clips_never_enter_the_junk_queue(self): """NFO 是资料边车；达到 120 MB 的同番号短版本归重复清理，不归垃圾。""" self.add(13, "local", r"R:\番号\点击观看.nfo", "other", 4096) self.add(14, "local", r"R:\番号\FC2-PPV-1110408.mp4", "video", 4 * 1024**3, 50 * 60, code="FC2-PPV-1110408") self.add(15, "local", r"R:\番号\FC2-PPV-1110408-2-4K修复.mp4", "video", 488 * 1024**2, 3 * 60, code="FC2-PPV-1110408") self.assertEqual(q_ads(self.contract, limit=200)["items"], []) def test_mib_archives_can_be_excluded_and_reconsidered_per_asset(self): """域名文件名可能是真资源；“不是垃圾”必须可持久排除且能撤销。""" self.add(20, "115", r"B:\\MIB\\Mib19.com.zip", "archive", 14 * 1024**3) self.add(21, "115", r"B:\\MIB\\Mib19.com(2).zip", "archive", 15 * 1024**3) self.add(22, "115", r"B:\\广告\\hayob9.com.jpg", "image", 28000) pending = q_ads(self.contract, limit=200) self.assertEqual(pending["counts"]["archive"], 2) self.assertEqual([item["id"] for item in q_ads( self.contract, limit=200, kind="archive")["items"]], [21, 20]) changed = w_batch(self.contract, {"ids": [20], "operation": "dismiss-junk"}) self.assertEqual(changed["changed"], 1) self.assertEqual(q_ads(self.contract, limit=200)["pending_total"], 2) dismissed = q_ads(self.contract, limit=200, status="dismissed") self.assertEqual([item["id"] for item in dismissed["items"]], [20]) self.assertEqual(dismissed["counts"]["archive"], 1) w_batch(self.contract, {"ids": [20], "operation": "reconsider-junk"}) self.assertEqual(q_ads(self.contract, limit=200)["pending_total"], 3) def test_one_scoring_serves_every_kind_status_and_page(self): """打分与类型、状态、分页无关：算一次，各种切法共用；一页里的条目改不到下一页。""" self.add(30, "115", r"B:\\广告\\hayob9.com.jpg", "image", 28000) self.add(31, "115", r"B:\\MIB\\Mib19.com.zip", "archive", 14 * 1024**3) with mock.patch.object(web_batch, "_scored_junk", wraps=web_batch._scored_junk) as scored: first = q_ads(self.contract, limit=1) archives = q_ads(self.contract, limit=200, kind="archive") dismissed = q_ads(self.contract, limit=200, status="dismissed") self.assertEqual(scored.call_count, 1) self.assertEqual(first["total"], 2) self.assertEqual([item["id"] for item in archives["items"]], [31]) self.assertEqual(dismissed["items"], []) first["items"][0]["why"] = ["被改过"] self.assertNotEqual(q_ads(self.contract, limit=1)["items"][0]["why"], ["被改过"]) self.assertEqual(q_ads( self.contract, limit=200, status="dismissed")["items"], []) def test_game_interstitials_in_a_release_directory_all_enter_the_queue(self): """`WAAA-415-uncensored-HD` 那批插页此前一张都没到门槛，只有正片自己入队。""" release = r"B:\番号\Wanz Factory\WAAA-415\WAAA-415-uncensored-HD" titles = ("爱姬远征", "天下布魔", "奇迹少女", "星隕計畫", "樱境物语", "欲神幻想", "腥空幻想", "工口.R18-成人遊戲", "三国志侵略版") for offset, title in enumerate(titles): self.add(30 + offset, "115", rf"{release}\{title}-免费18禁手游-扫码安装.jpg", "image", 1024**2) self.add(50, "115", rf"{release}\工口MH-扫描访问.png", "image", 966 * 1024) # 正片本体：体积与时长都在候选范围外，进不了队列，只作为目录里的邻居存在。 self.add(51, "115", rf"{release}\WAAA-415-UNCENSORED.mp4", "video", 4 * 1024**3, 8888) result = q_ads(self.contract, limit=200) self.assertEqual(result["total"], 10) self.assertNotIn(51, {item["id"] for item in result["items"]}) by_id = {item["id"]: item for item in result["items"]} # 游戏名有六个字，单看名字只够中间档；同目录另外九个同类推广名补齐证据。 self.assertIn("同目录另有 9 个同类推广名", by_id[38]["why"]) self.assertIn("整个名字都是推广语", by_id[30]["why"]) def test_bundled_photo_set_is_a_resource_not_an_ad(self): """域名+序号是打包渠道给整包起的名；几十上百张成套的图是资源。""" gallery = r"A:\创作者\森萝财团 整合用\森萝财团 X-019 肉丝换白丝 [103P1V-1.39GB]" for offset in range(BUNDLE_DIR_ASSETS): self.add(60 + offset, "local", rf"{gallery}\jitumi.pw({offset}).gif", "image", 5 * 1024**2) # 同样的名字只有一两张挤在别人的番号目录里，那就是插页。 self.add(90, "115", r"B:\番号\CAWD-241\jitumi.pw(1).gif", "image", 5 * 1024**2) result = q_ads(self.contract, limit=200) self.assertEqual([item["id"] for item in result["items"]], [90]) def test_cover_and_screenshot_outlive_their_promo_directory(self): """目录名带域名只说明从哪个站下的，说明不了目录里的图是广告。""" # 正片本体的体积在候选范围外，这里只作为截图的对照存在。 self.add(70, "115", r"B:\云下载\Jav.li_MIAD573_HD\MIAD573_02.wmv", "video", 800 * 1024**2, 3600, creator="Jav.li_MIAD573_HD") self.add(71, "115", r"B:\云下载\Jav.li_MIAD573_HD\MIAD573_02_s.jpg", "image", 40000, creator="Jav.li_MIAD573_HD") self.add(72, "115", r"B:\创作者\BNST033\aaxv.xyz-BNST033\BNST-033(1).jpg", "image", 40000, code="BNST-033") # 同一个广告包目录里，与番号无关的推广图仍然是广告。 self.add(73, "115", r"B:\创作者\BNST033\aaxv.xyz-BNST033\点击进入.jpg", "image", 40000) result = q_ads(self.contract, limit=200) self.assertEqual([item["id"] for item in result["items"]], [73]) def test_bluray_disc_image_and_its_poster_are_not_ads(self): """19.8 GB 的原盘和它的海报都是资源；同目录那个推广 rar 仍然是广告。 样本是 `B:\\云下载\\javme.me_LXVS-006-BD` 的实际三项。原盘在账本里是 `other`、没有 code、也没有时长，光看名字和目录只剩「域名+番号」这一条证据。 """ pack = r"B:\云下载\javme.me_LXVS-006-BD" self.add(91, "115", rf"{pack}\LXVS006-BD.iso", "other", 19778371584) self.add(92, "115", rf"{pack}\lxvs006pl.jpg", "image", 134351) self.add(93, "115", rf"{pack}\九色腾免费高清五码自拍在线观看不卡。.rar", "archive", 1362754) result = q_ads(self.contract, limit=200) self.assertEqual([item["id"] for item in result["items"]], [93]) def test_promo_cluster_needs_more_than_a_couple_of_neighbours(self): """两张同类插页不构成成群；`PROMO_CLUSTER_FILES` 是这条判据的下限。""" self.assertGreaterEqual(PROMO_CLUSTER_FILES, 3) release = r"B:\番号\Faleno\FSDSS-258\FSDSS-258.HD" for offset in range(PROMO_CLUSTER_FILES - 1): self.add(80 + offset, "115", rf"{release}\逆王传说 - 入侵女兒國{offset}-免费18禁手游-扫码安装.jpg", "image", 1024**2) result = q_ads(self.contract, limit=200) self.assertEqual(result["items"], []) def test_ad_flooded_folder_keeps_video_and_its_cover_only(self): """用户实例 `B:\\xxr\\0208 (23)`：整包只有正片和它的同名 PNG 是资源。 26 条广告各有各的自曝形态：装饰符小图、品牌推广卡、`.mhtml` 网页存档、 目录名自曝（`-APP`、`一键约炮`、`論壇文宣`）里的推广图。 """ base = r"B:\xxr\0208 (23)" self.add(100, "115", rf"{base}\梓怡-背著老公.mp4", "video", 985 * 1024**2, 1647) self.add(101, "115", rf"{base}\梓怡-背著老公.png", "image", 2379 * 1024) self.add(102, "115", rf"{base}\❤91短视频❤.jpg", "image", 600 * 1024) self.add(103, "115", rf"{base}\全国外围-一键约炮.jpg", "image", 606 * 1024) self.add(104, "115", rf"{base}\❤成人游戏-导航-.mhtml", "other", 6615 * 1024) self.add(105, "115", rf"{base}\_含羞草APP\❤含 羞 草❤.png", "image", 8 * 1024) self.add(106, "115", rf"{base}\論壇文宣\(_0高清国产精品资源获取.png", "image", 18 * 1024) result = q_ads(self.contract, limit=200) self.assertEqual( {item["id"] for item in result["items"]}, {102, 103, 104, 105, 106}, ) by_id = {item["id"]: item for item in result["items"]} self.assertIn("装饰符小图", by_id[102]["why"]) self.assertIn("推广品牌名", by_id[103]["why"]) self.assertIn("网页存档", by_id[104]["why"]) def test_brand_word_in_a_video_title_is_not_promo_by_itself(self): """`麻豆传媒` 一类品牌词也可以是正片标题的一部分，视频单独命中不算证据。""" self.add(110, "115", r"B:\番号\麻豆传媒 背著老公的妻子.mp4", "video", 400 * 1024**2, 900) result = q_ads(self.contract, limit=200) self.assertEqual(result["items"], []) def test_invalid_junk_filters_are_rejected(self): with self.assertRaisesRegex(ValueError, "invalid junk kind"): q_ads(self.contract, kind="document") with self.assertRaisesRegex(ValueError, "invalid junk status"): q_ads(self.contract, status="deleted") def test_small_installer_archive_name_enters_review_with_release_code(self): path = r'B:\MVP\Tokyo-Hot n1025-FHD\1024核工厂.rar' self.add(211, '115', path, 'archive', 4111) with closing(sqlite3.connect(self.db_path)) as db: db.execute('UPDATE asset SET code=? WHERE id=?', ('n1025', 211)) db.commit() items = q_ads(self.contract, limit=200)['items'] self.assertEqual([item['id'] for item in items], [211]) self.assertIn('推广安装包', items[0]['why']) self.assertEqual(junk._promo_name_reason('.rar', path, 128 * 1024), '') self.assertEqual(junk._promo_name_reason('.rar', r'B:\MVP\n1025.rar', 4111), '') def test_promotional_page_navigation_images_with_inherited_release_code(self): base = r'B:\MVP\1pon-092415_001-fhd' page = base + r'\花一分钟记住我们的网址 - sex5.xxx_files' self.add(201, '115', page + r'\banner_b.jpg', 'image', 312739) self.add(202, '115', base + r'\img\001.jpg', 'image', 46234) self.add(203, '115', page + r'\001.jpg', 'image', 46234) self.add(204, '115', base + r'\movie_files\logo.gif', 'image', 3439) with closing(sqlite3.connect(self.db_path)) as db: db.execute('UPDATE asset SET code=? WHERE id IN (201,202,203,204)', ('092415-001',)) db.commit() items = q_ads(self.contract, limit=200)['items'] self.assertEqual([item['id'] for item in items], [201]) self.assertIn('导航图片', items[0]['why']) self.assertEqual(junk._attachment_junk_reason('.gif', page + r'\1.gif', 974), '推广网页存档的横幅、标志或导航图片') self.assertEqual(junk._attachment_junk_reason('.gif', page + r'\1.gif', 6000), '') self.assertTrue(junk._attachment_junk_reason('.gif', page + r'\1000x90yunding.gif', 97843)) def add_download(self, task_id, code, folder, main): with self.contract.database.write_transaction() as connection: connection.execute( "INSERT INTO download_task(id,provider,source_uri,target,code,state,ledger_path,asset_id,updated_at) " "VALUES(?,'pikpak','magnet:?xt=urn:btih:fixture',?,?,'ingested',?,?,'now')", (task_id, f'115-{task_id}', code, folder, main)) def test_download_package_attachments_are_isolated_while_release_files_stay(self): # 样本取自 2026-11-03 的只读盘点：整理把正片移走后，推广附件留在落地目录。 self.add(1, '115', r'B:\S1\SSIS-839\SSIS-839 标题.mp4', 'video', 5 * 1024**3, 7200) self.add_download(1, 'SSIS-839', r'B:\云下载\SSIS-839', 1) base = r'B:\云下载\SSIS-839' + '\\' self.add(2, '115', base + '人间尤物.mp4', 'video', 78893257, 61) self.add(3, '115', base + '好站全在这.txt', 'other', 470) self.add(4, '115', base + r'其他作品合集\kpkp3.com_259LUXU-1555.jpg', 'image', 258032) self.add(5, '115', base + 'hjd2048.com-0907ssis839pl.jpg', 'image', 176840) self.add(6, '115', base + 'poster.jpg', 'image', 180000) self.add(7, '115', base + '回家指南综合包.zip', 'archive', 3165927) self.add(8, '115', base + '特典映像.mp4', 'video', 90 * 1024**2, 600) self.add(9, '115', base + '未探测.mp4', 'video', 4650760) # 落地目录就是下载根时，不能把根下所有文件都当成这个种子的附件。 self.add(10, '115', r'B:\云下载\notes.txt', 'other', 470) self.add_download(2, 'ABP-616', r'B:\云下载', 1) items = {item['id']: item for item in q_ads(self.contract, limit=200)['items']} self.assertEqual(set(items), {2, 3, 4, 7}) self.assertIn('SSIS-839 下载包里的附带短视频，正片长 120 分', items[2]['why']) self.assertIn('附带压缩包', items[7]['why']) self.assertLess(items[7]['score'], web_batch.AUTO_TRASH_SCORE) checked = junk.isolate(self.contract.database, asset_ids=list(range(1, 11)), protected_ids=[1]) self.assertEqual(sorted(checked['ids']), [2, 3, 4]) def test_untracked_folders_under_the_download_root_use_their_own_release_video(self): # 下载根由已有任务确定；没有任务行的旧下载按目录名里的番号认正片。 self.add(1, '115', r'B:\S1\SSIS-839\SSIS-839.mp4', 'video', 5 * 1024**3, 7200) self.add_download(1, 'SSIS-839', r'B:\云下载\SSIS-839', 1) base = r'B:\云下载\第一會所新片@SIS001@DDH-454' + '\\' self.add(10, '115', base + 'DDH-454.mp4', 'video', 3271300936, 4221.9) self.add(11, '115', base + '[资源推荐]！下载地址.txt', 'other', 154647) self.add(12, '115', base + '推特新晋约啪大神原创约炮自拍.jpg', 'image', 1410048) self.add(13, '115', base + 'cap_e_3_ddh-454.jpg', 'image', 275947) # 目录名带番号却找不到同番号正片的不算下载包；下载根以外的目录也不算。 self.add(20, '115', r'B:\云下载\ABC-123 合集\片段.mp4', 'video', 3 * 1024**3, 7200) self.add(21, '115', r'B:\云下载\ABC-123 合集\说明.txt', 'other', 400) self.add(30, '115', r'B:\日本\DDH-455\DDH-455.mp4', 'video', 3 * 1024**3, 4000) self.add(31, '115', r'B:\日本\DDH-455\推特新晋约啪大神原创约炮自拍.jpg', 'image', 1410048) items = {item['id']: item for item in q_ads(self.contract, limit=200)['items']} self.assertEqual(set(items), {11, 12}) self.assertIn('DDH-454 下载包里的附带文件', items[12]['why']) checked = junk.isolate(self.contract.database, asset_ids=[10, 11, 12, 13, 20, 21, 30, 31]) self.assertEqual(sorted(checked['ids']), [11, 12]) def test_download_attachment_rules_need_a_bound_main_and_a_long_enough_release(self): self.add(1, '115', r'B:\云下载\IPX-867\IPX-867.mp4', 'video', 300 * 1024**2, 1200) self.add_download(1, 'IPX-867', r'B:\云下载\IPX-867', 1) self.add(2, '115', r'B:\云下载\IPX-867\片头.mp4', 'video', 20 * 1024**2, 90) self.add(3, '115', r'B:\云下载\SONE-136\好站全在这.txt', 'other', 470) with self.contract.database.write_transaction() as connection: connection.execute( "INSERT INTO download_task(id,provider,source_uri,target,code,state,ledger_path,updated_at) " "VALUES(2,'pikpak','magnet:?xt=urn:btih:unbound','115-2','SONE-136','ingested',?,'now')", (r'B:\云下载\SONE-136',)) self.assertEqual(junk.isolate(self.contract.database, asset_ids=[1, 2, 3], protected_ids=[1])['ids'], []) def test_release_names_match_compact_padded_and_store_prefixed_spellings(self): for stem, code in (('hjd2048.com-0907mide580-h264', 'MIDE-580'), ('tkmj001jp', 'TKMJ-001'), ('hhd800.com@FC2-PPV-2753986_2', 'FC2-PPV-2753986'), ('STZY-019', '107STZY-019')): with self.subTest(stem=stem): self.assertTrue(junk.names_release(stem, code)) for stem, code in (('offkab@sukebei', 'FC2-PPV-4330669'), ('kpkp3.com_259LUXU-1555', 'SGKX-016'), ('MIDE5800', 'MIDE-580')): with self.subTest(stem=stem): self.assertFalse(junk.names_release(stem, code)) def test_identical_small_copies_across_release_folders_are_junk_evidence(self): for offset, folder in enumerate(('MIDA-117-UC', 'MIDA-154-UC', 'MIDA-234-UC')): self.add(20 + offset, '115', rf'B:\云下载\{folder}\情報.mp4', 'video', 11057812, 157.28) self.add(30 + offset, '115', rf'B:\云下载\{folder}\说明.txt', 'other', 136) self.add(40 + offset, '115', rf'B:\云下载\{folder}\ABC-123.jpg', 'image', 52000) self.add(50, '115', r'B:\云下载\MIDA-117-UC\片段.mp4', 'video', 2001226, 11) self.add(51, '115', r'B:\云下载\MIDA-154-UC\片段.mp4', 'video', 2001226, 11) self.add(52, '115', r'B:\云下载\MIDA-234-UC\片段.mp4', 'video', 2001226, 3600) # 创作者目录里每期都带的同一段片头不算：目录名不是发行包。 for offset, folder in enumerate(('第一期', '第二期', '第三期')): self.add(70 + offset, '115', rf'B:\创作者\{folder}\片头.mp4', 'video', 3000000, 8) items = {item['id']: item for item in q_ads(self.contract, limit=200)['items']} self.assertEqual(set(items), {20, 21, 22, 30, 31, 32}) self.assertIn('同名同体积的副本散布在 3 个目录', items[20]['why']) self.assertGreaterEqual(items[20]['score'], web_batch.AUTO_TRASH_SCORE) self.assertLess(items[30]['score'], web_batch.AUTO_TRASH_SCORE) def test_letter_spaced_promo_names_are_read_as_one_word(self): self.assertEqual(junk.despaced('x u u 6 2 . c o m'), 'xuu62.com') self.assertEqual(junk.despaced('最 新 位 址 獲 取'), '最新位址獲取') self.assertEqual(junk.despaced('No.017 - 서서 펠라_뒤치기'), 'No.017 - 서서 펠라_뒤치기') self.add(60, '115', r'B:\资源\x u u 9 2 .c o m.mp4', 'video', 23141112, 88.7) self.add(61, '115', r'B:\资源\最 新 位 址 獲 取：489155.com 收藏不迷路.txt', 'other', 136) self.add(62, '115', r'B:\资源\manko.fun.mp4', 'video', 11057812, 157.28) items = {item['id']: item for item in q_ads(self.contract, limit=200)['items']} self.assertEqual(set(items), {60, 61, 62}) self.assertTrue(all(item['score'] >= web_batch.AUTO_TRASH_SCORE for item in items.values())) if __name__ == "__main__": unittest.main()