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
from typing import Any

_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, Request                            # noqa: E402
from fastapi.responses import JSONResponse, PlainTextResponse, Response                # 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.post('/api/ask_run', summary='问答管线（同步跑一次，返回末态）')
    def ask_run(payload: dict | None = None):
        """P12：搬运自详情层 `_ask_worker`；详情层是"线程+全局状态"，这里同步返回末态。"""
        from . import ask_views
        p = payload or {}
        return ask_views.ask_run(q=p.get('q', ''), model=p.get('model', 'qwen3:8b'),
                                 cat=p.get('cat', ''), xrev=bool(p.get('xrev', False)),
                                 lang=p.get('lang', 'zh'))


    @app.post('/api/rpt_export_raw', summary='报告导出（POST：可带前端渲染的图片，保证与页面一致）')
    def rpt_export_raw_post(payload: dict | None = None):
        p = payload or {}
        from . import report_views
        data, mime, fn = report_views.export_bytes(
            fmt=str(p.get('fmt') or 'docx'), blocks=str(p.get('blocks') or ''), aud=str(p.get('aud') or 'in'),
            win=str(p.get('win') or '2026年'), narr=p.get('narr'), images=p.get('images') or None)
        from urllib.parse import quote
        return Response(content=data, media_type=mime,
                        headers={'Content-Disposition': "attachment; filename*=UTF-8''" + quote(fn)})


    @app.get('/api/rpt_export_raw', summary='报告导出（docx/xlsx 字节）')
    def rpt_export_raw(fmt: str = 'docx', blocks: str = '', aud: str = 'in', win: str = '2026年',
                       narr: str | None = None):
        """报告导出：返回字节（非 JSON）；组装逻辑见 service/report_views.py。"""
        from urllib.parse import quote
        from . import report_views
        data, mime, fn = report_views.export_bytes(fmt=fmt, blocks=blocks, aud=aud, win=win, narr=narr)
        return Response(content=data, media_type=mime,
                        headers={'Content-Disposition': "attachment; filename*=UTF-8''" + quote(fn)})


    @app.get('/api/{name}', summary='算法端点（见 /api/endpoints）')
    def algorithm(name: str, request: Request,
                  ttl: float | None = Query(None, description='缓存秒数（默认 300；0 = 不走缓存）')):
        """通用取数：**按登记表声明的参数**从查询串透传（P12 起不再写死参数名）。

        实逮：原先把参数写死成 span/since/month_from/all_turbines，新增端点（如 ont_list 的 type/t/limit）
        就只能改代码；改为按 spec 解析后，登记表成为唯一需要维护的清单。
        """
        spec = registry.ENDPOINTS.get(name)
        if not spec:
            raise HTTPException(status_code=404,
                                detail=f'未登记端点 {name!r}；可用: {", ".join(sorted(registry.ENDPOINTS))}')
        params: dict[str, Any] = {}
        for qname, (typ, fname, fixed) in (spec.get('args') or {}).items():
            if fixed is not None:
                continue
            if qname not in request.query_params:
                continue
            raw = request.query_params.get(qname)
            try:
                params[fname] = raw if typ == 'str' else (int(raw) if typ == 'int' else raw.lower() in ('1', 'true', 'yes'))
            except Exception:                                            # noqa: BLE001
                raise HTTPException(status_code=400, detail=f'参数 {qname} 类型应为 {typ}')
        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())