feat: add artifact library and improve public wiki
This commit is contained in:
1 parent
1b43205c1b
commit
dd0ee282df
16 files changed
+4081
-98
No files matched your search
@@ -1,11 +1,15 @@
|
||||
import asyncio
|
||||
import logging
|
||||
import mimetypes
|
||||
import time
|
||||
import zipfile
|
||||
from datetime import datetime, timezone
|
||||
from collections import OrderedDict
|
||||
from datetime import UTC, datetime
|
||||
from pathlib import Path
|
||||
from typing import Literal
|
||||
from urllib.parse import quote
|
||||
|
||||
from fastapi import APIRouter, HTTPException, Request
|
||||
from fastapi import APIRouter, HTTPException, Query, Request
|
||||
from fastapi.responses import FileResponse, PlainTextResponse, Response
|
||||
from pydantic import BaseModel
|
||||
|
||||
@@ -13,6 +17,7 @@ from app.gateway.authz import require_permission
|
||||
from app.gateway.path_utils import aresolve_thread_virtual_path
|
||||
from deerflow.config.paths import get_paths
|
||||
from deerflow.runtime.thread_paths import aresolve_path_user_id
|
||||
from deerflow.runtime.user_context import get_effective_user_id
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -38,6 +43,84 @@ EDITABLE_TEXT_SUFFIXES = {
|
||||
|
||||
_VIRTUAL_USER_DATA_PREFIX = "mnt/user-data"
|
||||
|
||||
ArtifactLibraryKind = Literal["all", "document", "spreadsheet", "text_data", "archive", "other"]
|
||||
|
||||
_MEDIA_MIME_PREFIXES = ("image/", "audio/", "video/")
|
||||
_MEDIA_SUFFIXES = {
|
||||
".aac",
|
||||
".aiff",
|
||||
".ape",
|
||||
".avi",
|
||||
".bmp",
|
||||
".flac",
|
||||
".gif",
|
||||
".heic",
|
||||
".ico",
|
||||
".jpeg",
|
||||
".jpg",
|
||||
".m4a",
|
||||
".m4v",
|
||||
".mkv",
|
||||
".mov",
|
||||
".mp3",
|
||||
".mp4",
|
||||
".ogg",
|
||||
".png",
|
||||
".svg",
|
||||
".tiff",
|
||||
".wav",
|
||||
".webm",
|
||||
".webp",
|
||||
".wma",
|
||||
}
|
||||
_DOCUMENT_SUFFIXES = {
|
||||
".doc",
|
||||
".docx",
|
||||
".epub",
|
||||
".odt",
|
||||
".pdf",
|
||||
".ppt",
|
||||
".pptx",
|
||||
".rtf",
|
||||
}
|
||||
_SPREADSHEET_SUFFIXES = {".csv", ".ods", ".tsv", ".xls", ".xlsx"}
|
||||
_TEXT_DATA_SUFFIXES = {
|
||||
".bash",
|
||||
".c",
|
||||
".conf",
|
||||
".cpp",
|
||||
".css",
|
||||
".go",
|
||||
".h",
|
||||
".html",
|
||||
".ini",
|
||||
".java",
|
||||
".js",
|
||||
".json",
|
||||
".jsonl",
|
||||
".log",
|
||||
".markdown",
|
||||
".md",
|
||||
".mmd",
|
||||
".py",
|
||||
".rs",
|
||||
".sh",
|
||||
".sql",
|
||||
".toml",
|
||||
".ts",
|
||||
".tsx",
|
||||
".txt",
|
||||
".xml",
|
||||
".yaml",
|
||||
".yml",
|
||||
}
|
||||
_ARCHIVE_SUFFIXES = {".7z", ".gz", ".rar", ".skill", ".tar", ".tgz", ".zip"}
|
||||
_ARTIFACT_LIBRARY_CACHE_TTL_SECONDS = 15.0
|
||||
_ARTIFACT_LIBRARY_CACHE_MAX_USERS = 128
|
||||
_ARTIFACT_LIBRARY_SCAN_CONCURRENCY = 8
|
||||
_ARTIFACT_LIBRARY_THREAD_BATCH_SIZE = 500
|
||||
_artifact_library_cache: OrderedDict[str, tuple[float, list[dict]]] = OrderedDict()
|
||||
|
||||
|
||||
def coerce_artifact_virtual_path(path: str) -> str:
|
||||
"""Normalize an artifacts URL path to a sandbox virtual path.
|
||||
@@ -60,6 +143,26 @@ def coerce_artifact_virtual_path(path: str) -> str:
|
||||
|
||||
class ArtifactUpdateRequest(BaseModel):
|
||||
content: str
|
||||
|
||||
|
||||
class ArtifactLibraryItem(BaseModel):
|
||||
thread_id: str
|
||||
thread_title: str
|
||||
agent_id: str | None = None
|
||||
thread_type: str | None = None
|
||||
path: str
|
||||
name: str
|
||||
size_bytes: int
|
||||
mime_type: str | None = None
|
||||
modified_at: str
|
||||
file_kind: str
|
||||
|
||||
|
||||
class ArtifactLibraryResponse(BaseModel):
|
||||
total: int
|
||||
offset: int
|
||||
limit: int
|
||||
items: list[ArtifactLibraryItem]
|
||||
|
||||
|
||||
def _build_content_disposition(disposition_type: str, filename: str) -> str:
|
||||
@@ -139,21 +242,184 @@ def collect_outputs_listing(outputs_dir: Path) -> list[dict]:
|
||||
resolved.relative_to(outputs_root)
|
||||
except (OSError, ValueError):
|
||||
continue
|
||||
relative = resolved.relative_to(outputs_root).as_posix()
|
||||
mime_type, _ = mimetypes.guess_type(resolved.name)
|
||||
stat = resolved.stat()
|
||||
try:
|
||||
relative = resolved.relative_to(outputs_root).as_posix()
|
||||
mime_type, _ = mimetypes.guess_type(resolved.name)
|
||||
stat = resolved.stat()
|
||||
except OSError:
|
||||
# A run may atomically replace or remove a file while this directory
|
||||
# is being scanned. Treat that transient entry as absent instead of
|
||||
# failing the complete thread/library listing.
|
||||
continue
|
||||
files.append(
|
||||
{
|
||||
"path": f"mnt/user-data/outputs/{relative}",
|
||||
"name": resolved.name,
|
||||
"size_bytes": stat.st_size,
|
||||
"mime_type": mime_type,
|
||||
"modified_at": datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc).isoformat(),
|
||||
"modified_at": datetime.fromtimestamp(stat.st_mtime, tz=UTC).isoformat(),
|
||||
}
|
||||
)
|
||||
|
||||
files.sort(key=lambda f: f["modified_at"], reverse=True)
|
||||
return files
|
||||
|
||||
|
||||
def artifact_library_kind(name: str) -> str:
|
||||
"""Classify one non-media artifact for the library filters."""
|
||||
suffix = Path(name).suffix.lower()
|
||||
if suffix in _DOCUMENT_SUFFIXES:
|
||||
return "document"
|
||||
if suffix in _SPREADSHEET_SUFFIXES:
|
||||
return "spreadsheet"
|
||||
if suffix in _TEXT_DATA_SUFFIXES:
|
||||
return "text_data"
|
||||
if suffix in _ARCHIVE_SUFFIXES:
|
||||
return "archive"
|
||||
return "other"
|
||||
|
||||
|
||||
def is_artifact_library_file(file: dict) -> bool:
|
||||
"""Return whether a generated file belongs in the document-only library."""
|
||||
mime_type = str(file.get("mime_type") or "").lower()
|
||||
if mime_type.startswith(_MEDIA_MIME_PREFIXES):
|
||||
return False
|
||||
return Path(str(file.get("name") or file.get("path") or "")).suffix.lower() not in _MEDIA_SUFFIXES
|
||||
|
||||
|
||||
async def _search_artifact_library_threads(thread_store) -> list[dict]:
|
||||
"""Load every normal conversation owned by the current request user."""
|
||||
rows: list[dict] = []
|
||||
seen: set[str] = set()
|
||||
offset = 0
|
||||
while True:
|
||||
batch = await thread_store.search(
|
||||
limit=_ARTIFACT_LIBRARY_THREAD_BATCH_SIZE,
|
||||
offset=offset,
|
||||
exclude_system=True,
|
||||
)
|
||||
if not batch:
|
||||
break
|
||||
added = 0
|
||||
for row in batch:
|
||||
thread_id = str(row.get("thread_id") or "")
|
||||
if not thread_id or thread_id in seen:
|
||||
continue
|
||||
seen.add(thread_id)
|
||||
rows.append(row)
|
||||
added += 1
|
||||
if len(batch) < _ARTIFACT_LIBRARY_THREAD_BATCH_SIZE or added == 0:
|
||||
break
|
||||
offset += len(batch)
|
||||
return rows
|
||||
|
||||
|
||||
async def _scan_artifact_library(thread_store) -> list[dict]:
|
||||
"""Scan generated outputs for all current-user conversation threads."""
|
||||
rows = await _search_artifact_library_threads(thread_store)
|
||||
paths = get_paths()
|
||||
semaphore = asyncio.Semaphore(_ARTIFACT_LIBRARY_SCAN_CONCURRENCY)
|
||||
|
||||
async def scan_thread(row: dict) -> list[dict]:
|
||||
thread_id = str(row.get("thread_id") or "")
|
||||
if not thread_id:
|
||||
return []
|
||||
async with semaphore:
|
||||
try:
|
||||
path_user_id = await aresolve_path_user_id(thread_id)
|
||||
outputs_dir = paths.sandbox_outputs_dir(thread_id, user_id=path_user_id)
|
||||
files = await asyncio.to_thread(collect_outputs_listing, outputs_dir)
|
||||
except Exception:
|
||||
logger.warning("Failed to scan artifacts for library thread %s", thread_id, exc_info=True)
|
||||
return []
|
||||
|
||||
metadata = row.get("metadata") if isinstance(row.get("metadata"), dict) else {}
|
||||
agent_id = metadata.get("agent_id")
|
||||
thread_type = metadata.get("thread_type")
|
||||
title = str(row.get("display_name") or "新对话")
|
||||
items: list[dict] = []
|
||||
for file in files:
|
||||
if not is_artifact_library_file(file):
|
||||
continue
|
||||
items.append(
|
||||
{
|
||||
"thread_id": thread_id,
|
||||
"thread_title": title,
|
||||
"agent_id": agent_id if isinstance(agent_id, str) and agent_id else None,
|
||||
"thread_type": thread_type if isinstance(thread_type, str) and thread_type else None,
|
||||
**file,
|
||||
"file_kind": artifact_library_kind(str(file.get("name") or "")),
|
||||
}
|
||||
)
|
||||
return items
|
||||
|
||||
nested = await asyncio.gather(*(scan_thread(row) for row in rows))
|
||||
items = [item for group in nested for item in group]
|
||||
items.sort(
|
||||
key=lambda item: (
|
||||
str(item.get("modified_at") or ""),
|
||||
str(item.get("thread_id") or ""),
|
||||
str(item.get("path") or ""),
|
||||
),
|
||||
reverse=True,
|
||||
)
|
||||
return items
|
||||
|
||||
|
||||
async def _cached_artifact_library(thread_store, *, refresh: bool) -> list[dict]:
|
||||
user_id = str(get_effective_user_id() or "default")
|
||||
now = time.monotonic()
|
||||
cached = _artifact_library_cache.get(user_id)
|
||||
if not refresh and cached and cached[0] > now:
|
||||
_artifact_library_cache.move_to_end(user_id)
|
||||
return cached[1]
|
||||
|
||||
items = await _scan_artifact_library(thread_store)
|
||||
_artifact_library_cache[user_id] = (now + _ARTIFACT_LIBRARY_CACHE_TTL_SECONDS, items)
|
||||
_artifact_library_cache.move_to_end(user_id)
|
||||
while len(_artifact_library_cache) > _ARTIFACT_LIBRARY_CACHE_MAX_USERS:
|
||||
_artifact_library_cache.popitem(last=False)
|
||||
return items
|
||||
|
||||
|
||||
@router.get(
|
||||
"/artifact-library",
|
||||
response_model=ArtifactLibraryResponse,
|
||||
summary="List Generated Files Across Conversations",
|
||||
description="Aggregate non-media files generated in the current user's normal conversations.",
|
||||
)
|
||||
@require_permission("threads", "read")
|
||||
async def list_artifact_library(
|
||||
request: Request,
|
||||
query: str | None = Query(default=None, max_length=200),
|
||||
file_kind: ArtifactLibraryKind = Query(default="all"),
|
||||
offset: int = Query(default=0, ge=0),
|
||||
limit: int = Query(default=50, ge=1, le=200),
|
||||
refresh: bool = Query(default=False),
|
||||
) -> ArtifactLibraryResponse:
|
||||
"""Return a flat, newest-first library of generated conversation files.
|
||||
|
||||
The filesystem remains the source of truth: every owned, non-system thread's
|
||||
``outputs`` directory is scanned, including files written without an
|
||||
explicit ``present_files`` call. Uploads and image/audio/video outputs are
|
||||
intentionally outside this collection.
|
||||
"""
|
||||
from app.gateway.deps import get_thread_store
|
||||
|
||||
items = await _cached_artifact_library(get_thread_store(request), refresh=refresh)
|
||||
normalized_query = (query or "").strip().casefold()
|
||||
filtered = [
|
||||
item
|
||||
for item in items
|
||||
if (file_kind == "all" or item["file_kind"] == file_kind) and (not normalized_query or normalized_query in str(item.get("name") or "").casefold() or normalized_query in str(item.get("thread_title") or "").casefold())
|
||||
]
|
||||
page = filtered[offset : offset + limit]
|
||||
return ArtifactLibraryResponse(
|
||||
total=len(filtered),
|
||||
offset=offset,
|
||||
limit=limit,
|
||||
items=[ArtifactLibraryItem.model_validate(item) for item in page],
|
||||
)
|
||||
|
||||
|
||||
@router.get(
|
||||
|
||||
@@ -55,6 +55,7 @@ from deerflow.config.agents_config import load_agent_config
|
||||
from deerflow.config.llmwiki_config import LlmWikiRuntimeOverride, normalize_service_url
|
||||
from deerflow.config.system_settings import load_system_settings, save_system_settings
|
||||
from deerflow.integrations.weknora.client import WeKnoraClient, WeKnoraError
|
||||
from deerflow.integrations.weknora.local_index.retrieval import is_fully_vectorized
|
||||
from deerflow.integrations.weknora.local_index.search import WikiIndexError
|
||||
from deerflow.integrations.weknora.runtime import build_weknora_client, get_resolved_llmwiki_runtime
|
||||
from deerflow.persistence.llmwiki import LlmWikiStore
|
||||
@@ -496,15 +497,7 @@ def _should_follow_weknora_binary_redirect(path: str) -> bool:
|
||||
"""Keep protected preview/image bytes inside the DeerFlow proxy."""
|
||||
|
||||
segments = _path_segments(path)
|
||||
return (
|
||||
len(segments) == 5
|
||||
and segments[:3] == ["api", "v1", "knowledge"]
|
||||
and segments[4] == "preview"
|
||||
) or (
|
||||
len(segments) == 5
|
||||
and segments[:3] == ["api", "v1", "knowledge-bases"]
|
||||
and segments[4] == "files"
|
||||
)
|
||||
return (len(segments) == 5 and segments[:3] == ["api", "v1", "knowledge"] and segments[4] == "preview") or (len(segments) == 5 and segments[:3] == ["api", "v1", "knowledge-bases"] and segments[4] == "files")
|
||||
|
||||
|
||||
def _forward_request_headers(request: Request) -> dict[str, str]:
|
||||
@@ -1833,6 +1826,9 @@ def _mapping_view(row: dict[str, Any], remote: dict[str, Any] | None, *, actor_u
|
||||
"knowledge_count": int(source.get("knowledge_count") or 0),
|
||||
"chunk_count": int(source.get("chunk_count") or 0),
|
||||
"processing_count": int(source.get("processing_count") or 0),
|
||||
"share_count": int(source.get("share_count") or 0),
|
||||
"is_processing": bool(source.get("is_processing") or int(source.get("processing_count") or 0) > 0),
|
||||
"is_temporary": bool(source.get("is_temporary", False)),
|
||||
"embedding_model_id": source.get("embedding_model_id"),
|
||||
"vector_store_id": source.get("vector_store_id"),
|
||||
"vector_store_name": source.get("vector_store_name"),
|
||||
@@ -1840,16 +1836,23 @@ def _mapping_view(row: dict[str, Any], remote: dict[str, Any] | None, *, actor_u
|
||||
"vector_store_engine_type": source.get("vector_store_engine_type"),
|
||||
"vector_store_status": source.get("vector_store_status"),
|
||||
"chunking_config": source.get("chunking_config") if isinstance(source.get("chunking_config"), dict) else {},
|
||||
"image_processing_config": (source.get("image_processing_config") if isinstance(source.get("image_processing_config"), dict) else {}),
|
||||
"indexing_strategy": source.get("indexing_strategy") if isinstance(source.get("indexing_strategy"), dict) else {},
|
||||
"capabilities": source.get("capabilities") if isinstance(source.get("capabilities"), dict) else {},
|
||||
"extract_config": source.get("extract_config") if isinstance(source.get("extract_config"), dict) else {},
|
||||
"faq_config": source.get("faq_config") if isinstance(source.get("faq_config"), dict) else {},
|
||||
"wiki_config": source.get("wiki_config") if isinstance(source.get("wiki_config"), dict) else {},
|
||||
"question_generation_config": (source.get("question_generation_config") if isinstance(source.get("question_generation_config"), dict) else {}),
|
||||
"auto_tag_config": source.get("auto_tag_config") if isinstance(source.get("auto_tag_config"), dict) else {},
|
||||
"vlm_config": source.get("vlm_config") if isinstance(source.get("vlm_config"), dict) else {},
|
||||
"asr_config": source.get("asr_config") if isinstance(source.get("asr_config"), dict) else {},
|
||||
"summary_model_id": source.get("summary_model_id"),
|
||||
"storage_provider": (source.get("storage_provider_config", {}).get("provider") if isinstance(source.get("storage_provider_config"), dict) else None),
|
||||
"created_at": row.get("created_at"),
|
||||
"updated_at": row.get("updated_at"),
|
||||
"published_at": row.get("published_at"),
|
||||
"remote_created_at": source.get("created_at"),
|
||||
"remote_updated_at": source.get("updated_at"),
|
||||
}
|
||||
|
||||
|
||||
@@ -1877,11 +1880,7 @@ async def _public_mapping(request: Request, mapping_id: str) -> dict[str, Any]:
|
||||
write=False,
|
||||
is_admin=False,
|
||||
)
|
||||
if (
|
||||
row is None
|
||||
or row.get("publication_status") != "published"
|
||||
or _is_conversation_deposit_mapping(row)
|
||||
):
|
||||
if row is None or row.get("publication_status") != "published" or _is_conversation_deposit_mapping(row):
|
||||
raise HTTPException(status_code=404, detail="Published knowledge base not found")
|
||||
return row
|
||||
|
||||
@@ -2537,24 +2536,260 @@ async def test_llmwiki_connection(request: Request, body: LlmWikiConnectionTest)
|
||||
return {"success": True, "provider": "weknora", "health": health}
|
||||
|
||||
|
||||
_PUBLIC_WIKI_SUMMARY_CONCURRENCY = 8
|
||||
_PUBLIC_WIKI_PAGE_SIZE = 500
|
||||
_PUBLIC_WIKI_PAGE_LIMIT = 100
|
||||
_GENERATED_PUBLIC_WIKI_INDEX_SLUG = "__index__"
|
||||
|
||||
|
||||
def _is_public_wiki_index_page(page: dict[str, Any]) -> bool:
|
||||
page_type = str(page.get("page_type") or "").strip().lower()
|
||||
title = str(page.get("title") or "").strip().lower()
|
||||
slug_name = str(page.get("slug") or "").strip("/").rsplit("/", 1)[-1].lower()
|
||||
return page_type == "index" or title in {"索引", "index"} or slug_name in {"index", "home", "readme"}
|
||||
|
||||
|
||||
def _public_wiki_default_page(pages: list[dict[str, Any]]) -> dict[str, Any] | None:
|
||||
return (
|
||||
next((page for page in pages if _is_public_wiki_index_page(page)), None)
|
||||
or next((page for page in pages if str(page.get("page_type") or "").lower() == "overview"), None)
|
||||
or next(
|
||||
(page for page in pages if not str(page.get("parent_slug") or "").strip() and int(page.get("depth") or 0) == 0),
|
||||
None,
|
||||
)
|
||||
or (pages[0] if pages else None)
|
||||
)
|
||||
|
||||
|
||||
def _count_public_wiki_values(pages: list[dict[str, Any]], key: str, fallback: str) -> dict[str, int]:
|
||||
counts: dict[str, int] = {}
|
||||
for page in pages:
|
||||
value = str(page.get(key) or fallback).strip().lower() or fallback
|
||||
counts[value] = counts.get(value, 0) + 1
|
||||
return counts
|
||||
|
||||
|
||||
def _public_wiki_time_bound(pages: list[dict[str, Any]], key: str, *, latest: bool) -> str | None:
|
||||
values = [str(page.get(key) or "").strip() for page in pages]
|
||||
values = [value for value in values if value]
|
||||
if not values:
|
||||
return None
|
||||
return max(values) if latest else min(values)
|
||||
|
||||
|
||||
async def _all_public_wiki_page_views(
|
||||
client: WeKnoraClient,
|
||||
remote_id: str,
|
||||
) -> tuple[list[dict[str, Any]], int, bool]:
|
||||
pages_by_slug: dict[str, dict[str, Any]] = {}
|
||||
remote_total = 0
|
||||
complete = False
|
||||
for page_number in range(1, _PUBLIC_WIKI_PAGE_LIMIT + 1):
|
||||
result = await client.list_wiki_pages(
|
||||
remote_id,
|
||||
page=page_number,
|
||||
page_size=_PUBLIC_WIKI_PAGE_SIZE,
|
||||
query="",
|
||||
)
|
||||
raw_pages = result.get("pages") if isinstance(result, dict) else []
|
||||
page_items = raw_pages if isinstance(raw_pages, list) else []
|
||||
for page_item in page_items:
|
||||
if not isinstance(page_item, dict):
|
||||
continue
|
||||
page = _wiki_page_view(page_item)
|
||||
if page is None:
|
||||
continue
|
||||
slug = str(page.get("slug") or "").strip()
|
||||
pages_by_slug[slug or f"__page_{len(pages_by_slug)}"] = page
|
||||
reported_total = int(result.get("total") or 0)
|
||||
remote_total = max(remote_total, reported_total, len(pages_by_slug))
|
||||
total_pages = max(1, int(result.get("total_pages") or 1))
|
||||
if not page_items or page_number >= total_pages or (reported_total > 0 and len(pages_by_slug) >= remote_total):
|
||||
complete = True
|
||||
break
|
||||
return list(pages_by_slug.values()), max(remote_total, len(pages_by_slug)), complete
|
||||
|
||||
|
||||
def _public_wiki_summary(
|
||||
mapping_id: str,
|
||||
pages: list[dict[str, Any]],
|
||||
*,
|
||||
total: int,
|
||||
complete: bool,
|
||||
) -> dict[str, Any]:
|
||||
explicit_index = next((page for page in pages if _is_public_wiki_index_page(page)), None)
|
||||
default_page = _public_wiki_default_page(pages)
|
||||
if explicit_index is not None:
|
||||
index_mode = "explicit"
|
||||
index_slug: str | None = str(explicit_index.get("slug") or "") or None
|
||||
elif pages:
|
||||
index_mode = "generated"
|
||||
index_slug = _GENERATED_PUBLIC_WIKI_INDEX_SLUG
|
||||
else:
|
||||
index_mode = "none"
|
||||
index_slug = None
|
||||
default_slug = index_slug or (str(default_page.get("slug") or "") if default_page else None)
|
||||
route = f"/embed/knowledge/{quote(mapping_id, safe='')}"
|
||||
if default_slug:
|
||||
route = f"{route}?{urlencode({'wikiSlug': default_slug})}"
|
||||
return {
|
||||
"status": "ready" if pages else "empty",
|
||||
"page_count": total,
|
||||
"loaded_page_count": len(pages),
|
||||
"stats_complete": complete,
|
||||
"page_type_counts": _count_public_wiki_values(pages, "page_type", "page"),
|
||||
"page_status_counts": _count_public_wiki_values(pages, "status", "unknown"),
|
||||
"index_mode": index_mode,
|
||||
"has_explicit_index": explicit_index is not None,
|
||||
"index_slug": index_slug,
|
||||
"default_slug": default_slug,
|
||||
"created_at": _public_wiki_time_bound(pages, "created_at", latest=False),
|
||||
"updated_at": _public_wiki_time_bound(pages, "updated_at", latest=True),
|
||||
"embed_route": route,
|
||||
}
|
||||
|
||||
|
||||
def _unavailable_public_wiki_summary(mapping_id: str) -> dict[str, Any]:
|
||||
return {
|
||||
"status": "unavailable",
|
||||
"page_count": 0,
|
||||
"loaded_page_count": 0,
|
||||
"stats_complete": False,
|
||||
"page_type_counts": {},
|
||||
"page_status_counts": {},
|
||||
"index_mode": "unknown",
|
||||
"has_explicit_index": False,
|
||||
"index_slug": None,
|
||||
"default_slug": None,
|
||||
"created_at": None,
|
||||
"updated_at": None,
|
||||
"embed_route": f"/embed/knowledge/{quote(mapping_id, safe='')}",
|
||||
}
|
||||
|
||||
|
||||
def _public_local_index_summary(
|
||||
request: Request,
|
||||
mapping: dict[str, Any],
|
||||
state: dict[str, Any] | None,
|
||||
) -> dict[str, Any]:
|
||||
app_state = request.app.state
|
||||
local_config = getattr(
|
||||
getattr(getattr(app_state, "config", None), "llmwiki", None),
|
||||
"local_wiki_index",
|
||||
None,
|
||||
)
|
||||
enabled = bool(getattr(local_config, "enabled", False))
|
||||
current = dict(state or {})
|
||||
state_name = str(current.get("state") or "not_synced")
|
||||
if state_name == "idle" and int(current.get("index_revision") or 0) == 0 and not current.get("last_completed_at") and not current.get("last_success_at"):
|
||||
state_name = "not_synced"
|
||||
embedding = getattr(app_state, "llmwiki_embedding", None)
|
||||
fingerprint = embedding.fingerprint if embedding is not None else None
|
||||
fully_vectorized = bool(mapping.get("wiki_index_enabled", True)) and is_fully_vectorized(
|
||||
current,
|
||||
fingerprint,
|
||||
)
|
||||
return {
|
||||
"enabled": enabled,
|
||||
"wiki_index_enabled": bool(mapping.get("wiki_index_enabled", True)),
|
||||
"state": state_name,
|
||||
"fully_vectorized": fully_vectorized,
|
||||
"retrieval_mode": "vector" if fully_vectorized else "wiki_api",
|
||||
"remote_page_count": int(current.get("remote_page_count") or 0),
|
||||
"local_page_count": int(current.get("local_page_count") or 0),
|
||||
"ready_page_count": int(current.get("ready_page_count") or 0),
|
||||
"failed_page_count": int(current.get("failed_page_count") or 0),
|
||||
"vector_count": int(current.get("vector_count") or 0),
|
||||
"last_started_at": current.get("last_started_at"),
|
||||
"last_completed_at": current.get("last_completed_at"),
|
||||
"last_success_at": current.get("last_success_at"),
|
||||
"updated_at": current.get("updated_at"),
|
||||
}
|
||||
|
||||
|
||||
@router.get("/api/public/llmwiki/knowledge-bases")
|
||||
async def list_public_knowledge_bases(request: Request) -> dict[str, Any]:
|
||||
"""List the minimal identity of every knowledge base published as a public Wiki."""
|
||||
"""List live, published Wiki libraries with safe metadata and index health."""
|
||||
|
||||
rows = await _store(request).list_visible(
|
||||
"__public_wiki__",
|
||||
scope="public",
|
||||
is_admin=False,
|
||||
)
|
||||
knowledge_bases = [
|
||||
{
|
||||
"id": str(row["id"]),
|
||||
"name": str(row.get("name") or ""),
|
||||
}
|
||||
for row in rows
|
||||
if row.get("publication_status") == "published"
|
||||
and not _is_conversation_deposit_mapping(row)
|
||||
rows = [
|
||||
row
|
||||
for row in await _store(request).list_visible(
|
||||
"__public_wiki__",
|
||||
scope="public",
|
||||
is_admin=False,
|
||||
)
|
||||
if row.get("publication_status") == "published" and not _is_conversation_deposit_mapping(row)
|
||||
]
|
||||
client = _client_or_503()
|
||||
try:
|
||||
remote_items = await client.list_knowledge_bases()
|
||||
except WeKnoraError as exc:
|
||||
_raise_upstream(exc)
|
||||
remote_by_id = {str(item.get("id")): item for item in remote_items if isinstance(item, dict) and item.get("id")}
|
||||
# A successful upstream list is authoritative: stale DeerFlow mappings are
|
||||
# omitted so discovery never advertises a Wiki that its detail route cannot open.
|
||||
rows = [row for row in rows if str(row.get("weknora_id") or "") in remote_by_id]
|
||||
mapping_ids = [str(row["id"]) for row in rows]
|
||||
index_states: dict[str, dict[str, Any]] = {}
|
||||
index_store = getattr(request.app.state, "llmwiki_index_store", None)
|
||||
if index_store is not None and mapping_ids:
|
||||
try:
|
||||
index_states = {str(state.get("knowledge_base_mapping_id") or ""): state for state in await index_store.list_index_status(mapping_ids) if isinstance(state, dict)}
|
||||
except Exception:
|
||||
logger.warning("Failed to read public Wiki local-index summaries", exc_info=True)
|
||||
|
||||
semaphore = asyncio.Semaphore(_PUBLIC_WIKI_SUMMARY_CONCURRENCY)
|
||||
|
||||
async def build_summary(row: dict[str, Any]) -> dict[str, Any]:
|
||||
mapping_id = str(row["id"])
|
||||
remote_id = str(row["weknora_id"])
|
||||
remote = remote_by_id[remote_id]
|
||||
async with semaphore:
|
||||
try:
|
||||
remote = {**remote, **await client.get_knowledge_base(remote_id)}
|
||||
except WeKnoraError as exc:
|
||||
if exc.status_code == 404:
|
||||
return {}
|
||||
logger.warning(
|
||||
"Failed to load public Wiki detail metadata: mapping_id=%s",
|
||||
mapping_id,
|
||||
exc_info=True,
|
||||
)
|
||||
try:
|
||||
pages, page_total, complete = await _all_public_wiki_page_views(client, remote_id)
|
||||
wiki = _public_wiki_summary(
|
||||
mapping_id,
|
||||
pages,
|
||||
total=page_total,
|
||||
complete=complete,
|
||||
)
|
||||
except WeKnoraError as exc:
|
||||
logger.warning(
|
||||
"Failed to load public Wiki page summary: mapping_id=%s",
|
||||
mapping_id,
|
||||
exc_info=True,
|
||||
)
|
||||
wiki = (
|
||||
_public_wiki_summary(mapping_id, [], total=0, complete=True)
|
||||
if exc.status_code == 404
|
||||
else _unavailable_public_wiki_summary(mapping_id)
|
||||
)
|
||||
return {
|
||||
**_mapping_view(
|
||||
row,
|
||||
remote,
|
||||
actor_user_id="__public_wiki__",
|
||||
is_admin=False,
|
||||
),
|
||||
"wiki": wiki,
|
||||
"local_index": _public_local_index_summary(
|
||||
request,
|
||||
row,
|
||||
index_states.get(mapping_id),
|
||||
),
|
||||
}
|
||||
|
||||
knowledge_bases = [summary for summary in await asyncio.gather(*(build_summary(row) for row in rows)) if summary]
|
||||
return {"knowledge_bases": knowledge_bases, "total": len(knowledge_bases)}
|
||||
|
||||
|
||||
@@ -2596,13 +2831,7 @@ async def list_public_wiki_pages(
|
||||
except WeKnoraError as exc:
|
||||
_raise_upstream(exc)
|
||||
raw_pages = result.get("pages") if isinstance(result, dict) else []
|
||||
pages = [
|
||||
view
|
||||
for page_item in (raw_pages if isinstance(raw_pages, list) else [])
|
||||
if isinstance(page_item, dict)
|
||||
for view in [_wiki_page_view(page_item)]
|
||||
if view is not None
|
||||
]
|
||||
pages = [view for page_item in (raw_pages if isinstance(raw_pages, list) else []) if isinstance(page_item, dict) for view in [_wiki_page_view(page_item)] if view is not None]
|
||||
return {**result, "pages": pages}
|
||||
|
||||
|
||||
@@ -2650,11 +2879,7 @@ def _public_wiki_index_item(value: Any) -> dict[str, Any] | None:
|
||||
"slug": slug,
|
||||
"title": str(value.get("title") or slug),
|
||||
"summary": str(value.get("summary") or ""),
|
||||
"category_path": (
|
||||
[str(item) for item in value.get("category_path", []) if str(item).strip()]
|
||||
if isinstance(value.get("category_path"), list)
|
||||
else []
|
||||
),
|
||||
"category_path": ([str(item) for item in value.get("category_path", []) if str(item).strip()] if isinstance(value.get("category_path"), list) else []),
|
||||
"wiki_path": str(value.get("wiki_path") or ""),
|
||||
"depth": int(value.get("depth") or 0),
|
||||
}
|
||||
@@ -2680,20 +2905,12 @@ async def _all_public_wiki_index(client: WeKnoraClient, remote_id: str) -> dict[
|
||||
version = max(version, int(payload.get("version") or 0))
|
||||
raw_groups = payload.get("groups") if isinstance(payload.get("groups"), list) else []
|
||||
group = next(
|
||||
(
|
||||
value
|
||||
for value in raw_groups
|
||||
if isinstance(value, dict) and value.get("type") == page_type
|
||||
),
|
||||
(value for value in raw_groups if isinstance(value, dict) and value.get("type") == page_type),
|
||||
{},
|
||||
)
|
||||
total = max(total, int(group.get("total") or 0))
|
||||
raw_items = group.get("items") if isinstance(group.get("items"), list) else []
|
||||
items.extend(
|
||||
item
|
||||
for value in raw_items
|
||||
if (item := _public_wiki_index_item(value)) is not None
|
||||
)
|
||||
items.extend(item for value in raw_items if (item := _public_wiki_index_item(value)) is not None)
|
||||
next_cursor = str(group.get("next_cursor") or "")
|
||||
if not next_cursor or next_cursor in seen_cursors:
|
||||
break
|
||||
@@ -3201,12 +3418,7 @@ async def search_llmwiki(request: Request, body: SearchRequest) -> dict[str, Any
|
||||
except WikiIndexError as exc:
|
||||
raise HTTPException(status_code=exc.status_code, detail={"code": exc.code, "message": str(exc)}) from None
|
||||
mapping_by_id = {str(row["id"]): row for row in rows}
|
||||
result["results"] = [
|
||||
_sanitize_conversation_deposit_value(item)
|
||||
if _is_conversation_deposit_mapping(mapping_by_id.get(str(item.get("knowledge_base_id") or ""), {}))
|
||||
else item
|
||||
for item in result.get("results") or []
|
||||
]
|
||||
result["results"] = [_sanitize_conversation_deposit_value(item) if _is_conversation_deposit_mapping(mapping_by_id.get(str(item.get("knowledge_base_id") or ""), {})) else item for item in result.get("results") or []]
|
||||
return result
|
||||
|
||||
|
||||
|
||||
Reference in new issue
Block a user