AiDOPWarehouse/doc/db/mdp/mock_api/mock_receipt_api.py

"""
双模式契约测试用 Mock API（S5 采购收货样板）。

严格对齐 MdpApiPullExecutor.cs 现有实现：
- GET，单次请求，不分页；
- Bearer Token 鉴权（api_auth_type='TOKEN'，token 见环境变量 MOCK_TOKEN）；
- 响应数组位于 data.list（对应 mdp_entity.response_data_path='data.list'）；
- 每行去重键 bizKey（对应 dedup_key_path='bizKey'）；
- 可选 ?cursor=<上批最后一行 bizKey>，用于演示增量：仅返回 bizKey 大于该游标的行。

启动：
    pip install -r requirements.txt
    set MOCK_TOKEN=uat-mock-token   # bash: export MOCK_TOKEN=uat-mock-token
    uvicorn mock_receipt_api:app --host 127.0.0.1 --port 8018
"""
import json
import os
from pathlib import Path

from fastapi import FastAPI, Header, HTTPException, Query

app = FastAPI(title="AIDOP Mock Receipt API", version="1.0.0")

_TOKEN = os.environ.get("MOCK_TOKEN", "uat-mock-token")
_SAMPLE_PATH = Path(__file__).with_name("sample_receipt.json")


def _load_rows() -> list[dict]:
    with _SAMPLE_PATH.open("r", encoding="utf-8") as f:
        return json.load(f)


def _check_auth(authorization: str | None) -> None:
    expected = f"Bearer {_TOKEN}"
    if authorization != expected:
        raise HTTPException(status_code=401, detail="invalid or missing bearer token")


@app.get("/api/receipt")
def get_receipt(
    cursor: str | None = Query(default=None, description="上批最后一行 bizKey（增量游标）"),
    authorization: str | None = Header(default=None),
):
    _check_auth(authorization)
    rows = _load_rows()
    # 按 bizKey 稳定排序，保证「最后一行 bizKey」作为游标可复现
    rows.sort(key=lambda r: r["bizKey"])
    if cursor:
        rows = [r for r in rows if r["bizKey"] > cursor]
    return {"code": 0, "message": "ok", "data": {"list": rows}}


@app.get("/health")
def health():
    return {"status": "UP"}