jpmschweitzerandClaude
a687b770ef
105 findings to zero. Most were mechanical — 67 unused imports, and assorted
f-strings without placeholders. Three groups needed a decision.
The 15 F821 "undefined name" were forward references, not runtime errors. Each
annotation is quoted — `-> "WikiService"`, `Optional["IngestionService"]` — with
the real import inside the function body to break an import cycle. A quoted
annotation is never evaluated, so the code ran; the names were simply
unresolvable to any checker. They now have a TYPE_CHECKING block, which costs
nothing at import time and keeps the cycle broken.
The 6 E402 split two ways. `import secrets`, `Security`, `Request` and
`HTTPBearer` in dependencies.py had drifted below several hundred lines of
factory functions for no reason — stdlib and fastapi, no cycle to avoid — and
moved up. The other three are deliberate and now say so: the VectorService and
GraphService aliases import back into dependencies.py, and main.py's routers
expect a configured app, so both must stay put.
Bare `except:` narrowed to `except Exception:` in three places, which stops them
swallowing KeyboardInterrupt and SystemExit.
The 5 unused locals were all genuinely dead. One is worth naming rather than
fixing: qdrant_client.delete()'s return value was bound and never read, so a
failed delete is indistinguishable from a successful one — the assignment is
gone, but nothing checks the status either way and that has not changed here.
`timing = {}` in _retrieve_parallel looked like it might mean the reported
per-leg timings were always zero; traced, and they come from output["timing"],
so the local was only vestigial.
426 passed, 29 skipped, unchanged. The app imports and the service aliases still
resolve, which is the check that mattered after moving imports in
dependencies.py.
The gate still prints "not gated here yet: test (T-56)" — lint is green, tests
remain unwired, and that is left visible rather than silently absent.
Co-Authored-By: Claude <noreply@anthropic.com>
605 lines
19 KiB
Python
605 lines
19 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 TYPE_CHECKING, 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
|
|
from src.models.wiki import (
|
|
WikiPage, WikiPageSummary, WikiPageList,
|
|
WikiPageCreate, WikiPageUpdate,
|
|
DossierInfo, DossierList
|
|
)
|
|
|
|
# Imported for annotations only. The real imports live inside the functions
|
|
# that use them, to break an import cycle; a quoted annotation is never
|
|
# evaluated at runtime, so the names were unresolvable to any checker. This
|
|
# block costs nothing at import time and makes them resolvable again.
|
|
if TYPE_CHECKING:
|
|
from src.services.hybrid_rag_service import HybridRAGService
|
|
from src.services.wiki_page_writer import WikiPageWriter
|
|
|
|
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
|
|
|
|
# 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
|
|
}
|