AI问数二期

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)

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,