ai-agent-admin/backend-fastapi/online_dev/report_manager/data_api.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""报表数据与预览 API"""
import base64
import logging
import re
from typing import Any, Dict

import httpx
from fastapi import APIRouter, Depends, File, HTTPException, Query, Request, UploadFile
from fastapi.responses import StreamingResponse
from sqlalchemy.ext.asyncio import AsyncSession

from app.database import get_db
from app.config import settings
from core.file_manager.service import FileManagerService
from online_dev.report_manager.exceptions import ReportServiceException
from online_dev.report_manager.schema import (
    ReportDownImgIn,
    ReportImportExcelOut,
    ReportPreviewIn,
    ReportPreviewOut,
    ReportUploadOut,
)
from online_dev.report_manager.service import ReportService
from online_dev.report_manager.version_service import ReportVersionService
from online_dev.report_manager.dataset_bridge import ReportDatasetBridge
from online_dev.report_manager.engine.chart_data import build_chart_data
from online_dev.report_manager.engine.convert import transform
from online_dev.report_manager.engine.expression_eval import detect_expression_cycles
from online_dev.report_manager.engine.export_excel import snapshot_to_xlsx_bytes
from online_dev.report_manager.engine.export_pdf import snapshot_to_pdf_bytes
from online_dev.report_manager.engine.import_excel import parse_excel_to_grid
from online_dev.report_manager.engine.watermark import build_watermark_payload
from online_dev.report_manager.engine.convert_lookup import build_lookup_cache_from_db
from online_dev.report_manager.engine.parameter_resolver import (
    build_system_params,
    merge_preview_params,
)
from online_dev.report_manager.engine.preview_guard import collect_preview_warnings

logger = logging.getLogger(__name__)

router = APIRouter(prefix="/report/data", tags=["报表数据"])


def _build_file_access_url(file_obj) -> str:
    if file_obj.url:
        return file_obj.url
    base_url = getattr(settings, "BASE_URL", "http://localhost:8000")
    if file_obj.storage_type == "local" and file_obj.storage_path:
        return f"{base_url}/api/file_manager/file/download?path={file_obj.storage_path}"
    return f"{base_url}/api/file_manager/url/{file_obj.id}"


def _flatten_query_list(query_list: list) -> list:
    """兼容 JNPF 按 sheet 包装与 ZQ 扁平 queryList。"""
    if not query_list:
        return []
    if isinstance(query_list[0], dict) and query_list[0].get("queryList") is not None:
        flat: list = []
        for block in query_list:
            if not isinstance(block, dict):
                continue
            for item in block.get("queryList") or []:
                if isinstance(item, dict):
                    flat.append(item)
        return flat
    return [x for x in query_list if isinstance(x, dict)]


def _default_params_from_query_list(query_list: list) -> Dict[str, Any]:
    """从 query_list 提取默认参数值"""
    params: Dict[str, Any] = {}
    for item in _flatten_query_list(query_list):
        field = item.get("field") or item.get("vModel") or item.get("prop")
        if not field:
            continue
        default_val = item.get("defaultValue")
        if default_val is None and "value" in item:
            default_val = item.get("value")
        if default_val is not None:
            params[field] = default_val
    return params


def _query_list_for_sheet(query_list: list, sheet_id: str) -> list:
    """按 sheet 过滤 query 项；扁平结构返回全部。"""
    if not query_list:
        return []
    if isinstance(query_list[0], dict) and query_list[0].get("queryList") is not None:
        for block in query_list:
            if str(block.get("sheet") or "") == str(sheet_id):
                return block.get("queryList") or []
        return _flatten_query_list(query_list)
    return query_list


def _parse_preview_draft_field(value: Any) -> Any:
    if value is None:
        return None
    return ReportVersionService._parse_json_field(value, None)


async def _build_preview(
    db: AsyncSession,
    version,
    template,
    params: Dict[str, Any],
    request=None,
    *,
    snapshot_override: Any = None,
    cells_override: Any = None,
    query_list_override: Any = None,
    sort_list_override: Any = None,
    column_list_override: Any = None,
    fence_list_override: Any = None,
    convert_config_override: Any = None,
) -> Dict[str, Any]:
    snapshot = (
        snapshot_override
        if snapshot_override is not None
        else (version.snapshot or {})
    )
    cells = cells_override if cells_override is not None else (version.cells or {})
    query_list = (
        query_list_override
        if query_list_override is not None
        else (version.query_list or [])
    )
    sort_list = (
        sort_list_override
        if sort_list_override is not None
        else (version.sort_list or [])
    )
    column_list = (
        column_list_override
        if column_list_override is not None
        else (version.column_list or [])
    )
    fence_list = (
        fence_list_override
        if fence_list_override is not None
        else (version.fence_list or [])
    )
    convert_config = (
        convert_config_override
        if convert_config_override is not None
        else (version.convert_config or {})
    )
    query_defaults = _default_params_from_query_list(query_list)
    user_id = getattr(getattr(request, "state", None), "user_id", None) if request else None
    user_name = ""
    dept_name = ""
    if user_id:
        try:
            from core.user.service import UserService
            from core.dept.service import DeptService

            user = await UserService.get(db, user_id)
            if user:
                user_name = user.name or user.username or ""
                if user.dept_id:
                    dept = await DeptService.get(db, user.dept_id)
                    if dept:
                        dept_name = dept.name or ""
        except Exception:
            pass
    system_params = build_system_params(
        user_id=str(user_id) if user_id else None,
        user_name=user_name,
        dept_name=dept_name,
    )
    merged_params = merge_preview_params(query_defaults, params, system_params)
    lookup = await build_lookup_cache_from_db(db)
    datasets = await ReportDatasetBridge.fetch_all(
        db,
        version.id,
        merged_params,
        sort_list=sort_list,
        version_convert=convert_config,
        lookup=lookup,
    )
    filled = transform(
        snapshot,
        cells,
        datasets,
        merged_params,
        column_list=column_list,
        fence_list=fence_list,
    )
    chart_data = build_chart_data(cells, datasets)
    watermark = build_watermark_payload(
        bool(template.allow_watermark),
        template.watermark_config or {},
        template_name=template.name or "",
    )
    warnings = list(detect_expression_cycles(cells, snapshot) or [])
    warnings.extend(collect_preview_warnings(datasets=datasets, snapshot=filled))
    return {
        "snapshot": filled,
        "cells": cells,
        "queryList": query_list,
        "chartData": chart_data,
        "allowExport": bool(template.allow_export),
        "allowPrint": bool(template.allow_print),
        "allowWatermark": watermark["show"],
        "watermarkConfig": watermark["config"],
        "watermark": watermark,
        "fullName": template.name,
        "warnings": warnings,
    }


@router.post("/{version_id}/preview", summary="预览（设计/运行）")
async def preview_version(
    request: Request,
    version_id: str,
    body: ReportPreviewIn,
    db: AsyncSession = Depends(get_db),
):
    try:
        version = await ReportVersionService.get(db, version_id)
        template = await ReportService.get(db, version.template_id)
        data = await _build_preview(db, version, template, body.params or {}, request)
        return data
    except ReportServiceException as e:
        raise HTTPException(status_code=400, detail=str(e))


@router.post("/preview-template", summary="按模板预览（启用中版本）")
async def preview_template(
    request: Request,
    body: ReportPreviewIn,
    template_id: str = Query(None, alias="templateId"),
    template_code: str = Query(None, alias="templateCode"),
    db: AsyncSession = Depends(get_db),
):
    try:
        if template_code:
            template = await ReportService.get_by_code(db, template_code)
        elif template_id:
            template = await ReportService.get(db, template_id)
        else:
            raise ReportServiceException("需要 template_id 或 template_code")
        version = await ReportVersionService.get_active(db, template.id)
        if not version:
            version = await ReportVersionService.get_designing_or_latest(db, template.id)
        if not version:
            raise ReportServiceException("没有可用版本")
        data = await _build_preview(db, version, template, body.params or {}, request)
        return data
    except ReportServiceException as e:
        raise HTTPException(status_code=400, detail=str(e))


@router.get("/query-list/{template_id}", summary="查询条件列表")
async def get_query_list(template_id: str, db: AsyncSession = Depends(get_db)):
    try:
        version = await ReportVersionService.get_active(db, template_id)
        if not version:
            version = await ReportVersionService.get_designing_or_latest(db, template_id)
        if not version:
            return {"queryList": []}
        return {"queryList": version.query_list or []}
    except ReportServiceException as e:
        raise HTTPException(status_code=400, detail=str(e))


@router.get("/query-list/code/{code}", summary="按编码获取查询条件")
async def get_query_list_by_code(code: str, db: AsyncSession = Depends(get_db)):
    try:
        template = await ReportService.get_by_code(db, code)
        version = await ReportVersionService.get_active(db, template.id)
        if not version:
            version = await ReportVersionService.get_designing_or_latest(db, template.id)
        if not version:
            return {"queryList": [], "templateId": template.id}
        return {
            "queryList": version.query_list or [],
            "templateId": template.id,
            "fullName": template.name,
        }
    except ReportServiceException as e:
        raise HTTPException(status_code=400, detail=str(e))


@router.post("/{version_id}/preview-design", summary="设计态预览")
async def preview_design(
    request: Request,
    version_id: str,
    body: ReportPreviewIn,
    db: AsyncSession = Depends(get_db),
):
    """设计器内预览；若传入 snapshot/cells 则使用当前编辑器草稿"""
    try:
        version = await ReportVersionService.get(db, version_id)
        template = await ReportService.get(db, version.template_id)
        draft_snapshot = _parse_preview_draft_field(body.snapshot)
        draft_cells = _parse_preview_draft_field(body.cells)
        draft_query_list = _parse_preview_draft_field(body.query_list)
        draft_sort_list = _parse_preview_draft_field(body.sort_list)
        draft_column_list = _parse_preview_draft_field(body.column_list)
        draft_fence_list = _parse_preview_draft_field(body.fence_list)
        draft_convert_config = _parse_preview_draft_field(body.convert_config)
        return await _build_preview(
            db,
            version,
            template,
            body.params or {},
            request,
            snapshot_override=draft_snapshot,
            cells_override=draft_cells,
            query_list_override=draft_query_list,
            sort_list_override=draft_sort_list,
            column_list_override=draft_column_list,
            fence_list_override=draft_fence_list,
            convert_config_override=draft_convert_config,
        )
    except ReportServiceException as e:
        raise HTTPException(status_code=400, detail=str(e))


@router.post("/{version_id}/export-excel-design", summary="设计态导出 Excel")
async def export_excel_design(
    request: Request,
    version_id: str,
    body: ReportPreviewIn,
    db: AsyncSession = Depends(get_db),
):
    """设计器内导出；若传入 snapshot/cells 则使用当前编辑器草稿"""
    try:
        version = await ReportVersionService.get(db, version_id)
        template = await ReportService.get(db, version.template_id)
        draft_snapshot = _parse_preview_draft_field(body.snapshot)
        draft_cells = _parse_preview_draft_field(body.cells)
        draft_query_list = _parse_preview_draft_field(body.query_list)
        draft_sort_list = _parse_preview_draft_field(body.sort_list)
        draft_column_list = _parse_preview_draft_field(body.column_list)
        draft_fence_list = _parse_preview_draft_field(body.fence_list)
        draft_convert_config = _parse_preview_draft_field(body.convert_config)
        return await _export_excel_response(
            db,
            version,
            template,
            body.params or {},
            request,
            snapshot_override=draft_snapshot,
            cells_override=draft_cells,
            query_list_override=draft_query_list,
            sort_list_override=draft_sort_list,
            column_list_override=draft_column_list,
            fence_list_override=draft_fence_list,
            convert_config_override=draft_convert_config,
        )
    except ReportServiceException as e:
        raise HTTPException(status_code=400, detail=str(e))


async def _export_excel_response(
    db: AsyncSession,
    version,
    template,
    params: Dict[str, Any],
    request=None,
    *,
    snapshot_override: Any = None,
    cells_override: Any = None,
    query_list_override: Any = None,
    sort_list_override: Any = None,
    column_list_override: Any = None,
    fence_list_override: Any = None,
    convert_config_override: Any = None,
):
    if not template.allow_export:
        raise ReportServiceException("该报表不允许导出")
    preview = await _build_preview(
        db,
        version,
        template,
        params,
        request,
        snapshot_override=snapshot_override,
        cells_override=cells_override,
        query_list_override=query_list_override,
        sort_list_override=sort_list_override,
        column_list_override=column_list_override,
        fence_list_override=fence_list_override,
        convert_config_override=convert_config_override,
    )
    watermark = preview.get("watermark") or {}
    wm_text = ""
    if watermark.get("show"):
        wm_text = str((watermark.get("config") or {}).get("content") or "")
    base_url = str(request.base_url).rstrip("/") if request else getattr(settings, "BASE_URL", "")
    from online_dev.report_manager.engine.export_excel_extras import build_fetch_url

    content = snapshot_to_xlsx_bytes(
        preview.get("snapshot") or {},
        watermark_text=wm_text,
        fetch_url=build_fetch_url(base_url),
    )
    filename = f"{template.code or 'report'}.xlsx"
    return StreamingResponse(
        iter([content]),
        media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
        headers={"Content-Disposition": f'attachment; filename="{filename}"'},
    )


async def _export_pdf_response(
    db: AsyncSession,
    version,
    template,
    params: Dict[str, Any],
    request=None,
):
    if not template.allow_export:
        raise ReportServiceException("该报表不允许导出")
    preview = await _build_preview(db, version, template, params, request)
    watermark = preview.get("watermark") or {}
    wm_text = ""
    if watermark.get("show"):
        wm_text = str((watermark.get("config") or {}).get("content") or "")
    content = snapshot_to_pdf_bytes(
        preview.get("snapshot") or {},
        title=template.name or "",
        watermark_text=wm_text,
    )
    filename = f"{template.code or 'report'}.pdf"
    return StreamingResponse(
        iter([content]),
        media_type="application/pdf",
        headers={"Content-Disposition": f'attachment; filename="{filename}"'},
    )


@router.post("/{version_id}/export-excel", summary="导出 Excel（按版本）")
async def export_excel_version(
    request: Request,
    version_id: str,
    body: ReportPreviewIn,
    db: AsyncSession = Depends(get_db),
):
    try:
        version = await ReportVersionService.get(db, version_id)
        template = await ReportService.get(db, version.template_id)
        return await _export_excel_response(db, version, template, body.params or {}, request)
    except ReportServiceException as e:
        raise HTTPException(status_code=400, detail=str(e))


@router.post("/upload/file", response_model=ReportUploadOut, summary="上传文件（图片等）")
async def upload_report_file(
    file: UploadFile = File(...),
    db: AsyncSession = Depends(get_db),
):
    try:
        content = await file.read()
        if not content:
            raise ReportServiceException("文件为空")
        filename = file.filename or "upload.bin"
        file_obj = await FileManagerService.upload_file(
            db=db,
            file_content=content,
            filename=filename,
            file_size=len(content),
            is_public=True,
            source="report",
        )
        return ReportUploadOut(name=file_obj.name, url=_build_file_access_url(file_obj))
    except ReportServiceException as e:
        raise HTTPException(status_code=400, detail=str(e))
    except Exception as e:
        logger.exception("报表文件上传失败")
        raise HTTPException(status_code=500, detail=str(e))


@router.post("/downImg", response_model=ReportUploadOut, summary="远端/Base64 图片转存")
async def download_remote_image(
    body: ReportDownImgIn,
    db: AsyncSession = Depends(get_db),
):
    try:
        img_value = (body.img_value or "").strip()
        if not img_value:
            raise ReportServiceException("图片内容为空")

        img_type = (body.img_type or "").upper()
        content: bytes
        ext = ".jpeg"

        if img_type == "BASE64":
            base64_img = img_value
            match = re.search(r"data:image/(\w+);base64,", img_value)
            if match:
                ext = f".{match.group(1)}"
                base64_img = img_value.split(",", 1)[-1]
            content = base64.b64decode(base64_img)
        else:
            async with httpx.AsyncClient(timeout=30.0, follow_redirects=True) as client:
                resp = await client.get(img_value)
                resp.raise_for_status()
                content = resp.content
                ctype = resp.headers.get("content-type", "")
                if "png" in ctype:
                    ext = ".png"
                elif "gif" in ctype:
                    ext = ".gif"
                elif "webp" in ctype:
                    ext = ".webp"

        if not content:
            raise ReportServiceException("无法获取图片数据")

        file_obj = await FileManagerService.upload_file(
            db=db,
            file_content=content,
            filename=f"report_img{ext}",
            file_size=len(content),
            is_public=True,
            source="report",
        )
        return ReportUploadOut(name=file_obj.name, url=_build_file_access_url(file_obj))
    except ReportServiceException as e:
        raise HTTPException(status_code=400, detail=str(e))
    except Exception as e:
        logger.exception("报表图片转存失败")
        raise HTTPException(status_code=500, detail=str(e))


@router.post("/import-excel", response_model=ReportImportExcelOut, summary="导入 Excel 到网格")
@router.post("/ImportExcel", response_model=ReportImportExcelOut, summary="导入 Excel（JNPF 兼容路径）")
async def import_excel_file(
    file: UploadFile = File(...),
):
    try:
        content = await file.read()
        if not content:
            raise ReportServiceException("文件为空")
        grid = parse_excel_to_grid(content)
        return ReportImportExcelOut(
            rowsCount=grid["rowsCount"],
            colsCount=grid["colsCount"],
            data=grid["data"],
        )
    except ReportServiceException as e:
        raise HTTPException(status_code=400, detail=str(e))
    except Exception as e:
        logger.exception("Excel 解析失败")
        raise HTTPException(status_code=400, detail=f"Excel 解析失败: {e}")


@router.post("/export-excel/template", summary="导出 Excel（按模板启用版本）")
async def export_excel_template(
    request: Request,
    body: ReportPreviewIn,
    template_id: str = Query(None, alias="templateId"),
    template_code: str = Query(None, alias="templateCode"),
    db: AsyncSession = Depends(get_db),
):
    try:
        if template_code:
            template = await ReportService.get_by_code(db, template_code)
        elif template_id:
            template = await ReportService.get(db, template_id)
        else:
            raise ReportServiceException("需要 template_id 或 template_code")
        version = await ReportVersionService.get_active(db, template.id)
        if not version:
            version = await ReportVersionService.get_designing_or_latest(db, template.id)
        if not version:
            raise ReportServiceException("没有可用版本")
        return await _export_excel_response(db, version, template, body.params or {}, request)
    except ReportServiceException as e:
        raise HTTPException(status_code=400, detail=str(e))


@router.post("/export-pdf/template", summary="导出 PDF（按模板启用版本）")
async def export_pdf_template(
    request: Request,
    body: ReportPreviewIn,
    template_id: str = Query(None, alias="templateId"),
    template_code: str = Query(None, alias="templateCode"),
    db: AsyncSession = Depends(get_db),
):
    try:
        if template_code:
            template = await ReportService.get_by_code(db, template_code)
        elif template_id:
            template = await ReportService.get(db, template_id)
        else:
            raise ReportServiceException("需要 template_id 或 template_code")
        version = await ReportVersionService.get_active(db, template.id)
        if not version:
            version = await ReportVersionService.get_designing_or_latest(db, template.id)
        if not version:
            raise ReportServiceException("没有可用版本")
        return await _export_pdf_response(db, version, template, body.params or {}, request)
    except ReportServiceException as e:
        raise HTTPException(status_code=400, detail=str(e))


@router.post("/export-pdf/template", summary="导出 PDF（按模板启用版本）")
async def export_pdf_template(
    request: Request,
    body: ReportPreviewIn,
    template_id: str = Query(None, alias="templateId"),
    template_code: str = Query(None, alias="templateCode"),
    db: AsyncSession = Depends(get_db),
):
    try:
        if template_code:
            template = await ReportService.get_by_code(db, template_code)
        elif template_id:
            template = await ReportService.get(db, template_id)
        else:
            raise ReportServiceException("需要 template_id 或 template_code")
        version = await ReportVersionService.get_active(db, template.id)
        if not version:
            version = await ReportVersionService.get_designing_or_latest(db, template.id)
        if not version:
            raise ReportServiceException("没有可用版本")
        return await _export_pdf_response(db, version, template, body.params or {}, request)
    except ReportServiceException as e:
        raise HTTPException(status_code=400, detail=str(e))