mtc/api.py

"""轻量数据查询接口，独立于 mqttc 进程运行。

启动: uvicorn api:app --host 0.0.0.0 --port 5900
"""
import os
import sys

CUR_DIR = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, CUR_DIR)
import conf

from fastapi import FastAPI, Query, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from sqlalchemy import create_engine, text
from datetime import datetime, timedelta
from typing import Optional

engine = create_engine(conf.DATABASE_URL, pool_size=5, max_overflow=10, pool_pre_ping=True)

ALLOWED_TABLES = {
    'mplogx', 'mplogx_xzzl', 'mplogx_tlxn',
    'mplogx_hknf_l1', 'mplogx_hknf_l2', 'mplogx_hknf_l3',
    'mplogx_wengfu',
}

app = FastAPI(title="MTC 数据查询接口", version="1.0.0")

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)


@app.get("/api/latest")
def get_latest(
    table: str = Query(..., description="表名，如 mplogx_wengfu"),
    mpoint_id: str = Query(..., description="测点ID"),
    limit: int = Query(1, ge=1, le=100),
):
    """获取指定测点的最新数据"""
    if table not in ALLOWED_TABLES:
        raise HTTPException(400, f"不允许的表名: {table}")
    sql = text(f"""
        SELECT timex, mpoint_id, val_float, val_str
        FROM "{table}"
        WHERE mpoint_id = :mpoint_id
        ORDER BY timex DESC
        LIMIT :limit
    """)
    with engine.connect() as conn:
        rows = conn.execute(sql, {"mpoint_id": mpoint_id, "limit": limit}).fetchall()
    return [
        {"timex": r.timex.isoformat(), "mpoint_id": r.mpoint_id,
         "val_float": r.val_float, "val_str": r.val_str}
        for r in rows
    ]


@app.get("/api/history")
def get_history(
    table: str = Query(..., description="表名"),
    mpoint_id: str = Query(..., description="测点ID"),
    start: str = Query(..., description="起始时间 yyyy-MM-dd HH:mm:ss"),
    end: Optional[str] = Query(None, description="结束时间，默认当前"),
    limit: int = Query(1000, ge=1, le=10000),
):
    """按时间范围查询测点历史数据"""
    if table not in ALLOWED_TABLES:
        raise HTTPException(400, f"不允许的表名: {table}")
    try:
        start_dt = datetime.strptime(start, "%Y-%m-%d %H:%M:%S")
    except ValueError:
        raise HTTPException(400, "start 格式错误，需要 yyyy-MM-dd HH:mm:ss")
    end_dt = datetime.now()
    if end:
        try:
            end_dt = datetime.strptime(end, "%Y-%m-%d %H:%M:%S")
        except ValueError:
            raise HTTPException(400, "end 格式错误，需要 yyyy-MM-dd HH:mm:ss")

    sql = text(f"""
        SELECT timex, mpoint_id, val_float, val_str
        FROM "{table}"
        WHERE mpoint_id = :mpoint_id
          AND timex >= :start AND timex < :end
        ORDER BY timex
        LIMIT :limit
    """)
    with engine.connect() as conn:
        rows = conn.execute(sql, {
            "mpoint_id": mpoint_id, "start": start_dt, "end": end_dt, "limit": limit
        }).fetchall()
    return [
        {"timex": r.timex.isoformat(), "mpoint_id": r.mpoint_id,
         "val_float": r.val_float, "val_str": r.val_str}
        for r in rows
    ]


@app.get("/api/mpoints")
def list_mpoints(
    table: str = Query(..., description="表名"),
    keyword: Optional[str] = Query(None, description="模糊搜索测点ID"),
    limit: int = Query(200, ge=1, le=1000),
):
    """列出表内的测点ID（去重）"""
    if table not in ALLOWED_TABLES:
        raise HTTPException(400, f"不允许的表名: {table}")
    if keyword:
        sql = text(f"""
            SELECT DISTINCT mpoint_id FROM "{table}"
            WHERE mpoint_id LIKE :kw
            ORDER BY mpoint_id LIMIT :limit
        """)
        params = {"kw": f"%{keyword}%", "limit": limit}
    else:
        sql = text(f"""
            SELECT DISTINCT mpoint_id FROM "{table}"
            ORDER BY mpoint_id LIMIT :limit
        """)
        params = {"limit": limit}
    with engine.connect() as conn:
        rows = conn.execute(sql, params).fetchall()
    return [r.mpoint_id for r in rows]


@app.get("/api/count")
def get_count():
    """查询每个 mplogx 表的总条数"""
    with engine.connect() as conn:
        tables = conn.execute(text("""
            SELECT tablename FROM pg_tables
            WHERE schemaname = current_schema() AND tablename LIKE 'mplogx%'
            ORDER BY tablename
        """)).fetchall()
        if not tables:
            return {"tables": [], "total": 0}
        results = []
        total = 0
        for (tablename,) in tables:
            count = conn.execute(text(
                f'SELECT COUNT(*) FROM "{tablename}"'
            )).scalar()
            results.append({"table_name": tablename, "row_count": count})
            total += count
    return {"tables": results, "total": total}


@app.get("/api/stats")
def get_stats():
    """各 mplogx 表的统计信息（复用 mqttc 中的逻辑）"""
    with engine.connect() as conn:
        tables = conn.execute(text("""
            SELECT tablename FROM pg_tables
            WHERE schemaname = current_schema() AND tablename LIKE 'mplogx%'
            ORDER BY tablename
        """)).fetchall()
        if not tables:
            return {"tables": [], "summary": {}}
        parts = []
        for (tablename,) in tables:
            parts.append(f"""
                SELECT '{tablename}' AS table_name,
                       COUNT(DISTINCT mpoint_id)::bigint AS point_count_1h,
                       approximate_row_count('{tablename}')::numeric AS approximate_row_count
                FROM "{tablename}"
                WHERE timex >= now() - interval '1 hour' AND timex < now()
            """)
        union_sql = "\nUNION ALL\n".join(parts)
        rows = conn.execute(text(f"""
            SELECT table_name, point_count_1h, approximate_row_count
            FROM ({union_sql}) AS t ORDER BY table_name
        """)).fetchall()

    result = []
    for table_name, point_count_1h, approx in rows:
        result.append({
            "table_name": table_name,
            "point_count_1h": int(point_count_1h or 0),
            "approximate_row_count": int(approx or 0),
        })
    total_points = sum(r["point_count_1h"] for r in result)
    total_rows = sum(r["approximate_row_count"] for r in result)
    return {
        "tables": result,
        "summary": {
            "point_count_1h_total": total_points,
            "approximate_row_count_total": total_rows,
            "approximate_row_count_total_yi": round(total_rows / 100000000, 2),
        },
    }