api.py

from __future__ import annotations

import argparse
import hmac
import json
import os
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from typing import Any

from . import __version__
from .engine import AuditEngine


MAX_BODY_BYTES = 1_000_000
ENGINE = AuditEngine()


def openapi(server_url: str) -> dict[str, Any]:
    return {
        "openapi": "3.0.3",
        "info": {
            "title": "研鉴 EvidenceOps API",
            "version": __version__,
            "description": "引用真实性、论断证据与实验复现要素审计工具。",
        },
        "servers": [{"url": server_url}],
        "components": {
            "securitySchemes": {
                "bearerAuth": {
                    "type": "http",
                    "scheme": "bearer",
                    "bearerFormat": "opaque API token",
                }
            }
        },
        "paths": {
            "/audit": {
                "post": {
                    "operationId": "auditResearchEvidence",
                    "summary": "审计论文或技术报告",
                    "requestBody": {
                        "required": True,
                        "content": {
                            "application/json": {
                                "schema": {
                                    "type": "object",
                                    "required": ["document_text"],
                                    "properties": {
                                        "document_text": {"type": "string", "maxLength": 100000},
                                        "references": {
                                            "type": "array",
                                            "maxItems": 50,
                                            "items": {"type": "string"},
                                        },
                                        "evidence": {
                                            "type": "array",
                                            "maxItems": 100,
                                            "items": {"type": "object"},
                                        },
                                        "live_resolution": {"type": "boolean"},
                                    },
                                }
                            }
                        },
                    },
                    "responses": {
                        "200": {"description": "结构化证据审计报告"},
                        "401": {"description": "认证失败"},
                        "400": {"description": "无效请求"},
                    },
                    "security": [{"bearerAuth": []}],
                }
            }
        },
    }


class Handler(BaseHTTPRequestHandler):
    server_version = "YanjianEvidenceOps/0.1"

    def _send(self, status: int, payload: dict[str, Any]) -> None:
        body = json.dumps(payload, ensure_ascii=False, indent=2).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self) -> None:
        if self.path == "/health":
            self._send(200, {"status": "ok", "version": __version__})
            return
        if self.path == "/openapi.json":
            host = self.headers.get("Host", "127.0.0.1:8765")
            self._send(200, openapi(f"http://{host}"))
            return
        self._send(404, {"error": "not_found"})

    def do_POST(self) -> None:
        if self.path != "/audit":
            self._send(404, {"error": "not_found"})
            return
        configured_token = os.environ.get("YANJIAN_API_TOKEN", "")
        if configured_token:
            supplied = self.headers.get("Authorization", "")
            expected = f"Bearer {configured_token}"
            if not hmac.compare_digest(supplied, expected):
                self._send(401, {"error": "unauthorized"})
                return
        try:
            content_type = self.headers.get("Content-Type", "")
            if "application/json" not in content_type:
                raise ValueError("Content-Type 必须是 application/json")
            length = int(self.headers.get("Content-Length", "0"))
            if length <= 0 or length > MAX_BODY_BYTES:
                raise ValueError("请求正文为空或超过 1MB")
            payload = json.loads(self.rfile.read(length).decode("utf-8"))
            if not isinstance(payload, dict):
                raise ValueError("请求正文必须是 JSON object")
            result = ENGINE.audit(payload)
        except (ValueError, TypeError, json.JSONDecodeError, UnicodeDecodeError) as exc:
            self._send(400, {"error": "invalid_request", "detail": str(exc)})
            return
        self._send(200, result)

    def log_message(self, format: str, *args: object) -> None:
        return


def serve(host: str = "127.0.0.1", port: int = 8765) -> None:
    if host not in {"127.0.0.1", "::1", "localhost"} and not os.environ.get(
        "YANJIAN_API_TOKEN"
    ):
        raise ValueError("非本机监听必须设置 YANJIAN_API_TOKEN")
    server = ThreadingHTTPServer((host, port), Handler)
    print(f"Yanjian EvidenceOps listening on http://{host}:{port}", flush=True)
    server.serve_forever()


def main() -> None:
    parser = argparse.ArgumentParser(description="研鉴 EvidenceOps HTTP API")
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("--port", type=int, default=8765)
    args = parser.parse_args()
    serve(args.host, args.port)


if __name__ == "__main__":
    main()