breakpointsoftware/document-parser

1from __future__ import annotations2 3import json4import os5from dataclasses import dataclass6from datetime import datetime, timezone7from typing import Any8 9from dotenv import load_dotenv10 11 12load_dotenv()13 14 15class FirebaseConfigError(RuntimeError):16    """Raised when Firebase configuration is missing or invalid."""17 18 19@dataclass20class CredentialsObject:21    """OAuth & API Keys for a tenant."""22    openai_api_key: str23    google_service_account_json: str24 25 26@dataclass27class RuleObject:28    """Processing rule for a tenant."""29    rule_id: str30    rule_name: str31    source_folder_id: str32    target_folder_id: str33    target_sheet_id: str34    sheet_tab_name: str35    parsing_prompt: str | None = None36    is_enabled: bool = True37 38    def to_dict(self) -> dict[str, Any]:39        return {40            "rule_id": self.rule_id,41            "rule_name": self.rule_name,42            "source_folder_id": self.source_folder_id,43            "target_folder_id": self.target_folder_id,44            "target_sheet_id": self.target_sheet_id,45            "sheet_tab_name": self.sheet_tab_name,46            "parsing_prompt": self.parsing_prompt,47            "is_enabled": self.is_enabled,48        }49 50    @staticmethod51    def from_dict(data: dict[str, Any]) -> RuleObject:52        return RuleObject(53            rule_id=data.get("rule_id", ""),54            rule_name=data.get("rule_name", ""),55            source_folder_id=data.get("source_folder_id", ""),56            target_folder_id=data.get("target_folder_id", ""),57            target_sheet_id=data.get("target_sheet_id", ""),58            sheet_tab_name=data.get("sheet_tab_name", ""),59            parsing_prompt=data.get("parsing_prompt"),60            is_enabled=data.get("is_enabled", True),61        )62 63 64@dataclass65class TenantConfig:66    """Complete tenant configuration (rules stored inside tenant document)."""67    tenant_id: str68    name: str69    active: bool70    credentials: CredentialsObject71    rules: list[RuleObject] | None = None72    created_at: str | None = None73 74    def to_dict(self) -> dict[str, Any]:75        return {76            "tenant_id": self.tenant_id,77            "name": self.name,78            "active": self.active,79            "credentials": {80                "openai_api_key": self.credentials.openai_api_key,81                "google_service_account_json": self.credentials.google_service_account_json,82            },83            "created_at": self.created_at,84        }85 86    @staticmethod87    def from_dict(data: dict[str, Any], tenant_id: str) -> TenantConfig:88        creds_data = data.get("credentials", {})89        credentials = CredentialsObject(90            openai_api_key=creds_data.get("openai_api_key", ""),91            google_service_account_json=creds_data.get("google_service_account_json", ""),92        )93        return TenantConfig(94            tenant_id=tenant_id,95            name=data.get("name", ""),96            active=data.get("active", True),97            credentials=credentials,98            rules=None,99            created_at=data.get("created_at"),100        )101 102 103class FirebaseTenantConfigManager:104    """Manages tenant configurations in Firebase Firestore.105    106    This is the TENANT CONFIG DB - separate from document processing.107    Other environments import this and use get_tenant() to fetch configs.108    """109 110    def __init__(self, collection_name: str | None = None):111        self.collection_name = (collection_name or os.getenv("FIREBASE_TENANTS_COLLECTION") or "tenants").strip()112        self._db = None113 114    def _build_credentials(self):115        from firebase_admin import credentials116 117        service_account_json = (os.getenv("FIREBASE_SERVICE_ACCOUNT_JSON") or "").strip()118        service_account_file = (os.getenv("FIREBASE_SERVICE_ACCOUNT_FILE") or "").strip()119 120        if not service_account_json:121            service_account_json = (os.getenv("GOOGLE_SERVICE_ACCOUNT_JSON") or "").strip()122        if not service_account_file:123            service_account_file = (os.getenv("GOOGLE_SERVICE_ACCOUNT_FILE") or "").strip()124 125        if service_account_json:126            try:127                info = json.loads(service_account_json)128            except json.JSONDecodeError as exc:129                raise FirebaseConfigError("FIREBASE_SERVICE_ACCOUNT_JSON is not valid JSON.") from exc130            return credentials.Certificate(info)131 132        if service_account_file:133            return credentials.Certificate(service_account_file)134 135        raise FirebaseConfigError(136            "Missing Firebase credentials. Set FIREBASE_SERVICE_ACCOUNT_JSON or FIREBASE_SERVICE_ACCOUNT_FILE."137        )138 139    def _get_db(self):140        if self._db is not None:141            return self._db142 143        import firebase_admin144        from firebase_admin import firestore145 146        if not firebase_admin._apps:147            cred = self._build_credentials()148            firebase_admin.initialize_app(cred)149 150        self._db = firestore.client()151        return self._db152 153    def get_tenant(self, tenant_id: str) -> TenantConfig | None:154        """Retrieve a tenant configuration by ID."""155        doc = self._get_db().collection(self.collection_name).document(tenant_id).get()156        if not doc.exists:157            return None158        data = doc.to_dict() or {}159        return TenantConfig.from_dict(data, tenant_id)160 161    def save_tenant(self, tenant: TenantConfig) -> None:162        """Save a tenant configuration."""163        payload = tenant.to_dict()164        if not payload.get("created_at"):165            payload["created_at"] = datetime.now(timezone.utc).isoformat()166        payload["updated_at"] = datetime.now(timezone.utc).isoformat()167        self._get_db().collection(self.collection_name).document(tenant.tenant_id).set(payload)168 169    def list_active_tenants(self) -> list[TenantConfig]:170        """Retrieve all active tenants."""171        docs = (172            self._get_db()173            .collection(self.collection_name)174            .where("active", "==", True)175            .stream()176        )177        tenants = []178        for doc in docs:179            try:180                tenant = TenantConfig.from_dict(doc.to_dict() or {}, doc.id)181                tenants.append(tenant)182            except Exception:183                continue184        return tenants185 186    def get_all_rules(self, tenant_id: str) -> list[RuleObject]:187        """Get all processing rules for a tenant from the rules subcollection."""188        tenant_doc = self._get_db().collection(self.collection_name).document(tenant_id)189        rules_subcoll = tenant_doc.collection("rules")190        rules = []191        for doc in rules_subcoll.stream():192            rule_data = doc.to_dict() or {}193            rules.append(RuleObject.from_dict(rule_data))194        return rules195 196    def get_enabled_rules(self, tenant_id: str) -> list[RuleObject]:197        """Get all enabled processing rules for a tenant."""198        all_rules = self.get_all_rules(tenant_id)199        return [rule for rule in all_rules if rule.is_enabled]200 201    def add_rule(self, tenant_id: str, rule: RuleObject) -> None:202        """Add a rule to a tenant's rules subcollection."""203        tenant_doc = self._get_db().collection(self.collection_name).document(tenant_id)204        rules_subcoll = tenant_doc.collection("rules")205        rules_subcoll.document(rule.rule_id).set(rule.to_dict())206 207    def remove_rule(self, tenant_id: str, rule_id: str) -> None:208        """Remove a rule from a tenant's rules subcollection."""209        tenant_doc = self._get_db().collection(self.collection_name).document(tenant_id)210        rules_subcoll = tenant_doc.collection("rules")211        rules_subcoll.document(rule_id).delete()212 213 214# Module-level convenience functions for backward compatibility215_manager = FirebaseTenantConfigManager()216 217 218def load_all_tenants() -> list[TenantConfig]:219    """Load all active tenants from Firebase."""220    return _manager.list_active_tenants()221 222 223def load_tenant_rules(tenant_id: str) -> list[RuleObject]:224    """Load all rules for a specific tenant."""225    return _manager.get_all_rules(tenant_id)226