feat: scaffold cupfox spider helpers

py/tests/test_茶杯狐.py

@@ -0,0 +1,50 @@
+import base64
+import unittest
+from importlib.machinery import SourceFileLoader
+from pathlib import Path
+from unittest.mock import patch
+
+
+ROOT = Path(__file__).resolve().parents[1]
+MODULE = SourceFileLoader("cupfox_spider", str(ROOT / "茶杯狐.py")).load_module()
+Spider = MODULE.Spider
+
+
+class TestCupfoxSpider(unittest.TestCase):
+    def setUp(self):
+        Spider._instance = None
+        self.spider = Spider()
+        self.spider.init()
+
+    def test_encode_and_decode_ids_keep_short_paths(self):
+        self.assertEqual(
+            self.spider._encode_detail_id("/movie/test-slug.html"),
+            "detail/test-slug",
+        )
+        self.assertEqual(
+            self.spider._decode_detail_id("detail/test-slug"),
+            "https://www.cupfox.ai/movie/test-slug.html",
+        )
+        self.assertEqual(
+            self.spider._encode_play_id("/play/abc123.html"),
+            "play/abc123",
+        )
+        self.assertEqual(
+            self.spider._decode_play_id("play/abc123"),
+            "https://www.cupfox.ai/play/abc123.html",
+        )
+
+    def test_merge_set_cookie_and_cookie_header(self):
+        jar = {}
+        self.spider._merge_set_cookie(jar, ["foo=1; Path=/", "bar=2; HttpOnly"])
+        self.assertEqual(jar, {"foo": "1", "bar": "2"})
+        self.assertEqual(self.spider._cookie_header(jar), "foo=1; bar=2")
+
+    def test_extract_firewall_token(self):
+        html = '<script>var token = encrypt("abcXYZ");</script>'
+        self.assertEqual(self.spider._extract_firewall_token(html), "abcXYZ")
+
+    @patch("cupfox_spider.random.randint", side_effect=[0, 1, 2, 3])
+    def test_firewall_encrypt_returns_base64_text(self, _mock_randint):
+        encoded = self.spider._cupfox_firewall_encrypt("PX")
+        self.assertEqual(base64.b64decode(encoded).decode("utf-8"), "PwXh7w")

py/茶杯狐.py

@@ -0,0 +1,82 @@
+# coding=utf-8
+import base64
+import random
+import re
+import sys
+from urllib.parse import urljoin
+
+from base.spider import Spider as BaseSpider
+
+sys.path.append("..")
+
+
+class Spider(BaseSpider):
+    def __init__(self):
+        self.name = "茶杯狐"
+        self.host = "https://www.cupfox.ai"
+        self.page_limit = 20
+        self.firewall_chars = "PXhw7UT1B0a9kQDKZsjIASmOezxYG4CHo5Jyfg2b8FLpEvRr3WtVnlqMidu6cN"
+        self.headers = {
+            "User-Agent": (
+                "Mozilla/5.0 (iPhone; CPU iPhone OS 18_3_2 like Mac OS X) "
+                "AppleWebKit/605.1.15 (KHTML, like Gecko) Version/18.3.1 "
+                "Mobile/15E148 Safari/604.1"
+            ),
+            "Referer": self.host + "/",
+            "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
+            "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
+        }
+
+    def init(self, extend=""):
+        return None
+
+    def getName(self):
+        return self.name
+
+    def _build_url(self, path):
+        return urljoin(self.host + "/", str(path or "").strip())
+
+    def _encode_detail_id(self, href):
+        matched = re.search(r"/movie/([^/?#]+)\.html", self._build_url(href))
+        return f"detail/{matched.group(1)}" if matched else ""
+
+    def _decode_detail_id(self, vod_id):
+        matched = re.search(r"^detail/([^/?#]+)$", str(vod_id or "").strip())
+        return self._build_url(f"/movie/{matched.group(1)}.html") if matched else ""
+
+    def _encode_play_id(self, href):
+        matched = re.search(r"/play/([^/?#]+)\.html", self._build_url(href))
+        return f"play/{matched.group(1)}" if matched else ""
+
+    def _decode_play_id(self, play_id):
+        matched = re.search(r"^play/([^/?#]+)$", str(play_id or "").strip())
+        return self._build_url(f"/play/{matched.group(1)}.html") if matched else ""
+
+    def _merge_set_cookie(self, cookie_jar, headers):
+        values = headers if isinstance(headers, list) else [headers]
+        for item in values:
+            first = str(item or "").split(";")[0]
+            if "=" not in first:
+                continue
+            name, value = first.split("=", 1)
+            if name.strip():
+                cookie_jar[name.strip()] = value.strip()
+
+    def _cookie_header(self, cookie_jar):
+        return "; ".join([f"{key}={value}" for key, value in cookie_jar.items()])
+
+    def _extract_firewall_token(self, html_text):
+        matched = re.search(r'var\s+token\s*=\s*encrypt\("([^"]+)"\)', str(html_text or ""))
+        return matched.group(1) if matched else ""
+
+    def _cupfox_firewall_encrypt(self, value):
+        encoded = ""
+        for char in str(value or ""):
+            index = self.firewall_chars.find(char)
+            mapped = char if index == -1 else self.firewall_chars[(index + 3) % 62]
+            encoded += (
+                self.firewall_chars[random.randint(0, 61)]
+                + mapped
+                + self.firewall_chars[random.randint(0, 61)]
+            )
+        return base64.b64encode(encoded.encode("utf-8")).decode("utf-8")