guanlan-src/app_algorithmModel/app_algorithmModel_guanlan/service/app.py

# -*- coding: utf-8 -*-
r"""算法服务（FastAPI）：把 `app_algorithmModel` 的既有函数包成 HTTP 接口 + OpenAPI。

启动：
    python scripts/algorithm_service.py --host 127.0.0.1 --port 18050
    （或 `python -m app_algorithmModel.app_algorithmModel_guanlan.service.app --port 18050`）

接口（全部 GET，只读）：
    /healthz                     存活与版本
    /api/endpoints               端点自述（登记表）
    /api/{端点名}                 见 service/registry.py（判级矩阵/可靠性/可用率/融合面/变桨面/曲线…）
    /docs · /openapi.json        Swagger UI 与 OpenAPI 文档（FastAPI 自带）

**只读保证**：带写盘副作用的参数在登记表里被固定成只读值（如 curves 的 write=False），
不会因为调一次接口改了盘上产物。
"""
from __future__ import annotations

import argparse
import json
import pathlib
import sys
import time

_HERE = pathlib.Path(__file__).resolve()
try:                                              # 走公共层的安装根（包结构异常时按位置上跳）
    from app_common.app_common_guanlan.api import install_root as _install_root
except ImportError:                               # pragma: no cover
    def _install_root(_f):
        return _HERE.parents[4]

ROOT = _install_root(__file__)
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

from . import jsonable, registry                                             # noqa: E402
from .deps import ensure_fastapi                                             # noqa: E402

fastapi, uvicorn = ensure_fastapi()
from fastapi import FastAPI, HTTPException, Query                            # noqa: E402
from fastapi.responses import JSONResponse, PlainTextResponse                # noqa: E402

try:                                     # 版本走公共层公开面（模块边界：不直连旧路径壳）
    from app_common.app_common_guanlan.api import version as V
    _VER = V.VERSION
except Exception:                                                            # noqa: BLE001
    _VER = 'unknown'

CACHE_TTL = 300.0            # 秒；判级矩阵这类重算是"按窗口"的，缓存与页面的按窗缓存同源
_cached: dict[str, tuple[float, str]] = {}


def create_app() -> FastAPI:
    app = FastAPI(title='观澜 · 算法服务', version=_VER,
                  description='算法层的 HTTP 封装（P10）：判级矩阵、可靠性、可用率、融合面、变桨面、曲线检查。'
                              '数据来源与页面同源，只读不写盘。',
                  docs_url='/docs', redoc_url='/redoc', openapi_url='/openapi.json')

    @app.get('/healthz', summary='存活与版本')
    def healthz():
        return {'ok': True, 'version': _VER, 'service': 'guanlan-algorithm',
                'endpoints': len(registry.ENDPOINTS), 'time': time.strftime('%Y-%m-%d %H:%M:%S')}

    @app.get('/api/endpoints', summary='端点自述')
    def endpoints():
        return {'count': len(registry.ENDPOINTS), 'items': registry.spec_list()}

    @app.get('/api/{name}', summary='算法端点（见 /api/endpoints）')
    def algorithm(name: str, span: str | None = Query(None), since: str | None = Query(None),
                  month_from: str | None = Query(None), all_turbines: bool | None = Query(None),
                  ttl: float | None = Query(None, description='缓存秒数（默认 300；0 = 不走缓存）')):
        spec = registry.ENDPOINTS.get(name)
        if not spec:
            raise HTTPException(status_code=404,
                                detail=f'未登记端点 {name!r}；可用: {", ".join(sorted(registry.ENDPOINTS))}')
        params = {k: v for k, v in dict(span=span, since=since, month_from=month_from,
                                        all_turbines=all_turbines).items() if v is not None}
        key = name + '|' + json.dumps(params, sort_keys=True)
        ttl = CACHE_TTL if ttl is None else max(0.0, ttl)
        now = time.time()
        if ttl and key in _cached:
            ts, body = _cached[key]
            if now - ts < ttl:
                return JSONResponse(content=json.loads(body), headers={'x-guanlan-cache': 'hit'})
        t0 = time.time()
        try:
            payload = registry.call(name, params)
        except Exception as e:                                               # noqa: BLE001
            raise HTTPException(status_code=500, detail=f'{type(e).__name__}: {e}') from e
        text = jsonable.dumps(payload)
        if ttl:
            _cached[key] = (now, text)
        return JSONResponse(content=json.loads(text),
                            headers={'x-guanlan-cache': 'miss', 'x-guanlan-seconds': '%.2f' % (time.time() - t0)})

    @app.get('/api/{name}/raw', summary='同一端点的确定性 JSON 文本（对拍用）',
             response_class=PlainTextResponse)
    def algorithm_raw(name: str):
        if name not in registry.ENDPOINTS:
            raise HTTPException(status_code=404, detail=f'未登记端点 {name!r}')
        return jsonable.dumps(registry.call(name, {}))

    return app


app = create_app()


def main(argv: list[str] | None = None) -> int:
    ap = argparse.ArgumentParser(description='观澜 · 算法服务（FastAPI）')
    ap.add_argument('--host', default='127.0.0.1')
    ap.add_argument('--port', type=int, default=18050)
    ap.add_argument('--reload', action='store_true')
    ap.add_argument('--log-level', default='info')
    a = ap.parse_args(argv)
    # 统一日志格式（用户令 2）：stdout/stderr 每行加 `时间戳 级别 组件` 前缀 ——
    # 与其它服务（serve/detail/cms/gateway/static_server）同一套设施。不这么做，日志门禁会报
    # "logs/xxx.log 末尾 N 行都不符合统一格式"（2026-09-29 实逮）。
    try:
        from app_common.app_common_guanlan.api import logfile as _lf
        _lf.prefix_stdout('algorithm')
    except Exception:                                                        # noqa: BLE001
        pass
    print(f'观澜 · 算法服务 v{_VER} → http://{a.host}:{a.port}/docs （{len(registry.ENDPOINTS)} 个端点）', flush=True)
    uvicorn.run(app if not a.reload else 'app_algorithmModel.app_algorithmModel_guanlan.service.app:app',
                host=a.host, port=a.port, reload=a.reload, log_level=a.log_level)
    return 0


if __name__ == '__main__':
    sys.exit(main())