0:20
Fiona - Paizuri (Nude)
1 week ago
Jkhomie1198 said: Gofile Click to expand...Genre spoilers first page have futa spoiler.
uploader1 said: Gofile Click to expand...thanks!
| VHSephi | |
| Aminaanimated Final_Fantasy_VIIInitialA Tifa_Lockhart | |
| Source | origin |
|---|---|
| 1280x720, 28.4s // 6.3MB // mp4 |
https://mega.nz/folder/abc
"}}).encode() bodies = {"posts": json.dumps(posts).encode(), "post/888": detail} def route(request): for key, body in bodies.items(): if request.url.endswith(key): return HttpResponse(200, {}, body) raise AssertionError(f"未预期的请求：{request.url}") result = KemonoConnector(transport=_routed(route)).fetch("fanbox/1") self.assertIn("888", [c.external_id for c in result.candidates]) self.assertEqual(result.probed, 1) def test_a_post_already_in_the_ledger_is_not_probed_again(self): """库里有的帖子当初判过保留；每次检查都再探一遍，请求数就跟着库存涨。""" posts = json.loads(KEMONO_POSTS) posts.append({"id": "888", "title": "March pack", "published": "2026-03-01T00:00:00", "file": {}, "attachments": [], "substring": "Here you go"}) def route(request): if "/post/" in request.url: raise AssertionError(f"未预期的详情请求：{request.url}") return HttpResponse(200, {}, json.dumps(posts).encode()) result = KemonoConnector(transport=_routed(route), enrich_skip={"888"}).fetch("fanbox/1") self.assertIn("888", [c.external_id for c in result.candidates]) self.assertEqual(result.probed, 0) def test_a_rate_limited_probe_stops_the_check(self): """探测被 429 挡回来时不能当成「保留」接着探下一帖，要让这次检查停下。""" posts = json.loads(KEMONO_POSTS) posts.extend({"id": str(ident), "title": f"pack {ident}", "published": "2026-03-01T00:00:00", "file": {}, "attachments": [], "substring": "Here"} for ident in (801, 802, 803)) probes = [] def route(request): if "/post/" in request.url: probes.append(request.url) return HttpResponse(429, {"Retry-After": "90"}, b"") return HttpResponse(200, {}, json.dumps(posts).encode()) with self.assertRaises(FollowSourceRateLimited) as raised: KemonoConnector(transport=_routed(route)).fetch("fanbox/1") self.assertEqual(raised.exception.retry_after, 90.0) self.assertEqual(len(probes), 1) def test_a_post_with_no_resource_at_all_is_dropped(self): """公告、感谢这类帖子既没有附件也没有网盘链接，抓了详情之后确认丢掉。""" posts = json.loads(KEMONO_POSTS) posts.append({"id": "999", "title": "Thanks for 10k followers!", "published": "2026-02-01T00:00:00", "file": {}, "attachments": [], "substring": "You are all wonderful"}) detail = json.dumps({"post": {"id": "999", "content": "See you next month
"}}).encode() def route(request): if request.url.endswith("post/999"): return HttpResponse(200, {}, detail) return HttpResponse(200, {}, json.dumps(posts).encode()) result = KemonoConnector(transport=_routed(route)).fetch("fanbox/1") self.assertEqual(result.skipped, 1) self.assertNotIn("999", [c.external_id for c in result.candidates]) def test_an_unreachable_probe_keeps_the_post(self): """抓不到详情就保留。 网络抖一下就删掉用户的一份更新，是拿一次失败的请求换一次不可见的数据丢失。 探测额度用完时同理——额度是内部限额，不该变成删数据的理由。 """ posts = json.loads(KEMONO_POSTS) posts.append({"id": "555", "title": "April pack", "published": "2026-04-01T00:00:00", "file": {}, "attachments": [], "substring": "Here"}) def route(request): if "post/555" in request.url: return HttpResponse(503, {}, b"") return HttpResponse(200, {}, json.dumps(posts).encode()) waits = [] result = KemonoConnector(transport=_routed(route), sleeper=waits.append).fetch("fanbox/1") self.assertIn("555", [c.external_id for c in result.candidates]) self.assertEqual(waits, [1, 2, 4, 8]) # 额度为 0 时不联网，也不删。 offline = KemonoConnector(max_probes=0, transport=_transport(body=json.dumps(posts).encode())) kept = offline.fetch("fanbox/1") self.assertIn("555", [c.external_id for c in kept.candidates]) self.assertEqual(kept.probed, 0) def test_a_release_named_after_a_poll_is_still_a_release(self): """**不要按标题关键词丢投票贴。** 2026-08-27 拿 LazyProcrastinator 的真实 50 条跑过一版 `poll|vote|survey|…` 正则，丢掉 18 条，其中包括 `Public Poll Release + Littlest Ramble`、`October Poll Animations Released` ——这位作者的正片就是按投票结果命名的。列表接口不给帖子类型字段， 标题不足以区分「投票贴」和「投票选出的成品」，误删一份 release 比多出一张卡片糟糕得多。这条测试就是为了挡住那个正则再被加回来。 """ posts = json.loads(KEMONO_POSTS) posts[0] = {**posts[0], "id": "777", "title": "Public Poll Release + Littlest Ramble"} result = KemonoConnector( transport=_transport(body=json.dumps(posts).encode())).fetch("fanbox/1") self.assertIn("777", [c.external_id for c in result.candidates]) self.assertEqual(result.skipped, 0) def test_paging_back_uses_an_offset_and_drops_the_conditional_headers(self): """往回翻要换页，而且**不能带条件请求头**。 `If-None-Match` 里存的是第一页的 etag。拿它去问第二页，站点很可能回 304， 用户点了「抓更早的」却什么都没发生——而且看起来和「确实没有更早的了」 一模一样。 2026-08-27 实测：kemono 的列表接口一页 50 条，`?o=50` 拿到的是第 51 条起， 与第一页零重叠。 """ seen = [] connector = KemonoConnector( transport=_transport(body=KEMONO_POSTS, record=seen)) connector.fetch("fanbox/30917150", etag='"k1"', last_modified="then", page=2) self.assertEqual(seen[0].url, "https://kemono.cr/api/v1/fanbox/user/30917150/posts?o=100") self.assertNotIn("If-None-Match", seen[0].headers) self.assertNotIn("If-Modified-Since", seen[0].headers) def test_the_first_page_still_sends_conditional_headers(self): # 常规检查必须保留条件请求：追更每天都在跑，没变就不该把整页再传一遍。 seen = [] KemonoConnector(transport=_transport(body=KEMONO_POSTS, record=seen)).fetch( "fanbox/30917150", etag='"k1"') self.assertEqual(seen[0].headers.get("If-None-Match"), '"k1"') self.assertNotIn("?o=", seen[0].url) def test_pawchive_returns_a_bare_list_and_uses_its_own_host(self): seen = [] connector = KemonoConnector(provider="pawchive", transport=_transport(body=KEMONO_POSTS, record=seen)) result = connector.fetch("fanbox/30917150") self.assertTrue(seen[0].url.startswith("https://pawchive.pw/")) self.assertEqual(result.provider, "pawchive") self.assertEqual(len(result.candidates), 2) def test_dict_shaped_payload_is_accepted(self): body = json.dumps({"posts": json.loads(KEMONO_POSTS)}).encode() result = KemonoConnector(transport=_transport(body=body)).fetch("fanbox/1") self.assertEqual(len(result.candidates), 2) def test_max_items_bounds_the_result(self): connector = KemonoConnector(max_items=1, transport=_transport(body=KEMONO_POSTS)) self.assertEqual(len(connector.fetch("fanbox/1").candidates), 1) def test_not_modified_returns_no_candidates(self): connector = KemonoConnector(transport=_transport(status=304)) result = connector.fetch("fanbox/1", etag='"k1"') self.assertTrue(result.not_modified) self.assertEqual(result.candidates, ()) def test_backfill_400_means_history_is_exhausted(self): connector = KemonoConnector(transport=_transport(status=400)) with self.assertRaises(FollowHistoryEnd): connector.fetch("fanbox/1", page=2) def test_malformed_ref_is_rejected_before_any_request(self): seen = [] connector = KemonoConnector(transport=_transport(record=seen)) for bad in ("30917150", "fanbox/", "../etc/passwd", "fanbox/a b"): with self.assertRaises(FollowSourceError): connector.fetch(bad) self.assertEqual(seen, []) def test_forbidden_status_names_credentials_or_bot_check(self): connector = KemonoConnector(transport=_transport(status=403)) with self.assertRaises(FollowSourceError) as caught: connector.fetch("fanbox/1") self.assertIn("403", str(caught.exception)) def test_unknown_kemono_host_is_rejected(self): with self.assertRaises(FollowSourceError): KemonoConnector(provider="nope") class Rule34VideoConnectorTests(unittest.TestCase): def _fetch(self, body=RULE34VIDEO_HTML, **kwargs): return Rule34VideoConnector(transport=_transport(body=body), **kwargs).fetch( "lazyprocrastinator") def test_creator_page_yields_deduplicated_videos_with_duration(self): result = self._fetch() self.assertEqual([c.external_id for c in result.candidates], ["4542721", "4542713"]) self.assertEqual(result.candidates[0].duration, 20.0) self.assertEqual(result.candidates[1].duration, 433.0) def test_empty_backfill_page_means_history_is_exhausted(self): connector = Rule34VideoConnector(transport=_transport(body=b"")) with self.assertRaises(FollowHistoryEnd): connector.fetch("lazyprocrastinator", page=4) def test_data_uri_placeholder_is_not_used_as_a_thumbnail(self): first = self._fetch().candidates[0] self.assertTrue(first.thumb_url.endswith("/3.jpg")) self.assertTrue(first.media_url.endswith("_preview.mp4/")) def test_relative_dates_are_marked_approximate(self): first = self._fetch().candidates[0] self.assertEqual(first.extra["published_precision"], "approximate") self.assertEqual(first.extra["added_text"], "1 week ago") self.assertIsNotNone(first.published_at) def test_missing_dates_are_reported_as_unknown_not_guessed(self): body = b'' candidate = self._fetch(body=body).candidates[0] self.assertIsNone(candidate.published_at) self.assertEqual(candidate.extra["published_precision"], "unknown") def test_empty_parse_is_an_error_not_an_empty_success(self): # 结构变了却报「本次没有更新」，会把站点改版静默吞掉。 with self.assertRaises(FollowSourceError): self._fetch(body=b"no videos here") def test_slug_is_validated(self): connector = Rule34VideoConnector(transport=_transport(body=RULE34VIDEO_HTML)) with self.assertRaises(FollowSourceError): connector.fetch("../models") def test_detail_enrichment_adds_full_cover_video_date_and_content_tags(self): seen = [] def transport(request, timeout, max_bytes): seen.append(request.url) body = (RULE34VIDEO_HTML if "/models/" in request.url else RULE34VIDEO_DETAIL_HTML) return HttpResponse(200, {}, body) result = Rule34VideoConnector(transport=transport).fetch("lazyprocrastinator") first = result.candidates[0] self.assertEqual(result.probed, 2) self.assertEqual(first.published_at, "2026-08-18T00:00:00Z") self.assertEqual(first.extra["published_precision"], "exact") self.assertEqual(first.duration, 433.0) self.assertEqual(first.extra["tags"], ["deep throat", "breast squeeze"]) self.assertEqual(first.extra["categories"], ["3D", "Final Fantasy"]) self.assertEqual(first.extra["tag_types"]["3D"], "metadata") self.assertEqual(first.extra["tag_types"]["Final Fantasy"], "copyright") self.assertTrue(first.thumb_url.endswith("/preview.jpg")) self.assertIn("4542721_360.mp4", first.media_url) def test_characters_follow_the_site_group_not_the_name_shape(self): """角色按站方的 Characters 分组记 character，名字带括号不算依据。 `d.mon (overwatch)` 这类标签站方列在 Tags 组里，照记 general。 """ groups = ( b'
