import hashlib import importlib import json import unicodedata import zipfile from dataclasses import dataclass from datetime import UTC, datetime from io import BytesIO from pathlib import Path, PurePath from secrets import token_urlsafe from uuid import uuid4 from xml.etree import ElementTree from pydantic import ValidationError from sqlalchemy import select from sqlalchemy.orm import Session from src.config import Settings, get_settings from src.models.auth import AuthorityLevel, SourceDocumentStatus, SourceScope from src.models.schemas import ( LEGAL_SOURCE_TYPE_LABELS_VI, DocumentListItem, DocumentMetadata, DocumentType, EffectiveStatus, IngestionStatus, LegalChunk, LegalDocument, LegalParseFlag, LegalSourceType, Purpose, RetentionPolicy, ScopedDocumentItem, utc_now_iso, ) from src.services.citations import CitationService from src.services.db import DocumentChunkRecord, new_id, utc_now from src.services.db import DocumentRecord as DbDocumentRecord from src.services.embeddings import has_real_api_key from src.services.legal_chunking import LegalChunker ALLOWED_UPLOAD_EXTENSIONS = {".pdf", ".docx", ".txt"} HTML_EXTENSION = ".html" PDF_EXTENSION = ".pdf" LEGAL_SOURCE_TYPE_BY_LABEL = { "hien phap": LegalSourceType.HIEN_PHAP, "bo luat": LegalSourceType.BO_LUAT, "luat": LegalSourceType.LUAT, "phap lenh": LegalSourceType.PHAP_LENH, "lenh": LegalSourceType.LENH, "nghi quyet": LegalSourceType.NGHI_QUYET, "nghi quyet lien tich": LegalSourceType.NGHI_QUYET_LIEN_TICH, "nghi dinh": LegalSourceType.NGHI_DINH, "quyet dinh": LegalSourceType.QUYET_DINH, "chi thi": LegalSourceType.CHI_THI, "thong tu": LegalSourceType.THONG_TU, "thong tu lien tich": LegalSourceType.THONG_TU_LIEN_TICH, "van ban hop nhat": LegalSourceType.VAN_BAN_HOP_NHAT, "dieu uoc quoc te": LegalSourceType.DIEU_UOC_QUOC_TE, "cong van": LegalSourceType.CONG_VAN, "cong dien": LegalSourceType.CONG_DIEN, "thong bao": LegalSourceType.THONG_BAO, "quy dinh": LegalSourceType.QUY_DINH, "quy che": LegalSourceType.QUY_CHE, "dieu le": LegalSourceType.DIEU_LE, "huong dan": LegalSourceType.HUONG_DAN, "khac": LegalSourceType.OTHER, } class DocumentServiceError(Exception): status_code = 400 code = "validation_error" message = "Document request is invalid." def __init__(self, message: str | None = None, details: dict | None = None) -> None: super().__init__(message or self.message) self.message = message or self.message self.details = details or {} class EmptyFileError(DocumentServiceError): status_code = 400 code = "empty_file" message = "Uploaded file is empty." class FileTooLargeError(DocumentServiceError): status_code = 413 code = "file_too_large" message = "Uploaded file exceeds the configured size limit." class UnsupportedFileTypeError(DocumentServiceError): status_code = 415 code = "unsupported_file_type" message = "Uploaded file type is not supported." class DocumentStoreUnavailableError(DocumentServiceError): status_code = 503 code = "document_store_unavailable" message = "Document store is unavailable." class IngestionUnavailableError(DocumentServiceError): status_code = 503 code = "ingestion_unavailable" message = "Document ingestion is unavailable." class MissingExtractionDependencyError(DocumentServiceError): status_code = 503 code = "missing_extraction_dependency" message = "Required document extraction dependency is unavailable." class EmptyExtractionError(DocumentServiceError): status_code = 422 code = "empty_extraction" message = "Document extraction returned empty text." class DocumentNotFoundError(DocumentServiceError): status_code = 404 code = "not_found" message = "Document was not found." @dataclass(frozen=True) class DocumentRecord: document_id: str filename: str document_type: DocumentType purpose: Purpose retention_policy: RetentionPolicy ingestion_status: IngestionStatus metadata: DocumentMetadata created_at: str @dataclass(frozen=True) class DocumentUploadInput: filename: str content: bytes document_type: DocumentType purpose: Purpose retention_policy: RetentionPolicy title: str | None metadata: DocumentMetadata @dataclass(frozen=True) class ExtractionResult: text: str file_type: str method: str warnings: list[str] @dataclass(frozen=True) class DocumentIngestionResult: upload_record: DocumentRecord extraction: ExtractionResult document: LegalDocument chunks: list[LegalChunk] citation_previews: list[str] warnings: list[str] @dataclass(frozen=True) class DocumentListFilters: page: int page_size: int document_type: DocumentType | None = None purpose: Purpose | None = None ingestion_status: IngestionStatus | None = None effective_status: EffectiveStatus | None = None q: str | None = None @dataclass(frozen=True) class ScopedUploadContext: source_scope: SourceScope session_id: str | None = None owner_user_id: str | None = None org_id: str | None = None uploaded_by: str | None = None authority_level: AuthorityLevel = AuthorityLevel.UPLOADED_CONTEXT document_status: SourceDocumentStatus = SourceDocumentStatus.ACTIVE def normalize_legal_source_label(value: str) -> str: normalized = unicodedata.normalize("NFKD", value) ascii_value = "".join(char for char in normalized if not unicodedata.combining(char)) ascii_value = ascii_value.replace("đ", "d").replace("Đ", "D") return " ".join(ascii_value.casefold().strip().replace("_", " ").split()) def legal_source_type_label_vi(value: LegalSourceType) -> str: return LEGAL_SOURCE_TYPE_LABELS_VI[value] def legal_source_type_from_source_value(value: str) -> LegalSourceType: slug_match = next((item for item in LegalSourceType if item.value == value), None) if slug_match is not None: return slug_match return LEGAL_SOURCE_TYPE_BY_LABEL.get(normalize_legal_source_label(value), LegalSourceType.OTHER) def parse_upload_metadata(metadata_json: str | None) -> dict: if metadata_json in {None, ""}: return {} try: parsed = json.loads(metadata_json) except json.JSONDecodeError as exc: raise DocumentServiceError( message="Metadata must be a valid JSON object.", details={"metadata": "invalid_json"}, ) from exc if not isinstance(parsed, dict): raise DocumentServiceError( message="Metadata must be a valid JSON object.", details={"metadata": "expected_object"}, ) return parsed def ensure_upload_directories(settings: Settings) -> None: """Safely ensure configured upload directories exist. Only creates directories if upload_storage_backend is local_disk. Does not delete existing uploads. """ if settings.upload_storage_backend != "local_disk": return for path_str in [settings.upload_dir, settings.guest_upload_dir, settings.org_upload_dir]: if path_str: Path(path_str).mkdir(parents=True, exist_ok=True) def is_safe_upload_path(path: Path | str, settings: Settings) -> bool: """Check if the resolved path is inside the allowed upload directories to prevent path traversal.""" try: resolved_path = Path(path).resolve() allowed_bases = [ Path(settings.upload_dir).resolve(), Path(settings.guest_upload_dir).resolve(), Path(settings.org_upload_dir).resolve(), ] if settings.app_env == "test": import tempfile allowed_bases.append(Path(tempfile.gettempdir()).resolve()) return any( resolved_path == base or base in resolved_path.parents for base in allowed_bases if base ) except Exception: return False def build_document_metadata(raw_metadata: dict, title: str | None) -> DocumentMetadata: metadata = dict(raw_metadata) if title is not None: metadata["title"] = title raw_loai_van_ban = metadata.pop("loai_van_ban", None) or metadata.get("raw_loai_van_ban") if raw_loai_van_ban is not None: metadata["raw_loai_van_ban"] = raw_loai_van_ban metadata["legal_source_type"] = legal_source_type_from_source_value(str(raw_loai_van_ban)) try: document_metadata = DocumentMetadata.model_validate(metadata) except ValidationError as exc: raise DocumentServiceError( message="Document metadata validation failed.", details={"errors": exc.errors(include_input=False, include_url=False)}, ) from exc if document_metadata.legal_source_type is not None: document_metadata.legal_source_type_label_vi = legal_source_type_label_vi( document_metadata.legal_source_type ) return document_metadata def extract_document_text_from_path(path: Path | str) -> ExtractionResult: source_path = Path(path) suffix = source_path.suffix.casefold() if suffix == PDF_EXTENSION: return _extract_pdf_text_from_path(source_path) if suffix == ".txt": return _extract_plain_text_from_path(source_path) if suffix == ".docx": return _extract_docx_text_from_path(source_path) raise UnsupportedFileTypeError(message=f"No extraction adapter is available for {suffix!r}.") def extract_document_text_from_bytes(content: bytes, *, filename: str) -> ExtractionResult: suffix = PurePath(filename).suffix.casefold() if suffix == PDF_EXTENSION: return _extract_pdf_text_from_bytes(content) if suffix == ".txt": return _extract_plain_text_from_bytes(content, file_type="txt") if suffix == ".docx": return _extract_docx_text_from_bytes(content) raise UnsupportedFileTypeError(message=f"No extraction adapter is available for {suffix!r}.") def ingest_upload_to_legal_chunks( *, upload: DocumentUploadInput, max_upload_bytes: int, ingestion_service: "InMemoryDocumentIngestionService | None" = None, source_uri: str | None = None, chunker: LegalChunker | None = None, citation_service: CitationService | None = None, include_citation_previews: bool = True, fail_on_empty: bool = False, ) -> DocumentIngestionResult: service = ingestion_service or InMemoryDocumentIngestionService() upload_record = service.upload_document(upload, max_upload_bytes=max_upload_bytes) extraction = extract_document_text_from_bytes(upload.content, filename=upload.filename) text = extraction.text.strip() document = _legal_document_from_upload( upload_record=upload_record, upload=upload, extraction=extraction, source_uri=source_uri or upload_record.filename, ready=bool(text), ) if not text: warnings = [*extraction.warnings, "Skipped chunking because extraction returned empty text."] if fail_on_empty: raise EmptyExtractionError(details={"warnings": warnings}) return DocumentIngestionResult( upload_record=upload_record, extraction=extraction, document=document, chunks=[], citation_previews=[], warnings=warnings, ) chunks = (chunker or LegalChunker()).chunk_text( document_id=document.document_id, text=extraction.text, ) previews = [] if include_citation_previews: service_for_citations = citation_service or CitationService() previews = [ service_for_citations.build_label(document=document, chunk=chunk) for chunk in chunks ] return DocumentIngestionResult( upload_record=upload_record, extraction=extraction, document=document, chunks=chunks, citation_previews=previews, warnings=list(extraction.warnings), ) def _legal_document_from_upload( *, upload_record: DocumentRecord, upload: DocumentUploadInput, extraction: ExtractionResult, source_uri: str, ready: bool, ) -> LegalDocument: metadata = upload.metadata parse_flags = [] if not any( [ metadata.document_number, metadata.legal_source_type, metadata.authority, metadata.effective_date, ] ): parse_flags.append(LegalParseFlag.METADATA_MISSING) return LegalDocument( document_id=upload_record.document_id, title=metadata.title or upload.title or upload_record.filename, document_type=upload_record.document_type, source_uri=source_uri, ingestion_status=IngestionStatus.READY if ready else IngestionStatus.FAILED, ingested_at=datetime.now(UTC), effective_status=metadata.effective_status or EffectiveStatus.UNKNOWN, document_number=metadata.document_number, legal_source_type=metadata.legal_source_type, authority=metadata.authority, metadata={ "filename": upload_record.filename, "file_size_bytes": len(upload.content), "file_type": extraction.file_type, "extraction_method": extraction.method, "upload_ingestion_status": upload_record.ingestion_status.value, "metadata_source": "provided-or-metadata-missing", }, parse_flags=parse_flags, ) def _extract_plain_text_from_path(path: Path) -> ExtractionResult: try: text = path.read_text(encoding="utf-8-sig") method = "Path.read_text(utf-8-sig)" except UnicodeDecodeError: text = path.read_text(encoding="utf-8") method = "Path.read_text(utf-8)" return ExtractionResult( text=_normalize_text_newlines(text), file_type=path.suffix.casefold().lstrip("."), method=method, warnings=[] if text.strip() else ["Text extraction returned empty text."], ) def _extract_plain_text_from_bytes(content: bytes, *, file_type: str) -> ExtractionResult: try: text = content.decode("utf-8-sig") method = "bytes.decode(utf-8-sig)" except UnicodeDecodeError: text = content.decode("utf-8") method = "bytes.decode(utf-8)" return ExtractionResult( text=_normalize_text_newlines(text), file_type=file_type, method=method, warnings=[] if text.strip() else ["Text extraction returned empty text."], ) def _extract_pdf_text_from_path(path: Path) -> ExtractionResult: data = path.read_bytes() return _extract_pdf_text(data, source=str(path)) def _extract_pdf_text_from_bytes(content: bytes) -> ExtractionResult: return _extract_pdf_text(content, source=None) def _extract_pdf_text(content: bytes, *, source: str | None) -> ExtractionResult: pypdf = _optional_import("pypdf") if pypdf is not None: reader_input = source if source is not None else BytesIO(content) reader = pypdf.PdfReader(reader_input) pages = [(page.extract_text() or "") for page in reader.pages] text = "\n\n".join(page.strip() for page in pages if page.strip()) return ExtractionResult( text=text, file_type="pdf", method="pypdf.PdfReader", warnings=[] if text else ["PDF extraction returned empty text."], ) pypdf2 = _optional_import("PyPDF2") if pypdf2 is not None: reader_input = source if source is not None else BytesIO(content) reader = pypdf2.PdfReader(reader_input) pages = [(page.extract_text() or "") for page in reader.pages] text = "\n\n".join(page.strip() for page in pages if page.strip()) return ExtractionResult( text=text, file_type="pdf", method="PyPDF2.PdfReader", warnings=[] if text else ["PDF extraction returned empty text."], ) fitz = _optional_import("fitz") if fitz is None: raise MissingExtractionDependencyError( message=( "PDF extraction dependency is missing. Install one parser, for example: " "python -m pip install pypdf" ) ) if source is not None: document = fitz.open(source) else: document = fitz.open(stream=content, filetype="pdf") with document: pages = [page.get_text("text") or "" for page in document] text = "\n\n".join(page.strip() for page in pages if page.strip()) return ExtractionResult( text=text, file_type="pdf", method="PyMuPDF(fitz)", warnings=[] if text else ["PDF extraction returned empty text."], ) def _extract_docx_text_from_path(path: Path) -> ExtractionResult: data = path.read_bytes() docx = _optional_import("docx") if docx is not None: document = docx.Document(str(path)) return _python_docx_result(document) return _extract_docx_text_with_stdlib(data) def _extract_docx_text_from_bytes(content: bytes) -> ExtractionResult: docx = _optional_import("docx") if docx is not None: document = docx.Document(BytesIO(content)) return _python_docx_result(document) return _extract_docx_text_with_stdlib(content) def _python_docx_result(document) -> ExtractionResult: text = "\n".join(paragraph.text for paragraph in document.paragraphs).strip() return ExtractionResult( text=text, file_type="docx", method="python-docx", warnings=[] if text else ["DOCX extraction returned empty text."], ) def _extract_docx_text_with_stdlib(content: bytes) -> ExtractionResult: try: with zipfile.ZipFile(BytesIO(content)) as archive: document_xml = archive.read("word/document.xml") except (KeyError, zipfile.BadZipFile) as exc: raise MissingExtractionDependencyError( message=( "DOCX extraction failed because the file is not a readable DOCX package. " "Install python-docx for broader DOCX compatibility: " "python -m pip install python-docx" ) ) from exc root = ElementTree.fromstring(document_xml) namespace = {"w": "http://schemas.openxmlformats.org/wordprocessingml/2006/main"} paragraphs = [] for paragraph in root.findall(".//w:p", namespace): pieces = [ node.text or "" for node in paragraph.findall(".//w:t", namespace) ] text = "".join(pieces).strip() if text: paragraphs.append(text) text = _normalize_text_newlines("\n".join(paragraphs)) return ExtractionResult( text=text, file_type="docx", method="stdlib zipfile+xml", warnings=[] if text else ["DOCX extraction returned empty text."], ) def _optional_import(module_name: str): try: return importlib.import_module(module_name) except ImportError: return None def _normalize_text_newlines(text: str) -> str: return text.replace("\r\n", "\n").replace("\r", "\n") class InMemoryDocumentIngestionService: def __init__(self) -> None: self._documents: list[DocumentRecord] = [] def reset(self) -> None: self._documents.clear() def upload_document( self, upload: DocumentUploadInput, max_upload_bytes: int, ) -> DocumentRecord: self._validate_upload(upload, max_upload_bytes) record = DocumentRecord( document_id=f"doc_{uuid4().hex}", filename=PurePath(upload.filename).name, document_type=upload.document_type, purpose=upload.purpose, retention_policy=upload.retention_policy, ingestion_status=IngestionStatus.QUEUED, metadata=upload.metadata, created_at=utc_now_iso(), ) self._documents.append(record) return record def list_documents(self, filters: DocumentListFilters) -> tuple[list[DocumentListItem], int]: filtered = list(self._documents) if filters.document_type is not None: filtered = [item for item in filtered if item.document_type == filters.document_type] if filters.purpose is not None: filtered = [item for item in filtered if item.purpose == filters.purpose] if filters.ingestion_status is not None: filtered = [ item for item in filtered if item.ingestion_status == filters.ingestion_status ] if filters.effective_status is not None: filtered = [ item for item in filtered if item.metadata.effective_status == filters.effective_status ] if filters.q: query = filters.q.casefold() filtered = [ item for item in filtered if query in (item.metadata.title or "").casefold() or query in item.filename.casefold() or query in (item.metadata.document_number or "").casefold() ] total = len(filtered) start = (filters.page - 1) * filters.page_size end = start + filters.page_size return [self._to_list_item(item) for item in filtered[start:end]], total def _validate_upload(self, upload: DocumentUploadInput, max_upload_bytes: int) -> None: filename = PurePath(upload.filename).name extension = PurePath(filename).suffix.casefold() if not upload.content: raise EmptyFileError() if len(upload.content) > max_upload_bytes: raise FileTooLargeError(details={"max_upload_bytes": max_upload_bytes}) if extension in ALLOWED_UPLOAD_EXTENSIONS: return if extension == HTML_EXTENSION: approved_import = bool(upload.metadata.model_extra.get("approved_corpus_import")) if upload.purpose == Purpose.CORPUS and approved_import: return raise UnsupportedFileTypeError() @staticmethod def _to_list_item(record: DocumentRecord) -> DocumentListItem: return DocumentListItem( document_id=record.document_id, title=record.metadata.title or record.filename, filename=record.filename, document_type=record.document_type, purpose=record.purpose, retention_policy=record.retention_policy, ingestion_status=record.ingestion_status, effective_status=record.metadata.effective_status, created_at=record.created_at, ) document_ingestion_service = InMemoryDocumentIngestionService() PRIVATE_COLLECTION_BY_SCOPE = { SourceScope.GUEST_SESSION: "guest_sessions", SourceScope.ORG_PRIVATE: "org_documents", } def _private_collection_for_scope(scope: SourceScope) -> str: try: if scope == SourceScope.ORG_PRIVATE: from src.config import get_settings return get_settings().private_chroma_collection_name return PRIVATE_COLLECTION_BY_SCOPE[scope] except KeyError as exc: raise DocumentServiceError( "Unsupported private document scope. Personal/user_private libraries are disabled.", details={"source_scope": scope.value}, ) from exc class DocumentLibraryService: def __init__( self, db: Session, settings: Settings | None = None, ) -> None: self.db = db self.settings = settings or get_settings() def ensure_guest_session_id(self, session_id: str | None) -> str: cleaned = str(session_id or "").strip() if cleaned: return cleaned return f"guest_{token_urlsafe(32)}" def upload_scoped_document( self, upload: DocumentUploadInput, *, context: ScopedUploadContext, mime_type: str = "", ) -> tuple[DbDocumentRecord, int, list[str]]: if not self.settings.enable_private_document_library: raise DocumentStoreUnavailableError("Private document libraries are disabled.") storage_path = self._write_upload_file(upload, context) source_uri = f"private://documents/{context.source_scope.value}/{Path(storage_path).name}" ingestion = ingest_upload_to_legal_chunks( upload=upload, max_upload_bytes=self.settings.max_upload_bytes, source_uri=source_uri, fail_on_empty=True, ) self._apply_private_metadata( ingestion.document, ingestion.chunks, context=context, filename=ingestion.upload_record.filename, mime_type=mime_type, ) collection_name = _private_collection_for_scope(context.source_scope) # Ingestion Flow: Save metadata + chunk_text to Postgres first record = DbDocumentRecord( id=ingestion.document.document_id, source_scope=context.source_scope.value, owner_user_id=context.owner_user_id, org_id=context.org_id, session_id=context.session_id, uploaded_by=context.uploaded_by, title=ingestion.document.title, filename=ingestion.upload_record.filename, mime_type=mime_type, storage_path=storage_path, source_uri=source_uri, authority_level=context.authority_level.value, document_status=SourceDocumentStatus.PENDING_INDEX.value, document_type=upload.document_type.value, purpose=upload.purpose.value, retention_policy=upload.retention_policy.value, effective_date=upload.metadata.effective_date, ) self.db.add(record) for chunk in ingestion.chunks: self.db.add( DocumentChunkRecord( id=new_id("dchk"), document_id=record.id, chunk_id=chunk.chunk_id, chunk_index=chunk.chunk_index, chroma_collection=collection_name, chroma_id=chunk.chunk_id, token_count=chunk.token_count, chunk_text=chunk.content, metadata_json=chunk.metadata, citation_label=None, citation_debug=None, content_hash=hashlib.sha256(chunk.content.encode("utf-8")).hexdigest(), ) ) self.db.commit() self.db.refresh(record) # Call retrieval service /index from src.services.retrieval_service_client import RetrievalServiceClient chunks_payload = [ { "chunk_id": chunk.chunk_id, "chunk_index": chunk.chunk_index, "text": chunk.content, "metadata": chunk.metadata, } for chunk in ingestion.chunks ] client = RetrievalServiceClient() success = client.index_chunks( document_id=record.id, organization_id=context.org_id or "", collection_name=collection_name, chunks=chunks_payload, ) if success: record.document_status = SourceDocumentStatus.ACTIVE.value for chunk_rec in record.chunks: meta = dict(chunk_rec.metadata_json) if chunk_rec.metadata_json else {} meta["document_status"] = SourceDocumentStatus.ACTIVE.value chunk_rec.metadata_json = meta else: record.document_status = SourceDocumentStatus.INDEXING_FAILED.value for chunk_rec in record.chunks: meta = dict(chunk_rec.metadata_json) if chunk_rec.metadata_json else {} meta["document_status"] = SourceDocumentStatus.INDEXING_FAILED.value chunk_rec.metadata_json = meta self.db.add(record) self.db.commit() self.db.refresh(record) return record, len(ingestion.chunks), ingestion.warnings def list_documents(self, context: ScopedUploadContext) -> list[DbDocumentRecord]: statement = self._scoped_statement(context).order_by(DbDocumentRecord.created_at.desc()) return list(self.db.scalars(statement).all()) def get_document(self, document_id: str, context: ScopedUploadContext) -> DbDocumentRecord: record = self.db.scalar( self._scoped_statement(context).where(DbDocumentRecord.id == document_id) ) if record is None: raise DocumentNotFoundError() return record def delete_document(self, document_id: str, context: ScopedUploadContext) -> DbDocumentRecord: record = self.get_document(document_id, context) record.deleted_at = utc_now() record.document_status = SourceDocumentStatus.OUTDATED.value self.db.add(record) self._delete_vectors(record) # Best-effort physical file deletion if record.storage_path: try: if is_safe_upload_path(record.storage_path, self.settings): file_path = Path(record.storage_path).resolve() if file_path.is_file(): file_path.unlink() else: import logging logging.getLogger(__name__).warning( "Skipped physical file deletion for document %s: path %s is outside allowed upload directories.", record.id, record.storage_path, ) except Exception as exc: import logging logging.getLogger(__name__).warning( "Best-effort physical file deletion failed for document %s (path=%s): %s", record.id, record.storage_path, exc, ) self.db.commit() self.db.refresh(record) return record @staticmethod def to_item(record: DbDocumentRecord) -> ScopedDocumentItem: return ScopedDocumentItem( document_id=record.id, source_scope=SourceScope(record.source_scope), title=record.title, filename=record.filename, mime_type=record.mime_type, authority_level=AuthorityLevel(record.authority_level), document_status=SourceDocumentStatus(record.document_status), document_type=DocumentType(record.document_type), purpose=Purpose(record.purpose), retention_policy=RetentionPolicy(record.retention_policy), created_at=record.created_at.isoformat(), updated_at=record.updated_at.isoformat(), effective_date=record.effective_date, ) def _scoped_statement(self, context: ScopedUploadContext): statement = select(DbDocumentRecord).where( DbDocumentRecord.source_scope == context.source_scope.value, DbDocumentRecord.deleted_at.is_(None), ) if context.source_scope == SourceScope.GUEST_SESSION: statement = statement.where(DbDocumentRecord.session_id == context.session_id) elif context.source_scope == SourceScope.USER_PRIVATE: statement = statement.where(DbDocumentRecord.owner_user_id == context.owner_user_id) elif context.source_scope == SourceScope.ORG_PRIVATE: statement = statement.where(DbDocumentRecord.org_id == context.org_id) return statement def _write_upload_file(self, upload: DocumentUploadInput, context: ScopedUploadContext) -> str: directory = self._upload_directory(context) directory.mkdir(parents=True, exist_ok=True) safe_name = PurePath(upload.filename).name or "upload.bin" path = directory / f"{uuid4().hex}_{safe_name}" path.write_bytes(upload.content) return str(path) def _upload_directory(self, context: ScopedUploadContext) -> Path: if context.source_scope == SourceScope.GUEST_SESSION: return Path(self.settings.guest_upload_dir) / str(context.session_id) if context.source_scope == SourceScope.ORG_PRIVATE: return Path(self.settings.org_upload_dir) / str(context.org_id) raise DocumentServiceError("Unsupported document scope.") def _apply_private_metadata( self, document: LegalDocument, chunks: list[LegalChunk], *, context: ScopedUploadContext, filename: str, mime_type: str, ) -> None: org_id = context.org_id or "" document_id = document.document_id source_uri = document.source_uri or f"org://{org_id}/documents/{document_id}" metadata = { "source_scope": context.source_scope.value, "owner_user_id": context.owner_user_id or "", "org_id": org_id, "organization_id": org_id, "session_id": context.session_id or "", "uploaded_by": context.uploaded_by or "", "filename": filename, "mime_type": mime_type, "authority_level": context.authority_level.value, "document_status": context.document_status.value, "document_id": document_id, "source_uri": source_uri, } document.metadata.update(metadata) if not document.source_uri: document.source_uri = source_uri for chunk in chunks: chunk.metadata.update(metadata) chunk.metadata["chunk_id"] = chunk.chunk_id chunk.metadata["chunk_index"] = chunk.chunk_index def _index_private_chunks( self, *, document: LegalDocument, chunks: list[LegalChunk], collection_name: str, ) -> None: from src.services.corpus_indexing import ChromaIndexAdapter, build_indexable_records explicit_provider = self.settings.embedding_provider allow_fake_embeddings = ( explicit_provider == "auto" and not ( has_real_api_key(self.settings.openai_api_key) or has_real_api_key(self.settings.google_api_key) ) ) adapter = ChromaIndexAdapter( persist_dir=self.settings.private_chroma_persist_dir, collection_name=collection_name, embedding_provider=explicit_provider, allow_fake_embeddings=allow_fake_embeddings, ) try: adapter.upsert(build_indexable_records(document, chunks)) except Exception as exc: raise DocumentStoreUnavailableError( "Could not index uploaded document in Chroma. " "Check EMBEDDING_PROVIDER and rebuild the private Chroma collection if needed.", details={"collection_name": collection_name, "reason": str(exc)}, ) from exc def _delete_vectors(self, record: DbDocumentRecord) -> None: collection_name = _private_collection_for_scope(SourceScope(record.source_scope)) try: from src.services.retrieval_service_client import RetrievalServiceClient client = RetrievalServiceClient() client.delete_document(record.id, collection_name) except Exception: return