Files
AILab/bbit_ai/app/routers/CubeReport.py
T
2026-07-16 13:55:21 +08:00

648 lines
21 KiB
Python

import asyncio
import json
from typing import Any
from uuid import UUID
from fastapi import APIRouter, Depends, File, HTTPException, Query, UploadFile
from fastapi.responses import Response, StreamingResponse
import db.sqlserver as sqlserver
from config.cubeReport import get_cube_report_settings
from config.security import get_user_id_from_token
from db.postgres.cube_report import (
delete_cube_report_state,
delete_saved_cube_report,
get_cube_report_state,
get_saved_cube_report,
get_saved_report_by_conversation,
list_cube_report_states,
list_saved_cube_reports,
list_saved_report_links,
rename_cube_report_state,
rename_saved_cube_report,
save_cube_report_from_state,
upsert_cube_report_state,
)
from models.BaseResponse import BaseResponse
from models.CubeReportRequest import (
RenameCubeReportConversationRequest,
SendCubeReportMessageRequest,
)
from service.cube_report import (
delete_dify_conversation,
get_dify_conversation,
get_dify_conversation_variables,
get_dify_messages,
get_dify_parameters,
get_latest_cube_response,
list_dify_conversations_page,
normalize_cube_load,
normalize_cube_response,
preview_dify_file,
query_cube_export,
query_cube_page,
rename_dify_conversation,
stop_dify_task,
stream_dify_chat,
upload_dify_file,
)
from routers.dify_export_router import build_xlsx, sanitize_filename, upload_xlsx
from service.cube_metadata import (
get_cube_metadata_status,
refresh_cube_metadata_cache,
)
cubeReportRouter = APIRouter(prefix="/cube-report")
def _sse(payload: dict[str, Any]) -> str:
return f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"
def _scope(user_id: UUID) -> dict[str, Any]:
scope = sqlserver.get_user_company_scope(str(user_id))
if not scope["companies"]:
raise HTTPException(status_code=403, detail="当前用户没有可用的数据租户")
return {
"canRefreshMetadata": bool(scope["canGlobal"]),
"companies": scope["companies"],
}
def _validate_scope(
user_id: UUID,
*,
tenant_id: str | None,
) -> tuple[str, str]:
scope = _scope(user_id)
companies = {item["id"]: item["name"] for item in scope["companies"]}
if not tenant_id or tenant_id not in companies:
raise HTTPException(status_code=403, detail="无权访问所选租户")
return tenant_id, companies[tenant_id]
async def _save_cube_state(
*,
conversation_id: str,
user_id: UUID,
session: dict[str, Any],
cube_response: dict[str, Any],
request_content: str = "",
) -> None:
tenant_id = str(session.get("tenantId") or "").strip()
if not tenant_id:
raise HTTPException(status_code=422, detail="当前会话缺少租户信息")
await asyncio.to_thread(
upsert_cube_report_state,
conversation_id=conversation_id,
user_id=str(user_id),
title=str(session.get("title") or "新数据对话"),
tenant_id=tenant_id,
tenant_name=str(session.get("tenantName") or ""),
query=cube_response["query"],
limit_source=str(cube_response.get("limitSource") or "system"),
response_message=str(cube_response.get("message") or ""),
request_content=request_content,
)
async def _resolve_cube_state(
conversation_id: str,
user_id: UUID,
session: dict[str, Any],
) -> dict[str, Any] | None:
state = await asyncio.to_thread(
get_cube_report_state, conversation_id, str(user_id)
)
if state:
session_tenant_id = str(session.get("tenantId") or "")
if not session_tenant_id or state.get("tenantId") != session_tenant_id:
raise HTTPException(
status_code=409,
detail="本地 Cube 状态与当前会话租户不一致,已阻止数据查询",
)
return state
cube_response = await get_latest_cube_response(conversation_id, str(user_id))
if not cube_response or not cube_response["status"]:
variables = await get_dify_conversation_variables(
conversation_id, str(user_id)
)
query = normalize_cube_load(variables.get("temp_request"))
if not query:
return None
cube_response = {
"query": query,
"limitSource": str(variables.get("temp_limit_source") or "system"),
"message": "",
"status": True,
}
await _save_cube_state(
conversation_id=conversation_id,
user_id=user_id,
session=session,
cube_response=cube_response,
)
return await asyncio.to_thread(
get_cube_report_state, conversation_id, str(user_id)
)
async def _resolve_saved_report(
report_id: str,
user_id: UUID,
) -> tuple[dict[str, Any], str]:
report = await asyncio.to_thread(
get_saved_cube_report,
report_id,
str(user_id),
)
if not report:
raise HTTPException(status_code=404, detail="收藏报表不存在或无权访问")
tenant_id, tenant_name = _validate_scope(
user_id,
tenant_id=report["tenantId"],
)
report["tenantName"] = tenant_name
return report, tenant_id
@cubeReportRouter.get("/scope")
def get_scope(user_id: UUID = Depends(get_user_id_from_token)):
return BaseResponse(data=_scope(user_id))
@cubeReportRouter.get("/meta/status")
async def get_metadata_status(
user_id: UUID = Depends(get_user_id_from_token),
):
scope = _scope(user_id)
if not scope["canRefreshMetadata"]:
raise HTTPException(status_code=403, detail="当前用户没有刷新 Cube 元数据的权限")
return BaseResponse(data=await get_cube_metadata_status())
@cubeReportRouter.post("/meta/refresh")
async def refresh_metadata(
user_id: UUID = Depends(get_user_id_from_token),
):
scope = _scope(user_id)
if not scope["canRefreshMetadata"]:
raise HTTPException(status_code=403, detail="当前用户没有刷新 Cube 元数据的权限")
try:
data = await refresh_cube_metadata_cache()
except HTTPException:
raise
except Exception as error:
raise HTTPException(
status_code=503,
detail=f"Cube 元数据缓存刷新失败:{str(error)[:500]}",
) from error
return BaseResponse(data=data)
@cubeReportRouter.get("/app-parameters")
async def get_app_parameters(user_id: UUID = Depends(get_user_id_from_token)):
_scope(user_id)
return BaseResponse(data=await get_dify_parameters())
@cubeReportRouter.get("/sessions")
async def list_sessions(
tenantId: str = Query(min_length=1, max_length=64),
cursor: str | None = Query(default=None, max_length=128),
pageSize: int = Query(default=20, ge=1, le=50),
keyword: str = Query(default="", max_length=100),
user_id: UUID = Depends(get_user_id_from_token),
):
tenant_id, _ = _validate_scope(user_id, tenant_id=tenantId)
page, states = await asyncio.gather(
list_dify_conversations_page(
str(user_id),
tenant_id=tenant_id,
cursor=cursor,
page_size=pageSize,
keyword=keyword,
),
asyncio.to_thread(list_cube_report_states, str(user_id)),
)
conversations = page["items"]
links = await asyncio.to_thread(
list_saved_report_links,
str(user_id),
[item["id"] for item in conversations],
)
for item in conversations:
item["hasData"] = item["id"] in states
item["reportId"] = links.get(item["id"])
return BaseResponse(data=page)
@cubeReportRouter.get("/sessions/{conversation_id}")
async def get_session(
conversation_id: str,
user_id: UUID = Depends(get_user_id_from_token),
):
session, messages = await asyncio.gather(
get_dify_conversation(conversation_id, str(user_id)),
get_dify_messages(conversation_id, str(user_id)),
)
_validate_scope(user_id, tenant_id=session["tenantId"])
state = await _resolve_cube_state(conversation_id, user_id, session)
session["hasData"] = bool(state and state.get("query"))
saved_report = await asyncio.to_thread(
get_saved_report_by_conversation,
conversation_id,
str(user_id),
)
session["reportId"] = saved_report["id"] if saved_report else None
return BaseResponse(data={"session": session, "messages": messages})
@cubeReportRouter.get("/sessions/{conversation_id}/data")
async def get_session_data(
conversation_id: str,
page: int = Query(default=1, ge=1),
user_id: UUID = Depends(get_user_id_from_token),
):
page_size = get_cube_report_settings().default_page_size
session = await get_dify_conversation(conversation_id, str(user_id))
tenant_id, _ = _validate_scope(user_id, tenant_id=session["tenantId"])
state = await _resolve_cube_state(conversation_id, user_id, session)
if not state or not state.get("query"):
return BaseResponse(
data={
"columns": [],
"rows": [],
"page": page,
"pageSize": page_size,
"total": 0,
"sql": "",
}
)
data = await query_cube_page(
base_load=state["query"],
tenant_id=tenant_id,
page=page,
page_size=page_size,
limit_source=state["limitSource"],
)
data["title"] = session.get("title") or state.get("title")
return BaseResponse(data=data)
@cubeReportRouter.get("/reports")
async def list_reports(
tenantId: str = Query(min_length=1, max_length=64),
page: int = Query(default=1, ge=1),
pageSize: int = Query(default=8, ge=1, le=50),
keyword: str = Query(default="", max_length=100),
user_id: UUID = Depends(get_user_id_from_token),
):
tenant_id, _ = _validate_scope(user_id, tenant_id=tenantId)
items, total = await asyncio.to_thread(
list_saved_cube_reports,
user_id=str(user_id),
tenant_id=tenant_id,
page=page,
page_size=pageSize,
keyword=keyword.strip(),
)
return BaseResponse(
data={
"items": items,
"page": page,
"pageSize": pageSize,
"total": total,
}
)
@cubeReportRouter.get("/reports/{report_id}")
async def get_report(
report_id: str,
user_id: UUID = Depends(get_user_id_from_token),
):
report, _ = await _resolve_saved_report(report_id, user_id)
return BaseResponse(data=report)
@cubeReportRouter.get("/reports/{report_id}/data")
async def get_report_data(
report_id: str,
page: int = Query(default=1, ge=1),
user_id: UUID = Depends(get_user_id_from_token),
):
report, tenant_id = await _resolve_saved_report(report_id, user_id)
page_size = get_cube_report_settings().default_page_size
data = await query_cube_page(
base_load=report["query"],
tenant_id=tenant_id,
page=page,
page_size=page_size,
limit_source=report["limitSource"],
)
data["title"] = report["title"]
return BaseResponse(data=data)
async def _export_cube_result(
*,
query: dict[str, Any],
tenant_id: str,
limit_source: str,
title: str,
) -> BaseResponse:
export_data = await query_cube_export(
base_load=query,
tenant_id=tenant_id,
limit_source=limit_source,
)
filename = sanitize_filename(title or "Cube查询结果")
contents, sheet_count = await asyncio.to_thread(
build_xlsx, export_data["rows"], export_data["headers"]
)
object_name, url = await asyncio.to_thread(upload_xlsx, contents, filename)
return BaseResponse(
data={
"url": url,
"filename": filename,
"bucketName": "dify-export",
"objectName": object_name,
"rowCount": len(export_data["rows"]),
"sheetCount": sheet_count,
}
)
@cubeReportRouter.post("/reports/{report_id}/export")
async def export_report_data(
report_id: str,
user_id: UUID = Depends(get_user_id_from_token),
):
report, tenant_id = await _resolve_saved_report(report_id, user_id)
return await _export_cube_result(
query=report["query"],
tenant_id=tenant_id,
limit_source=report["limitSource"],
title=report["title"],
)
@cubeReportRouter.post("/sessions/{conversation_id}/favorite")
async def favorite_session(
conversation_id: str,
user_id: UUID = Depends(get_user_id_from_token),
):
session = await get_dify_conversation(conversation_id, str(user_id))
_validate_scope(user_id, tenant_id=session["tenantId"])
state = await _resolve_cube_state(conversation_id, user_id, session)
if not state or not state.get("query"):
raise HTTPException(status_code=409, detail="当前会话还没有可收藏的数据请求")
report = await asyncio.to_thread(
save_cube_report_from_state,
conversation_id=conversation_id,
user_id=str(user_id),
title=str(session.get("title") or state.get("title") or "未命名报表"),
)
if not report:
raise HTTPException(status_code=409, detail="收藏报表失败,请先生成有效数据")
return BaseResponse(data=report)
@cubeReportRouter.post("/reports/{report_id}/name")
async def rename_report(
report_id: str,
request: RenameCubeReportConversationRequest,
user_id: UUID = Depends(get_user_id_from_token),
):
await _resolve_saved_report(report_id, user_id)
await asyncio.to_thread(
rename_saved_cube_report,
report_id,
str(user_id),
request.name.strip(),
)
report, _ = await _resolve_saved_report(report_id, user_id)
return BaseResponse(data=report)
@cubeReportRouter.delete("/reports/{report_id}")
async def delete_report(
report_id: str,
user_id: UUID = Depends(get_user_id_from_token),
):
await _resolve_saved_report(report_id, user_id)
await asyncio.to_thread(delete_saved_cube_report, report_id, str(user_id))
return BaseResponse(data={"result": "success"})
@cubeReportRouter.post("/sessions/{conversation_id}/export")
async def export_session_data(
conversation_id: str,
user_id: UUID = Depends(get_user_id_from_token),
):
session = await get_dify_conversation(conversation_id, str(user_id))
tenant_id, _ = _validate_scope(user_id, tenant_id=session["tenantId"])
state = await _resolve_cube_state(conversation_id, user_id, session)
if not state or not state.get("query"):
raise HTTPException(status_code=404, detail="当前会话还没有可导出的数据请求")
return await _export_cube_result(
query=state["query"],
tenant_id=tenant_id,
limit_source=state["limitSource"],
title=str(session.get("title") or state.get("title") or "Cube查询结果"),
)
@cubeReportRouter.post("/messages/stream")
async def send_message_stream(
request: SendCubeReportMessageRequest,
user_id: UUID = Depends(get_user_id_from_token),
):
conversation_id = request.conversationId
if conversation_id:
session = await get_dify_conversation(conversation_id, str(user_id))
tenant_id, tenant_name = _validate_scope(
user_id,
tenant_id=session["tenantId"],
)
else:
tenant_id, tenant_name = _validate_scope(
user_id,
tenant_id=request.tenantId,
)
session = {
"title": "新数据对话",
"tenantId": tenant_id,
"tenantName": tenant_name,
}
files = [item.model_dump() for item in request.files]
inputs = {
"tenant_id": tenant_id,
"tenant_name": tenant_name,
}
async def event_stream():
active_conversation_id = conversation_id
answer = ""
task_id = None
workflow_status_sent = False
try:
yield _sse({"type": "status", "text": "正在理解问题…"})
async for event in stream_dify_chat(
user_id=str(user_id),
content=request.content.strip(),
conversation_id=conversation_id,
inputs=inputs,
files=files,
):
event_type = event.get("event")
active_conversation_id = (
event.get("conversation_id") or active_conversation_id
)
current_task_id = event.get("task_id")
if current_task_id and current_task_id != task_id:
task_id = current_task_id
yield _sse({"type": "task", "taskId": task_id})
if event_type in {"message", "agent_message"}:
delta = str(event.get("answer") or "")
answer += delta
elif event_type == "message_replace":
answer = str(event.get("answer") or "")
elif (
event_type in {"workflow_started", "node_started"}
and not workflow_status_sent
):
workflow_status_sent = True
yield _sse({"type": "status", "text": "正在生成数据请求…"})
elif event_type == "error":
raise RuntimeError(
str(event.get("message") or "Dify 工作流执行失败")
)
cube_response = normalize_cube_response(answer)
has_data = bool(cube_response and cube_response["status"])
if has_data and active_conversation_id and cube_response:
await _save_cube_state(
conversation_id=active_conversation_id,
user_id=user_id,
session=session,
cube_response=cube_response,
request_content=request.content.strip(),
)
display_message = (
cube_response["message"]
if cube_response
else answer or "工作流没有返回可识别的数据请求"
)
yield _sse({"type": "message_replace", "content": display_message})
yield _sse(
{
"type": "complete",
"conversationId": active_conversation_id,
"hasData": has_data,
}
)
except Exception as error:
message = error.detail if isinstance(error, HTTPException) else str(error)
yield _sse({"type": "error", "message": message})
settings = get_cube_report_settings()
return StreamingResponse(
event_stream(),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"X-Accel-Buffering": "no",
"X-Page-Size": str(settings.default_page_size),
},
)
@cubeReportRouter.post("/tasks/{task_id}/stop")
async def stop_task(
task_id: str,
user_id: UUID = Depends(get_user_id_from_token),
):
return BaseResponse(data=await stop_dify_task(task_id, str(user_id)))
@cubeReportRouter.post("/files/upload")
async def upload_file(
file: UploadFile = File(...),
user_id: UUID = Depends(get_user_id_from_token),
):
filename = (file.filename or "").lower()
if not filename.endswith((".xls", ".xlsx")):
raise HTTPException(status_code=415, detail="当前仅支持 XLS、XLSX 文件")
return BaseResponse(data=await upload_dify_file(file, str(user_id)))
@cubeReportRouter.post("/sessions/{conversation_id}/name")
async def rename_session(
conversation_id: str,
request: RenameCubeReportConversationRequest,
user_id: UUID = Depends(get_user_id_from_token),
):
session = await get_dify_conversation(conversation_id, str(user_id))
_validate_scope(user_id, tenant_id=session["tenantId"])
data = await rename_dify_conversation(
conversation_id, str(user_id), request.name.strip()
)
await asyncio.to_thread(
rename_cube_report_state,
conversation_id,
str(user_id),
request.name.strip(),
)
return BaseResponse(data=data)
@cubeReportRouter.delete("/sessions/{conversation_id}")
async def delete_session(
conversation_id: str,
user_id: UUID = Depends(get_user_id_from_token),
):
session = await get_dify_conversation(conversation_id, str(user_id))
_validate_scope(user_id, tenant_id=session["tenantId"])
saved_report = await asyncio.to_thread(
get_saved_report_by_conversation,
conversation_id,
str(user_id),
)
if saved_report:
raise HTTPException(
status_code=409,
detail="该会话已收藏为报表,请先在报表中心取消收藏",
)
await delete_dify_conversation(conversation_id, str(user_id))
await asyncio.to_thread(
delete_cube_report_state, conversation_id, str(user_id)
)
return BaseResponse(data={"result": "success"})
@cubeReportRouter.get("/files/{file_id}/preview")
async def preview_file(
file_id: str,
conversationId: str = Query(min_length=1),
asAttachment: bool = Query(default=False),
user_id: UUID = Depends(get_user_id_from_token),
):
messages = await get_dify_messages(conversationId, str(user_id))
owned = any(
file["id"] == file_id
for message in messages
for file in message.get("files", [])
)
if not owned:
raise HTTPException(status_code=404, detail="附件不存在或无权访问")
content, headers = await preview_dify_file(
file_id, as_attachment=asAttachment
)
media_type = headers.pop("content-type", "application/octet-stream")
return Response(content=content, media_type=media_type, headers=headers)