Algorithm API Wrapper

算法脚本封装为标准 API 工程

目标

把一个散乱或研究型算法工程封装成可部署、可测试、接口清晰的 FastAPI 工程。工作时要持续从用户每次追加提示中抓关键点，并把这些关键点落实到代码结构、接口参数、配置、输出、日志和验证流程中。

工作原则

• 先读参考工程，再动手。重点看 main.py、api/urls.py、conf/config_*.yaml、core/、util/、requirements.txt。
• 先找算法真实调用链。不要把原工程整包照搬成 API；只保留接口实际用到的函数、模块和依赖。
• 传入和输出参数形式默认优先保持原始算法一致。不要为了“标准化”擅自删参数、改字段名、改返回结构。
• 只有用户明确要求“去除”“改为配置”“返回改成某格式”时，才收敛请求参数或改变输出格式。
• 用户每次说“去除”“保留”“改成”“不要生成”时，要追踪到所有相关位置：模型字段、服务封装、配置、README、测试请求、返回结构。
• 返回结构优先保持原算法的成功码、消息、结果对象/路径等语义；用户给出指定格式时，严格匹配用户格式。
• 每个算法任务入口都必须同时使用 print(..., flush=True) 和日志输出两类执行追踪：开始时输出实际传参的具体内容，结束时输出最终返回对象；成功与失败分支都不能遗漏。
• Linux 和 Windows 都要考虑：路径用 Path 处理，接口返回路径可按用户要求统一为 /，部署路径用环境变量覆盖。
• 改动后必须验证：至少运行 python -m compileall，再做导入测试；有样例数据时做一次真实接口或 TestClient 调用，并验证标准输出和日志中都包含传参与最终返回结果。

标准工程结构

在 E:\QHSW\api\_frame 这类承载多个算法工程的统一 API 框架中，core/ 下必须按工程拆分独立目录， 每个工程目录、服务脚本、api/urls.py 中的接口以及 conf/ 下的 YAML 配置一一对应。不要把不同工程的 服务脚本平铺在 core/ 根目录，也不要让多个工程共用一个通用配置文件。

推荐结构：

api_project/
├── main.py                 # FastAPI 入口
├── api/
│   └── urls.py             # 各工程的请求模型、接口及对应服务导入
├── conf/
│   ├── config_probability_service.yaml
│   └── config_xxx_service.yaml
├── core/
│   ├── 01probability_service/
│   │   └── probability_service.py
│   └── NNxxx_service/
│       └── xxx_service.py
├── swagger/                # 本地 Swagger UI 静态资源，构建 API 工程时必须添加
│   ├── css/
│   │   └── swagger-ui.css
│   └── js/
│       └── swagger-ui-bundle.js
├── util/
│   ├── xxx_utils.py        # 日志、校验、坐标范围等辅助函数
│   └── Lib/                # 实际调用到的算法函数脚本
├── logs/                   # 如果用户要求保留日志，运行时生成
├── output/                 # 最终结果
└── requirements.txt

core 的每个子目录只承载一个工程的服务编排脚本；util 存放实际调用的算法函数和辅助函数；无用的 原算法文件要删除。工程目录名遵循现有编号和工程命名，例如 01probability_service；服务脚本去掉编号， 例如 probability_service.py。如果用户或参考工程已经指定编号，必须沿用，不得自行重新编号。

以下映射是强约束：

每增加或迁移一个工程，都要同时确认这四项映射，不能只创建服务目录而遗漏接口或配置。编号前缀只用于 工程目录，不进入服务脚本名和配置文件名。目录名以数字开头时，不能生成语法无效的 from core.01probability_service ...；应沿用参考框架现有的加载方式，或使用 importlib.import_module() 按字符串动态导入，并做一次实际导入验证。

Swagger 资源模板

本 skill 自带标准 Swagger UI 静态资源模板：

assets/swagger/
├── css/
│   └── swagger-ui.css
└── js/
    └── swagger-ui-bundle.js

每次构建 API 工程时，直接把 assets/swagger/ 复制到目标工程根目录的 swagger/。不要临时写占位版 swagger-ui.css 或 swagger-ui-bundle.js，也不要依赖 CDN。复制后验证两个文件存在且不是空文件。

关键点提取

用户的每次提示通常属于以下类型，处理时要定位对应代码面：

当用户最新提示与前面冲突时，以最新提示为准，并检查旧说明是否仍残留在 README 或示例请求中。

封装流程

1. 读取上下文

   • 列出算法工程文件。
   • 读取参考 API 工程的 main.py、api/urls.py、conf/config_*.yaml。
   • 建立现有工程目录、服务脚本、路由和配置文件的对应表，确认新工程应使用的编号和名称。
   • 找算法入口和真实调用链。

2. 识别输入输出

   • 读取样例数据结构，例如 NetCDF 的 dims、coords、data_vars。
   • 明确必需变量、文件数量、文件命名和时间解析方式。
   • 确认输出文件名、输出格式和是否是标准 JSON。

3. 构建 API 骨架

   • main.py 创建 FastAPI app，挂载路由和 CORS。
   • 每次构建 API 工程时都要从本 skill 的 assets/swagger/ 复制完整 swagger/ 文件夹内容， 至少包含 swagger/css/swagger-ui.css 和 swagger/js/swagger-ui-bundle.js。
   • main.py 默认关闭 FastAPI CDN 文档页：FastAPI(..., docs_url=None)，用 SWAGGER_DIR = Path(__file__).resolve().parent / "swagger" 定位静态目录，挂载 StaticFiles 到 /swagger，并用 get_swagger_ui_html 自定义 /docs， 让页面引用 /swagger/js/swagger-ui-bundle.js 和 /swagger/css/swagger-ui.css。
   • api/urls.py 定义 Pydantic 请求模型，并为每个工程提供独立的对应接口。
   • api/urls.py 中所有会出现在 OpenAPI/Swagger 的路由装饰器都必须添加 summary，例如 @app.post("/bump-index/", tags=[...], summary="飞机颠簸")。 summary 使用简短中文业务名称，和接口实际算法能力一致。
   • core/NNxxx_service/xxx_service.py 调用该工程的算法流程；不同工程必须放在不同子目录。
   • conf/config_xxx_service.yaml 管理该工程的部署和算法配置；不同工程必须使用不同 YAML 文件。
   • 接口、服务目录、服务脚本和 YAML 配置必须按工程一一对应；以 probability 为例，使用 core/01probability_service/probability_service.py、api/urls.py 中的 probability 接口和 conf/config_probability_service.yaml。
   • README.md 目录树和启动说明必须写明 swagger/ 文件夹以及 /docs 访问地址。

4. 设计请求参数

   • 先按原算法的 JSON 配置、CLI 参数或函数签名设计请求体，保持字段名、类型和默认语义一致。
   • 区分三类参数：算法业务参数、部署环境参数、内部临时参数。
   • 算法业务参数默认保留在请求体；部署环境参数可放配置文件或环境变量；内部临时参数不要暴露给 API。
   • 用户明确要求减少传参时，再把固定参数迁到配置文件或环境变量。
   • 如果用户要求删除旧字段，可设置 extra="forbid"，避免旧参数被默默接受。

5. 迁移函数和清理源码

   • 先在 core/ 下为当前工程创建独立目录，再把服务编排入口放入该目录；不得写到 core/ 根目录。
   • 把 API 实际调用的函数脚本迁到 util/。
   • 保留间接依赖，例如算法链路里的工具函数、插值函数、HDF/GeoJSON 转换函数。
   • 删除未使用的 SLA、绘图、示例 JSON、旧 README、旧入口等文件。
   • 修正导入路径，如 from util.Lib.xxx import ...；数字开头的工程目录使用参考框架既有加载方式或 importlib.import_module()，不得写无效的静态 import。

6. 输出和日志

   • 默认保持原算法输出语义，例如返回状态码、消息和算法结果列表/文件路径。
   • 若原算法输出为文件路径列表，API 可返回列表；若原算法输出对象，API 可返回对象。
   • 文件型算法接口默认将最终响应收敛成 code、message、outpath；用户明确指定其他格式时，严格按用户格式。
   • 在 core/NNxxx_service/xxx_service.py 的实际算法任务入口记录执行追踪，不能只在 api/urls.py 记录：
     • 调用算法前，把经过接口校验、即将传给算法的完整业务参数序列化后，同时写入 print(..., flush=True) 和 LOGGER.info(...)。
     • 返回前先构造唯一的最终响应对象，再把该对象同时写入 print 和日志，最后 return response。
     • 成功、已知业务异常和未知异常都必须走最终响应输出；未知异常仍使用 LOGGER.exception 保留堆栈。
     • 控制台与日志使用相同 JSON 内容，设置 ensure_ascii=False 和 default=str，保证中文、Path、日期等值可读且不会因序列化失败影响任务。
     • 业务字段、阈值、输入输出路径等具体传参不得用省略号或仅输出字段名。若请求确实包含密码、令牌或密钥，只对这些敏感字段脱敏，其他内容完整输出。
     • 请求日志和响应日志各输出一次，避免路由层与服务层重复打印。并发任务有 requestId 时，把它放在日志前缀中便于关联。
   • 用户要求的文件型接口最终返回格式例如：

{
  "code": 200,
  "message": "运行成功",
  "outpath": "E:/path/to/result.json"
}

• 日志如果保留，写到配置的 LOG_DIR；是否把日志路径放进接口返回，取决于原算法契约或用户要求。
• 不需要临时配置文件时，不生成 tmp/*.json。

服务层参考写法：

import json
import logging
from pathlib import Path
from typing import Any

LOGGER = logging.getLogger(__name__)


def _emit_task_trace(event: str, payload: dict[str, Any]) -> None:
    body = json.dumps(payload, ensure_ascii=False, default=str)
    line = f"{event} | {body}"
    print(line, flush=True)
    LOGGER.info("%s", line)


def run_xxx_task(request: dict[str, Any]) -> dict[str, Any]:
    _emit_task_trace("algorithm.request", request)
    try:
        result_path = run_algorithm(request)
        response = {
            "code": 200,
            "message": "任务运行成功",
            "outpath": Path(result_path).as_posix(),
        }
    except XxxBusinessError as exc:
        LOGGER.exception("算法任务业务失败")
        response = {"code": 500, "message": str(exc), "outpath": None}
    except Exception as exc:
        LOGGER.exception("算法任务运行失败")
        response = {
            "code": 500,
            "message": f"任务运行失败：{exc}",
            "outpath": None,
        }
    _emit_task_trace("algorithm.response", response)
    return response

示例中的 run_algorithm 和 XxxBusinessError 要替换成当前工程的真实调用与异常类型；不要照抄占位符。

7. 跨平台处理

   • 内部路径用 pathlib.Path。
   • 默认输出路径优先可用环境变量覆盖。
   • Windows 示例可用 E:/...；Linux 示例用 /data/...。
   • 服务器部署时提醒配置输入数据目录和输出目录。

8. 验证

   • 运行：

python -m compileall api conf core util

• 验证 Swagger 静态资源存在：

python -c "from pathlib import Path; assert Path('swagger/css/swagger-ui.css').is_file(); assert Path('swagger/js/swagger-ui-bundle.js').is_file(); print('swagger ok')"

• 做导入检查；若工程目录有数字前缀，按实际路由使用的动态加载方式检查：

python -c "import importlib; m = importlib.import_module('core.01probability_service.probability_service'); print('imports ok')"

• 检查 api/urls.py 的当前接口只调用对应工程的服务，并读取对应的 conf/config_xxx_service.yaml。

• 有样例数据时，用 TestClient 或本地端口调一次接口。

• 分别覆盖成功和失败分支，捕获标准输出与日志，确认两处都能看到完整请求内容和最终响应中的 code、message、outpath，且每个任务只输出一次请求追踪和一次响应追踪。

• 验证输出文件能被 json.load 读取；若输出含 np.float64(...) 或 NaN，修正转换逻辑。

配置文件写法

每个工程使用独立的 conf/config_<service_name>.yaml。例如 core/01probability_service/probability_service.py 只读取 conf/config_probability_service.yaml；其他工程 分别读取自己的 config_xxx_service.yaml，不得退化成共享的 config.yaml 或共享 Python 配置模块。

YAML 按算法需要组织字段，字段名和默认语义优先保持原算法一致，例如：

data_type: H2D_REG
output_path: output/probability_service
log_dir: logs/probability_service

服务脚本用 Path 从框架根目录定位自己的 YAML，不依赖当前工作目录；需要环境变量覆盖时，只覆盖当前 工程对应的配置值，并沿用参考工程已有的加载规则。若参考框架没有加载器，可采用等价的最小实现：

import os
from pathlib import Path

import yaml

FRAME_ROOT = Path(__file__).resolve().parents[2]
CONFIG_PATH = Path(
    os.environ.get(
        "PROBABILITY_SERVICE_CONFIG",
        FRAME_ROOT / "conf" / "config_probability_service.yaml",
    )
)

with CONFIG_PATH.open("r", encoding="utf-8") as file:
    CONFIG = yaml.safe_load(file) or {}

路径层级必须按实际服务文件位置复核，不能机械照抄 parents[2]。同时把 YAML 解析依赖加入工程依赖清单， 除非参考工程已有等价配置加载机制。

接口函数说明

所有纳入 Swagger 的接口路由都要在装饰器里写 summary，避免 Swagger 页面只展示函数名。 summary 要短而明确，例如“飞机积冰”“飞机颠簸”“版本信息”；如果接口不进入 OpenAPI， 例如自定义 /docs 且 include_in_schema=False，可以不写。

在 async def xxx(param: InputModel): 下面写简短 docstring，说明：

• 这个接口做什么。
• 请求体每个字段含义。
• 哪些字段来自原算法输入，哪些字段被迁移到配置文件或环境变量。
• 返回值如何对应原算法输出。

示例：

async def eddy_h2d(param: H2DEddyRequest):
    """
    H2D 中尺度涡检测接口。

    - inputPaths: H2D 输入文件列表，至少包含两个 NetCDF 文件。
      通常第一个文件包含 U 向流速变量 CUR，第二个文件包含 V 向流速变量 CVR。
    - 其他字段是否放入请求体，以原算法配置和用户要求为准。
    - 返回 results 与原算法输出保持一致；用户指定格式时按指定格式返回。
    """

常见坑

• 原算法主入口会导入不相关重依赖，例如 SLA 分支的 py_eddy_tracker。只封装当前数据类型所需链路，避免接口启动被无关依赖卡住。
• 旧算法输出可能不是标准 JSON，例如拼接出 np.float64(...)。必须改为 json.dump 并显式转 float。
• SciPy 版本可能导致 integrate.cumtrapz 缺失，可兼容为 integrate.cumulative_trapezoid。
• 有些候选涡边界为空时原算法会异常，应增加空数组保护，让算法跳过或走兜底逻辑。
• 不要过早“优化”接口参数。先保持原算法契约，等用户确认后再迁移到配置或环境变量。
• 不要让 Swagger 示例与实际请求体不一致；每次参数变化都同步 README 和测试请求。
• 不要在多个异常分支里直接 return，否则容易漏掉最终结果的 print 或日志；先统一赋值给 response，追踪后再返回。