feat: scaffold jikanyingshi spider
This commit is contained in:
@@ -0,0 +1,52 @@
|
||||
import base64
|
||||
import unittest
|
||||
from importlib.machinery import SourceFileLoader
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import patch
|
||||
|
||||
try:
|
||||
from Crypto.Cipher import AES
|
||||
from Crypto.Util.Padding import pad
|
||||
except ModuleNotFoundError:
|
||||
from Cryptodome.Cipher import AES
|
||||
from Cryptodome.Util.Padding import pad
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
MODULE = SourceFileLoader("jikanyingshi_spider", str(ROOT / "即看影视.py")).load_module()
|
||||
Spider = MODULE.Spider
|
||||
|
||||
|
||||
def build_sk_payload(value, key, iv):
|
||||
cipher = AES.new(key.encode("utf-8"), AES.MODE_CBC, iv.encode("utf-8"))
|
||||
encrypted = cipher.encrypt(pad(value.encode("utf-8"), AES.block_size)).hex()
|
||||
return "FROMSKZZJM" + encrypted
|
||||
|
||||
|
||||
class TestJiKanYingShiSpider(unittest.TestCase):
|
||||
def setUp(self):
|
||||
Spider._instance = None
|
||||
self.spider = Spider()
|
||||
self.spider.init()
|
||||
|
||||
@patch.object(Spider, "fetch")
|
||||
def test_resolve_api_host_reads_appipr_txt(self, mock_fetch):
|
||||
mock_fetch.return_value = SimpleNamespace(text="https://appsky2025999.ideasz.net\n")
|
||||
self.assertEqual(self.spider._resolve_api_host(), "https://appsky2025999.ideasz.net")
|
||||
|
||||
def test_sk_decrypt_returns_plain_text_without_prefix(self):
|
||||
self.assertEqual(self.spider._sk_decrypt("plain-text"), "plain-text")
|
||||
|
||||
def test_sk_decrypt_decodes_prefixed_ciphertext(self):
|
||||
encrypted = build_sk_payload('{"msg":"ok"}', self.spider.aes_key, self.spider.aes_iv)
|
||||
self.assertEqual(self.spider._sk_decrypt(encrypted), '{"msg":"ok"}')
|
||||
|
||||
def test_ck_encrypt_returns_base64_ascii(self):
|
||||
result = self.spider._ck_encrypt("https://host##5483##1700000000000##ckzmbc")
|
||||
decoded = base64.b64decode(result.encode("utf-8")).decode("utf-8")
|
||||
self.assertTrue(decoded)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
+63
@@ -0,0 +1,63 @@
|
||||
# coding=utf-8
|
||||
import base64
|
||||
import hashlib
|
||||
import sys
|
||||
|
||||
try:
|
||||
from Crypto.Cipher import AES
|
||||
from Crypto.Util.Padding import pad, unpad
|
||||
except ModuleNotFoundError:
|
||||
from Cryptodome.Cipher import AES
|
||||
from Cryptodome.Util.Padding import pad, unpad
|
||||
|
||||
from base.spider import Spider as BaseSpider
|
||||
|
||||
sys.path.append("..")
|
||||
|
||||
|
||||
class Spider(BaseSpider):
|
||||
def __init__(self):
|
||||
self.name = "即看影视"
|
||||
self.host_config_url = "https://skyappdata-1321528676.cos.accelerate.myqcloud.com/4kapp/appipr.txt"
|
||||
self.aes_key = "ygcnbckhcuvygdyb"
|
||||
self.aes_iv = "4023892775143708"
|
||||
self.ck_key = "ygcnbcrvaervztmw"
|
||||
self.ck_iv = "1212164105143708"
|
||||
self.user_agent = "Dart/2.10 (dart:io)"
|
||||
self.headers = {
|
||||
"User-Agent": self.user_agent,
|
||||
"Accept-Encoding": "gzip",
|
||||
}
|
||||
self.api_host = ""
|
||||
self.auth_token = ""
|
||||
self.config_data = {}
|
||||
self.is_ready = False
|
||||
|
||||
def init(self, extend=""):
|
||||
return None
|
||||
|
||||
def getName(self):
|
||||
return self.name
|
||||
|
||||
def _resolve_api_host(self):
|
||||
response = self.fetch(self.host_config_url, headers=self.headers, timeout=10, verify=False)
|
||||
return str(getattr(response, "text", "")).strip().rstrip("/")
|
||||
|
||||
def _md5(self, text):
|
||||
return hashlib.md5(str(text).encode("utf-8")).hexdigest()
|
||||
|
||||
def _sk_decrypt(self, text):
|
||||
value = str(text or "")
|
||||
prefix = "FROMSKZZJM"
|
||||
if not value.startswith(prefix):
|
||||
return value
|
||||
encrypted_hex = value[len(prefix):]
|
||||
cipher = AES.new(self.aes_key.encode("utf-8"), AES.MODE_CBC, self.aes_iv.encode("utf-8"))
|
||||
return unpad(cipher.decrypt(bytes.fromhex(encrypted_hex)), AES.block_size).decode("utf-8")
|
||||
|
||||
def _ck_encrypt(self, text):
|
||||
first = base64.b64encode(str(text).encode("utf-8"))
|
||||
second = base64.b64encode(first)
|
||||
cipher = AES.new(self.ck_key.encode("utf-8"), AES.MODE_CBC, self.ck_iv.encode("utf-8"))
|
||||
encrypted = cipher.encrypt(pad(second, AES.block_size)).hex()
|
||||
return base64.b64encode(encrypted.encode("utf-8")).decode("utf-8")
|
||||
Reference in New Issue
Block a user