proposa-ai/scripts/test_api_callback.py

"""向对接系统发送一条与 Proposa API 当前格式一致的测试回调。

直接运行只预览 JSON；确认内容后增加 ``--send`` 才会真正发送。
最常修改的参数集中在下方“可编辑配置”区域。
"""

from __future__ import annotations

import argparse
import json
import sys
from dataclasses import dataclass
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen


# ============================ 可编辑配置 ============================
CALLBACK_URL = "http://121.43.55.7:10026/shenqin/tender/callback"
CALLBACK_STATUS = "completed"  # 可改为 "failed"
TXB_ID = "4c9e748b-331f-468b-a735-d6748bc50cfd"
REQUEST_ID = "test_callback_request_001"
RESULT_PATH = "output/api_jobs/test_callback_request_001/测试招标文件.docx"
FINAL_REVIEW_URL = (
    "http://127.0.0.1:8000/api/v1/jobs/test_callback_request_001/file"
)
ERROR_MESSAGE = "测试失败：Step1-6 处理进程异常退出（退出码 7）"
TIMEOUT_SECONDS = 30

# 文件名、大小和 SHA-256 均为测试数据，可按需直接修改。
FILES = {
    "TENDER_FILE": {
        "filename": "测试招标文件.pdf",
        "size": 123456,
        "sha256": "1" * 64,
    },
    "PROCUREMENT_FILE": {
        "filename": "测试采购需求.docx",
        "size": 234567,
        "sha256": "2" * 64,
    },
    "REFERENCE_BID": {
        "filename": "测试参考投书.docx",
        "size": 345678,
        "sha256": "3" * 64,
    },
}
# ===================================================================


@dataclass(frozen=True)
class CallbackResponse:
    status: int
    body: str


def build_callback_payload(status: str = CALLBACK_STATUS) -> dict:
    """按 src/http_api.py 当前成功/失败回调字段生成测试载荷。"""
    if status not in {"completed", "failed"}:
        raise ValueError("status 只能是 completed 或 failed")

    payload = {
        "request_id": REQUEST_ID,
        "txbId": TXB_ID,
        "status": status,
        "resultPath": RESULT_PATH,
        "files": FILES,
    }
    if status == "completed":
        payload["final_review_url"] = FINAL_REVIEW_URL
    else:
        payload["resultPath"] = f"error: {ERROR_MESSAGE}"
        payload["error"] = ERROR_MESSAGE
    return payload


def send_callback(url: str, payload: dict) -> CallbackResponse:
    """使用与正式 API 相同的 POST JSON 编码发送，并返回接收端响应。"""
    body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
    request = Request(
        url,
        data=body,
        headers={"Content-Type": "application/json; charset=utf-8"},
        method="POST",
    )
    try:
        with urlopen(request, timeout=TIMEOUT_SECONDS) as response:
            response_body = response.read().decode("utf-8", errors="replace")
            return CallbackResponse(status=response.status, body=response_body)
    except HTTPError as exc:
        response_body = exc.read().decode("utf-8", errors="replace")
        return CallbackResponse(status=exc.code, body=response_body)
    except URLError as exc:
        raise RuntimeError(f"无法连接回调地址: {exc.reason}") from exc


def main() -> int:
    if hasattr(sys.stdout, "reconfigure"):
        sys.stdout.reconfigure(encoding="utf-8")
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--send", action="store_true", help="实际发送；不传时只预览")
    parser.add_argument("--url", default=CALLBACK_URL, help="临时覆盖回调 URL")
    parser.add_argument(
        "--status",
        choices=("completed", "failed"),
        default=CALLBACK_STATUS,
        help="生成成功或失败回调",
    )
    args = parser.parse_args()
    payload = build_callback_payload(args.status)

    print(f"回调 URL: {args.url}")
    print(json.dumps(payload, ensure_ascii=False, indent=2))
    if not args.send:
        print("\n当前仅预览，确认后增加 --send 才会发送。")
        return 0

    response = send_callback(args.url, payload)
    print(f"\nHTTP 状态: {response.status}")
    print(f"响应内容: {response.body or '<空>'}")
    business_ok = True
    try:
        response_json = json.loads(response.body)
    except (json.JSONDecodeError, TypeError):
        response_json = None
    if isinstance(response_json, dict) and "code" in response_json:
        business_code = response_json["code"]
        business_ok = str(business_code) in {"0", "200"}
        print(f"业务码: {business_code}（{'成功' if business_ok else '失败'}）")
    return 0 if 200 <= response.status < 300 and business_ok else 1


if __name__ == "__main__":
    raise SystemExit(main())