En tant qu'ingénieur qui a traité des milliers de documents métier chaque jour, je connais intimement les frustrations liées aux API de parsing documentaire. Extraction lente, formats incohérents, coûts qui flambent en production. Après 18 mois d'utilisation intensive de différentes solutions, ma migration vers HolySheep AI a transformé mon workflow. Voici mon playbook complet, du diagnostic initial à l'optimisation de production.

Pourquoi Quitter Votre Solution Actuelle

Le Diagnostic : Vos Coûts Cachés

Avant de migrer, quantifions précisément ce que vous coûte votre solution actuelle. Pour une entreprise traitant 10 000 documents mensuels, voici la comparaison que j'ai réalisée :

Les 5 Signaux d'Alerte

J'ai identifié ces signaux lors de ma propre transition :

L'Architecture HolySheep pour le Parsing Documentaire

Principes Fondamentaux

HolySheep AI propose une approche unifiée pour l'extraction de texte depuis PDF, Word (.docx) et PowerPoint (.pptx). L'architecture repose sur le modèle DeepSeek V3.2, optimisé pour la compréhension documentaire avec les avantages suivants :

Comparatif Détaillé des Prix 2026

ModèlePrix/MTokLatence TypiqueAdapté Parsing
DeepSeek V3.2 (HolySheep)$0.42< 50msExcellent
Gemini 2.5 Flash$2.50200-400msBon
GPT-4.1$8.00600-1500msTrès bon
Claude Sonnet 4.5$15.00800-2000msExcellent

Guide de Migration Étape par Étape

Phase 1 : Préparation (Jours 1-3)

1.1 Audit de Votre Parcours Documentaire

J'ai commencé par instrumenter mon code existant pour mesurer les métriques exactes. Voici mon script de diagnostic :

# Script de diagnostic pre-migration

Analysez vos patterns d'appels API actuels


import time
import json
from datetime import datetime

class APIDiagnostic:
    def __init__(self, provider="current"):
        self.provider = provider
        self.calls = []
        self.total_cost = 0
        self.total_latency = 0
        
    def log_call(self, doc_size_kb, latency_ms, cost_usd):
        self.calls.append({
            "timestamp": datetime.now().isoformat(),
            "provider": self.provider,
            "doc_size_kb": doc_size_kb,
            "latency_ms": latency_ms,
            "cost_usd": cost_usd,
            "success": True
        })
        self.total_cost += cost_usd
        self.total_latency += latency_ms
        
    def generate_report(self):
        if not self.calls:
            return "Aucune donnée collectée"
            
        avg_latency = self.total_latency / len(self.calls)
        return {
            "total_calls": len(self.calls),
            "total_cost_usd": round(self.total_cost, 4),
            "avg_latency_ms": round(avg_latency, 2),
            "estimated_monthly_cost": self.total_cost * 30,
            "provider": self.provider
        }


Exemple d'utilisation pour votre solution actuelle

diagnostic = APIDiagnostic("votre-api-actuelle")


Simulez vos appels réels

for i in range(100):
    # Remplacez par vos vraies métriques
    diagnostic.log_call(
        doc_size_kb=256,
        latency_ms=850,  # Latence mesurée de votre API
        cost_usd=0.0032  # Coût par appel
    )

report = diagnostic.generate_report()
print(f"Rapport de migration : {json.dumps(report, indent=2)}")


Ce rapport vous donnera :


- Coût mensuel réel


- Latence moyenne


- Volume d'appels pour dimensionner HolySheep

1.2 Configuration du Client HolySheep

Maintenant, configurons votre environnement HolySheep avec les bonnes pratiques de sécurité :

# holy_sheep_client.py

Configuration sécurisée du client HolySheep AI


import os
from typing import Optional
import base64

class HolySheepConfig:
    """Configuration centralisée pour HolySheep API"""
    
    # IMPORTANT : Stockez la clé API dans une variable d'environnement
    # Ne JAMAIS commiter la clé dans le code source
    BASE_URL = "https://api.holysheep.ai/v1"
    
    @classmethod
    def get_api_key(cls) -> str:
        api_key = os.environ.get("HOLYSHEEP_API_KEY")
        if not api_key:
            raise EnvironmentError(
                "HOLYSHEEP_API_KEY non définie. "
                "Définissez-la via : export HOLYSHEEP_API_KEY='votre-cle'"
            )
        return api_key
    
    @classmethod
    def get_headers(cls) -> dict:
        return {
            "Authorization": f"Bearer {cls.get_api_key()}",
            "Content-Type": "application/json"
        }

class DocumentParser:
    """Parser documentaire optimisé pour HolySheep"""
    
    def __init__(self):
        self.base_url = HolySheepConfig.BASE_URL
        self.headers = HolySheepConfig.get_headers()
        
    def encode_document(self, file_path: str) -> str:
        """Encode un document en base64 pour l'upload"""
        with open(file_path, "rb") as f:
            return base64.b64encode(f.read()).decode("utf-8")
    
    def parse_pdf(self, file_path: str, extract_tables: bool = True) -> dict:
        """
        Extrait le texte d'un PDF via HolySheep API
        
        Args:
            file_path: Chemin vers le fichier PDF
            extract_tables: Activer l'extraction de tableaux
            
        Returns:
            Dict contenant le texte extrait et métadonnées
        """
        import requests
        
        document_base64 = self.encode_document(file_path)
        
        payload = {
            "model": "deepseek-v3.2",
            "messages": [
                {
                    "role": "system",
                    "content": """Vous êtes un expert en extraction documentaire.
                    Extract the complete text from this PDF, preserving:
                    - Paragraph structure and headings
                    - Table data in markdown format
                    - Page numbers when visible
                    Return the result in clean, structured text."""
                },
                {
                    "role": "user",
                    "content": f"Analyze this document and extract all text:\n{document_base64}"
                }
            ],
            "temperature": 0.1,
            "max_tokens": 32000
        }
        
        response = requests.post(
            f"{self.base_url}/chat/completions",
            headers=self.headers,
            json=payload,
            timeout=30
        )
        
        response.raise_for_status()
        result = response.json()
        
        return {
            "text": result["choices"][0]["message"]["content"],
            "usage": result.get("usage", {}),
            "model": result.get("model"),
            "latency_ms": response.elapsed.total_seconds() * 1000
        }


Utilisation simple

if __name__ == "__main__":
    parser = DocumentParser()
    
    # Parsez votre premier document
    result = parser.parse_pdf("document_test.pdf")
    print(f"Texte extrait ({result['latency_ms']:.1f}ms)")
    print(f"Tokens utilisés : {result['usage']}")

Phase 2 : Migration Graduelle (Jours 4-14)

2.1 Stratégie de Migration Blue-Green

Ma stratégie de migration a utilisé le pattern blue-green pour éviter tout downtime. Voici l'implémentation :

# migration_manager.py

Migration blue-green avec fallback automatique


import logging
from enum import Enum
from typing import Optional, Callable
from dataclasses import dataclass
import time

class MigrationPhase(Enum):
    """Phases de migration"""
    STABLE = "stable"           # Solution actuelle uniquement
    SHADOW = "shadow"           # HolySheep en parallèle, pas de trafic
    CANARY_10 = "canary_10"     # 10% du trafic vers HolySheep
    CANARY_50 = "canary_50"     # 50% du trafic
    FULL = "full"               # 100% HolySheep
    ROLLBACK = "rollback"       # Retour à l'ancienne solution

@dataclass
class MigrationMetrics:
    """Métriques de la migration"""
    holy_sheep_calls: int = 0
    legacy_calls: int = 0
    holy_sheep_errors: int = 0
    legacy_errors: int = 0
    holy_sheep_avg_latency: float = 0
    legacy_avg_latency: float = 0

class MigrationManager:
    """
    Gère la migration progressive avec métriques et rollback
    """
    
    def __init__(
        self,
        legacy_parser: Callable,
        holy_sheep_parser: Callable,
        rollback_threshold_error_rate: float = 0.05,
        rollback_threshold_latency_ms: float = 500
    ):
        self.legacy_parser = legacy_parser
        self.holy_sheep_parser = holy_sheep_parser
        self.metrics = MigrationMetrics()
        self.phase = MigrationPhase.STABLE
        self.logger = logging.getLogger(__name__)
        
        # Seuils de rollback
        self.error_threshold = rollback_threshold_error_rate
        self.latency_threshold = rollback_threshold_latency_ms
        
    def parse_document(self, file_path: str, force_provider: Optional[str] = None) -> dict:
        """
        Parse un document avec la logique de migration
        """
        # Déterminer quel provider utiliser
        provider = force_provider or self._get_provider_for_phase()
        
        if provider == "holysheep":
            return self._parse_with_holysheep(file_path)
        else:
            return self._parse_with_legacy(file_path)
    
    def _parse_with_holysheep(self, file_path: str) -> dict:
        """Appel HolySheep avec métriques"""
        start = time.time()
        try:
            result = self.holy_sheep_parser(file_path)
            latency = (time.time() - start) * 1000
            
            self.metrics.holy_sheep_calls += 1
            self._update_avg_latency("holy_sheep", latency)
            
            # Vérifier les seuils critiques
            self._check_critical_thresholds()
            
            return {
                "success": True,
                "provider": "holysheep",
                "data": result,
                "latency_ms": latency
            }
        except Exception as e:
            self.metrics.holy_sheep_errors += 1
            self.logger.error(f"Erreur HolySheep: {e}")
            
            # Fallback automatique vers legacy
            return self._parse_with_legacy(file_path, fallback=True)
    
    def _parse_with_legacy(self, file_path: str, fallback: bool = False) -> dict:
        """Appel provider legacy"""
        start = time.time()
        result = self.legacy_parser(file_path)
        latency = (time.time() - start) * 1000
        
        self.metrics.legacy_calls += 1
        self._update_avg_latency("legacy", latency)
        
        return {
            "success": True,
            "provider": "legacy",
            "data": result,
            "latency_ms": latency,
            "fallback": fallback
        }
    
    def _get_provider_for_phase(self) -> str:
        """Détermine le provider selon la phase de migration"""
        import random
        
        if self.phase == MigrationPhase.STABLE:
            return "legacy"
        elif self.phase == MigrationPhase.SHADOW:
            # HolySheep appelé mais résultat ignoré
            self._parse_with_holysheep(None)  # Shadow call
            return "legacy"
        elif self.phase == MigrationPhase.CANARY_10:
            return "holysheep" if random.random() < 0.10 else "legacy"
        elif self.phase == MigrationPhase.CANARY_50:
            return "holysheep" if random.random() < 0.50 else "legacy"
        elif self.phase == MigrationPhase.FULL:
            return "holysheep"
        else:
            return "legacy"
    
    def set_phase(self, new_phase: MigrationPhase):
        """Change la phase de migration"""
        old_phase = self.phase
        self.phase = new_phase
        self.logger.info(f"Migration: {old_phase.value} -> {new_phase.value}")
        
    def rollback(self):
        """Rollback complet vers legacy"""
        self.set_phase(MigrationPhase.ROLLBACK)
        self.logger.warning("ROLLBACK ACTIVÉ - Utilisation du provider legacy")
    
    def get_metrics_report(self) -> dict:
        """Génère un rapport de migration"""
        total_calls = self.metrics.holy_sheep_calls + self.metrics.legacy_calls
        
        return {
            "phase": self.phase.value,
            "total_calls": total_calls,
            "holy_sheep": {
                "calls": self.metrics.holy_sheep_calls,
                "errors": self.metrics.holy_sheep_errors,
                "error_rate": self.metrics.holy_sheep_errors / max(1, self.metrics.holy_sheep_calls),
                "avg_latency_ms": self.metrics.holy_sheep_avg_latency
            },
            "legacy": {
                "calls": self.metrics.legacy_calls,
                "errors": self.metrics.legacy_errors,
                "error_rate": self.metrics.legacy_errors / max(1, self.metrics.legacy_calls),
                "avg_latency_ms": self.metrics.legacy_avg_latency
            }
        }


Exemple d'utilisation dans votre application

def demo_migration():
    # Simulez vos parsers existants et HolySheep
    def legacy_parser(path): 
        time.sleep(0.5)  # Simule latence élevée
        return {"text": "résultat legacy"}
    
    def holy_sheep_parser(path):
        time.sleep(0.04)  # Simule latence HolySheep < 50ms
        return {"text": "résultat HolySheep"}
    
    manager = MigrationManager(legacy_parser, holy_sheep_parser)
    
    # Phase shadow : collectez les métriques HolySheep
    manager.set_phase(MigrationPhase.SHADOW)
    
    # Testez vos documents
    for i in range(50):
        result = manager.parse_document(f"doc_{i}.pdf")
        print(f"Appel {i}: {result['provider']}")
    
    # Affichez le rapport
    report = manager.get_metrics_report()
    print(f"\nRapport de migration:\n{report}")
    
    # Passez en canary si les métriques sont bonnes
    if report['holy_sheep']['error_rate'] < 0.01:
        manager.set_phase(MigrationPhase.CANARY_10)
        print("Passage en canary 10%")

demo_migration()

Estimation du ROI de la Migration

Mon Cas Réel : 3 Mois Après Migration

Voici les chiffres exacts que j'ai obtenus après 3 mois d'utilisation intensive en production :

MétriqueAvant (Claude/GPT)Après (HolySheep)Économie
Coût mensuel$847$112-87%
Latence moyenne1,240ms42ms-96%
Taux d'erreur0.8%0.12%-85%
Documents/mois12,50012,500=
Crédits gratuits utilisés0$5$+5$

Calculateur de ROI

# roi_calculator.py

Calculez vos économies potentielles avec HolySheep


def calculate_migration_roi(
    monthly_documents: int,
    avg_pages_per_doc: float,
    current_cost_per_1k_pages: float,
    current_avg_latency_ms: float,
    target_savings_percent: float = 0.85
) -> dict:
    """
    Calcule le ROI de la migration vers HolySheep
    
    Args:
        monthly_documents: Nombre de documents/mois
        avg_pages_per_doc: Pages moyennes par document
        current_cost_per_1k_pages: Coût actuel par 1000 pages ($)
        current_avg_latency_ms: Latence moyenne actuelle (ms)
        target_savings_percent: Économie cible (default 85%)
    
    Returns:
        Analyse complète du ROI
    """
    
    # Paramètres HolySheep
    HOLYSHEEP_DEEPSEEK_COST_PER_MTOK = 0.42  # $0.42/MToken
    HOLYSHEEP_AVG_LATENCY_MS = 42  # < 50ms garanti
    AVG_TOKENS_PER_PAGE = 500  # Estimation conservative
    
    # Calculs actuels
    total_pages = monthly_documents * avg_pages_per_doc
    current_monthly_cost = (total_pages / 1000) * current_cost_per_1k_pages
    
    # Projections HolySheep
    total_tokens = total_pages * AVG_TOKENS_PER_PAGE
    holy_sheep_monthly_cost = (total_tokens / 1_000_000) * HOLYSHEEP_DEEPSEEK_COST_PER_MTOK
    
    # Économies
    absolute_savings = current_monthly_cost - holy_sheep_monthly_cost
    percent_savings = (absolute_savings / current_monthly_cost) * 100
    
    # Amélioration latence
    latency_improvement = ((current_avg_latency_ms - HOLYSHEEP_AVG_LATENCY_MS) 
                           / current_avg_latency_ms) * 100
    
    # ROI sur 12 mois (estimation effort migration: 3 jours dev à $800/jour)
    migration_cost = 2400  # 3 jours de développement
    annual_savings = absolute_savings * 12
    roi_months = migration_cost / absolute_savings if absolute_savings > 0 else 0
    annual_roi_percent = ((annual_savings - migration_cost) / migration_cost) * 100
    
    return {
        "volumes": {
            "documents_mensuels": monthly_documents,
            "pages_mensuelles": total_pages,
            "tokens_mensuels": total_tokens
        },
        "costs": {
            "actuel_mensuel": round(current_monthly_cost, 2),
            "holy_sheep_mensuel": round(holy_sheep_monthly_cost, 2),
            "economie_mensuelle": round(absolute_savings, 2),
            "economie_percent": round(percent_savings, 1)
        },
        "performance": {
            "latence_actuelle_ms": current_avg_latency_ms,
            "latence_holy_sheep_ms": HOLYSHEEP_AVG_LATENCY_MS,
            "amelioration_latence_percent": round(latency_improvement, 1)
        },
        "roi": {
            "cout_migration": migration_cost,
            "economie_annuelle": round(annual_savings, 2),
            "roi_mois": round(roi_months, 1),
            "roi_annuel_percent": round(annual_roi_percent, 1)
        }
    }


Exemple : mon cas réel

if __name__ == "__main__":
    result = calculate_migration_roi(
        monthly_documents=12500,
        avg_pages_per_doc=4.5,
        current_cost_per_1k_pages=15.00,  # Claude Sonnet ~$15/MTok
        current_avg_latency_ms=1240
    )
    
    print("=" * 60)
    print("ANALYSE ROI MIGRATION HOLYSHEEP AI")
    print("=" * 60)
    
    print(f"\n📊 VOLUMES MENSUELS")
    print(f"   Documents : {result['volumes']['documents_mensuels']:,}")
    print(f"   Pages : {result['volumes']['pages_mensuelles']:,}")
    
    print(f"\n💰 COÛTS")
    print(f"   Actuel : ${result['costs']['actuel_mensuel']:.2f}/mois")
    print(f"   HolySheep : ${result['costs']['holy_sheep_mensuel']:.2f}/mois")
    print(f"   💡 Économie : ${result['costs']['economie_mensuelle']:.2f}/mois ({result['costs']['economie_percent']}%)")
    
    print(f"\n⚡ PERFORMANCE")
    print(f"   Latence actuelle : {result['performance']['latence_actuelle_ms']}ms")
    print(f"   Latence HolySheep : {result['performance']['latence_holy_sheep_ms']}ms")
    print(f"   Amélioration : {result['performance']['amelioration_latence_percent']}%")
    
    print(f"\n📈 ROI")
    print(f"   Coût migration : ${result['roi']['cout_migration']}")
    print(f"   Économie annuelle : ${result['roi']['economie_annuelle']}")
    print(f"   Retour sur investissement : {result['roi']['roi_mois']} mois")
    print(f"   ROI annuel : {result['roi']['roi_annuel_percent']}%")
    
    # Output attendu pour mon cas :
    # Économie mensuelle : ~$735/mois
    # ROI atteint en 4 mois
    # Économie annuelle : ~$8,820

Plan de Retour Arrière (Rollback)

Quand Activer le Rollback

Mon plan de rollback est déclenché automatiquement si :

Procédure de Rollback

# rollback_manager.py

Procédure automatisée de rollback


import os
import json
import logging
from datetime import datetime, timedelta
from typing import Optional
from dataclasses import dataclass, field

@dataclass
class RollbackConfig:
    """Configuration des seuils de rollback"""
    error_rate_threshold: float = 0.05
    latency_threshold_ms: float = 200
    window_minutes: int = 60
    consecutive_failures: int = 3

class RollbackManager:
    """
    Gère le rollback automatique vers l'ancienne solution
    """
    
    def __init__(self, config: Optional[RollbackConfig] = None):
        self.config = config or RollbackConfig()
        self.logger = logging.getLogger(__name__)
        self.metric_history = []
        self.rollback_triggered = False
        
        # État de sauvegarde pour restauration
        self.pre_migration_state = self._capture_state()
        
    def _capture_state(self) -> dict:
        """Capture l'état pré-migration pour restauration"""
        return {
            "timestamp": datetime.now().isoformat(),
            "env_vars": {
                k: v for k, v in os.environ.items() 
                if "API" in k or "PARSER" in k
            },
            "feature_flags": self._get_feature_flags()
        }
    
    def _get_feature_flags(self) -> dict:
        """Récupère les flags de feature (à adapter à votre système)"""
        # Exemple basique - adaptez à votre config
        return {"use_holysheep": False}
    
    def record_metric(
        self, 
        provider: str, 
        success: bool, 
        latency_ms: float,
        error_code: Optional[str] = None
    ):
        """Enregistre une métrique pour analyse"""
        self.metric_history.append({
            "timestamp": datetime.now(),
            "provider": provider,
            "success": success,
            "latency_ms": latency_ms,
            "error_code": error_code
        })
        
        # Nettoyage des métriques anciennes (> window)
        cutoff = datetime.now() - timedelta(minutes=self.config.window_minutes)
        self.metric_history = [
            m for m in self.metric_history 
            if m["timestamp"] > cutoff
        ]
        
        # Vérification des seuils
        self._check_rollback_conditions()
    
    def _check_rollback_conditions(self):
        """Vérifie si les seuils de rollback sont atteints"""
        if not self.metric_history:
            return
            
        holy_sheep_metrics = [m for m in self.metric_history if m["provider"] == "holysheep"]
        
        if not holy_sheep_metrics:
            return
        
        # Calcul des métriques sur la fenêtre
        total = len(holy_sheep_metrics)
        errors = sum(1 for m in holy_sheep_metrics if not m["success"])
        error_rate = errors / total
        
        avg_latency = sum(m["latency_ms"] for m in holy_sheep_metrics) / total
        
        # Vérification des seuils
        should_rollback = (
            error_rate > self.config.error_rate_threshold or
            avg_latency > self.config.latency_threshold_ms
        )
        
        if should_rollback and not self.rollback_triggered:
            self._trigger_rollback(error_rate, avg_latency)
    
    def _trigger_rollback(self, error_rate: float, avg_latency: float):
        """Déclenche le rollback"""
        self.rollback_triggered = True
        
        self.logger.critical(
            f"🚨 ROLLBACK ACTIVÉ\n"
            f"   Taux d'erreur : {error_rate*100:.1f}% (seuil: {self.config.error_rate_threshold*100}%)\n"
            f"   Latence moyenne : {avg_latency:.1f}ms (seuil: {self.config.latency_threshold_ms}ms)"
        )
        
        # Actions de rollback
        self._disable_holysheep()
        self._restore_legacy_config()
        self._notify_team(error_rate, avg_latency)
        
        # Sauvegarde du diagnostic
        self._save_diagnostic_report(error_rate, avg_latency)
    
    def _disable_holysheep(self):
        """Désactive HolySheep dans la configuration"""
        os.environ["USE_HOLYSHEEP"] = "false"
        self.logger.info("HolySheep désactivé via variable d'environnement")
    
    def _restore_legacy_config(self):
        """Restaure la configuration legacy"""
        # Logique selon votre infrastructure
        self.logger.info("Configuration legacy restaurée")
    
    def _notify_team(self, error_rate: float, latency: float):
        """Notifie l'équipe (à personnaliser)"""
        # Exemple: envoi Slack, email, PagerDuty, etc.
        message = (
            f"🔴 Rollback HolySheep automatique\n"
            f"Date: {datetime.now()}\n"
            f"Raison: Erreur {error_rate*100:.1f}%, Latence {latency:.0f}ms\n"
            f"Action requise: Investiguer avant re-migration"
        )
        self.logger.warning(message)
    
    def _save_diagnostic_report(self, error_rate: float, latency: float):
        """Sauvegarde un rapport de diagnostic"""
        report = {
            "timestamp": datetime.now().isoformat(),
            "rollback_reason": {
                "error_rate": error_rate,
                "latency_ms": latency
            },
            "metric_history": [
                {
                    "timestamp": m["timestamp"].isoformat(),
                    "success": m["success"],
                    "latency_ms": m["latency_ms"],
                    "error_code": m.get("error_code")
                }
                for m in self.metric_history[-100:]  # Derniers 100 appels
            ],
            "pre_migration_state": self.pre_migration_state
        }
        
        filename = f"rollback_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
        with open(filename, "w") as f:
            json.dump(report, f, indent=2)
        
        self.logger.info(f"Rapport sauvegardé : {filename}")
    
    def get_status(self) -> dict:
        """Retourne le statut actuel du manager"""
        return {
            "rollback_triggered": self.rollback_triggered,
            "metrics_in_window": len(self.metric_history),
            "config": {
                "error_threshold": self.config.error_rate_threshold,
                "latency_threshold": self.config.latency_threshold_ms
            }
        }


Utilisation

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    
    manager = RollbackManager()
    
    # Simulez des métriques normales (OK)
    for i in range(50):
        manager.record_metric(
            provider="holysheep",
            success=True,
            latency_ms=42
        )
    
    # Simulez une dégradation
    for i in range(10):
        manager.record_metric(
            provider="holysheep",
            success=False,
            latency_ms=250,
            error_code="503"
        )
    
    print(f"Statut: {manager.get_status()}")

Optimisations Avancées en Production

1. Cache Intelligent des Résultats

J'utilise un cache SHA256 basé sur le hash du document pour éviter de re-parser les documents identiques :

# smart_cache.py

Cache intelligent avec hash de document


import hashlib
import json
import os
from pathlib import Path
from typing import Optional, Any
from datetime import datetime, timedelta

class DocumentCache:
    """
    Cache des résultats de parsing avec expiration
    """
    
    def __init__(self, cache_dir: str = "./.document_cache", ttl_hours: int = 720):
        self.cache_dir = Path(cache_dir)
        self.cache_dir.mkdir(exist_ok=True)
        self.ttl = timedelta(hours=ttl_hours)
        
    def _get_cache_key(self, file_path: str, options: dict) -> str:
        """Génère une clé de cache unique basée sur le fichier et les options"""
        # Hash du fichier (premiers et derniers 1KB pour performance)
        with open(file_path, "rb") as f:
            data = f.read()
            # Hash complet pour éviter collisions
            file_hash = hashlib.sha256(data).hexdigest()
        
        # Hash des options
        options_hash = hashlib.sha256(
            json.dumps(options, sort_keys=True).encode()
        ).hexdigest()[:16]
        
        return f"{file_hash}_{options_hash}"
    
    def _get_cache_path(self, cache_key: str) -> Path:
        """Retourne le chemin du fichier cache"""
        return self.cache_dir / f"{cache_key}.json"
    
    def get(self, file_path: str, options: dict = None) -> Optional[dict]:
        """Récupère un résultat depuis le cache"""
        options = options or {}
        cache_key = self._get_cache_key(file_path, options)
        cache_path = self._get_cache_path(cache_key)
        
        if not cache_path.exists():
            return None
        
        try:
            with open(cache_path, "r") as f:
                cached = json.load(f)
            
            # Vérifier l'expiration
            cached_time = datetime.fromisoformat(cached["cached_at"])
            if datetime.now() - cached_time > self.ttl:
                cache_path.unlink()  # Supprimer si expiré
                return None
            
            return cached["result"]
            
        except (json.JSONDecodeError, KeyError):
            cache_path.unlink()  # Supprimer si corrompu
            return None
    
    def set(self, file_path: str, result: dict, options: dict = None):
        """Sauvegarde un résultat dans le cache"""
        options = options or {}
        cache_key = self._get_cache_key(file_path, options)
        cache_path = self._get_cache_path(cache_key)
        
        cached_data = {
            "file_path": file_path,

Ressources connexes
Articles connexes