feat: scaffold heimao app spider
This commit is contained in:
@@ -0,0 +1,36 @@
|
||||
import unittest
|
||||
from importlib.machinery import SourceFileLoader
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
MODULE = SourceFileLoader("heimao_spider", str(ROOT / "黑猫APP.py")).load_module()
|
||||
Spider = MODULE.Spider
|
||||
|
||||
|
||||
class TestHeiMaoAppSpider(unittest.TestCase):
|
||||
def setUp(self):
|
||||
Spider._instance = None
|
||||
self.spider = Spider()
|
||||
self.spider.init()
|
||||
|
||||
def test_home_video_content_returns_empty_list(self):
|
||||
self.assertEqual(self.spider.homeVideoContent(), {"list": []})
|
||||
|
||||
def test_build_api_url_uses_default_getappapi_path(self):
|
||||
self.assertEqual(
|
||||
self.spider._build_api_url("initV119"),
|
||||
"http://app1-0-0.87333.cc/api.php/getappapi.index/initV119",
|
||||
)
|
||||
|
||||
def test_aes_encrypt_and_decrypt_round_trip(self):
|
||||
encrypted = self.spider._aes_encrypt('{"msg":"ok"}')
|
||||
self.assertNotEqual(encrypted, '{"msg":"ok"}')
|
||||
self.assertEqual(self.spider._aes_decrypt(encrypted), '{"msg":"ok"}')
|
||||
|
||||
def test_replace_code_normalizes_common_ocr_misreads(self):
|
||||
self.assertEqual(self.spider._replace_code("5y6口"), "5960")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
+67
@@ -0,0 +1,67 @@
|
||||
# coding=utf-8
|
||||
import base64
|
||||
import sys
|
||||
|
||||
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
|
||||
from cryptography.hazmat.primitives.padding import PKCS7
|
||||
|
||||
from base.spider import Spider as BaseSpider
|
||||
|
||||
sys.path.append("..")
|
||||
|
||||
|
||||
class Spider(BaseSpider):
|
||||
def __init__(self):
|
||||
self.name = "黑猫APP"
|
||||
self.host = "http://app1-0-0.87333.cc"
|
||||
self.api_path = "/api.php/getappapi"
|
||||
self.user_agent = "okhttp/3.10.0"
|
||||
self.aes_key = "VwsHxkCViDXEExWa"
|
||||
self.aes_iv = "VwsHxkCViDXEExWa"
|
||||
|
||||
def init(self, extend=""):
|
||||
self.search_api = "searchList"
|
||||
self.init_api = "initV119"
|
||||
self.search_verify = False
|
||||
return None
|
||||
|
||||
def getName(self):
|
||||
return self.name
|
||||
|
||||
def homeVideoContent(self):
|
||||
return {"list": []}
|
||||
|
||||
def _build_api_url(self, endpoint):
|
||||
path = str(endpoint or "").lstrip("/")
|
||||
return f"{self.host}{self.api_path}.index/{path}"
|
||||
|
||||
def _aes_encrypt(self, text):
|
||||
padder = PKCS7(128).padder()
|
||||
data = padder.update(str(text).encode("utf-8")) + padder.finalize()
|
||||
cipher = Cipher(algorithms.AES(self.aes_key.encode("utf-8")), modes.CBC(self.aes_iv.encode("utf-8")))
|
||||
encryptor = cipher.encryptor()
|
||||
return base64.b64encode(encryptor.update(data) + encryptor.finalize()).decode("utf-8")
|
||||
|
||||
def _aes_decrypt(self, text):
|
||||
data = base64.b64decode(str(text or ""))
|
||||
cipher = Cipher(algorithms.AES(self.aes_key.encode("utf-8")), modes.CBC(self.aes_iv.encode("utf-8")))
|
||||
decryptor = cipher.decryptor()
|
||||
padded = decryptor.update(data) + decryptor.finalize()
|
||||
unpadder = PKCS7(128).unpadder()
|
||||
return (unpadder.update(padded) + unpadder.finalize()).decode("utf-8")
|
||||
|
||||
def _replace_code(self, text):
|
||||
replacements = {
|
||||
"y": "9",
|
||||
"口": "0",
|
||||
"q": "0",
|
||||
"u": "0",
|
||||
"o": "0",
|
||||
">": "1",
|
||||
"d": "0",
|
||||
"b": "8",
|
||||
"已": "2",
|
||||
"D": "0",
|
||||
"五": "5",
|
||||
}
|
||||
return "".join(replacements.get(char, char) for char in str(text or ""))
|
||||
Reference in New Issue
Block a user