feat: add _api_post with encrypted API communication

Co-Authored-By: Claude Opus 4.7 <[email protected]>
This commit is contained in:
Harold
2026-04-20 12:58:35 +08:00
co-authored by Claude Opus 4.7
parent 2cb8545052
commit f190f770c4
2 changed files with 35 additions and 0 deletions
+16
View File
@@ -1,3 +1,4 @@
import json
import unittest
from importlib.machinery import SourceFileLoader
from pathlib import Path
@@ -22,3 +23,18 @@ class TestJingyuSpider(unittest.TestCase):
import re
encrypted = self.spider._aes_encrypt("hello")
self.assertTrue(re.match(r'^[A-Za-z0-9+/=]+$', encrypted))
def test_api_post_decrypts_response(self):
payload = {"result": "ok"}
encrypted = self.spider._aes_encrypt(json.dumps(payload))
class FakeResponse:
status_code = 200
encoding = "utf-8"
def json(self):
return {"data": encrypted}
self.spider.post = lambda url, **kwargs: FakeResponse()
self.spider.host = "http://test.com"
result = self.spider._api_post("someEndpoint")
self.assertEqual(result, payload)
+19
View File
@@ -47,3 +47,22 @@ class Spider(BaseSpider):
cipher = AES.new(key, AES.MODE_CBC, iv)
decrypted = unpad(cipher.decrypt(raw), AES.block_size)
return decrypted.decode("utf-8")
def _api_post(self, endpoint, payload=None):
if payload is None:
payload = {}
ep = f"/{endpoint}" if not endpoint.startswith("/") else endpoint
url = f"{self.host}{self.api_path}{ep}"
headers = {
"User-Agent": self.ua,
"Accept-Encoding": "gzip",
}
rsp = self.post(url, json=payload, headers=headers, timeout=15, verify=False)
data = rsp.json().get("data")
if not data:
return None
try:
return json.loads(self._aes_decrypt(data))
except Exception as e:
self.log(f"JSON解析失败: {e}")
return None