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>
234 lines
9.7 KiB
Python
234 lines
9.7 KiB
Python
"""
|
|
Pydantic models for Wiki.js operations.
|
|
|
|
Models for:
|
|
- Wiki pages (CRUD operations)
|
|
- Dossiers (tag-based collections)
|
|
- Search results
|
|
"""
|
|
|
|
from pydantic import BaseModel, Field, field_validator
|
|
from typing import Optional, List, Dict, Any
|
|
from datetime import datetime
|
|
|
|
|
|
# Base models
|
|
class WikiPageBase(BaseModel):
|
|
"""Base wiki page fields."""
|
|
title: str = Field(..., min_length=1, max_length=500, description="Page title")
|
|
description: Optional[str] = Field(None, max_length=1000, description="Page description")
|
|
tags: List[str] = Field(default_factory=list, description="Tags (for dossier organization)")
|
|
is_published: bool = Field(default=True, description="Whether page is published")
|
|
|
|
@field_validator("tags")
|
|
@classmethod
|
|
def validate_tags(cls, v: List[str]) -> List[str]:
|
|
"""Validate and clean tags."""
|
|
# Remove empty tags and strip whitespace
|
|
cleaned = [tag.strip() for tag in v if tag.strip()]
|
|
# Ensure uniqueness
|
|
return list(set(cleaned))
|
|
|
|
|
|
class WikiPageCreate(WikiPageBase):
|
|
"""Request model for creating a wiki page."""
|
|
content: str = Field(..., description="Page content (markdown)")
|
|
path: str = Field(..., min_length=1, max_length=500, description="Page path (e.g., '/projects/library-desk')")
|
|
editor: str = Field(default="markdown", description="Editor type")
|
|
user: Optional[str] = Field(None, description="User identifier (defaults to configured user)")
|
|
|
|
@field_validator("path")
|
|
@classmethod
|
|
def validate_path(cls, v: str) -> str:
|
|
"""Validate page path."""
|
|
# Ensure path starts with /
|
|
if not v.startswith("/"):
|
|
v = f"/{v}"
|
|
# Remove trailing slash
|
|
if v.endswith("/") and v != "/":
|
|
v = v.rstrip("/")
|
|
return v
|
|
|
|
|
|
class WikiPageUpdate(BaseModel):
|
|
"""Request model for updating a wiki page."""
|
|
content: Optional[str] = Field(None, description="Updated content")
|
|
title: Optional[str] = Field(None, min_length=1, max_length=500, description="Updated title")
|
|
description: Optional[str] = Field(None, max_length=1000, description="Updated description")
|
|
tags: Optional[List[str]] = Field(None, description="Updated tags")
|
|
|
|
@field_validator("tags")
|
|
@classmethod
|
|
def validate_tags(cls, v: Optional[List[str]]) -> Optional[List[str]]:
|
|
"""Validate and clean tags."""
|
|
if v is None:
|
|
return None
|
|
cleaned = [tag.strip() for tag in v if tag.strip()]
|
|
return list(set(cleaned))
|
|
|
|
|
|
class WikiPage(WikiPageBase):
|
|
"""Response model for a wiki page."""
|
|
id: int = Field(..., description="Page ID")
|
|
path: str = Field(..., description="Page path")
|
|
content: Optional[str] = Field(None, description="Page content")
|
|
created_at: Optional[str] = Field(None, description="Creation timestamp")
|
|
updated_at: Optional[str] = Field(None, description="Last update timestamp")
|
|
editor: Optional[str] = Field(None, description="Editor type")
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
class WikiPageSummary(BaseModel):
|
|
"""Summarized wiki page (for list responses)."""
|
|
id: int = Field(..., description="Page ID")
|
|
path: str = Field(..., description="Page path")
|
|
title: str = Field(..., description="Page title")
|
|
description: Optional[str] = Field(None, description="Page description")
|
|
tags: List[str] = Field(default_factory=list, description="Page tags")
|
|
updated_at: Optional[str] = Field(None, description="Last update timestamp")
|
|
is_published: bool = Field(..., description="Publication status")
|
|
|
|
|
|
class WikiPageList(BaseModel):
|
|
"""Response model for list of pages."""
|
|
pages: List[WikiPageSummary] = Field(..., description="List of pages")
|
|
total: int = Field(..., description="Total number of pages")
|
|
filtered_by_tag: Optional[str] = Field(None, description="Tag filter applied")
|
|
user: str = Field(..., description="User namespace")
|
|
|
|
|
|
# Dossier models
|
|
class DossierCreate(BaseModel):
|
|
"""Request model for creating a dossier."""
|
|
name: str = Field(..., min_length=1, max_length=100, description="Dossier name (becomes a tag)")
|
|
title: str = Field(..., min_length=1, max_length=200, description="Human-readable title")
|
|
description: str = Field(..., min_length=1, description="Dossier description")
|
|
create_index_page: bool = Field(default=True, description="Create an index page for the dossier")
|
|
user: Optional[str] = Field(None, description="User identifier")
|
|
|
|
@field_validator("name")
|
|
@classmethod
|
|
def validate_name(cls, v: str) -> str:
|
|
"""Validate dossier name (will be used as tag)."""
|
|
# Convert to lowercase, replace spaces with hyphens
|
|
name = v.lower().strip()
|
|
name = name.replace(" ", "-")
|
|
# Remove special characters except hyphens and underscores
|
|
name = "".join(c for c in name if c.isalnum() or c in "-_")
|
|
if not name:
|
|
raise ValueError("Dossier name must contain alphanumeric characters")
|
|
return name
|
|
|
|
|
|
class DossierInfo(BaseModel):
|
|
"""Response model for dossier information."""
|
|
name: str = Field(..., description="Dossier name (tag)")
|
|
title: str = Field(..., description="Dossier title")
|
|
description: str = Field(..., description="Dossier description")
|
|
page_count: int = Field(..., description="Number of pages in dossier")
|
|
index_page_id: Optional[int] = Field(None, description="ID of index page")
|
|
index_page_path: Optional[str] = Field(None, description="Path to index page")
|
|
created_at: Optional[str] = Field(None, description="Creation timestamp")
|
|
|
|
|
|
class DossierList(BaseModel):
|
|
"""Response model for list of dossiers."""
|
|
dossiers: List[DossierInfo] = Field(..., description="List of dossiers")
|
|
total: int = Field(..., description="Total number of dossiers")
|
|
user: str = Field(..., description="User namespace")
|
|
|
|
|
|
# Search models
|
|
class WikiSearchResult(BaseModel):
|
|
"""Search result item."""
|
|
id: int = Field(..., description="Page ID")
|
|
path: str = Field(..., description="Page path")
|
|
title: str = Field(..., description="Page title")
|
|
description: Optional[str] = Field(None, description="Page description")
|
|
relevance: Optional[float] = Field(None, description="Search relevance score")
|
|
|
|
|
|
class WikiSearchResponse(BaseModel):
|
|
"""Response model for search results."""
|
|
results: List[WikiSearchResult] = Field(..., description="Search results")
|
|
query: str = Field(..., description="Search query")
|
|
total: int = Field(..., description="Total results found")
|
|
|
|
|
|
# Move/rename models
|
|
class WikiPageMove(BaseModel):
|
|
"""Request model for moving/renaming a page."""
|
|
new_path: str = Field(..., min_length=1, description="New page path")
|
|
locale: str = Field(default="en", description="Page locale")
|
|
|
|
@field_validator("new_path")
|
|
@classmethod
|
|
def validate_new_path(cls, v: str) -> str:
|
|
"""Validate new path."""
|
|
if not v.startswith("/"):
|
|
v = f"/{v}"
|
|
if v.endswith("/") and v != "/":
|
|
v = v.rstrip("/")
|
|
return v
|
|
|
|
|
|
# Response models for operations
|
|
class WikiOperationResponse(BaseModel):
|
|
"""Generic response for wiki operations."""
|
|
success: bool = Field(..., description="Whether operation succeeded")
|
|
message: str = Field(..., description="Operation message")
|
|
page_id: Optional[int] = Field(None, description="Page ID (if applicable)")
|
|
page_path: Optional[str] = Field(None, description="Page path (if applicable)")
|
|
|
|
|
|
class DossierOperationResponse(BaseModel):
|
|
"""Response for dossier operations."""
|
|
success: bool = Field(..., description="Whether operation succeeded")
|
|
message: str = Field(..., description="Operation message")
|
|
dossier_name: str = Field(..., description="Dossier name")
|
|
index_page_id: Optional[int] = Field(None, description="Index page ID (if created)")
|
|
index_page_path: Optional[str] = Field(None, description="Index page path (if created)")
|
|
|
|
|
|
# Smart create models (HybridRAG-powered page creation)
|
|
class WikiSmartCreateRequest(BaseModel):
|
|
"""Request model for smart page creation with research."""
|
|
topic: str = Field(..., min_length=1, max_length=500, description="Topic to research and create page about")
|
|
path: Optional[str] = Field(None, description="Page path (auto-generated from topic if not provided)")
|
|
tags: List[str] = Field(default_factory=list, description="Tags for the page")
|
|
user: Optional[str] = Field(None, description="User identifier")
|
|
include_web_research: bool = Field(default=True, description="Include web search results")
|
|
include_wiki_search: bool = Field(default=True, description="Include existing wiki knowledge")
|
|
|
|
@field_validator("tags")
|
|
@classmethod
|
|
def validate_tags(cls, v: List[str]) -> List[str]:
|
|
"""Validate and clean tags."""
|
|
cleaned = [tag.strip() for tag in v if tag.strip()]
|
|
return list(set(cleaned))
|
|
|
|
@field_validator("path")
|
|
@classmethod
|
|
def validate_path(cls, v: Optional[str]) -> Optional[str]:
|
|
"""Validate page path if provided."""
|
|
if v is None:
|
|
return None
|
|
# Ensure path starts with /
|
|
if not v.startswith("/"):
|
|
v = f"/{v}"
|
|
# Remove trailing slash
|
|
if v.endswith("/") and v != "/":
|
|
v = v.rstrip("/")
|
|
return v
|
|
|
|
|
|
class WikiSmartCreateResponse(BaseModel):
|
|
"""Response model for smart page creation."""
|
|
page: WikiPage = Field(..., description="Created wiki page")
|
|
research_summary: Dict[str, Any] = Field(..., description="Summary of research used")
|
|
sources_used: int = Field(..., description="Number of sources incorporated")
|
|
search_id: Optional[str] = Field(None, description="HybridRAG search ID for reference")
|
|
entity_linking: Dict[str, int] = Field(default_factory=dict, description="Entity linking statistics")
|