From a84173066eabf7ca12f85a1384bbdf99c8f1ccdf Mon Sep 17 00:00:00 2001 From: BBIT-Kai <2911862937@qq.com> Date: Thu, 16 Jul 2026 13:55:21 +0800 Subject: [PATCH] =?UTF-8?q?AI=E9=97=AE=E6=95=B0=E4=BA=8C=E6=9C=9F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- bbit_ai/app/app.py | 15 +- bbit_ai/app/config/cubeReport.py | 4 + bbit_ai/app/db/postgres/cube_report.py | 230 ++++- bbit_ai/app/models/CubeReportRequest.py | 1 - bbit_ai/app/routers/CubeReport.py | 331 +++++-- bbit_ai/app/service/cube_metadata.py | 314 +++++++ bbit_ai/app/service/cube_report.py | 163 +++- bbit_ai/sql/cube_report_state.sql | 38 +- vue2/apps/web-antd/src/api/llm/report-cube.ts | 123 ++- .../llm/report/report-cube-chat/index.vue | 814 ++++++++++++++---- 10 files changed, 1731 insertions(+), 302 deletions(-) create mode 100644 bbit_ai/app/service/cube_metadata.py diff --git a/bbit_ai/app/app.py b/bbit_ai/app/app.py index 869e832..9903cb3 100644 --- a/bbit_ai/app/app.py +++ b/bbit_ai/app/app.py @@ -27,6 +27,7 @@ from routers.dify_export_router import difyRouter from service.RabbitMQ import ( mq_client, ) +from service.cube_metadata import warm_cube_metadata_cache @asynccontextmanager @@ -35,10 +36,16 @@ async def lifespan(app: FastAPI): await mq_client.init() # 启动消费者 await mq_client.start_all_consumer() - yield - # 应用关闭时关闭 MQ 连接 - if mq_client._connection: - await mq_client._connection.close() + cube_metadata_task = asyncio.create_task(warm_cube_metadata_cache()) + try: + yield + finally: + if not cube_metadata_task.done(): + cube_metadata_task.cancel() + await asyncio.gather(cube_metadata_task, return_exceptions=True) + # 应用关闭时关闭 MQ 连接 + if mq_client._connection: + await mq_client._connection.close() async def ai_lab(): diff --git a/bbit_ai/app/config/cubeReport.py b/bbit_ai/app/config/cubeReport.py index 4db516f..a8c3046 100644 --- a/bbit_ai/app/config/cubeReport.py +++ b/bbit_ai/app/config/cubeReport.py @@ -11,6 +11,7 @@ class CubeReportSettings: request_timeout_seconds: float default_page_size: int max_page_size: int + metadata_refresh_seconds: int def get_cube_report_settings() -> CubeReportSettings: @@ -30,4 +31,7 @@ def get_cube_report_settings() -> CubeReportSettings: ), default_page_size=int(os.getenv("CUBE_REPORT_DEFAULT_PAGE_SIZE", "20")), max_page_size=int(os.getenv("CUBE_REPORT_MAX_PAGE_SIZE", "100")), + metadata_refresh_seconds=int( + os.getenv("CUBE_REPORT_METADATA_REFRESH_SECONDS", "3600") + ), ) diff --git a/bbit_ai/app/db/postgres/cube_report.py b/bbit_ai/app/db/postgres/cube_report.py index dfc35d0..f6d4e50 100644 --- a/bbit_ai/app/db/postgres/cube_report.py +++ b/bbit_ai/app/db/postgres/cube_report.py @@ -1,4 +1,5 @@ from typing import Any +from uuid import uuid4 from psycopg.types.json import Jsonb @@ -35,9 +36,9 @@ def get_cube_report_state( with conn.cursor() as cur: cur.execute( """ - SELECT title, tenant_id, tenant_name, allow_global, + SELECT title, tenant_id, tenant_name, current_cube, limit_source, response_message, - created_at, updated_at + request_content, created_at, updated_at FROM ai_cube_report_states WHERE dify_conversation_id = %s AND created_by = %s """, @@ -51,10 +52,10 @@ def get_cube_report_state( "title": row[0], "tenantId": str(row[1]) if row[1] is not None else None, "tenantName": row[2], - "allowGlobal": row[3], - "query": row[4], - "limitSource": row[5], - "message": row[6], + "query": row[3], + "limitSource": row[4], + "message": row[5], + "requestContent": row[6], "createdAt": row[7].isoformat() if row[7] else None, "updatedAt": row[8].isoformat() if row[8] else None, } @@ -65,12 +66,12 @@ def upsert_cube_report_state( conversation_id: str, user_id: str, title: str, - tenant_id: str | None, + tenant_id: str, tenant_name: str, - allow_global: bool, query: dict[str, Any], limit_source: str, response_message: str, + request_content: str, ) -> None: normalized_source = "user" if limit_source == "user" else "system" with pg_pool.getConn() as conn: @@ -79,17 +80,17 @@ def upsert_cube_report_state( """ INSERT INTO ai_cube_report_states ( dify_conversation_id, created_by, title, tenant_id, - tenant_name, allow_global, current_cube, limit_source, - response_message, created_at, updated_at + tenant_name, current_cube, limit_source, + response_message, request_content, created_at, updated_at ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW()) ON CONFLICT (dify_conversation_id) DO UPDATE SET title = EXCLUDED.title, tenant_id = EXCLUDED.tenant_id, tenant_name = EXCLUDED.tenant_name, - allow_global = EXCLUDED.allow_global, current_cube = EXCLUDED.current_cube, limit_source = EXCLUDED.limit_source, response_message = EXCLUDED.response_message, + request_content = EXCLUDED.request_content, updated_at = NOW() WHERE ai_cube_report_states.created_by = EXCLUDED.created_by """, @@ -99,10 +100,33 @@ def upsert_cube_report_state( title[:200] or "新数据对话", tenant_id, tenant_name[:200], - allow_global, Jsonb(query), normalized_source, response_message, + request_content, + ), + ) + cur.execute( + """ + UPDATE ai_cube_saved_reports + SET requirement = %s, + tenant_id = %s, + tenant_name = %s, + cube_query = %s, + limit_source = %s, + response_message = %s, + updated_at = NOW() + WHERE dify_conversation_id = %s AND created_by = %s + """, + ( + request_content, + tenant_id, + tenant_name[:200], + Jsonb(query), + normalized_source, + response_message, + conversation_id, + user_id, ), ) @@ -132,3 +156,185 @@ def delete_cube_report_state(conversation_id: str, user_id: str) -> None: """, (conversation_id, user_id), ) + + +def _saved_report_row(row) -> dict[str, Any]: + return { + "id": str(row[0]), + "conversationId": str(row[1]), + "title": row[2], + "requirement": row[3], + "tenantId": str(row[4]), + "tenantName": row[5], + "query": row[6], + "limitSource": row[7], + "message": row[8], + "createdAt": row[9].isoformat() if row[9] else None, + "updatedAt": row[10].isoformat() if row[10] else None, + } + + +def list_saved_cube_reports( + *, + user_id: str, + tenant_id: str, + page: int, + page_size: int, + keyword: str = "", +) -> tuple[list[dict[str, Any]], int]: + conditions = ["created_by = %s", "tenant_id = %s"] + params: list[Any] = [user_id, tenant_id] + if keyword: + conditions.append("(title ILIKE %s OR requirement ILIKE %s)") + value = f"%{keyword}%" + params.extend([value, value]) + where_clause = " AND ".join(conditions) + offset = (page - 1) * page_size + + with pg_pool.getConn() as conn: + with conn.cursor() as cur: + cur.execute( + f"SELECT COUNT(*) FROM ai_cube_saved_reports WHERE {where_clause}", + tuple(params), + ) + total = int(cur.fetchone()[0]) + cur.execute( + f""" + SELECT id, dify_conversation_id, title, requirement, + tenant_id, tenant_name, cube_query, limit_source, + response_message, created_at, updated_at + FROM ai_cube_saved_reports + WHERE {where_clause} + ORDER BY updated_at DESC, id DESC + LIMIT %s OFFSET %s + """, + (*params, page_size, offset), + ) + items = [_saved_report_row(row) for row in cur.fetchall()] + return items, total + + +def get_saved_cube_report(report_id: str, user_id: str) -> dict[str, Any] | None: + with pg_pool.getConn() as conn: + with conn.cursor() as cur: + cur.execute( + """ + SELECT id, dify_conversation_id, title, requirement, + tenant_id, tenant_name, cube_query, limit_source, + response_message, created_at, updated_at + FROM ai_cube_saved_reports + WHERE id = %s AND created_by = %s + """, + (report_id, user_id), + ) + row = cur.fetchone() + return _saved_report_row(row) if row else None + + +def get_saved_report_by_conversation( + conversation_id: str, user_id: str +) -> dict[str, Any] | None: + with pg_pool.getConn() as conn: + with conn.cursor() as cur: + cur.execute( + """ + SELECT id, dify_conversation_id, title, requirement, + tenant_id, tenant_name, cube_query, limit_source, + response_message, created_at, updated_at + FROM ai_cube_saved_reports + WHERE dify_conversation_id = %s AND created_by = %s + """, + (conversation_id, user_id), + ) + row = cur.fetchone() + return _saved_report_row(row) if row else None + + +def list_saved_report_links( + user_id: str, conversation_ids: list[str] +) -> dict[str, str]: + if not conversation_ids: + return {} + with pg_pool.getConn() as conn: + with conn.cursor() as cur: + cur.execute( + """ + SELECT dify_conversation_id, id + FROM ai_cube_saved_reports + WHERE created_by = %s + AND dify_conversation_id = ANY(%s::varchar[]) + """, + (user_id, conversation_ids), + ) + return {str(row[0]): str(row[1]) for row in cur.fetchall()} + + +def save_cube_report_from_state( + *, + conversation_id: str, + user_id: str, + title: str, +) -> dict[str, Any] | None: + report_id = str(uuid4()) + with pg_pool.getConn() as conn: + with conn.cursor() as cur: + cur.execute( + """ + INSERT INTO ai_cube_saved_reports ( + id, dify_conversation_id, created_by, title, requirement, + tenant_id, tenant_name, cube_query, limit_source, + response_message, created_at, updated_at + ) + SELECT %s, dify_conversation_id, created_by, %s, + request_content, tenant_id, tenant_name, current_cube, + limit_source, response_message, NOW(), NOW() + FROM ai_cube_report_states + WHERE dify_conversation_id = %s + AND created_by = %s + AND tenant_id IS NOT NULL + ON CONFLICT (created_by, dify_conversation_id) DO UPDATE SET + requirement = EXCLUDED.requirement, + tenant_id = EXCLUDED.tenant_id, + tenant_name = EXCLUDED.tenant_name, + cube_query = EXCLUDED.cube_query, + limit_source = EXCLUDED.limit_source, + response_message = EXCLUDED.response_message, + updated_at = NOW() + RETURNING id, dify_conversation_id, title, requirement, + tenant_id, tenant_name, cube_query, limit_source, + response_message, created_at, updated_at + """, + ( + report_id, + title[:200] or "未命名报表", + conversation_id, + user_id, + ), + ) + row = cur.fetchone() + return _saved_report_row(row) if row else None + + +def rename_saved_cube_report(report_id: str, user_id: str, title: str) -> None: + with pg_pool.getConn() as conn: + with conn.cursor() as cur: + cur.execute( + """ + UPDATE ai_cube_saved_reports + SET title = %s, updated_at = NOW() + WHERE id = %s AND created_by = %s + """, + (title[:200], report_id, user_id), + ) + + +def delete_saved_cube_report(report_id: str, user_id: str) -> None: + with pg_pool.getConn() as conn: + with conn.cursor() as cur: + cur.execute( + """ + DELETE FROM ai_cube_saved_reports + WHERE id = %s AND created_by = %s + """, + (report_id, user_id), + ) diff --git a/bbit_ai/app/models/CubeReportRequest.py b/bbit_ai/app/models/CubeReportRequest.py index 8d40856..627766a 100644 --- a/bbit_ai/app/models/CubeReportRequest.py +++ b/bbit_ai/app/models/CubeReportRequest.py @@ -12,7 +12,6 @@ class CubeReportFileInput(BaseModel): class SendCubeReportMessageRequest(BaseModel): content: str = Field(default="", max_length=8000) conversationId: str | None = None - allowGlobal: bool = False tenantId: str | None = Field(default=None, max_length=64) tenantName: str | None = Field(default=None, max_length=200) files: list[CubeReportFileInput] = Field(default_factory=list, max_length=5) diff --git a/bbit_ai/app/routers/CubeReport.py b/bbit_ai/app/routers/CubeReport.py index 3339a85..8a9ca23 100644 --- a/bbit_ai/app/routers/CubeReport.py +++ b/bbit_ai/app/routers/CubeReport.py @@ -11,9 +11,16 @@ 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 @@ -28,7 +35,7 @@ from service.cube_report import ( get_dify_messages, get_dify_parameters, get_latest_cube_response, - list_dify_conversations, + list_dify_conversations_page, normalize_cube_load, normalize_cube_response, preview_dify_file, @@ -40,6 +47,10 @@ from service.cube_report import ( 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") @@ -52,21 +63,18 @@ 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 scope + return { + "canRefreshMetadata": bool(scope["canGlobal"]), + "companies": scope["companies"], + } def _validate_scope( user_id: UUID, *, - allow_global: bool, tenant_id: str | None, -) -> tuple[str | None, str]: +) -> tuple[str, str]: scope = _scope(user_id) - if allow_global: - if not scope["canGlobal"]: - raise HTTPException(status_code=403, detail="当前用户没有全局查询权限") - return None, "不限租户" - 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="无权访问所选租户") @@ -79,18 +87,22 @@ async def _save_cube_state( 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=session.get("tenantId"), + tenant_id=tenant_id, tenant_name=str(session.get("tenantName") or ""), - allow_global=bool(session.get("allowGlobal")), 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, ) @@ -103,6 +115,12 @@ async def _resolve_cube_state( 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)) @@ -130,11 +148,59 @@ async def _resolve_cube_state( ) +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) @@ -142,14 +208,34 @@ async def get_app_parameters(user_id: UUID = Depends(get_user_id_from_token)): @cubeReportRouter.get("/sessions") -async def list_sessions(user_id: UUID = Depends(get_user_id_from_token)): - conversations, states = await asyncio.gather( - list_dify_conversations(str(user_id)), +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 - return BaseResponse(data=conversations) + item["reportId"] = links.get(item["id"]) + return BaseResponse(data=page) @cubeReportRouter.get("/sessions/{conversation_id}") @@ -161,13 +247,15 @@ async def get_session( get_dify_conversation(conversation_id, str(user_id)), get_dify_messages(conversation_id, str(user_id)), ) - _validate_scope( - user_id, - allow_global=session["allowGlobal"], - tenant_id=session["tenantId"], - ) + _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}) @@ -175,15 +263,11 @@ async def get_session( async def get_session_data( conversation_id: str, page: int = Query(default=1, ge=1), - pageSize: int = Query(default=20, ge=1, le=100), 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, - allow_global=session["allowGlobal"], - tenant_id=session["tenantId"], - ) + 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( @@ -191,7 +275,7 @@ async def get_session_data( "columns": [], "rows": [], "page": page, - "pageSize": pageSize, + "pageSize": page_size, "total": 0, "sql": "", } @@ -199,37 +283,82 @@ async def get_session_data( data = await query_cube_page( base_load=state["query"], tenant_id=tenant_id, - allow_global=session["allowGlobal"], page=page, - page_size=pageSize, + page_size=page_size, limit_source=state["limitSource"], ) data["title"] = session.get("title") or state.get("title") return BaseResponse(data=data) -@cubeReportRouter.post("/sessions/{conversation_id}/export") -async def export_session_data( - conversation_id: str, +@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), ): - session = await get_dify_conversation(conversation_id, str(user_id)) - tenant_id, _ = _validate_scope( - user_id, - allow_global=session["allowGlobal"], - 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="当前会话还没有可导出的数据请求") - - export_data = await query_cube_export( - base_load=state["query"], + 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, - allow_global=session["allowGlobal"], - limit_source=state["limitSource"], + page=page, + page_size=pageSize, + keyword=keyword.strip(), ) - filename = sanitize_filename(session.get("title") or "Cube查询结果") + 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"] ) @@ -246,6 +375,87 @@ async def export_session_data( ) +@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, @@ -254,31 +464,25 @@ async def send_message_stream( conversation_id = request.conversationId if conversation_id: session = await get_dify_conversation(conversation_id, str(user_id)) - allow_global = session["allowGlobal"] tenant_id, tenant_name = _validate_scope( user_id, - allow_global=allow_global, tenant_id=session["tenantId"], ) else: - allow_global = request.allowGlobal tenant_id, tenant_name = _validate_scope( user_id, - allow_global=allow_global, tenant_id=request.tenantId, ) session = { "title": "新数据对话", "tenantId": tenant_id, "tenantName": tenant_name, - "allowGlobal": allow_global, } files = [item.model_dump() for item in request.files] inputs = { - "allow_global": allow_global, - "tenant_id": "__ALL__" if allow_global else tenant_id, - "tenant_name": "不限租户" if allow_global else tenant_name, + "tenant_id": tenant_id, + "tenant_name": tenant_name, } async def event_stream(): @@ -327,6 +531,7 @@ async def send_message_stream( user_id=user_id, session=session, cube_response=cube_response, + request_content=request.content.strip(), ) display_message = ( cube_response["message"] @@ -382,7 +587,8 @@ async def rename_session( request: RenameCubeReportConversationRequest, user_id: UUID = Depends(get_user_id_from_token), ): - await get_dify_conversation(conversation_id, str(user_id)) + 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() ) @@ -400,7 +606,18 @@ async def delete_session( conversation_id: str, user_id: UUID = Depends(get_user_id_from_token), ): - await get_dify_conversation(conversation_id, str(user_id)) + 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) diff --git a/bbit_ai/app/service/cube_metadata.py b/bbit_ai/app/service/cube_metadata.py new file mode 100644 index 0000000..e50d6ff --- /dev/null +++ b/bbit_ai/app/service/cube_metadata.py @@ -0,0 +1,314 @@ +import asyncio +import hashlib +import json +import logging +from datetime import datetime, timezone +from typing import Any +from uuid import uuid4 + +import httpx +from fastapi import HTTPException + +from config.cubeReport import get_cube_report_settings +from config.redis import redis_client + +CUBE_METADATA_CACHE_KEY = "cube-report:metadata:v1" +CUBE_METADATA_LOCK_KEY = "cube-report:metadata:refresh-lock:v1" +_TIME_GRANULARITIES = { + "day", + "hour", + "minute", + "month", + "quarter", + "second", + "week", + "year", +} + +logger = logging.getLogger(__name__) +_refresh_lock = asyncio.Lock() +_background_tasks: set[asyncio.Task[Any]] = set() + + +def _utc_now() -> datetime: + return datetime.now(timezone.utc) + + +def _member_name(cube_name: str, value: Any) -> str: + name = str(value or "").strip() + if not name: + return "" + if "." not in name and cube_name: + return f"{cube_name}.{name}" + return name + + +def _member_title(member: dict[str, Any], member_name: str) -> str: + for key in ("shortTitle", "short_title", "title"): + title = str(member.get(key) or "").strip() + if title: + return title + return member_name + + +def normalize_cube_metadata(payload: Any) -> dict[str, Any]: + """将 Cube /meta 响应压缩为成员名到展示标题的映射。""" + if not isinstance(payload, dict): + raise ValueError("Cube 元数据响应不是 JSON 对象") + cubes = payload.get("cubes") + if not isinstance(cubes, list): + raise ValueError("Cube 元数据缺少 cubes 数组") + + members: dict[str, str] = {} + cube_count = 0 + for cube in cubes: + if not isinstance(cube, dict): + continue + cube_name = str(cube.get("name") or "").strip() + if not cube_name: + continue + cube_count += 1 + for collection_name in ("dimensions", "measures", "segments"): + collection = cube.get(collection_name) + if not isinstance(collection, list): + continue + for member in collection: + if not isinstance(member, dict): + continue + name = _member_name(cube_name, member.get("name")) + if name: + members[name] = _member_title(member, name) + + if not members: + raise ValueError("Cube 元数据中没有可用成员") + + schema_json = json.dumps( + members, + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + ) + return { + "version": 1, + "members": members, + "cubeCount": cube_count, + "memberCount": len(members), + "refreshedAt": _utc_now().isoformat(), + "schemaHash": hashlib.sha256(schema_json.encode("utf-8")).hexdigest(), + } + + +async def _read_cache() -> dict[str, Any] | None: + try: + value = await asyncio.to_thread( + redis_client.get_value, CUBE_METADATA_CACHE_KEY + ) + except Exception as error: + logger.warning("读取 Cube 元数据缓存失败：%s", error) + return None + if not isinstance(value, dict) or not isinstance(value.get("members"), dict): + return None + return value + + +async def _write_cache(value: dict[str, Any]) -> None: + await asyncio.to_thread( + redis_client.set_value, + CUBE_METADATA_CACHE_KEY, + value, + ) + + +async def _cache_is_refreshing() -> bool: + try: + return bool( + await asyncio.to_thread( + redis_client.redis.exists, CUBE_METADATA_LOCK_KEY + ) + ) + except Exception: + return _refresh_lock.locked() + + +def _is_stale(cache: dict[str, Any]) -> bool: + value = cache.get("refreshedAt") + try: + refreshed_at = datetime.fromisoformat(str(value).replace("Z", "+00:00")) + if refreshed_at.tzinfo is None: + refreshed_at = refreshed_at.replace(tzinfo=timezone.utc) + except (TypeError, ValueError): + return True + max_age = max(get_cube_report_settings().metadata_refresh_seconds, 60) + return (_utc_now() - refreshed_at).total_seconds() >= max_age + + +def _status( + cache: dict[str, Any] | None, + *, + changed: bool | None = None, + refreshing: bool = False, +) -> dict[str, Any]: + result: dict[str, Any] = { + "available": bool(cache), + "cubeCount": int(cache.get("cubeCount") or 0) if cache else 0, + "memberCount": int(cache.get("memberCount") or 0) if cache else 0, + "refreshedAt": cache.get("refreshedAt") if cache else None, + "refreshing": refreshing, + "stale": _is_stale(cache) if cache else True, + } + if changed is not None: + result["changed"] = changed + return result + + +async def _fetch_cube_metadata() -> dict[str, Any]: + settings = get_cube_report_settings() + headers = ( + {"Authorization": f"Bearer {settings.cube_api_token}"} + if settings.cube_api_token + else {} + ) + timeout = httpx.Timeout(settings.request_timeout_seconds, connect=15.0) + async with httpx.AsyncClient(timeout=timeout) as client: + response = await client.get( + f"{settings.cube_api_base}/meta", + headers=headers, + ) + if response.is_error: + raise HTTPException( + status_code=502, + detail=f"Cube 元数据获取失败：{response.text[:500]}", + ) + try: + return normalize_cube_metadata(response.json()) + except (TypeError, ValueError) as error: + raise HTTPException(status_code=502, detail=str(error)) from error + + +async def _acquire_distributed_lock(token: str) -> bool: + settings = get_cube_report_settings() + expires = max(int(settings.request_timeout_seconds) + 30, 60) + try: + return bool( + await asyncio.to_thread( + redis_client.redis.set, + CUBE_METADATA_LOCK_KEY, + token, + nx=True, + ex=expires, + ) + ) + except Exception as error: + logger.warning("Cube 元数据刷新锁不可用，将使用进程内锁：%s", error) + return True + + +async def _release_distributed_lock(token: str) -> None: + script = ( + "if redis.call('get', KEYS[1]) == ARGV[1] then " + "return redis.call('del', KEYS[1]) else return 0 end" + ) + try: + await asyncio.to_thread( + redis_client.redis.eval, + script, + 1, + CUBE_METADATA_LOCK_KEY, + token, + ) + except Exception as error: + logger.warning("释放 Cube 元数据刷新锁失败：%s", error) + + +async def refresh_cube_metadata_cache() -> dict[str, Any]: + """从 Cube 重新加载元数据，并以单个 Redis 值原子替换旧缓存。""" + async with _refresh_lock: + token = uuid4().hex + acquired = await _acquire_distributed_lock(token) + if not acquired: + return _status( + await _read_cache(), + changed=False, + refreshing=True, + ) + try: + previous = await _read_cache() + current = await _fetch_cube_metadata() + await _write_cache(current) + return _status( + current, + changed=( + not previous + or previous.get("schemaHash") != current.get("schemaHash") + ), + ) + finally: + await _release_distributed_lock(token) + + +async def get_cube_metadata_status() -> dict[str, Any]: + return _status( + await _read_cache(), + refreshing=await _cache_is_refreshing(), + ) + + +def _track_background_task(task: asyncio.Task[Any]) -> None: + _background_tasks.add(task) + task.add_done_callback(_background_tasks.discard) + + +async def _background_refresh() -> None: + try: + await refresh_cube_metadata_cache() + except Exception as error: + logger.warning("后台刷新 Cube 元数据失败：%s", error) + + +def _schedule_background_refresh() -> None: + if any(not task.done() for task in _background_tasks): + return + _track_background_task(asyncio.create_task(_background_refresh())) + + +async def get_cube_member_titles() -> dict[str, str]: + cache = await _read_cache() + if not cache: + try: + await refresh_cube_metadata_cache() + cache = await _read_cache() + except Exception as error: + logger.warning("Cube 元数据缓存回源失败：%s", error) + return {} + elif _is_stale(cache): + _schedule_background_refresh() + + members = cache.get("members") if cache else None + if not isinstance(members, dict): + return {} + return { + str(key): str(value) + for key, value in members.items() + if key and value + } + + +def cube_member_candidates(member: str, column: str) -> list[str]: + """生成表头查找候选项，兼容 SQL 别名和时间粒度成员。""" + candidates: list[str] = [] + for value in (member, member.replace("__", "."), column.replace("__", ".")): + if value and value not in candidates: + candidates.append(value) + parts = value.rsplit(".", 1) + if len(parts) == 2 and parts[1].lower() in _TIME_GRANULARITIES: + if parts[0] not in candidates: + candidates.append(parts[0]) + return candidates + + +async def warm_cube_metadata_cache() -> None: + """应用启动时预热；失败只记录日志，不影响主服务可用性。""" + try: + await refresh_cube_metadata_cache() + except Exception as error: + logger.warning("启动预热 Cube 元数据失败：%s", error) diff --git a/bbit_ai/app/service/cube_report.py b/bbit_ai/app/service/cube_report.py index 12f29e3..8b90889 100644 --- a/bbit_ai/app/service/cube_report.py +++ b/bbit_ai/app/service/cube_report.py @@ -11,6 +11,7 @@ from fastapi import HTTPException, UploadFile from config.cubeReport import get_cube_report_settings from db.sqlserver import execute_parameterized_scalar, execute_parameterized_sql +from service.cube_metadata import cube_member_candidates, get_cube_member_titles def _require_dify_key() -> str: @@ -143,15 +144,11 @@ async def get_dify_parameters() -> dict[str, Any]: def normalize_conversation(item: dict[str, Any]) -> dict[str, Any]: inputs = item.get("inputs") if isinstance(item.get("inputs"), dict) else {} - allow_global = _as_bool(inputs.get("allow_global")) return { "id": str(item.get("id") or ""), "title": str(item.get("name") or "新数据对话"), - "tenantId": None if allow_global else str(inputs.get("tenant_id") or ""), - "tenantName": "不限租户" - if allow_global - else str(inputs.get("tenant_name") or "未指定租户"), - "allowGlobal": allow_global, + "tenantId": str(inputs.get("tenant_id") or ""), + "tenantName": str(inputs.get("tenant_name") or "未指定租户"), "createdAt": _iso_timestamp(item.get("created_at")), "updatedAt": _iso_timestamp(item.get("updated_at")), "inputs": inputs, @@ -188,19 +185,118 @@ async def list_dify_conversations( conversations.extend(item for item in batch if isinstance(item, dict)) if not payload.get("has_more"): break - last_id = str(batch[-1].get("id") or "") + last_item = batch[-1] + last_id = ( + str(last_item.get("id") or "") + if isinstance(last_item, dict) + else "" + ) if not last_id: break return [normalize_conversation(item) for item in conversations] +async def list_dify_conversations_page( + user_id: str, + *, + tenant_id: str, + cursor: str | None, + page_size: int, + keyword: str = "", + max_scanned: int = 500, +) -> dict[str, Any]: + """按 Dify 游标向后扫描，收集当前租户的一页会话。""" + settings = get_cube_report_settings() + key = _require_dify_key() + items: list[dict[str, Any]] = [] + last_id = cursor + has_more = True + scanned = 0 + normalized_keyword = keyword.strip().lower() + + async with httpx.AsyncClient(timeout=_timeout()) as client: + while has_more and len(items) < page_size and scanned < max_scanned: + response = await client.get( + f"{settings.dify_api_base}/conversations", + headers=_headers(key), + params={ + "user": user_id, + "limit": min(100, max_scanned - scanned), + "sort_by": "-updated_at", + **({"last_id": last_id} if last_id else {}), + }, + ) + _raise_dify_error(response) + payload = response.json() + batch = payload.get("data", []) + if not isinstance(batch, list) or not batch: + has_more = False + break + + stopped_inside_batch = False + for index, raw_item in enumerate(batch): + if not isinstance(raw_item, dict): + continue + scanned += 1 + last_id = str(raw_item.get("id") or last_id or "") + item = normalize_conversation(raw_item) + if item["tenantId"] != tenant_id: + continue + if normalized_keyword and normalized_keyword not in item[ + "title" + ].lower(): + continue + items.append(item) + if len(items) >= page_size: + stopped_inside_batch = index < len(batch) - 1 + break + + has_more = stopped_inside_batch or bool(payload.get("has_more")) + if not last_id: + has_more = False + + return { + "items": items, + "nextCursor": last_id if has_more else None, + "hasMore": has_more, + } + + async def get_dify_conversation( conversation_id: str, user_id: str ) -> dict[str, Any]: - conversations = await list_dify_conversations(user_id) - for item in conversations: - if item["id"] == conversation_id: - return item + settings = get_cube_report_settings() + key = _require_dify_key() + last_id: str | None = None + scanned = 0 + async with httpx.AsyncClient(timeout=_timeout()) as client: + while scanned < 5000: + response = await client.get( + f"{settings.dify_api_base}/conversations", + headers=_headers(key), + params={ + "user": user_id, + "limit": 100, + "sort_by": "-updated_at", + **({"last_id": last_id} if last_id else {}), + }, + ) + _raise_dify_error(response) + payload = response.json() + batch = payload.get("data", []) + if not isinstance(batch, list) or not batch: + break + for raw_item in batch: + if not isinstance(raw_item, dict): + continue + scanned += 1 + if str(raw_item.get("id") or "") == conversation_id: + return normalize_conversation(raw_item) + if not payload.get("has_more"): + break + last_id = str(batch[-1].get("id") or "") + if not last_id: + break raise HTTPException(status_code=404, detail="Dify 会话不存在或无权访问") @@ -684,8 +780,16 @@ def _display_sql(raw_sql: str, values: list[Any]) -> str: return re.sub(r"@_(\d+)", replace, raw_sql) -def _column_title(column: str, aliases: dict[str, Any]) -> str: +def _column_title( + column: str, + aliases: dict[str, Any], + member_titles: dict[str, str], +) -> str: member = str(aliases.get(column) or column) + for candidate in cube_member_candidates(member, column): + title = member_titles.get(candidate) + if title: + return title return member @@ -693,7 +797,6 @@ async def query_cube_page( *, base_load: dict[str, Any], tenant_id: str | None, - allow_global: bool, page: int, page_size: int, limit_source: str | None, @@ -702,10 +805,9 @@ async def query_cube_page( page_size = min(max(page_size, 1), settings.max_page_size) page_offset = (page - 1) * page_size canonical_query = deepcopy(base_load) - if not allow_global: - if not tenant_id: - raise HTTPException(status_code=422, detail="当前会话缺少租户信息") - ensure_tenant_filter(canonical_query, tenant_id) + if not tenant_id: + raise HTTPException(status_code=422, detail="当前会话缺少租户信息") + ensure_tenant_filter(canonical_query, tenant_id) is_user_limit = str(limit_source).lower() == "user" semantic_limit = ( @@ -739,9 +841,10 @@ async def query_cube_page( for key in ("limit", "offset", "order", "total"): count_query.pop(key, None) - page_sql, count_sql = await asyncio.gather( + page_sql, count_sql, member_titles = await asyncio.gather( _generate_cube_sql(page_query), _generate_cube_sql(count_query), + get_cube_member_titles(), ) executable_sql, bindings = _prepare_cube_sql(page_sql["sql"], page_sql["params"]) count_executable, count_bindings = _prepare_cube_sql( @@ -761,7 +864,11 @@ async def query_cube_page( columns = [ { "key": key, - "title": _column_title(key, page_sql["aliases"]), + "title": _column_title( + key, + page_sql["aliases"], + member_titles, + ), "type": "text", } for key in keys @@ -795,25 +902,29 @@ async def query_cube_export( *, base_load: dict[str, Any], tenant_id: str | None, - allow_global: bool, limit_source: str | None, ) -> dict[str, Any]: query = deepcopy(base_load) - if not allow_global: - if not tenant_id: - raise HTTPException(status_code=422, detail="当前会话缺少租户信息") - ensure_tenant_filter(query, tenant_id) + if not tenant_id: + raise HTTPException(status_code=422, detail="当前会话缺少租户信息") + ensure_tenant_filter(query, tenant_id) if str(limit_source).lower() != "user": query.pop("limit", None) query.pop("offset", None) query.pop("total", None) - sql_info = await _generate_cube_sql(query) + sql_info, member_titles = await asyncio.gather( + _generate_cube_sql(query), + get_cube_member_titles(), + ) executable_sql, bindings = _prepare_cube_sql(sql_info["sql"], sql_info["params"]) keys, rows = await asyncio.to_thread( execute_parameterized_sql, executable_sql, bindings ) - headers = {key: _column_title(key, sql_info["aliases"]) for key in keys} + headers = { + key: _column_title(key, sql_info["aliases"], member_titles) + for key in keys + } return { "headers": headers, "rows": rows, diff --git a/bbit_ai/sql/cube_report_state.sql b/bbit_ai/sql/cube_report_state.sql index 00660c1..42ddce3 100644 --- a/bbit_ai/sql/cube_report_state.sql +++ b/bbit_ai/sql/cube_report_state.sql @@ -3,23 +3,51 @@ CREATE TABLE IF NOT EXISTS ai_cube_report_states ( dify_conversation_id VARCHAR(128) PRIMARY KEY, created_by UUID NOT NULL, title VARCHAR(200) NOT NULL DEFAULT '新数据对话', - tenant_id VARCHAR(64), + tenant_id VARCHAR(64) NOT NULL, tenant_name VARCHAR(200) NOT NULL DEFAULT '', - allow_global BOOLEAN NOT NULL DEFAULT FALSE, current_cube JSONB NOT NULL, limit_source VARCHAR(16) NOT NULL DEFAULT 'system', response_message TEXT NOT NULL DEFAULT '', + request_content TEXT NOT NULL DEFAULT '', created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), CONSTRAINT ck_ai_cube_report_states_limit_source - CHECK (limit_source IN ('system', 'user')), - CONSTRAINT ck_ai_cube_report_states_scope - CHECK (allow_global OR tenant_id IS NOT NULL) + CHECK (limit_source IN ('system', 'user')) ); +ALTER TABLE ai_cube_report_states + ADD COLUMN IF NOT EXISTS request_content TEXT NOT NULL DEFAULT ''; + CREATE INDEX IF NOT EXISTS idx_ai_cube_report_states_user_updated ON ai_cube_report_states (created_by, updated_at DESC); COMMENT ON TABLE ai_cube_report_states IS 'Cube 数据助手当前查询状态；聊天历史由 Dify 保存'; COMMENT ON COLUMN ai_cube_report_states.current_cube IS '工作流最后一次成功返回的原始 Cube query'; COMMENT ON COLUMN ai_cube_report_states.limit_source IS 'system=系统预览限制，user=用户明确指定限制'; +COMMENT ON COLUMN ai_cube_report_states.request_content IS '最近一次成功生成 Cube query 的用户需求'; + +CREATE TABLE IF NOT EXISTS ai_cube_saved_reports ( + id UUID PRIMARY KEY, + dify_conversation_id VARCHAR(128) NOT NULL, + created_by UUID NOT NULL, + title VARCHAR(200) NOT NULL, + requirement TEXT NOT NULL DEFAULT '', + tenant_id VARCHAR(64) NOT NULL, + tenant_name VARCHAR(200) NOT NULL DEFAULT '', + cube_query JSONB NOT NULL, + limit_source VARCHAR(16) NOT NULL DEFAULT 'system', + response_message TEXT NOT NULL DEFAULT '', + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + CONSTRAINT uq_ai_cube_saved_reports_user_conversation + UNIQUE (created_by, dify_conversation_id), + CONSTRAINT ck_ai_cube_saved_reports_limit_source + CHECK (limit_source IN ('system', 'user')) +); + +CREATE INDEX IF NOT EXISTS idx_ai_cube_saved_reports_user_tenant_updated + ON ai_cube_saved_reports (created_by, tenant_id, updated_at DESC); + +COMMENT ON TABLE ai_cube_saved_reports IS '用户收藏的 Cube 报表；每个 Dify 会话最多收藏一个报表'; +COMMENT ON COLUMN ai_cube_saved_reports.requirement IS '最近一次成功生成该报表的用户需求'; +COMMENT ON COLUMN ai_cube_saved_reports.cube_query IS '收藏报表当前使用的原始 Cube query'; diff --git a/vue2/apps/web-antd/src/api/llm/report-cube.ts b/vue2/apps/web-antd/src/api/llm/report-cube.ts index 84329ee..db08764 100644 --- a/vue2/apps/web-antd/src/api/llm/report-cube.ts +++ b/vue2/apps/web-antd/src/api/llm/report-cube.ts @@ -4,10 +4,20 @@ import { useAccessStore } from '@vben/stores'; import { pyRequestClient } from '#/api/request'; export interface CubeReportScope { - canGlobal: boolean; + canRefreshMetadata: boolean; companies: Array<{ id: string; name: string }>; } +export interface CubeReportMetaStatus { + available: boolean; + changed?: boolean; + cubeCount: number; + memberCount: number; + refreshedAt: null | string; + refreshing: boolean; + stale: boolean; +} + export interface CubeReportAppParameters { fileUpload: Record; openingStatement: string; @@ -18,16 +28,43 @@ export interface CubeReportAppParameters { } export interface CubeReportSession { - allowGlobal: boolean; createdAt: null | string; hasData?: boolean; id: string; - tenantId: null | string; + reportId?: null | string; + tenantId: string; tenantName: string; title: string; updatedAt: null | string; } +export interface CubeReportSessionPage { + hasMore: boolean; + items: CubeReportSession[]; + nextCursor: null | string; +} + +export interface CubeSavedReport { + conversationId: string; + createdAt: null | string; + id: string; + limitSource: 'system' | 'user'; + message: string; + query: Record; + requirement: string; + tenantId: string; + tenantName: string; + title: string; + updatedAt: null | string; +} + +export interface CubeSavedReportPage { + items: CubeSavedReport[]; + page: number; + pageSize: number; + total: number; +} + export interface CubeReportFile { belongsTo?: 'assistant' | 'user'; id: string; @@ -83,7 +120,6 @@ export type CubeReportStreamEvent = | { text: string; type: 'status' }; export interface SendCubeReportMessage { - allowGlobal: boolean; content: string; conversationId?: null | string; files: CubeReportFile[]; @@ -116,14 +152,49 @@ export async function getCubeReportScope() { return pyRequestClient.get('/llm/cube-report/scope'); } +export async function getCubeReportMetaStatus() { + return pyRequestClient.get( + '/llm/cube-report/meta/status', + ); +} + +export async function refreshCubeReportMeta() { + return pyRequestClient.post( + '/llm/cube-report/meta/refresh', + ); +} + export async function getCubeReportAppParameters() { return pyRequestClient.get( '/llm/cube-report/app-parameters', ); } -export async function getCubeReportSessions() { - return pyRequestClient.get('/llm/cube-report/sessions'); +export async function getCubeReportSessions(params: { + cursor?: string; + keyword?: string; + pageSize?: number; + tenantId: string; +}) { + return pyRequestClient.get( + '/llm/cube-report/sessions', + { params }, + ); +} + +export async function getCubeReports(params: { + keyword?: string; + page: number; + pageSize: number; + tenantId: string; +}) { + return pyRequestClient.get('/llm/cube-report/reports', { + params, + }); +} + +export async function getCubeReport(id: string) { + return pyRequestClient.get(`/llm/cube-report/reports/${id}`); } export async function getCubeReportSession(id: string) { @@ -133,22 +204,41 @@ export async function getCubeReportSession(id: string) { }>(`/llm/cube-report/sessions/${id}`); } -export async function getCubeReportData( - id: string, - params: { page: number; pageSize: number }, -) { +export async function getCubeReportData(id: string, params: { page: number }) { return pyRequestClient.get( `/llm/cube-report/sessions/${id}/data`, { params }, ); } +export async function getCubeSavedReportData( + id: string, + params: { page: number }, +) { + return pyRequestClient.get( + `/llm/cube-report/reports/${id}/data`, + { params }, + ); +} + export async function exportCubeReportSession(id: string) { return pyRequestClient.post( `/llm/cube-report/sessions/${id}/export`, ); } +export async function exportCubeSavedReport(id: string) { + return pyRequestClient.post( + `/llm/cube-report/reports/${id}/export`, + ); +} + +export async function favoriteCubeReportSession(id: string) { + return pyRequestClient.post( + `/llm/cube-report/sessions/${id}/favorite`, + ); +} + export async function renameCubeReportSession(id: string, name: string) { return pyRequestClient.post( `/llm/cube-report/sessions/${id}/name`, @@ -162,6 +252,19 @@ export async function deleteCubeReportSession(id: string) { ); } +export async function renameCubeSavedReport(id: string, name: string) { + return pyRequestClient.post( + `/llm/cube-report/reports/${id}/name`, + { name }, + ); +} + +export async function deleteCubeSavedReport(id: string) { + return pyRequestClient.delete<{ result: string }>( + `/llm/cube-report/reports/${id}`, + ); +} + export async function stopCubeReportTask(taskId: string) { return pyRequestClient.post<{ result: string }>( `/llm/cube-report/tasks/${taskId}/stop`, diff --git a/vue2/apps/web-antd/src/views/llm/report/report-cube-chat/index.vue b/vue2/apps/web-antd/src/views/llm/report/report-cube-chat/index.vue index 0c03217..68a3576 100644 --- a/vue2/apps/web-antd/src/views/llm/report/report-cube-chat/index.vue +++ b/vue2/apps/web-antd/src/views/llm/report/report-cube-chat/index.vue @@ -5,18 +5,26 @@ import type { CubeReportColumn, CubeReportFile, CubeReportMessage, + CubeReportMetaStatus, CubeReportPage, CubeReportSession, CubeReportStreamEvent, + CubeSavedReport, } from '#/api'; -import { computed, nextTick, onBeforeUnmount, onMounted, ref } from 'vue'; +import { + computed, + nextTick, + onBeforeUnmount, + onMounted, + ref, + watch, +} from 'vue'; import { IconifyIcon } from '@vben/icons'; import { Button, - Empty, Input, message, Modal, @@ -24,7 +32,6 @@ import { Segmented, Select, Spin, - Switch, Table, Tag, Textarea, @@ -36,15 +43,22 @@ import * as api from '#/api'; type WorkspaceView = 'chat' | 'data' | 'split'; +const reports = ref([]); +const reportPage = ref(1); +const reportPageSize = 6; +const reportTotal = ref(0); const sessions = ref([]); +const sessionCursor = ref(); +const sessionsHaveMore = ref(false); const companies = ref>([]); const openingStatement = ref('您好，我是 F8 数据助手，请问有什么可以帮到您？'); const suggestedQuestions = ref([]); -const canGlobal = ref(false); -const allowGlobal = ref(false); +const canRefreshMetadata = ref(false); const selectedCompanyId = ref(); const currentSessionId = ref(null); const currentSession = ref(null); +const currentReportId = ref(null); +const currentReport = ref(null); const messages = ref([]); const dataPage = ref({ columns: [], @@ -59,10 +73,14 @@ const resizing = ref(false); const sessionKeyword = ref(''); const inputMessage = ref(''); const pendingFiles = ref([]); +const loadingReports = ref(false); const loadingSessions = ref(false); +const loadingMoreSessions = ref(false); const loadingConversation = ref(false); const loadingData = ref(false); const exporting = ref(false); +const favoriting = ref(false); +const refreshingCubeMetadata = ref(false); const uploading = ref(false); const sending = ref(false); const stopping = ref(false); @@ -74,23 +92,30 @@ const renameValue = ref(''); const renaming = ref(false); const chatContainer = ref(null); const panelContainer = ref(null); +const tableRegion = ref(null); +const tableScrollY = ref(240); const fileInput = ref(null); - -const filteredSessions = computed(() => { - const keyword = sessionKeyword.value.trim().toLowerCase(); - if (!keyword) return sessions.value; - return sessions.value.filter((item) => - `${item.title} ${item.tenantName}`.toLowerCase().includes(keyword), - ); -}); +const cubeMetadataStatus = ref(null); +let tableResizeObserver: null | ResizeObserver = null; +let searchTimer: ReturnType | undefined; const selectedCompany = computed(() => companies.value.find((item) => item.id === selectedCompanyId.value), ); -const scopeLocked = computed(() => Boolean(currentSessionId.value)); +const currentTitle = computed( + () => + currentReport.value?.title || + currentSession.value?.title || + '开始一次数据探索', +); +const activeSavedReportId = computed( + () => currentReportId.value || currentSession.value?.reportId || null, +); const hasData = computed( () => - Boolean(currentSession.value?.hasData) || dataPage.value.columns.length > 0, + Boolean(currentReport.value?.query) || + Boolean(currentSession.value?.hasData) || + dataPage.value.columns.length > 0, ); const tableColumns = computed(() => dataPage.value.columns.map((column: CubeReportColumn) => ({ @@ -106,6 +131,14 @@ const dataPanelStyle = computed(() => ? { width: `${splitPercent.value}%` } : undefined, ); +const cubeMetadataTooltip = computed(() => { + if (refreshingCubeMetadata.value) return '正在刷新 Cube 表头缓存…'; + const status = cubeMetadataStatus.value; + if (!status?.refreshedAt) return '刷新 Cube 表头缓存'; + return `刷新 Cube 表头缓存 · ${status.memberCount} 个字段 · ${dayjs( + status.refreshedAt, + ).format('MM-DD HH:mm')}`; +}); const suggestionIcons = [ 'lucide:map-pinned', 'lucide:list-ordered', @@ -139,7 +172,7 @@ function resetData() { dataPage.value = { columns: [], page: 1, - pageSize: dataPage.value.pageSize || 20, + pageSize: 20, rows: [], total: 0, }; @@ -148,12 +181,21 @@ function resetData() { async function loadScope() { const scope = await api.getCubeReportScope(); companies.value = scope.companies; - canGlobal.value = scope.canGlobal; + canRefreshMetadata.value = scope.canRefreshMetadata; if (!selectedCompanyId.value && companies.value[0]) { selectedCompanyId.value = companies.value[0].id; } } +async function loadCubeMetadataStatus() { + if (!canRefreshMetadata.value) return; + try { + cubeMetadataStatus.value = await api.getCubeReportMetaStatus(); + } catch { + cubeMetadataStatus.value = null; + } +} + async function loadAppParameters() { const parameters = await api.getCubeReportAppParameters(); if (parameters.openingStatement) { @@ -162,40 +204,90 @@ async function loadAppParameters() { suggestedQuestions.value = parameters.suggestedQuestions.filter(Boolean); } -async function loadSessions() { - loadingSessions.value = true; +async function loadReports(page = reportPage.value) { + if (!selectedCompanyId.value) return; + loadingReports.value = true; try { - sessions.value = await api.getCubeReportSessions(); + const result = await api.getCubeReports({ + keyword: sessionKeyword.value.trim(), + page, + pageSize: reportPageSize, + tenantId: selectedCompanyId.value, + }); + reports.value = result.items; + reportPage.value = result.page; + reportTotal.value = result.total; } finally { - loadingSessions.value = false; + loadingReports.value = false; } } +async function loadSessions(reset = true) { + if (!selectedCompanyId.value) return; + const loadingRef = reset ? loadingSessions : loadingMoreSessions; + loadingRef.value = true; + try { + const result = await api.getCubeReportSessions({ + cursor: reset ? undefined : sessionCursor.value, + keyword: sessionKeyword.value.trim(), + pageSize: 20, + tenantId: selectedCompanyId.value, + }); + sessions.value = reset + ? result.items + : [ + ...sessions.value, + ...result.items.filter( + (item) => !sessions.value.some((session) => session.id === item.id), + ), + ]; + sessionCursor.value = result.nextCursor || undefined; + sessionsHaveMore.value = result.hasMore; + } finally { + loadingRef.value = false; + } +} + +async function reloadNavigation() { + reportPage.value = 1; + await Promise.all([loadReports(1), loadSessions(true)]); +} + function createNewConversation() { if (sending.value) return; currentSessionId.value = null; currentSession.value = null; + currentReportId.value = null; + currentReport.value = null; messages.value = []; pendingFiles.value = []; inputMessage.value = ''; streamStatus.value = ''; currentTaskId.value = undefined; activeView.value = 'chat'; - allowGlobal.value = false; resetData(); } +async function handleTenantChange() { + createNewConversation(); + await reloadNavigation(); +} + async function loadConversation(id: string) { - if (sending.value || id === currentSessionId.value) return; + if ( + sending.value || + (id === currentSessionId.value && !currentReportId.value) + ) { + return; + } loadingConversation.value = true; currentSessionId.value = id; + currentReportId.value = null; + currentReport.value = null; try { const detail = await api.getCubeReportSession(id); currentSession.value = detail.session; messages.value = detail.messages; - allowGlobal.value = detail.session.allowGlobal; - selectedCompanyId.value = - detail.session.tenantId || selectedCompanyId.value; pendingFiles.value = []; resetData(); activeView.value = detail.session.hasData ? 'split' : 'chat'; @@ -209,10 +301,50 @@ async function loadConversation(id: string) { } } +async function loadReport(id: string) { + if (sending.value || id === currentReportId.value) return; + loadingConversation.value = true; + currentReportId.value = id; + try { + const report = await api.getCubeReport(id); + currentReport.value = report; + currentSessionId.value = report.conversationId; + try { + const detail = await api.getCubeReportSession(report.conversationId); + currentSession.value = detail.session; + messages.value = detail.messages; + } catch { + currentSession.value = { + createdAt: report.createdAt, + hasData: true, + id: report.conversationId, + reportId: report.id, + tenantId: report.tenantId, + tenantName: report.tenantName, + title: report.title, + updatedAt: report.updatedAt, + }; + messages.value = []; + } + pendingFiles.value = []; + resetData(); + activeView.value = 'split'; + await loadData(1); + await scrollToBottom(); + } catch (error) { + currentReportId.value = null; + currentReport.value = null; + currentSessionId.value = null; + throw error; + } finally { + loadingConversation.value = false; + } +} + async function sendMessage() { const content = inputMessage.value.trim(); if ((!content && pendingFiles.value.length === 0) || sending.value) return; - if (!currentSession.value && !allowGlobal.value && !selectedCompany.value) { + if (!currentSession.value && !selectedCompany.value) { message.warning('请先选择租户'); return; } @@ -251,7 +383,6 @@ async function sendMessage() { try { await api.streamCubeReportMessage( { - allowGlobal: currentSession.value?.allowGlobal ?? allowGlobal.value, content, conversationId: currentSessionId.value, files: sentFiles, @@ -267,28 +398,23 @@ async function sendMessage() { completedWithData = event.hasData; assistantMessage.status = 'completed'; if (event.hasData) activeView.value = 'split'; - break; } case 'message_delta': { assistantMessage.content += event.content; void scrollToBottom(); - break; } case 'message_replace': { assistantMessage.content = event.content; - break; } case 'status': { streamStatus.value = event.text; - break; } case 'task': { currentTaskId.value = event.taskId; - break; } case 'error': { @@ -309,8 +435,10 @@ async function sendMessage() { const detail = await api.getCubeReportSession(completedConversationId); currentSession.value = detail.session; messages.value = detail.messages; - allowGlobal.value = detail.session.allowGlobal; - await loadSessions(); + if (currentReportId.value) { + currentReport.value = await api.getCubeReport(currentReportId.value); + } + await reloadNavigation(); if (completedWithData || detail.session.hasData) { activeView.value = 'split'; await loadData(1); @@ -355,17 +483,15 @@ async function loadData(page = dataPage.value.page) { if (!currentSessionId.value) return; loadingData.value = true; try { - dataPage.value = await api.getCubeReportData(currentSessionId.value, { - page, - pageSize: dataPage.value.pageSize, - }); + dataPage.value = currentReportId.value + ? await api.getCubeSavedReportData(currentReportId.value, { page }) + : await api.getCubeReportData(currentSessionId.value, { page }); } finally { loadingData.value = false; } } -async function handlePageChange(page: number, pageSize: number) { - dataPage.value.pageSize = pageSize; +async function handlePageChange(page: number) { await loadData(page); } @@ -383,7 +509,9 @@ async function exportCurrentData() { if (!currentSessionId.value || exporting.value) return; exporting.value = true; try { - const result = await api.exportCubeReportSession(currentSessionId.value); + const result = currentReportId.value + ? await api.exportCubeSavedReport(currentReportId.value) + : await api.exportCubeReportSession(currentSessionId.value); const anchor = document.createElement('a'); anchor.href = result.url; anchor.download = result.filename; @@ -399,6 +527,67 @@ async function exportCurrentData() { } } +async function toggleFavorite() { + if (!currentSessionId.value || favoriting.value) return; + if (activeSavedReportId.value) { + const reportId = activeSavedReportId.value; + Modal.confirm({ + title: '取消收藏这个报表？', + content: '只会从报表中心移除，不会删除原始对话。', + okText: '取消收藏', + cancelText: '返回', + async onOk() { + favoriting.value = true; + try { + await api.deleteCubeSavedReport(reportId); + currentReportId.value = null; + currentReport.value = null; + if (currentSession.value) currentSession.value.reportId = null; + await loadReports(1); + message.success('已取消收藏'); + } finally { + favoriting.value = false; + } + }, + }); + return; + } + favoriting.value = true; + try { + const report = await api.favoriteCubeReportSession(currentSessionId.value); + if (currentSession.value) currentSession.value.reportId = report.id; + await loadReports(1); + message.success('已收藏到报表中心'); + } finally { + favoriting.value = false; + } +} + +async function refreshCubeMetadata() { + if (!canRefreshMetadata.value || refreshingCubeMetadata.value) return; + refreshingCubeMetadata.value = true; + try { + const status = await api.refreshCubeReportMeta(); + cubeMetadataStatus.value = status; + if (status.refreshing) { + message.info('Cube 元数据正在刷新，请稍后再试'); + return; + } + if (hasData.value) await loadData(dataPage.value.page); + message.success( + status.changed + ? `Cube 表头已更新，共 ${status.memberCount} 个字段` + : `Cube 表头已是最新，共 ${status.memberCount} 个字段`, + ); + } catch (error) { + message.error( + error instanceof Error ? error.message : 'Cube 表头缓存刷新失败', + ); + } finally { + refreshingCubeMetadata.value = false; + } +} + async function handleFileSelect(event: Event) { const input = event.target as HTMLInputElement; const file = input.files?.[0]; @@ -427,30 +616,61 @@ async function downloadFile(file: CubeReportFile) { } function openRename() { - if (!currentSession.value) return; - renameValue.value = currentSession.value.title; + if (!currentSession.value && !currentReport.value) return; + renameValue.value = + currentReport.value?.title || currentSession.value?.title || ''; renameOpen.value = true; } async function confirmRename() { const name = renameValue.value.trim(); - if (!currentSessionId.value || !name) return; + if (!name) return; renaming.value = true; try { - await api.renameCubeReportSession(currentSessionId.value, name); - if (currentSession.value) currentSession.value.title = name; - const item = sessions.value.find( - (session) => session.id === currentSessionId.value, - ); - if (item) item.title = name; + if (currentReportId.value) { + const report = await api.renameCubeSavedReport( + currentReportId.value, + name, + ); + currentReport.value = report; + const item = reports.value.find((entry) => entry.id === report.id); + if (item) item.title = name; + message.success('报表名称已更新'); + } else if (currentSessionId.value) { + await api.renameCubeReportSession(currentSessionId.value, name); + if (currentSession.value) currentSession.value.title = name; + const item = sessions.value.find( + (session) => session.id === currentSessionId.value, + ); + if (item) item.title = name; + message.success('会话名称已更新'); + } renameOpen.value = false; - message.success('会话名称已更新'); } finally { renaming.value = false; } } -function deleteCurrentConversation() { +function deleteCurrentEntity() { + if (currentReportId.value) { + const reportId = currentReportId.value; + Modal.confirm({ + title: '从报表中心移除？', + content: '原始对话会保留，可以之后再次收藏。', + okText: '移除', + okType: 'danger', + cancelText: '取消', + async onOk() { + await api.deleteCubeSavedReport(reportId); + currentReportId.value = null; + currentReport.value = null; + if (currentSession.value) currentSession.value.reportId = null; + await loadReports(1); + message.success('报表已移除'); + }, + }); + return; + } if (!currentSessionId.value) return; const id = currentSessionId.value; Modal.confirm({ @@ -500,15 +720,42 @@ function stopResize() { resizing.value = false; } +function observeTableRegion() { + tableResizeObserver?.disconnect(); + if (!tableRegion.value) return; + tableResizeObserver = new ResizeObserver(([entry]) => { + if (!entry) return; + tableScrollY.value = Math.max( + 120, + Math.floor(entry.contentRect.height - 48), + ); + }); + tableResizeObserver.observe(tableRegion.value); +} + +watch(tableRegion, () => nextTick(observeTableRegion)); +watch(sessionKeyword, () => { + if (searchTimer) clearTimeout(searchTimer); + searchTimer = setTimeout(() => void reloadNavigation(), 320); +}); + onMounted(async () => { window.addEventListener('pointermove', handlePointerMove); window.addEventListener('pointerup', stopResize); - await Promise.all([loadScope(), loadSessions(), loadAppParameters()]); + await loadScope(); + await Promise.all([ + reloadNavigation(), + loadAppParameters(), + loadCubeMetadataStatus(), + ]); + await nextTick(observeTableRegion); }); onBeforeUnmount(() => { window.removeEventListener('pointermove', handlePointerMove); window.removeEventListener('pointerup', stopResize); + tableResizeObserver?.disconnect(); + if (searchTimer) clearTimeout(searchTimer); }); @@ -535,29 +782,9 @@ onBeforeUnmount(() => { 新建数据对话 -
-
全局查询
- - - -
-