breakpointsoftware/document-parser

1from __future__ import annotations2 3import hashlib4import json5import os6from datetime import datetime, timezone7from pathlib import Path8from typing import Any9 10from dotenv import load_dotenv11 12 13load_dotenv()14 15 16class FirebaseConfigError(RuntimeError):17    """Raised when Firebase configuration is missing or invalid."""18 19 20class FirebaseProcessedFilesTracker:21    def __init__(22        self,23        tenant_id: str,24        rule_id: str,25    ):26        self.tenant_id = (tenant_id or "").strip()27        self.rule_id = (rule_id or "").strip()28        29        if not self.tenant_id:30            raise FirebaseConfigError("tenant_id is required")31        if not self.rule_id:32            raise FirebaseConfigError("rule_id is required")33        34        # Multi-tenant rule-based tracking: tenants/{tenant_id}/rules_executions/{rule_id}/processed_documents/{file_hash}35        self.collection_name = f"{self.tenant_id}/{self.rule_id}_executions/processed_documents"36        self._db = None37 38    def _build_credentials(self):39        from firebase_admin import credentials40 41        service_account_json = (os.getenv("FIREBASE_SERVICE_ACCOUNT_JSON") or "").strip()42        service_account_file = (os.getenv("FIREBASE_SERVICE_ACCOUNT_FILE") or "").strip()43 44        # Reuse Google service account settings when dedicated Firebase keys are not set.45        if not service_account_json:46            service_account_json = (os.getenv("GOOGLE_SERVICE_ACCOUNT_JSON") or "").strip()47        if not service_account_file:48            service_account_file = (os.getenv("GOOGLE_SERVICE_ACCOUNT_FILE") or "").strip()49 50        if service_account_json:51            try:52                info = json.loads(service_account_json)53            except json.JSONDecodeError as exc:54                raise FirebaseConfigError("FIREBASE_SERVICE_ACCOUNT_JSON is not valid JSON.") from exc55            return credentials.Certificate(info)56 57        if service_account_file:58            return credentials.Certificate(service_account_file)59 60        raise FirebaseConfigError(61            "Missing Firebase credentials. Set FIREBASE_SERVICE_ACCOUNT_JSON or FIREBASE_SERVICE_ACCOUNT_FILE "62            "(or reuse GOOGLE_SERVICE_ACCOUNT_JSON/GOOGLE_SERVICE_ACCOUNT_FILE)."63        )64 65    def _get_db(self):66        if self._db is not None:67            return self._db68 69        import firebase_admin70        from firebase_admin import firestore71 72        if not firebase_admin._apps:73            cred = self._build_credentials()74            firebase_admin.initialize_app(cred)75 76        self._db = firestore.client()77        return self._db78 79    def compute_file_hash(self, file_path: Path) -> str:80        sha256 = hashlib.sha256()81        with file_path.open("rb") as stream:82            for chunk in iter(lambda: stream.read(1024 * 1024), b""):83                sha256.update(chunk)84        return sha256.hexdigest()85 86    def is_processed(self, file_hash: str) -> bool:87        doc = self._get_db().collection(self.collection_name).document(file_hash).get()88        return bool(doc.exists)89 90    def mark_processed(self, file_hash: str, source_file: str) -> None:91        payload = {92            "file_hash": file_hash,93            "source_file": source_file,94            "processed_at": datetime.now(timezone.utc).isoformat(),95        }96        self._get_db().collection(self.collection_name).document(file_hash).set(payload)97 98    def get_document_record(self, document_id: str) -> dict[str, Any] | None:99        doc = self._get_db().collection(self.collection_name).document(document_id).get()100        if not doc.exists:101            return None102        return doc.to_dict() or {}103 104    def save_document_record(105        self,106        file_hash: str,107        document_id: str,108        source_file: str,109        modification_date: str | None,110        status: str,111        parsed_data: dict[str, Any] | None = None,112    ) -> None:113        now = datetime.now(timezone.utc).isoformat()114        payload: dict[str, Any] = {115            "file_hash": file_hash,116            "document_id": document_id,117            "source_file": source_file,118            "modificationDate": modification_date,119            "status": status,120            "updated_at": now,121        }122 123        if parsed_data is not None:124            payload["parsed_data"] = parsed_data125 126        existing = self.get_document_record(file_hash)127        if existing is None:128            payload["created_at"] = now129 130        # Use file_hash as the document ID in Firebase131        self._get_db().collection(self.collection_name).document(file_hash).set(payload, merge=True)132        133        # Also mark the file as processed in the documents collection134        self.mark_processed(file_hash, source_file)135 136    def mark_document_sent(self, document_id: str) -> None:137        now = datetime.now(timezone.utc).isoformat()138        payload = {139            "status": "Sent",140            "sent_at": now,141            "updated_at": now,142        }143        self._get_db().collection(self.collection_name).document(document_id).set(payload, merge=True)144 145    def list_documents_by_statuses(self, statuses: list[str]) -> list[dict[str, Any]]:146        unique_statuses = [status for status in dict.fromkeys(statuses) if str(status).strip()]147        if not unique_statuses:148            return []149 150        collection = self._get_db().collection(self.collection_name)151        documents: list[dict[str, Any]] = []152 153        try:154            query = collection.where("status", "in", unique_statuses)155            for doc in query.stream():156                payload = doc.to_dict() or {}157                payload.setdefault("document_id", doc.id)158                documents.append(payload)159            return documents160        except Exception:161            # Fallback for environments where the 'in' operator is unavailable or restricted.162            for doc in collection.stream():163                payload = doc.to_dict() or {}164                status = str(payload.get("status") or "")165                if status in unique_statuses:166                    payload.setdefault("document_id", doc.id)167                    documents.append(payload)168 169        return documents170 171 172def build_firebase_tracker(173    tenant_id: str,174    rule_id: str,175) -> FirebaseProcessedFilesTracker:176    enabled = (os.getenv("FIREBASE_TRACK_PROCESSED") or "true").strip().lower()177    if enabled in {"0", "false", "no", "off"}:178        raise FirebaseConfigError("Firebase tracking is disabled (FIREBASE_TRACK_PROCESSED=false)")179 180    return FirebaseProcessedFilesTracker(tenant_id=tenant_id, rule_id=rule_id)181 182 183def check_files_already_processed(184    tracker: FirebaseProcessedFilesTracker | None, file_paths: list[str]185) -> tuple[list[str], list[tuple[str, str]], str | None]:186    if tracker is None:187        return file_paths, [], None188 189    new_files: list[str] = []190    skipped_files: list[tuple[str, str]] = []191 192    try:193        for file_path in file_paths:194            path = Path(file_path)195            file_hash = tracker.compute_file_hash(path)196            if tracker.is_processed(file_hash):197                skipped_files.append((path.name, file_hash))198            else:199                new_files.append(file_path)200    except Exception as exc:201        return file_paths, [], f"Firebase tracking unavailable. Processing all files. Details: {exc}"202 203    return new_files, skipped_files, None204 205 206def check_drive_documents_to_process(207    tracker: FirebaseProcessedFilesTracker | None,208    documents: list[dict[str, Any]],209) -> tuple[list[dict[str, Any]], list[dict[str, Any]], str | None]:210    """211    Returns (documents_to_process, skipped_documents, warning).212 213    Process:214    1. Compute file hash from file content215    2. Use hash as unique document identifier216    3. Check if document (by hash) already exists217    4. Compare modification dates for changes218 219    - File already processed (by hash) -> skipped220    - New document (no hash record) -> status "Parsed"221    - Existing document with modified date changed -> status "Modified"222    - Existing document with same modified date -> skipped223    """224    if tracker is None:225        docs = [dict(doc, target_status="Parsed") for doc in documents]226        return docs, [], None227 228    to_process: list[dict[str, Any]] = []229    skipped: list[dict[str, Any]] = []230 231    try:232        for document in documents:233            document_id = str(document.get("document_id") or "").strip()234            modified_time = str(document.get("modificationDate") or "").strip() or None235            local_path = str(document.get("local_path") or "").strip()236 237            if not document_id or not local_path:238                skipped.append({**document, "skip_reason": "missing_document_id_or_path"})239                continue240 241            # Compute file hash from file content - this becomes the unique identifier242            try:243                path = Path(local_path)244                if not path.exists():245                    skipped.append({**document, "skip_reason": "file_not_found"})246                    continue247                file_hash = tracker.compute_file_hash(path)248            except Exception as exc:249                skipped.append({**document, "skip_reason": f"hash_computation_failed: {exc}"})250                continue251 252            # Use hash as the document identifier253            existing = tracker.get_document_record(file_hash)254            if existing is None:255                # New file - process it256                to_process.append({257                    **document,258                    "file_hash": file_hash,259                    "target_status": "Parsed",260                    "hash_document_id": file_hash,261                })262                continue263 264            # Existing document - check if modified265            existing_modified = str(existing.get("modificationDate") or "").strip() or None266            if existing_modified != modified_time:267                to_process.append({268                    **document,269                    "file_hash": file_hash,270                    "target_status": "Modified",271                    "hash_document_id": file_hash,272                })273            else:274                skipped.append({275                    **document,276                    "skip_reason": "already_processed_same_content",277                    "file_hash": file_hash,278                })279    except Exception as exc:280        docs = [dict(doc, target_status="Parsed") for doc in documents]281        return docs, [], f"Firebase tracking unavailable. Processing all Drive files. Details: {exc}"282 283    return to_process, skipped, None284 285