jpmschweitzerandClaude Opus 4.5
e05e7aeae3
Add POST /wiki/pages/smart-create endpoint that combines research with content generation for the librarian agent: - Run HybridRAG search on topic (wiki + graph + web) - Use LLM (WikiPageWriter) to synthesize findings into wiki content - Create page with proper attribution and sources - Schedule background tasks for vector/graph indexing - Apply bidirectional entity linking (forward + backward links) New files: - src/services/entity_linking_utils.py - shared entity linking helper Modified: - src/models/wiki.py - WikiSmartCreateRequest/Response models - src/services/wiki_service.py - smart_create_page() method - src/routers/wiki.py - /pages/smart-create endpoint 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
597 lines
18 KiB
Python
597 lines
18 KiB
Python
"""
|
|
Wiki service layer for Library Desk.
|
|
|
|
Handles business logic for wiki operations with:
|
|
- Multi-tenant path scoping
|
|
- Dossier management (tag-based)
|
|
- Page CRUD operations
|
|
- Search functionality
|
|
"""
|
|
|
|
from typing import List, Optional, Dict, Any
|
|
import logging
|
|
|
|
from src.clients.wikijs_client import WikiJSClient
|
|
from src.core.multi_tenancy import get_wikijs_namespace, validate_user_id, DEFAULT_USER
|
|
from src.models.wiki import (
|
|
WikiPage, WikiPageSummary, WikiPageList,
|
|
WikiPageCreate, WikiPageUpdate,
|
|
DossierInfo, DossierList
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class WikiService:
|
|
"""
|
|
Service layer for wiki operations.
|
|
|
|
Responsibilities:
|
|
- Enforce multi-tenant path scoping
|
|
- Convert between client and API models
|
|
- Handle dossier (tag) operations
|
|
- Provide business logic layer
|
|
"""
|
|
|
|
def __init__(self, wiki_client: WikiJSClient):
|
|
"""
|
|
Initialize wiki service.
|
|
|
|
Args:
|
|
wiki_client: Initialized Wiki.js client
|
|
"""
|
|
self.wiki_client = wiki_client
|
|
|
|
def _get_user_namespace(self, user: str) -> str:
|
|
"""
|
|
Get user's wiki namespace with validation.
|
|
|
|
Args:
|
|
user: User identifier
|
|
|
|
Returns:
|
|
Wiki.js namespace path
|
|
|
|
Raises:
|
|
ValueError: If user ID is invalid
|
|
"""
|
|
if not validate_user_id(user):
|
|
raise ValueError(f"Invalid user ID: {user}")
|
|
return get_wikijs_namespace(user)
|
|
|
|
def _ensure_user_path(self, path: str, user: str) -> str:
|
|
"""
|
|
Ensure path is within user's namespace.
|
|
|
|
Args:
|
|
path: Requested page path
|
|
user: User identifier
|
|
|
|
Returns:
|
|
Full path within user namespace
|
|
|
|
Example:
|
|
>>> self._ensure_user_path("/projects/foo", "jpmschweitzer")
|
|
'/users/jpmschweitzer/projects/foo'
|
|
"""
|
|
namespace = self._get_user_namespace(user)
|
|
|
|
# If path already starts with namespace, return as-is
|
|
if path.startswith(namespace):
|
|
return path
|
|
|
|
# Remove leading slash from path if present
|
|
path = path.lstrip("/")
|
|
|
|
# Combine namespace and path
|
|
return f"{namespace}/{path}"
|
|
|
|
async def list_pages(
|
|
self,
|
|
user: str,
|
|
tag: Optional[str] = None,
|
|
limit: int = 50
|
|
) -> WikiPageList:
|
|
"""
|
|
List pages for a user, optionally filtered by tag.
|
|
|
|
Args:
|
|
user: User identifier
|
|
tag: Optional tag filter (dossier)
|
|
limit: Maximum pages to return
|
|
|
|
Returns:
|
|
WikiPageList with pages and metadata
|
|
"""
|
|
namespace = self._get_user_namespace(user)
|
|
|
|
# Get pages with filtering
|
|
pages = await self.wiki_client.list_pages(
|
|
path_prefix=namespace,
|
|
tags=[tag] if tag else None,
|
|
limit=limit
|
|
)
|
|
|
|
# Convert to summary format
|
|
summaries = [
|
|
WikiPageSummary(
|
|
id=p["id"],
|
|
path=p["path"],
|
|
title=p["title"],
|
|
description=p.get("description"),
|
|
tags=p.get("tags", []),
|
|
updated_at=p.get("updatedAt"),
|
|
is_published=p.get("isPublished", True)
|
|
)
|
|
for p in pages
|
|
]
|
|
|
|
return WikiPageList(
|
|
pages=summaries,
|
|
total=len(summaries),
|
|
filtered_by_tag=tag,
|
|
user=user
|
|
)
|
|
|
|
async def get_page(self, page_id: int, user: str) -> Optional[WikiPage]:
|
|
"""
|
|
Get a single page by ID.
|
|
|
|
Args:
|
|
page_id: Page ID
|
|
user: User identifier (for validation)
|
|
|
|
Returns:
|
|
WikiPage or None if not found or access denied
|
|
|
|
Note: Validates that page belongs to user's namespace
|
|
"""
|
|
page = await self.wiki_client.get_page(page_id)
|
|
|
|
if not page:
|
|
return None
|
|
|
|
# Validate page is in user's namespace
|
|
namespace = self._get_user_namespace(user)
|
|
page_path = "/" + page["path"].lstrip("/") # Normalize path with leading slash
|
|
if not page_path.startswith(namespace):
|
|
logger.warning(f"User {user} attempted to access page outside namespace: {page['path']}")
|
|
return None
|
|
|
|
return WikiPage(
|
|
id=page["id"],
|
|
path=page["path"],
|
|
title=page["title"],
|
|
description=page.get("description"),
|
|
content=page.get("content"),
|
|
tags=page.get("tags", []),
|
|
created_at=page.get("createdAt"),
|
|
updated_at=page.get("updatedAt"),
|
|
is_published=page.get("isPublished", True),
|
|
editor=page.get("editor")
|
|
)
|
|
|
|
async def create_page(self, page_data: WikiPageCreate) -> WikiPage:
|
|
"""
|
|
Create a new wiki page.
|
|
|
|
Args:
|
|
page_data: Page creation data
|
|
|
|
Returns:
|
|
Created WikiPage
|
|
|
|
Raises:
|
|
ValueError: If creation fails
|
|
"""
|
|
user = page_data.user or DEFAULT_USER
|
|
|
|
# Ensure path is in user's namespace
|
|
full_path = self._ensure_user_path(page_data.path, user)
|
|
|
|
try:
|
|
created = await self.wiki_client.create_page(
|
|
path=full_path,
|
|
title=page_data.title,
|
|
content=page_data.content,
|
|
description=page_data.description or "",
|
|
tags=page_data.tags,
|
|
is_published=page_data.is_published,
|
|
editor=page_data.editor
|
|
)
|
|
|
|
# Fetch full page details
|
|
page = await self.wiki_client.get_page(created["id"])
|
|
if not page:
|
|
raise ValueError("Page created but could not be retrieved")
|
|
|
|
return WikiPage(
|
|
id=page["id"],
|
|
path=page["path"],
|
|
title=page["title"],
|
|
description=page.get("description"),
|
|
content=page.get("content"),
|
|
tags=page.get("tags", []),
|
|
created_at=page.get("createdAt"),
|
|
updated_at=page.get("updatedAt"),
|
|
is_published=page.get("isPublished", True),
|
|
editor=page.get("editor")
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to create page: {e}", exc_info=True)
|
|
raise ValueError(f"Failed to create page: {str(e)}")
|
|
|
|
async def update_page(
|
|
self,
|
|
page_id: int,
|
|
page_data: WikiPageUpdate,
|
|
user: str
|
|
) -> WikiPage:
|
|
"""
|
|
Update an existing page.
|
|
|
|
Args:
|
|
page_id: Page ID to update
|
|
page_data: Update data
|
|
user: User identifier (for validation)
|
|
|
|
Returns:
|
|
Updated WikiPage
|
|
|
|
Raises:
|
|
ValueError: If page not found or update fails
|
|
"""
|
|
# Verify page exists and belongs to user
|
|
existing = await self.get_page(page_id, user)
|
|
if not existing:
|
|
raise ValueError(f"Page {page_id} not found or access denied")
|
|
|
|
try:
|
|
await self.wiki_client.update_page(
|
|
page_id=page_id,
|
|
content=page_data.content,
|
|
title=page_data.title,
|
|
description=page_data.description,
|
|
tags=page_data.tags,
|
|
is_published=True # Always keep pages published for internal wiki
|
|
)
|
|
|
|
# Fetch updated page
|
|
updated = await self.get_page(page_id, user)
|
|
if not updated:
|
|
raise ValueError("Page updated but could not be retrieved")
|
|
|
|
return updated
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to update page {page_id}: {e}", exc_info=True)
|
|
raise ValueError(f"Failed to update page: {str(e)}")
|
|
|
|
async def delete_page(self, page_id: int, user: str) -> bool:
|
|
"""
|
|
Delete a page.
|
|
|
|
Args:
|
|
page_id: Page ID to delete
|
|
user: User identifier (for validation)
|
|
|
|
Returns:
|
|
True if deleted successfully
|
|
|
|
Raises:
|
|
ValueError: If page not found or deletion fails
|
|
"""
|
|
# Verify page exists and belongs to user
|
|
existing = await self.get_page(page_id, user)
|
|
if not existing:
|
|
raise ValueError(f"Page {page_id} not found or access denied")
|
|
|
|
try:
|
|
await self.wiki_client.delete_page(page_id)
|
|
logger.info(f"Deleted page {page_id} for user {user}")
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to delete page {page_id}: {e}", exc_info=True)
|
|
raise ValueError(f"Failed to delete page: {str(e)}")
|
|
|
|
async def search_pages(
|
|
self,
|
|
query: str,
|
|
user: str,
|
|
limit: int = 20
|
|
) -> List[WikiPageSummary]:
|
|
"""
|
|
Search pages in user's namespace.
|
|
|
|
Args:
|
|
query: Search query
|
|
user: User identifier
|
|
limit: Maximum results
|
|
|
|
Returns:
|
|
List of matching pages
|
|
"""
|
|
namespace = self._get_user_namespace(user)
|
|
|
|
results = await self.wiki_client.search_pages(
|
|
query=query,
|
|
path_prefix=namespace
|
|
)
|
|
|
|
# Convert to summaries (limit results)
|
|
return [
|
|
WikiPageSummary(
|
|
id=r["id"],
|
|
path=r["path"],
|
|
title=r["title"],
|
|
description=r.get("description"),
|
|
tags=[], # Search results don't include tags
|
|
updated_at=None,
|
|
is_published=True
|
|
)
|
|
for r in results[:limit]
|
|
]
|
|
|
|
async def move_page(
|
|
self,
|
|
page_id: int,
|
|
new_path: str,
|
|
user: str
|
|
) -> bool:
|
|
"""
|
|
Move/rename a page.
|
|
|
|
Args:
|
|
page_id: Page ID to move
|
|
new_path: New path (within user namespace)
|
|
user: User identifier
|
|
|
|
Returns:
|
|
True if moved successfully
|
|
|
|
Raises:
|
|
ValueError: If operation fails
|
|
"""
|
|
# Verify page exists and belongs to user
|
|
existing = await self.get_page(page_id, user)
|
|
if not existing:
|
|
raise ValueError(f"Page {page_id} not found or access denied")
|
|
|
|
# Ensure new path is in user's namespace
|
|
full_new_path = self._ensure_user_path(new_path, user)
|
|
|
|
try:
|
|
success = await self.wiki_client.move_page(page_id, full_new_path)
|
|
if success:
|
|
logger.info(f"Moved page {page_id} to {full_new_path}")
|
|
return success
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to move page {page_id}: {e}", exc_info=True)
|
|
raise ValueError(f"Failed to move page: {str(e)}")
|
|
|
|
# Dossier operations (tag-based)
|
|
|
|
async def list_dossiers(self, user: str) -> DossierList:
|
|
"""
|
|
List all dossiers (unique tags) for a user.
|
|
|
|
Args:
|
|
user: User identifier
|
|
|
|
Returns:
|
|
DossierList with all dossiers
|
|
"""
|
|
# Get all pages for user
|
|
pages = await self.list_pages(user, limit=1000)
|
|
|
|
# Collect unique tags
|
|
tag_counts: Dict[str, int] = {}
|
|
for page in pages.pages:
|
|
for tag in page.tags:
|
|
tag_counts[tag] = tag_counts.get(tag, 0) + 1
|
|
|
|
# Create dossier info for each tag
|
|
dossiers = [
|
|
DossierInfo(
|
|
name=tag,
|
|
title=tag.replace("-", " ").title(),
|
|
description=f"Dossier for {tag}",
|
|
page_count=count,
|
|
index_page_id=None,
|
|
index_page_path=None,
|
|
created_at=None
|
|
)
|
|
for tag, count in tag_counts.items()
|
|
]
|
|
|
|
return DossierList(
|
|
dossiers=sorted(dossiers, key=lambda d: d.page_count, reverse=True),
|
|
total=len(dossiers),
|
|
user=user
|
|
)
|
|
|
|
async def get_dossier_pages(
|
|
self,
|
|
dossier_name: str,
|
|
user: str,
|
|
limit: int = 100
|
|
) -> WikiPageList:
|
|
"""
|
|
Get all pages in a dossier (by tag).
|
|
|
|
Args:
|
|
dossier_name: Dossier name (tag)
|
|
user: User identifier
|
|
limit: Maximum pages
|
|
|
|
Returns:
|
|
WikiPageList filtered by dossier tag
|
|
"""
|
|
return await self.list_pages(user, tag=dossier_name, limit=limit)
|
|
|
|
async def smart_create_page(
|
|
self,
|
|
topic: str,
|
|
user: str,
|
|
path: Optional[str],
|
|
tags: List[str],
|
|
hybrid_rag_service: "HybridRAGService",
|
|
wiki_page_writer: "WikiPageWriter",
|
|
include_web: bool = True,
|
|
include_wiki: bool = True
|
|
) -> tuple["WikiPage", Dict[str, Any]]:
|
|
"""
|
|
Create wiki page with research from HybridRAG.
|
|
|
|
This method combines research + content generation + page creation:
|
|
1. Run HybridRAG search on topic
|
|
2. Format results for WikiPageWriter
|
|
3. Generate page content with LLM
|
|
4. Create page in Wiki.js
|
|
5. Return page + research summary
|
|
|
|
Args:
|
|
topic: Topic to research and create page about
|
|
user: User identifier
|
|
path: Optional page path (auto-generated from topic if not provided)
|
|
tags: Tags for the page
|
|
hybrid_rag_service: HybridRAG service for multi-source search
|
|
wiki_page_writer: WikiPageWriter for LLM content generation
|
|
include_web: Include web search results
|
|
include_wiki: Include existing wiki knowledge
|
|
|
|
Returns:
|
|
Tuple of (created WikiPage, research summary dict)
|
|
"""
|
|
from src.models.hybrid_rag import HybridRAGConfig
|
|
|
|
logger.info(f"Smart create page: topic='{topic}', user='{user}'")
|
|
|
|
# Step 1: Run HybridRAG search on the topic
|
|
config = HybridRAGConfig(
|
|
enable_vector=include_wiki,
|
|
enable_graph=include_wiki,
|
|
enable_web=include_web,
|
|
enable_reranking=True,
|
|
enable_enrichment=True,
|
|
final_result_count=15 # Get more results for rich content
|
|
)
|
|
|
|
search_response = await hybrid_rag_service.search(
|
|
query=topic,
|
|
user=user,
|
|
config=config
|
|
)
|
|
|
|
logger.info(
|
|
f"HybridRAG search completed: {search_response.total_results} results, "
|
|
f"search_id={search_response.search_id}"
|
|
)
|
|
|
|
# Step 2: Format results for WikiPageWriter
|
|
source_information = []
|
|
wiki_results_count = 0
|
|
web_results_count = 0
|
|
graph_entities_count = 0
|
|
|
|
for result in search_response.results:
|
|
source_type = result.source_type
|
|
|
|
if "web" in source_type:
|
|
web_results_count += 1
|
|
source_information.append({
|
|
"title": result.title,
|
|
"url": result.url or "",
|
|
"content": result.content[:500] if result.content else ""
|
|
})
|
|
elif "vector" in source_type or "graph" in source_type:
|
|
wiki_results_count += 1
|
|
# For wiki results, use page path as URL
|
|
source_information.append({
|
|
"title": result.title,
|
|
"url": f"/{result.page_path}" if result.page_path else "",
|
|
"content": result.content[:500] if result.content else ""
|
|
})
|
|
|
|
# Count entities from related dossiers
|
|
if result.related_dossiers:
|
|
graph_entities_count += len(result.related_dossiers)
|
|
|
|
# Step 3: Generate page content with LLM
|
|
# Use topic as summary and let WikiPageWriter create structured content
|
|
topic_summary = f"Research findings about: {topic}"
|
|
if search_response.keywords:
|
|
topic_summary += f"\n\nKey concepts: {', '.join(search_response.keywords.core_keywords)}"
|
|
|
|
# Extract entities from search results for knowledge graph linking
|
|
entities = []
|
|
if search_response.keywords and search_response.keywords.core_keywords:
|
|
entities = search_response.keywords.core_keywords[:10]
|
|
|
|
# Get related documents for cross-linking
|
|
related_docs = []
|
|
for result in search_response.results[:5]:
|
|
if result.page_path:
|
|
related_docs.append(f"[{result.title}](/{result.page_path})")
|
|
|
|
content = await wiki_page_writer.create_page(
|
|
title=topic,
|
|
topic_summary=topic_summary,
|
|
source_information=source_information[:10], # Limit sources
|
|
entities=entities,
|
|
related_docs=related_docs
|
|
)
|
|
|
|
logger.info(f"Generated page content: {len(content)} characters")
|
|
|
|
# Step 4: Auto-generate path from topic if not provided
|
|
if not path:
|
|
# Convert topic to kebab-case path
|
|
import re
|
|
path_slug = topic.lower()
|
|
path_slug = re.sub(r'[^\w\s-]', '', path_slug) # Remove special chars
|
|
path_slug = re.sub(r'\s+', '-', path_slug) # Spaces to hyphens
|
|
path_slug = re.sub(r'-+', '-', path_slug) # Multiple hyphens to single
|
|
path_slug = path_slug.strip('-')
|
|
|
|
# Infer category from tags or use reference
|
|
category = "reference"
|
|
if tags:
|
|
category = tags[0].lower()
|
|
|
|
path = f"/{category}/{path_slug}"
|
|
|
|
# Step 5: Create page using existing create_page method
|
|
from src.models.wiki import WikiPageCreate
|
|
|
|
page_data = WikiPageCreate(
|
|
title=topic,
|
|
path=path,
|
|
content=content,
|
|
description=f"Research summary about {topic}",
|
|
tags=tags,
|
|
user=user
|
|
)
|
|
|
|
page = await self.create_page(page_data)
|
|
|
|
logger.info(f"Created page: id={page.id}, path={page.path}")
|
|
|
|
# Build research summary
|
|
research_summary = {
|
|
"wiki_results": wiki_results_count,
|
|
"web_results": web_results_count,
|
|
"graph_entities": graph_entities_count,
|
|
"keywords_extracted": len(search_response.keywords.core_keywords) if search_response.keywords else 0,
|
|
"timing_ms": search_response.timing.total_ms if search_response.timing else 0
|
|
}
|
|
|
|
return page, {
|
|
"research_summary": research_summary,
|
|
"sources_used": len(source_information),
|
|
"search_id": search_response.search_id
|
|
}
|