浅念
40 lines
1.5 KiB
Python
40 lines
1.5 KiB
Python
"""本地联调假 UniService 接收端(仅测试用,不进生产)。
|
|
|
|
用法: python scripts/mock_uniservice.py [端口,默认9099]
|
|
预期请求头 X-Api-Key: test-key-123;收到推送打印前 2000 字符,回 accepted。
|
|
"""
|
|
import json
|
|
import sys
|
|
from http.server import BaseHTTPRequestHandler, HTTPServer
|
|
|
|
KEY = "test-key-123"
|
|
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def do_POST(self):
|
|
n = int(self.headers.get("Content-Length", 0))
|
|
body = json.loads(self.rfile.read(n) or b"{}")
|
|
msgs = body.get("messages", [])
|
|
print(f"\n=== 收到推送 {len(msgs)} 条 (X-Api-Key={self.headers.get('X-Api-Key')}) ===")
|
|
print(json.dumps(msgs, ensure_ascii=False, indent=2)[:2000])
|
|
if self.headers.get("X-Api-Key") != KEY:
|
|
code, resp = 401, {"success": False, "error": "unauthorized"}
|
|
else:
|
|
code, resp = 200, {"success": True, "data": {
|
|
"accepted": len(msgs), "duplicated": 0, "failed": []}}
|
|
data = json.dumps(resp).encode()
|
|
self.send_response(code)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(data)))
|
|
self.end_headers()
|
|
self.wfile.write(data)
|
|
|
|
def log_message(self, *a): # 静默默认访问日志,只留推送打印
|
|
pass
|
|
|
|
|
|
if __name__ == "__main__":
|
|
port = int(sys.argv[1]) if len(sys.argv) > 1 else 9099
|
|
print(f"mock UniService on http://127.0.0.1:{port} (期待 X-Api-Key: {KEY})")
|
|
HTTPServer(("127.0.0.1", port), Handler).serve_forever()
|