Spaces:
Runtime error
Runtime error
| """ | |
| Service for managing official policies (admin only). | |
| """ | |
| from sqlalchemy.orm import Session | |
| from typing import List, Optional | |
| import os | |
| import shutil | |
| from app.database.models import OfficialPolicy | |
| from app.utils.helpers import generate_id | |
| from app.utils.validators import validate_file_type, validate_file_size | |
| from app.services.rag_service import rag_service | |
| class PolicyService: | |
| """Service for official policy management.""" | |
| def upload_policy( | |
| db: Session, | |
| file, | |
| title: str, | |
| filename: str, | |
| admin_user_id: str, | |
| description: Optional[str] = None, | |
| category: Optional[str] = None | |
| ) -> OfficialPolicy: | |
| """ | |
| Upload an official policy document. | |
| Args: | |
| db: Database session | |
| file: File object | |
| title: Policy title | |
| filename: Original filename | |
| admin_user_id: Admin user ID | |
| description: Optional description | |
| category: Optional category | |
| Returns: | |
| Created OfficialPolicy object | |
| Raises: | |
| ValueError: If file validation fails | |
| """ | |
| # Validate file type | |
| if not validate_file_type(filename, ['pdf', 'txt', 'docx']): | |
| raise ValueError("Invalid file type. Only PDF, TXT, and DOCX files are allowed.") | |
| # Get file size | |
| file.seek(0, 2) | |
| file_size = file.tell() | |
| file.seek(0) | |
| # Validate file size (20MB limit for policies) | |
| if not validate_file_size(file_size, max_size_mb=20): | |
| raise ValueError("File size exceeds 20MB limit.") | |
| # Generate policy ID | |
| policy_id = generate_id() | |
| # Determine file type | |
| file_extension = filename.rsplit('.', 1)[1].lower() if '.' in filename else 'unknown' | |
| # Create upload directory | |
| upload_dir = os.path.join(os.path.dirname(__file__), "..", "..", "data", "policies") | |
| os.makedirs(upload_dir, exist_ok=True) | |
| # Save file | |
| file_path = os.path.join(upload_dir, f"{policy_id}_{filename}") | |
| with open(file_path, "wb") as buffer: | |
| shutil.copyfileobj(file, buffer) | |
| # Create policy record | |
| policy = OfficialPolicy( | |
| id=policy_id, | |
| title=title, | |
| description=description, | |
| filename=filename, | |
| file_path=file_path, | |
| file_type=file_extension, | |
| file_size=file_size, | |
| category=category, | |
| uploaded_by=admin_user_id, | |
| is_active=1 | |
| ) | |
| db.add(policy) | |
| db.commit() | |
| db.refresh(policy) | |
| return policy | |
| def process_policy_content( | |
| db: Session, | |
| policy_id: str, | |
| content: str | |
| ) -> int: | |
| """ | |
| Process policy content for RAG (store in vector DB with special collection). | |
| Args: | |
| db: Database session | |
| policy_id: Policy ID | |
| content: Extracted text content | |
| Returns: | |
| Number of chunks created | |
| """ | |
| policy = db.query(OfficialPolicy).filter(OfficialPolicy.id == policy_id).first() | |
| if not policy: | |
| raise ValueError("Policy not found") | |
| # Process with RAG service (using a special "official_policies" user_id) | |
| num_chunks = rag_service.process_document( | |
| document_id=policy.id, | |
| filename=f"[POLICY] {policy.title}", | |
| content=content, | |
| user_id="official_policies", # Special ID for policies | |
| db=db | |
| ) | |
| return num_chunks | |
| def get_all_policies(db: Session, active_only: bool = True) -> List[OfficialPolicy]: | |
| """ | |
| Get all official policies. | |
| Args: | |
| db: Database session | |
| active_only: Only return active policies | |
| Returns: | |
| List of OfficialPolicy objects | |
| """ | |
| query = db.query(OfficialPolicy) | |
| if active_only: | |
| query = query.filter(OfficialPolicy.is_active == 1) | |
| return query.order_by(OfficialPolicy.created_at.desc()).all() | |
| def get_policy_by_id(db: Session, policy_id: str) -> Optional[OfficialPolicy]: | |
| """ | |
| Get policy by ID. | |
| Args: | |
| db: Database session | |
| policy_id: Policy ID | |
| Returns: | |
| OfficialPolicy object or None | |
| """ | |
| return db.query(OfficialPolicy).filter(OfficialPolicy.id == policy_id).first() | |
| def update_policy( | |
| db: Session, | |
| policy_id: str, | |
| title: Optional[str] = None, | |
| description: Optional[str] = None, | |
| category: Optional[str] = None, | |
| is_active: Optional[bool] = None | |
| ) -> Optional[OfficialPolicy]: | |
| """ | |
| Update policy metadata. | |
| Args: | |
| db: Database session | |
| policy_id: Policy ID | |
| title: New title | |
| description: New description | |
| category: New category | |
| is_active: New active status | |
| Returns: | |
| Updated OfficialPolicy object or None | |
| """ | |
| policy = db.query(OfficialPolicy).filter(OfficialPolicy.id == policy_id).first() | |
| if not policy: | |
| return None | |
| if title is not None: | |
| policy.title = title | |
| if description is not None: | |
| policy.description = description | |
| if category is not None: | |
| policy.category = category | |
| if is_active is not None: | |
| policy.is_active = 1 if is_active else 0 | |
| db.commit() | |
| db.refresh(policy) | |
| return policy | |
| def delete_policy(db: Session, policy_id: str) -> bool: | |
| """ | |
| Delete a policy and its chunks. | |
| Args: | |
| db: Database session | |
| policy_id: Policy ID | |
| Returns: | |
| True if deleted, False if not found | |
| """ | |
| policy = db.query(OfficialPolicy).filter(OfficialPolicy.id == policy_id).first() | |
| if not policy: | |
| return False | |
| # Delete file from filesystem | |
| if os.path.exists(policy.file_path): | |
| os.remove(policy.file_path) | |
| # Delete chunks from vector database | |
| rag_service.delete_document_chunks(policy_id) | |
| # Delete database record | |
| db.delete(policy) | |
| db.commit() | |
| return True | |
| # Global policy service instance | |
| policy_service = PolicyService() | |