feat: add czzy iframe player parsers

py/tests/test_czzy.py

@@ -1,8 +1,12 @@
 import unittest
+from base64 import b64encode
 from importlib.machinery import SourceFileLoader
 from pathlib import Path
 from unittest.mock import patch
 
+from Cryptodome.Cipher import AES
+from Cryptodome.Util.Padding import pad
+
 
 ROOT = Path(__file__).resolve().parents[1]
 MODULE = SourceFileLoader("czzy_spider", str(ROOT / "厂长资源.py")).load_module()
@@ -147,6 +151,55 @@ class TestCZZYSpider(unittest.TestCase):
         self.assertEqual(result["list"][0]["vod_id"], "/movie/example.html")
         self.assertEqual(result["list"][0]["vod_name"], "示例影片")
 
+    def test_extract_iframe_src(self):
+        html = '<iframe src="/player-v2/test"></iframe>'
+        self.assertEqual(
+            self.spider._extract_iframe_src(html, "https://www.czzy89.com"),
+            "https://www.czzy89.com/player-v2/test",
+        )
+
+    def test_extract_player_url_prefers_mysvg_then_art_url(self):
+        self.assertEqual(
+            self.spider._extract_player_url_from_iframe("var mysvg='https://video.example/a.m3u8';"),
+            "https://video.example/a.m3u8",
+        )
+        self.assertEqual(
+            self.spider._extract_player_url_from_iframe("art.url='https://video.example/b.m3u8';"),
+            "https://video.example/b.m3u8",
+        )
+
+    def test_extract_player_url_decodes_data_payload(self):
+        original = "https://video.example/data.m3u8"
+        middle = len(original) // 2
+        obfuscated = original[:middle] + "ABCDEFG" + original[middle:]
+        encoded = "".join("{:02x}".format(ord(ch)) for ch in obfuscated)[::-1]
+        html = 'var config = {"data":"%s"};' % encoded
+        self.assertEqual(self.spider._extract_player_url_from_iframe(html), original)
+
+    def test_extract_player_url_decrypts_player_payload(self):
+        payload = '{"url":"https://video.example/encrypted.m3u8"}'
+        key = b"VFBTzdujpR9FWBhe"
+        iv = b"1234567890abcdef"
+        cipher = AES.new(key, AES.MODE_CBC, iv)
+        encrypted = b64encode(cipher.encrypt(pad(payload.encode("utf-8"), AES.block_size))).decode("utf-8")
+        html = 'var player="%s";var rand="%s";' % (encrypted, iv.decode("utf-8"))
+        self.assertEqual(
+            self.spider._extract_player_url_from_iframe(html),
+            "https://video.example/encrypted.m3u8",
+        )
+
+    def test_extract_player_url_supports_wp_nonce_fallback(self):
+        html = """
+        <script>
+        window.wp_nonce = "token";
+        var config = { url: 'https://video.example/wp.m3u8' };
+        </script>
+        """
+        self.assertEqual(
+            self.spider._extract_player_url_from_iframe(html),
+            "https://video.example/wp.m3u8",
+        )
+
 
 if __name__ == "__main__":
     unittest.main()

py/厂长资源.py

@@ -1,7 +1,13 @@
 # coding=utf-8
+import base64
+import json
+import re
 import sys
 from urllib.parse import quote, urljoin
 
+from Cryptodome.Cipher import AES
+from Cryptodome.Util.Padding import unpad
+
 from base.spider import Spider as BaseSpider
 
 sys.path.append("..")
@@ -209,6 +215,66 @@ class Spider(BaseSpider):
         }
         return {"list": [vod]}
 
+    def _extract_iframe_src(self, html, host):
+        match = re.search(r"<iframe[^>]+src=['\"]([^'\"]+)['\"]", html, re.I)
+        if not match:
+            return ""
+        return self._normalize_url(match.group(1), host)
+
+    def _decode_data_url(self, value):
+        try:
+            encrypted = value[::-1]
+            temp = ""
+            for idx in range(0, len(encrypted), 2):
+                pair = encrypted[idx:idx + 2]
+                if len(pair) == 2:
+                    temp += chr(int(pair, 16))
+            middle = (len(temp) - 7) // 2
+            return temp[:middle] + temp[middle + 7:]
+        except Exception:
+            return ""
+
+    def _decrypt_player_payload(self, cipher_text, iv):
+        try:
+            cipher = AES.new(b"VFBTzdujpR9FWBhe", AES.MODE_CBC, iv.encode("utf-8"))
+            raw = base64.b64decode(cipher_text)
+            value = unpad(cipher.decrypt(raw), AES.block_size).decode("utf-8")
+            return json.loads(value).get("url", "")
+        except Exception:
+            return ""
+
+    def _extract_player_url_from_iframe(self, html):
+        match = re.search(
+            r"var\s+player\s*=\s*[\"']([^\"']+)[\"'].*?var\s+rand\s*=\s*[\"']([^\"']+)[\"']",
+            html,
+            re.S,
+        )
+        if match:
+            value = self._decrypt_player_payload(match.group(1), match.group(2))
+            if value:
+                return value
+
+        match = re.search(r"[\"']data[\"']\s*:\s*[\"']([^\"']+)[\"']", html)
+        if match:
+            value = self._decode_data_url(match.group(1))
+            if value:
+                return value
+
+        match = re.search(r"\bmysvg\b\s*=\s*[\"']([^\"']+)[\"']", html, re.I)
+        if match:
+            return match.group(1)
+
+        match = re.search(r"art\.url\s*=\s*[\"']([^\"']+)[\"']", html, re.I)
+        if match:
+            return match.group(1)
+
+        if "window.wp_nonce" in html:
+            match = re.search(r"url\s*:\s*[\"']([^\"']+)[\"']", html, re.I)
+            if match:
+                return match.group(1)
+
+        return ""
+
     def categoryContent(self, tid, pg, filter, extend):
         path = self.category_paths.get(tid, self.category_paths["movie"]).format(pg=pg)
         html, host = self._request_html(path, expect_xpath="//a[@href]")