feat: scaffold mofang app spider
This commit is contained in:
@@ -0,0 +1,32 @@
|
||||
import unittest
|
||||
from importlib.machinery import SourceFileLoader
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
MODULE = SourceFileLoader("mofang_spider", str(ROOT / "魔方APP.py")).load_module()
|
||||
Spider = MODULE.Spider
|
||||
|
||||
|
||||
class TestMoFangAppSpider(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://www.613mf4k12.top/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")
|
||||
+64
@@ -0,0 +1,64 @@
|
||||
# coding=utf-8
|
||||
import base64
|
||||
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 = "魔方APP"
|
||||
self.host = "http://www.613mf4k12.top"
|
||||
self.api_path = "/api.php/getappapi"
|
||||
self.user_agent = "okhttp/3.10.0"
|
||||
self.aes_key = "1234567887654321"
|
||||
self.aes_iv = "1234567887654321"
|
||||
|
||||
def init(self, extend=""):
|
||||
self.init_api = "initV119"
|
||||
self.search_api = "searchList"
|
||||
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):
|
||||
cipher = AES.new(self.aes_key.encode("utf-8"), AES.MODE_CBC, self.aes_iv.encode("utf-8"))
|
||||
return base64.b64encode(cipher.encrypt(pad(str(text).encode("utf-8"), AES.block_size))).decode("utf-8")
|
||||
|
||||
def _aes_decrypt(self, text):
|
||||
cipher = AES.new(self.aes_key.encode("utf-8"), AES.MODE_CBC, self.aes_iv.encode("utf-8"))
|
||||
return unpad(cipher.decrypt(base64.b64decode(str(text or ""))), AES.block_size).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