# Algorithm API Wrapper

> 将已有算法脚本封装成标准 FastAPI API 调用工程。用于用户要求参考既有 API 工程、整理算法脚本、生成接口服务、调整传参方式、迁移工具函数、删除无用源码、配置环境变量、统一返回格式、保留日志和验证 Linux/Windows 部署兼容性时。

- Skill: `codeboy-bot/algorithm-api-wrapper` (Agent Skill, multi-file: 3 files)
- Install (CLI): `npx skillmds@latest add codeboy-bot/algorithm-api-wrapper`
- Raw SKILL.md: https://api.skillmd.com/api/skills/codeboy-bot/algorithm-api-wrapper/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Integrations & APIs
- Author: Codeboy-bot (https://skillmd.com/u/codeboy-bot)
- Updated: 2026-09-22
- Page: https://skillmd.com/skills/codeboy-bot/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/` 根目录，也不要让多个工程共用一个通用配置文件。

推荐结构：

```text
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`。如果用户或参考工程已经指定编号，必须沿用，不得自行重新编号。

以下映射是强约束：

| 工程目录 | 服务脚本 | 路由位置 | 独立配置 |
| --- | --- | --- | --- |
| `core/01probability_service/` | `probability_service.py` | `api/urls.py` 中 probability 对应接口 | `conf/config_probability_service.yaml` |
| `core/NNxxx_service/` | `xxx_service.py` | `api/urls.py` 中 xxx 对应接口 | `conf/config_xxx_service.yaml` |

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

## Swagger 资源模板

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

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

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

## 关键点提取

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

| 用户提示 | 要落地的位置 |
| --- | --- |
| 参考某个 API 工程 | 读取参考工程入口、路由、配置、返回格式、依赖 |
| 输入数据是什么 | 读取样例数据变量、维度、坐标、文件命名规则 |
| 参数保持原算法一致 | 读取原算法 JSON/CLI/函数签名，按原始字段设计 Pydantic 模型 |
| 参数去除/改为配置 | 仅在用户明确要求时修改 Pydantic 模型、服务封装、配置文件、README、测试请求 |
| 写成环境变量 | 保持每个工程的独立 YAML 配置边界，再按参考工程方式支持环境变量覆盖 |
| 新增/迁移一个工程 | 同步创建 `core/NNxxx_service/xxx_service.py`、`api/urls.py` 对应接口和 `conf/config_xxx_service.yaml` |
| 不生成 tmp/log/output | 分清临时配置、日志、最终结果；按用户最新要求保留或删除 |
| 返回格式修改 | 仅在用户明确指定时修改 API 层返回，不改变算法内部结果结构 |
| 输出执行传参和结果 | 在服务层任务入口同时用 `print` 与 `LOGGER.info` 输出完整请求和最终响应 |
| Linux 支持 | 移除硬编码 Windows 路径默认依赖，保留环境变量覆盖 |
| 删除无用源码 | 根据导入链保留实际依赖，迁移后编译和导入验证 |

当用户最新提示与前面冲突时，以最新提示为准，并检查旧说明是否仍残留在 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` 时，把它放在日志前缀中便于关联。
   - 用户要求的文件型接口最终返回格式例如：

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

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

服务层参考写法：

```python
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. **验证**
   - 运行：

```bash
python -m compileall api conf core util
```

   - 验证 Swagger 静态资源存在：

```bash
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')"
```

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

```bash
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 按算法需要组织字段，字段名和默认语义优先保持原算法一致，例如：

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

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

```python
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，说明：

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

示例：

```python
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`，追踪后再返回。