#!/usr/bin/env python3 """Zero-dependency mock for the commercial-intelligence monitor APIs.""" from __future__ import annotations import argparse import json from datetime import datetime from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from urllib.parse import parse_qs, unquote, urlparse AUTHORIZATION = "Bearer admin" PREFIX = "/consumer/monitor/task" CALENDAR_ROWS = [ { "createdTime": "2026-09-28 09:10:00", "description": "重点会议进入准备窗口，需持续跟踪相关议题和参与方动态。", "direction": "T", "endTime": "2026-10-01 18:00:00", "nodeName": "重点会议", "recUuid": "calendar-t-001", "startTime": "2026-09-29 10:00:00", "title": "台海方向重点会议准备节点", "updateTime": "2026-09-28 09:20:00", }, { "createdTime": "2026-09-28 09:30:00", "description": "公开活动开始前的关键观察窗口。", "direction": "T", "endTime": "2026-10-05 20:00:00", "nodeName": "公开活动", "recUuid": "calendar-t-002", "startTime": "2026-10-03 08:30:00", "title": "台海方向公开活动观察节点", "updateTime": "2026-09-28 09:35:00", }, ] EVENT_ROWS = { "热点": [ { "id": 101, "m_title": "台海方向热点活动升温", "m_publish_time_date": "2026-09-29 08:15:00", "source": "热点", "m_content": "相关活动密度上升，公开信息关注度持续增加。", "import_num": 9, "direction": "台", }, { "id": 102, "m_title": "重点区域公开讨论增多", "m_publish_time_date": "2026-09-29 09:20:00", "source": "热点", "m_content": "多个公开渠道围绕重点区域展开讨论。", "import_num": 6, "direction": "台", }, ], "威胁": [ { "id": 201, "m_title": "外部干扰因素增加", "m_publish_time_date": "2026-09-29 07:40:00", "source": "威胁", "m_content": "外部干扰活动可能对既定安排形成压力。", "import_num": 8, "direction": "台", } ], "战机": [ { "id": 301, "m_title": "信息窗口出现", "m_publish_time_date": "2026-09-29 06:50:00", "source": "战机", "m_content": "公开信息集中释放，形成进一步研判和行动衔接窗口。", "import_num": 7, "direction": "台", } ], } ENTITY_ROWS = { 101: { "eid": "Q22686", "name": "示例关键人物", "etype": "human", "text": "与热点活动相关的关键人物。", "imagePath": "/mock/images/q22686.png", "num": 12, "label": "重点关注", "reason": "在多个相关事件中被重复提及。", "score": 92, }, 201: { "eid": "ORG-1001", "name": "示例关联机构", "etype": "organization", "text": "与威胁事件有关联的机构。", "imagePath": "/mock/images/org-1001.png", "num": 8, "label": "风险关联", "reason": "与高重要度威胁事件存在关联。", "score": 86, }, 301: { "eid": "LOC-3001", "name": "示例重点区域", "etype": "location", "text": "战机事件涉及的重点区域。", "imagePath": "/mock/images/loc-3001.png", "num": 6, "label": "机会窗口", "reason": "在战机信息中具有较高关联度。", "score": 81, }, } def _contains(row: dict, keyword: str, fields: tuple[str, ...]) -> bool: if not keyword: return True return any(keyword.lower() in str(row.get(field, "")).lower() for field in fields) def _in_range(row_time: str, start: str, end: str) -> bool: try: value = datetime.strptime(row_time, "%Y-%m-%d %H:%M:%S") if start and value < datetime.strptime(start, "%Y-%m-%d %H:%M:%S"): return False if end and value > datetime.strptime(end, "%Y-%m-%d %H:%M:%S"): return False except ValueError: return True return True class Handler(BaseHTTPRequestHandler): server_version = "MonitorMock/1.0" def log_message(self, fmt: str, *args) -> None: print(f"[mock-monitor] {self.address_string()} {fmt % args}") def _json(self, status: int, payload: dict) -> None: raw = json.dumps(payload, ensure_ascii=False).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(raw))) self.end_headers() self.wfile.write(raw) def _authorized(self) -> bool: if self.headers.get("Authorization", "") == AUTHORIZATION: return True self._json(403, {"data": None, "msg": "Authorization 必须为 Bearer admin", "state": 403}) return False def do_GET(self) -> None: # noqa: N802 if not self._authorized(): return parsed = urlparse(self.path) path = parsed.path.rstrip("/") query = {key: values[-1] for key, values in parse_qs(parsed.query, keep_blank_values=True).items()} if path == f"{PREFIX}/calendar": self._calendar(query) return if path == f"{PREFIX}/eventMonitorByTime": self._events(query) return if path.startswith(f"{PREFIX}/realNetwork/list/"): self._real_network(unquote(path.rsplit("/", 1)[-1])) return if path.startswith(PREFIX + "/"): self._task_detail(unquote(path.rsplit("/", 1)[-1])) return self._json(404, {"data": None, "msg": "接口不存在", "state": 404}) def do_POST(self) -> None: # noqa: N802 if not self._authorized(): return path = urlparse(self.path).path.rstrip("/") if path != f"{PREFIX}/entites": self._json(404, {"data": None, "msg": "接口不存在", "state": 404}) return try: length = int(self.headers.get("Content-Length", "0")) body = json.loads(self.rfile.read(length) or b"[]") if not isinstance(body, list): raise ValueError("请求体必须是事件 id 数组") ids = [int(value) for value in body] except (ValueError, json.JSONDecodeError) as error: self._json(400, {"data": [], "msg": str(error), "state": 400}) return rows = [ENTITY_ROWS[event_id] for event_id in ids if event_id in ENTITY_ROWS] self._json(200, {"data": rows, "msg": "操作成功", "state": 200}) def _task_detail(self, task_id: str) -> None: numeric_id = int(task_id) if task_id.isdigit() else task_id self._json( 200, { "data": { "id": numeric_id, "taskName": "台海商业情报分析", "taskDesc": "收集台海方向重要节点、热点、威胁、战机、目标及衔接任务。", "direction": "T", "taskType": "商业情报", "createTime": "2026-09-28 10:00:00", "updateTime": "2026-09-29 09:00:00", }, "msg": "操作成功", "state": 200, }, ) def _calendar(self, query: dict[str, str]) -> None: direction = query.get("direction", "") keyword = query.get("searchText", "") start = query.get("startDate", "") end = query.get("endDate", "") rows = [ row for row in CALENDAR_ROWS if (not direction or row["direction"] == direction) and _contains(row, keyword, ("title", "description", "nodeName")) and _in_range(row["startTime"], start, end) ] page_num = max(1, int(query.get("pageNum", "1") or 1)) page_size = max(1, int(query.get("pageSize", "10") or 10)) offset = (page_num - 1) * page_size self._json(200, {"data": rows[offset : offset + page_size], "msg": "操作成功", "state": 200, "sumCount": len(rows)}) def _events(self, query: dict[str, str]) -> None: discriminant = query.get("discriminant", "热点") keyword = query.get("keyword", "") start = query.get("startTime", "") end = query.get("endTime", "") direction = query.get("direction", "") direction_label = {"U": "美", "T": "台", "J": "日", "F": "菲"}.get(direction, direction) rows = [ row for row in EVENT_ROWS.get(discriminant, []) if (not direction_label or row["direction"] == direction_label) and _contains(row, keyword, ("m_title", "m_content")) and _in_range(row["m_publish_time_date"], start, end) ] size = max(1, int(query.get("size", "50") or 50)) self._json(200, {"data": rows[:size], "msg": "操作成功", "state": 200}) def _real_network(self, task_id: str) -> None: self._json( 200, { "data": [ { "taskId": f"{task_id}-A01", "connectionType": "打击", "taskName": "重点目标关联处置", "taskDirection": "T", "taskLevel": 1, "taskStatus": 20, "taskContent": "围绕重点目标形成处置任务。", "sourceType": 0, "createTime": "2026-09-29 09:10:00", }, { "taskId": f"{task_id}-S01", "connectionType": "塑势", "taskName": "公开信息正面塑势", "taskDirection": "T", "taskLevel": 2, "taskStatus": 10, "taskContent": "结合热点事件形成公开信息塑势任务。", "sourceType": 1, "createTime": "2026-09-29 09:30:00", }, ], "msg": "操作成功", "state": 200, }, ) def main() -> int: parser = argparse.ArgumentParser(description="商业情报监测接口本地模拟服务") parser.add_argument("--host", default="127.0.0.1") parser.add_argument("--port", type=int, default=18080) args = parser.parse_args() server = ThreadingHTTPServer((args.host, args.port), Handler) print(f"mock monitor API listening on http://{args.host}:{args.port}/consumer", flush=True) try: server.serve_forever() except KeyboardInterrupt: pass finally: server.server_close() return 0 if __name__ == "__main__": raise SystemExit(main())