KI Dokumentenverarbeitung: 92% Zeitersparnis + €76k ROI [OCR+NLP 2025]

IDP_Pipeline_Rechnungen:
  Input-Layer:
    - E-Mail-Postfach (IMAP-Sync)
    - Upload-Portal (Web-Interface)
    - Scanner-Integration (Network Scan to Folder)
    - Formate: PDF, PNG, JPG, TIFF
  
  Document-Intelligence (Azure):
    OCR_Engine:
      - Model: prebuilt-invoice (Azure)
      - Accuracy: 98,2% bei deutschen Rechnungen
      - Felder: Kreditor, RE-Nr, Datum, Betrag, USt, IBAN
      - Processing: \\<3 Sek/Rechnung
    
    Custom_Validation:
      - Dublettenprüfung (RE-Nr + Kreditor)
      - Plausibilitätschecks (Betrag, USt-Sätze)
      - Kreditor-Whitelist-Abgleich
      - GoBD-Konformität
  
  ERP-Integration:
    - System: DATEV Unternehmen Online
    - Method: REST-API
    - Auto-Buchung bei Confidence >\>90%
    - Human-Review bei \<90%
    - Belege anhängen (PDF/A Langzeitarchiv)
  
  Dashboard:
    - Verarbeitungsstatus (Echtzeit)
    - Fehlerquote & Confidence
    - Kosteneinsparung (€ & Zeit)
    - Export für DATEV

Kosten:
  - Azure Document Intelligence: €280/Mt (1.200 Docs)
  - Storage (Langzeitarchiv): €25/Mt
  - Custom Development: €18.000 (einmalig)
  - **Total Jahr 1:** €21.660

# Production-Ready Invoice Processing

from azure.ai.formrecognizer import DocumentAnalysisClient
from azure.core.credentials import AzureKeyCredential
import requests

class IntelligentInvoiceProcessor:
    def __init__(self, config):
        # Azure Document Intelligence
        self.doc_client = DocumentAnalysisClient(
            endpoint=config['azure_endpoint'],
            credential=AzureKeyCredential(config['azure_key'])
        )
        
        # ERP Connector (DATEV)
        self.erp = DATEVConnector(config['datev_api'])
        
        # Storage (für GoBD-Archiv)
        self.archive = DocumentArchive(config['storage'])
        
    def process_invoice(self, invoice_pdf: bytes, filename: str):
        """
        Vollautomatische Rechnungsverarbeitung
        """
        # 1. OCR + Strukturierte Extraktion
        poller = self.doc_client.begin_analyze_document(
            "prebuilt-invoice", 
            invoice_pdf
        )
        result = poller.result()
        
        # 2. Daten extrahieren
        invoice_data = self._extract_invoice_fields(result)
        
        # 3. Validierung & Dublettencheck
        validation = self._validate_invoice(invoice_data)
        
        if not validation['is_valid']:
            return self._route_to_manual_review(invoice_data, validation['issues'])
        
        # 4. Kreditor-Matching (Stammdaten)
        vendor = self._match_vendor(invoice_data['vendor_name'], invoice_data.get('vendor_vat'))
        
        if not vendor:
            return self._create_new_vendor_workflow(invoice_data)
        
        # 5. ERP-Buchung erstellen
        booking_entry = {
            'vendor_id': vendor['datev_id'],
            'invoice_number': invoice_data['invoice_number'],
            'invoice_date': invoice_data['invoice_date'],
            'due_date': invoice_data['due_date'],
            'net_amount': invoice_data['net_amount'],
            'vat_amount': invoice_data['vat_amount'],
            'total_amount': invoice_data['total_amount'],
            'currency': 'EUR',
            'line_items': invoice_data['line_items'],
            'attachment': invoice_pdf
        }
        
        # 6. Confidence-basierte Auto-Buchung
        if validation['confidence'] >= 0.90:
            # Automatische Buchung
            erp_result = self.erp.create_invoice_entry(booking_entry)
            status = 'auto_booked'
        else:
            # Zur Prüfung vormerken
            erp_result = self.erp.create_draft_entry(booking_entry)
            status = 'pending_review'
        
        # 7. GoBD-konformes Archiv
        archive_id = self.archive.store_document(
            document=invoice_pdf,
            metadata=invoice_data,
            retention_years=10  # Gesetzliche Aufbewahrungspflicht
        )
        
        # 8. Audit-Log
        self._log_processing({
            'filename': filename,
            'status': status,
            'confidence': validation['confidence'],
            'erp_reference': erp_result.get('booking_id'),
            'archive_id': archive_id,
            'processing_time_ms': (datetime.now() - start_time).total_seconds() * 1000
        })
        
        return {
            'status': status,
            'invoice_data': invoice_data,
            'erp_booking_id': erp_result.get('booking_id'),
            'confidence': validation['confidence'],
            'processing_time': f"{(datetime.now() - start_time).total_seconds():.1f}s"
        }
    
    def _extract_invoice_fields(self, azure_result):
        """
        Strukturierte Datenextraktion aus Azure-Ergebnis
        """
        doc = azure_result.documents[0]
        fields = doc.fields
        
        return {
            'vendor_name': fields.get('VendorName', {}).value,
            'vendor_address': fields.get('VendorAddress', {}).value,
            'vendor_vat': fields.get('VendorTaxId', {}).value,
            'invoice_number': fields.get('InvoiceId', {}).value,
            'invoice_date': fields.get('InvoiceDate', {}).value,
            'due_date': fields.get('DueDate', {}).value,
            'net_amount': fields.get('SubTotal', {}).value,
            'vat_amount': fields.get('TotalTax', {}).value,
            'total_amount': fields.get('InvoiceTotal', {}).value,
            'currency': fields.get('CurrencyCode', {}).value or 'EUR',
            'line_items': self._extract_line_items(fields.get('Items', {}).value or []),
            'payment_terms': fields.get('PaymentTerm', {}).value,
            'iban': self._extract_iban(fields.get('BillingAddress', {}).value)
        }
    
    def _validate_invoice(self, invoice_data):
        """
        Intelligente Validierung
        """
        issues = []
        confidence = 1.0
        
        # 1. Pflichtfelder prüfen
        required_fields = ['vendor_name', 'invoice_number', 'invoice_date', 'total_amount']
        for field in required_fields:
            if not invoice_data.get(field):
                issues.append(f"Fehlendes Pflichtfeld: {field}")
                confidence -= 0.3
        
        # 2. Dublettencheck
        duplicate = self.erp.check_duplicate_invoice(
            invoice_data['vendor_name'], 
            invoice_data['invoice_number']
        )
        if duplicate:
            issues.append(f"Duplikat! Bereits gebucht: {duplicate['booking_date']}")
            confidence = 0.0
        
        # 3. Plausibilitätsprüfung
        if invoice_data.get('net_amount') and invoice_data.get('vat_amount'):
            calculated_vat = invoice_data['net_amount'] * 0.19
            actual_vat = invoice_data['vat_amount']
            
            if abs(calculated_vat - actual_vat) > 1.0:  # Toleranz €1
                issues.append(f"USt-Abweichung: {actual_vat:.2f}€ (erwartet {calculated_vat:.2f}€)")
                confidence -= 0.15
        
        # 4. Datumslogik
        if invoice_data.get('invoice_date') and invoice_data.get('due_date'):
            if invoice_data['due_date'] < invoice_data['invoice_date']:
                issues.append("Fälligkeitsdatum vor Rechnungsdatum")
                confidence -= 0.1
        
        return {
            'is_valid': confidence >= 0.70,
            'confidence': confidence,
            'issues': issues
        }

# Deployment: Batch-Processing
processor = IntelligentInvoiceProcessor(config)

invoices_folder = Path("/mnt/scanner-input/")
for invoice_file in invoices_folder.glob("*.pdf"):
    result = processor.process_invoice(invoice_file.read_bytes(), invoice_file.name)
    
    if result['status'] == 'auto_booked':
        print(f"✅ {invoice_file.name}: Auto-gebucht in {result['processing_time']}")
        invoice_file.rename(f"/mnt/processed/{invoice_file.name}")
    else:
        print(f"⚠️ {invoice_file.name}: Prüfung erforderlich (Confidence: {result['confidence']:.0%})")
        invoice_file.rename(f"/mnt/review/{invoice_file.name}")

# Contract Analysis AI

from azure.ai.language.questionanswering import QuestionAnsweringClient
import spacy

class ContractAnalysisAI:
    def __init__(self, config):
        self.nlp = spacy.load("de_core_news_lg")  # Deutsche NLP
        self.qa_client = QuestionAnsweringClient(config['azure_endpoint'], AzureKeyCredential(config['key']))
        
    def analyze_contract(self, contract_pdf: bytes):
        """
        KI-gestützte Vertragsanalyse
        """
        # 1. PDF → Text
        contract_text = self._extract_text_from_pdf(contract_pdf)
        
        # 2. Strukturanalyse
        structure = self._analyze_structure(contract_text)
        
        # 3. Klausel-Erkennung (NER)
        clauses = self._extract_clauses(contract_text)
        
        # 4. Risikobewertung
        risks = self._assess_risks(clauses)
        
        # 5. Compliance-Check
        compliance = self._check_compliance(clauses)
        
        # 6. Q&A für spezifische Fragen
        key_facts = {
            'Vertragspartner': self._extract_answer(contract_text, "Wer sind die Vertragsparteien?"),
            'Laufzeit': self._extract_answer(contract_text, "Wie lange läuft der Vertrag?"),
            'Kündigungsfrist': self._extract_answer(contract_text, "Welche Kündigungsfrist gilt?"),
            'Haftung': self._extract_answer(contract_text, "Welche Haftungsbeschränkungen gelten?"),
            'Zahlungsbedingungen': self._extract_answer(contract_text, "Welche Zahlungsbedingungen sind vereinbart?")
        }
        
        return {
            'contract_type': structure['type'],
            'parties': structure['parties'],
            'key_facts': key_facts,
            'risk_score': risks['overall_score'],
            'risk_details': risks['details'],
            'compliance_status': compliance['status'],
            'requires_legal_review': risks['overall_score'] > 0.6 or not compliance['compliant'],
            'processing_time': '12 min (vorher: 2,5h)'
        }

# Case: Dienstleistervertrag-Prüfung
result = contract_ai.analyze_contract(contract_pdf)

print(f"Vertragstyp: {result['contract_type']}")
print(f"Risiko-Score: {result['risk_score']:.0%}")
print(f"Legal Review nötig: {'Ja' if result['requires_legal_review'] else 'Nein'}")

Privacy_by_Design:
  Datenminimierung:
    - Nur notwendige Felder extrahieren
    - PII-Erkennung & Pseudonymisierung
    - Automatische Löschung nach Aufbewahrungszeit
  
  Verschlüsselung:
    - In-Transit: TLS 1.3
    - At-Rest: AES-256
    - Backup: Verschlüsselt (EU-Region)
  
  Betroffenenrechte:
    - Auskunft: Export aller verarbeiteten Docs
    - Löschung: Sofortige Entfernung aus Archiv
    - Portabilität: JSON/CSV Export

EU_Datenresidenz:
  Azure_Region: "West Europe (NL) oder Germany West Central"
  Storage: "Azure Storage EU (GRS)"
  No_Data_Transfer_Outside_EU: true