"""Admin API routes for taxonomy governance (Phase 5). Implements admin dashboard and tag promotion per Spec Section 4.4. """ from typing import Any, Optional import structlog from fastapi import APIRouter, Depends, HTTPException, Query, status from pydantic import BaseModel, Field from sqlalchemy.ext.asyncio import AsyncSession from src.api.dependencies import get_db_session from src.extraction_v2.taxonomy_loader import TaxonomyLoader from src.services.tag_governance_service import ( DynamicTagAnalyzer, TagPromotionService, ) logger = structlog.get_logger(__name__) router = APIRouter(prefix="/v1/admin/taxonomy", tags=["admin", "taxonomy"]) # Request/Response schemas class TrendingTagResponse(BaseModel): """Response schema for trending tags.""" name: str = Field(..., description="Tag name (e.g., 'topic:gdpr')") usage_count: int = Field(..., description="Total usage count across all records") document_count: int = Field(..., description="Number of documents using this tag") verified_ratio: float = Field( ..., ge=0.0, le=1.0, description="Ratio of verified to total uses" ) first_seen: Optional[str] = Field(None, description="Date first seen (YYYY-MM-DD)") last_seen: Optional[str] = Field(None, description="Date last seen (YYYY-MM-DD)") class TrendingTagsListResponse(BaseModel): """Response schema for list of trending tags.""" trending_tags: list[TrendingTagResponse] total_count: int filters_applied: dict[str, Any] class TagUsageStatsResponse(BaseModel): """Response schema for tag usage statistics.""" total_tags: int = Field(..., description="Total number of tag assignments") by_namespace: dict[str, int] = Field( ..., description="Tag count by namespace (req, topic, etc.)" ) by_status: dict[str, int] = Field( ..., description="Tag count by status (verified, pending, rejected)" ) dynamic_tags: int = Field( ..., description="Count of dynamic namespace tags (topic, entity, tech, domain)" ) class PromoteTagRequest(BaseModel): """Request schema for tag promotion.""" old_tag: str = Field( ..., description="Current dynamic tag to promote (e.g., 'topic:gdpr')", examples=["topic:gdpr"], ) new_tag: str = Field( ..., description="New canonical tag in standard namespace (e.g., 'compliance:regulatory:gdpr')", examples=["compliance:regulatory:gdpr"], ) description: str = Field( ..., description="Description of the tag for taxonomy file", examples=["GDPR compliance requirements"], ) examples: list[str] = Field( default_factory=list, description="Example usages of the tag", ) profile: Optional[str] = Field( None, description="Optional profile to add tag to (fintech, healthcare, automotive)", ) add_to_taxonomy: bool = Field( True, description="Whether to add to taxonomy config file" ) class PromoteTagResponse(BaseModel): """Response schema for tag promotion.""" success: bool old_tag: str new_tag: str records_updated: int = Field(..., description="Number of records updated") added_to_taxonomy: bool = Field( ..., description="Whether tag was added to config file" ) message: Optional[str] = None @router.get( "/trending", response_model=TrendingTagsListResponse, summary="Get trending dynamic tags", description="Identify trending dynamic tags that are candidates for promotion to standard taxonomy. Per Spec Section 4.4.2.", ) async def get_trending_tags( min_usage: int = Query( 50, ge=1, description="Minimum number of times tag must be used", ), min_documents: int = Query( 3, ge=1, description="Minimum number of documents tag must appear in", ), namespace: Optional[str] = Query( None, description="Filter by namespace (topic, entity, tech, domain)", ), db: AsyncSession = Depends(get_db_session), ) -> TrendingTagsListResponse: """Get trending dynamic tags for promotion consideration. This endpoint helps admins identify popular dynamic tags that should be promoted to the standard taxonomy. Trending criteria: - Usage count >= min_usage (default: 50) - Appears in >= min_documents documents (default: 3) - High verified ratio indicates user acceptance Args: min_usage: Minimum usage count threshold min_documents: Minimum document count threshold namespace: Optional namespace filter db: Database session Returns: TrendingTagsListResponse with list of trending tags """ analyzer = DynamicTagAnalyzer(db) trending = await analyzer.get_trending_tags( min_usage=min_usage, min_documents=min_documents, namespace=namespace, ) logger.info( "trending_tags_retrieved", count=len(trending), min_usage=min_usage, min_documents=min_documents, namespace=namespace, ) return TrendingTagsListResponse( trending_tags=[TrendingTagResponse(**tag) for tag in trending], total_count=len(trending), filters_applied={ "min_usage": min_usage, "min_documents": min_documents, "namespace": namespace, }, ) @router.get( "/stats", response_model=TagUsageStatsResponse, summary="Get tag usage statistics", description="Get overall tag usage statistics by namespace and status.", ) async def get_tag_stats( db: AsyncSession = Depends(get_db_session), ) -> TagUsageStatsResponse: """Get overall tag usage statistics. Provides dashboard-level statistics: - Total tag assignments - Distribution by namespace - Distribution by status - Count of dynamic tags Args: db: Database session Returns: TagUsageStatsResponse with usage statistics """ analyzer = DynamicTagAnalyzer(db) stats = await analyzer.get_tag_usage_stats() logger.info( "tag_stats_retrieved", total_tags=stats["total_tags"], dynamic_tags=stats["dynamic_tags"], ) return TagUsageStatsResponse(**stats) @router.post( "/promote", response_model=PromoteTagResponse, status_code=status.HTTP_200_OK, summary="Promote a dynamic tag to standard taxonomy", description="Promote a dynamic tag to canonical namespace and optionally add to taxonomy config. Per Spec Section 4.4.3.", ) async def promote_tag( request: PromoteTagRequest, db: AsyncSession = Depends(get_db_session), ) -> PromoteTagResponse: """Promote a dynamic tag to standard taxonomy. Promotion workflow (Spec Section 4.4.3): 1. Validate new tag format and namespace 2. Batch update: Rename tag in all records 3. Add to taxonomy config file (if requested) 4. Update history with promotion entry Example: - Old: "topic:gdpr" - New: "compliance:regulatory:gdpr" - Profile: "fintech" (optional) Args: request: Promotion request with old/new tags db: Database session Returns: PromoteTagResponse with success status and update count Raises: HTTPException: 400 if validation fails, 404 if tag not found """ # Load taxonomy for validation taxonomy_loader = TaxonomyLoader(profile=request.profile) promotion_service = TagPromotionService(db, taxonomy_loader) # Perform promotion (database migration) success, records_updated, error = await promotion_service.promote_tag( old_tag=request.old_tag, new_tag=request.new_tag, profile=request.profile, ) if not success: if "not found" in error.lower(): raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=error, ) else: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=error, ) # Add to taxonomy file if requested added_to_taxonomy = False if request.add_to_taxonomy: taxonomy_success, taxonomy_error = promotion_service.add_to_taxonomy_file( tag=request.new_tag, description=request.description, examples=request.examples, profile=request.profile, ) if taxonomy_success: added_to_taxonomy = True else: logger.warning( "failed_to_add_to_taxonomy", tag=request.new_tag, error=taxonomy_error, ) logger.info( "tag_promoted", old_tag=request.old_tag, new_tag=request.new_tag, records_updated=records_updated, added_to_taxonomy=added_to_taxonomy, profile=request.profile, ) return PromoteTagResponse( success=True, old_tag=request.old_tag, new_tag=request.new_tag, records_updated=records_updated, added_to_taxonomy=added_to_taxonomy, message=f"Successfully promoted {request.old_tag} to {request.new_tag}. " f"Updated {records_updated} records." + ( " Added to taxonomy config." if added_to_taxonomy else " (Not added to taxonomy config)" ), )