# Beispiel: E-Commerce Produktbilder analysieren
from google.cloud import vision

def categorize_product_image(image_path):
    client = vision.ImageAnnotatorClient()

    with open(image_path, 'rb') as image_file:
        content = image_file.read()

    image = vision.Image(content=content)
    response = client.label_detection(image=image)
    labels = response.label_annotations

    categories = []
    for label in labels:
        if label.score > 0.7:  # Nur sichere Erkennungen
            categories.append({
                'name': label.description,
                'confidence': label.score
            })

    return categories

# Ergebnis: ['Kleidung', 'T-Shirt', 'Mode', 'Baumwolle']

def analyze_damage_claim(image_path):
    client = vision.ImageAnnotatorClient()

    # Bild laden und analysieren
    with open(image_path, 'rb') as image_file:
        content = image_file.read()

    image = vision.Image(content=content)

    # Objekterkennung für Fahrzeugschäden
    objects = client.object_localization(image=image).localized_object_annotations

    # Text extraction für Kennzeichen/Dokumente
    texts = client.text_detection(image=image).text_annotations

    damage_assessment = {
        'vehicle_parts': [],
        'damage_type': [],
        'extracted_text': []
    }

    for obj in objects:
        if obj.score > 0.6:
            damage_assessment['vehicle_parts'].append({
                'part': obj.name,
                'confidence': obj.score,
                'location': obj.bounding_poly
            })

    return damage_assessment

def process_medical_document(image_path):
    client = vision.ImageAnnotatorClient()

    with open(image_path, 'rb') as image_file:
        content = image_file.read()

    image = vision.Image(content=content)

    # Document Text Detection für strukturierte Dokumente
    response = client.document_text_detection(image=image)
    document = response.full_text_annotation

    # Medizinische Begriffe extrahieren
    medical_terms = []
    for page in document.pages:
        for block in page.blocks:
            for paragraph in block.paragraphs:
                for word in paragraph.words:
                    word_text = ''.join([
                        symbol.text for symbol in word.symbols
                    ])
                    # Medizinische Fachbegriffe identifizieren
                    if is_medical_term(word_text):
                        medical_terms.append(word_text)

    return {
        'full_text': document.text,
        'medical_terms': medical_terms,
        'confidence': response.text_annotations[0].confidence if response.text_annotations else 0
    }

def quality_control_check(product_image_path, reference_image_path):
    client = vision.ImageAnnotatorClient()

    # Produktbild analysieren
    with open(product_image_path, 'rb') as image_file:
        content = image_file.read()

    image = vision.Image(content=content)

    # Objekterkennung für Defekte
    response = client.object_localization(image=image)
    objects = response.localized_object_annotations

    # Label detection für Qualitätsmerkmale
    labels_response = client.label_detection(image=image)
    labels = labels_response.label_annotations

    quality_score = 100
    defects = []

    for label in labels:
        if any(defect in label.description.lower() for defect in ['crack', 'scratch', 'damage', 'defect']):
            quality_score -= (label.score * 30)
            defects.append({
                'type': label.description,
                'severity': label.score
            })

    return {
        'quality_score': max(0, quality_score),
        'defects': defects,
        'passed': quality_score >= 85
    }

# Google Cloud SDK installieren
curl https://sdk.cloud.google.com | bash
exec -l $SHELL

# Neues Projekt erstellen
gcloud projects create my-vision-project --name="Vision API Project"
gcloud config set project my-vision-project

# Vision API aktivieren
gcloud services enable vision.googleapis.com

# Service Account erstellen
gcloud iam service-accounts create vision-service-account \
    --description="Service account for Vision API" \
    --display-name="Vision Service Account"

# Schlüssel generieren
gcloud iam service-accounts keys create ~/vision-key.json \
    --iam-account=vision-service-account@my-vision-project.iam.gserviceaccount.com

# Virtual Environment erstellen
python3 -m venv vision-env
source vision-env/bin/activate  # Linux/Mac
# vision-env\Scripts\activate   # Windows

# Google Cloud Vision installieren
pip install google-cloud-vision
pip install Pillow  # Für Bildverarbeitung
pip install pandas  # Für Datenanalyse

import os
from google.cloud import vision

# Service Account Key setzen
os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = '/path/to/vision-key.json'

# Client initialisieren
client = vision.ImageAnnotatorClient()
print("Google Cloud Vision API erfolgreich verbunden!")

import io
import os
from google.cloud import vision
from google.cloud.vision_v1 import types
import pandas as pd

class GermanDocumentOCR:
    def __init__(self):
        self.client = vision.ImageAnnotatorClient()

    def extract_text_from_image(self, image_path, language_hints=['de', 'en']):
        """
        Extrahiert Text aus einem Bild mit deutscher Sprachoptimierung
        """
        with io.open(image_path, 'rb') as image_file:
            content = image_file.read()

        image = vision.Image(content=content)

        # Image context für bessere deutsche Texterkennung
        image_context = vision.ImageContext(language_hints=language_hints)

        # Document text detection für strukturierte Dokumente
        response = self.client.document_text_detection(
            image=image,
            image_context=image_context
        )

        if response.error.message:
            raise Exception(f'Fehler bei der Texterkennung: {response.error.message}')

        document = response.full_text_annotation

        # Strukturierte Datenextraktion
        extracted_data = {
            'full_text': document.text,
            'pages': len(document.pages),
            'confidence': self._calculate_confidence(document),
            'word_count': len(document.text.split()),
            'blocks': []
        }

        # Block-weise Analyse für bessere Strukturierung
        for page in document.pages:
            for block in page.blocks:
                block_text = self._extract_block_text(block)
                if block_text.strip():
                    extracted_data['blocks'].append({
                        'text': block_text,
                        'confidence': block.confidence,
                        'bbox': self._get_bounding_box(block.bounding_box)
                    })

        return extracted_data

    def _calculate_confidence(self, document):
        """Berechnet die durchschnittliche Erkennungsgenauigkeit"""
        total_confidence = 0
        word_count = 0

        for page in document.pages:
            for block in page.blocks:
                for paragraph in block.paragraphs:
                    for word in paragraph.words:
                        total_confidence += word.confidence
                        word_count += 1

        return total_confidence / word_count if word_count > 0 else 0

    def _extract_block_text(self, block):
        """Extrahiert Text aus einem Block"""
        text = ""
        for paragraph in block.paragraphs:
            for word in paragraph.words:
                word_text = ''.join([symbol.text for symbol in word.symbols])
                text += word_text + " "
            text += "\n"
        return text

    def _get_bounding_box(self, bounding_box):
        """Konvertiert Bounding Box zu Koordinaten"""
        return {
            'x1': bounding_box.vertices[0].x,
            'y1': bounding_box.vertices[0].y,
            'x2': bounding_box.vertices[2].x,
            'y2': bounding_box.vertices[2].y
        }

    def process_multiple_documents(self, image_folder, output_csv=None):
        """Verarbeitet mehrere Dokumente in einem Ordner"""
        results = []

        for filename in os.listdir(image_folder):
            if filename.lower().endswith(('.png', '.jpg', '.jpeg', '.pdf', '.tiff')):
                try:
                    file_path = os.path.join(image_folder, filename)
                    result = self.extract_text_from_image(file_path)
                    result['filename'] = filename
                    results.append(result)
                    print(f"✓ {filename} erfolgreich verarbeitet")
                except Exception as e:
                    print(f"✗ Fehler bei {filename}: {str(e)}")

        # Optional: Ergebnisse als CSV speichern
        if output_csv:
            df = pd.DataFrame([{
                'filename': r['filename'],
                'word_count': r['word_count'],
                'confidence': r['confidence'],
                'pages': r['pages'],
                'preview': r['full_text'][:100] + '...' if len(r['full_text']) > 100 else r['full_text']
            } for r in results])
            df.to_csv(output_csv, index=False)
            print(f"Ergebnisse gespeichert in: {output_csv}")

        return results

# Verwendung des OCR-Tools
if __name__ == "__main__":
    ocr = GermanDocumentOCR()

    # Einzelnes Dokument verarbeiten
    result = ocr.extract_text_from_image("rechnung.jpg")
    print(f"Erkannter Text (Konfidenz: {result['confidence']:.2f}):")
    print(result['full_text'])

    # Mehrere Dokumente verarbeiten
    results = ocr.process_multiple_documents(
        image_folder="./documents",
        output_csv="ocr_results.csv"
    )

500 Rechnungen × 5 Minuten = 2.500 Minuten
2.500 Minuten ÷ 60 = 41,7 Stunden
41,7 Stunden × 25€/Stunde = 1.042€/Monat

500 OCR-Requests = 0,75€ (unter Frei-Kontingent größtenteils kostenlos)
Setup und Wartung = 200€/Monat
Gesamt = 200,75€/Monat

def batch_process_images(image_list, batch_size=16):
    """Verarbeitet Bilder in Batches für bessere Effizienz"""
    client = vision.ImageAnnotatorClient()
    results = []

    for i in range(0, len(image_list), batch_size):
        batch = image_list[i:i + batch_size]
        requests = []

        for image_path in batch:
            with open(image_path, 'rb') as image_file:
                content = image_file.read()

            image = vision.Image(content=content)
            requests.append({
                'image': image,
                'features': [{'type_': vision.Feature.Type.TEXT_DETECTION}]
            })

        # Batch-Request senden
        response = client.batch_annotate_images(requests=requests)
        results.extend(response.responses)

    return results

# Nur benötigte Features verwenden
features = [
    {'type_': vision.Feature.Type.TEXT_DETECTION, 'max_results': 10},
    # {'type_': vision.Feature.Type.FACE_DETECTION},  # Nur wenn nötig
]

from PIL import Image
import io

def optimize_image_for_api(image_path, max_size=(1024, 1024), quality=85):
    """Optimiert Bilder für API-Verwendung"""
    with Image.open(image_path) as img:
        # Größe reduzieren
        img.thumbnail(max_size, Image.Resampling.LANCZOS)

        # Als JPEG mit reduzierter Qualität speichern
        output_buffer = io.BytesIO()
        img.save(output_buffer, format='JPEG', quality=quality, optimize=True)

        return output_buffer.getvalue()

✅ Einverständniserklärung der betroffenen Personen einholen
✅ Datenschutzerklärung um Vision API Nutzung erweitern
✅ Data Processing Agreement (DPA) mit Google abschließen
✅ Zweckbindung: Nur für deklarierte Zwecke nutzen
✅ Datenminimierung: Nur notwendige Bilddaten übertragen
✅ Löschkonzept: Regelmäßige Löschung nicht mehr benötigter Daten
✅ Technische und organisatorische Maßnahmen (TOMs) implementieren

import hashlib
import json
from datetime import datetime, timedelta

class GDPRCompliantVisionAPI:
    def __init__(self):
        self.client = vision.ImageAnnotatorClient()
        self.processing_log = []

    def process_image_with_consent(self, image_path, consent_id, purpose, retention_days=30):
        """
        Verarbeitet Bild mit DSGVO-Dokumentation
        """
        # Bildinhalt hashen für Logging (keine Bilddaten speichern)
        image_hash = self._hash_image(image_path)

        # Verarbeitungslog erstellen
        log_entry = {
            'timestamp': datetime.now().isoformat(),
            'image_hash': image_hash,
            'consent_id': consent_id,
            'purpose': purpose,
            'retention_until': (datetime.now() + timedelta(days=retention_days)).isoformat(),
            'data_minimization': True,
            'gdpr_compliant': True
        }

        try:
            # Bildverarbeitung
            with open(image_path, 'rb') as image_file:
                content = image_file.read()

            image = vision.Image(content=content)

            # Nur notwendige Features verwenden (Datenminimierung)
            response = self.client.text_detection(image=image)

            # Erfolg loggen
            log_entry['status'] = 'success'
            log_entry['features_used'] = ['text_detection']

            # Sensible Daten anonymisieren
            result = self._anonymize_sensitive_data(response)

        except Exception as e:
            log_entry['status'] = 'error'
            log_entry['error'] = str(e)
            result = None

        # Log speichern
        self.processing_log.append(log_entry)
        self._save_processing_log()

        return result

    def _hash_image(self, image_path):
        """Erstellt Hash vom Bildinhalt für Logging"""
        hasher = hashlib.sha256()
        with open(image_path, 'rb') as f:
            for chunk in iter(lambda: f.read(4096), b""):
                hasher.update(chunk)
        return hasher.hexdigest()

    def _anonymize_sensitive_data(self, response):
        """Entfernt oder anonymisiert sensible Daten"""
        # Beispiel: Email-Adressen, Telefonnummern anonymisieren
        import re

        if response.text_annotations:
            text = response.text_annotations[0].description

            # Email-Adressen anonymisieren
            text = re.sub(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
                         '[EMAIL_ANONYMIZED]', text)

            # Telefonnummern anonymisieren
            text = re.sub(r'\b\d{3,4}[-.\s]?\d{3,4}[-.\s]?\d{3,6}\b',
                         '[PHONE_ANONYMIZED]', text)

            return {
                'text': text,
                'confidence': response.text_annotations[0].confidence if response.text_annotations else 0,
                'anonymized': True
            }

        return None

    def _save_processing_log(self):
        """Speichert Verarbeitungslog für Audits"""
        with open('gdpr_processing_log.json', 'w') as f:
            json.dump(self.processing_log, f, indent=2)

    def delete_expired_logs(self):
        """Löscht abgelaufene Logs gemäß Retention Policy"""
        current_time = datetime.now()
        self.processing_log = [
            log for log in self.processing_log
            if datetime.fromisoformat(log['retention_until']) > current_time
        ]
        self._save_processing_log()

import pytesseract
from PIL import Image

def tesseract_ocr_german(image_path):
    """Kostenlose OCR-Alternative mit Tesseract"""
    img = Image.open(image_path)

    # Deutsche Sprache konfigurieren
    custom_config = r'--oem 3 --psm 6 -l deu'
    text = pytesseract.image_to_string(img, config=custom_config)

    return text

# Vor- und Nachteile:
# ✅ Kostenlos und offline nutzbar
# ✅ DSGVO-konform (lokale Verarbeitung)
# ❌ Niedrigere Genauigkeit als Cloud-APIs
# ❌ Mehr Setup und Wartung erforderlich

from paddleocr import PaddleOCR

def paddle_ocr_german(image_path):
    """Alternative OCR mit PaddleOCR"""
    ocr = PaddleOCR(use_angle_cls=True, lang='german')
    result = ocr.ocr(image_path, cls=True)

    extracted_text = ""
    for idx in range(len(result)):
        res = result[idx]
        for line in res:
            extracted_text += line[1][0] + "\n"

    return extracted_text

import asyncio
from concurrent.futures import ThreadPoolExecutor
from google.cloud import vision

class AsyncVisionProcessor:
    def __init__(self, max_workers=10):
        self.client = vision.ImageAnnotatorClient()
        self.executor = ThreadPoolExecutor(max_workers=max_workers)

    async def process_images_async(self, image_paths):
        """Verarbeitet mehrere Bilder parallel"""
        loop = asyncio.get_event_loop()

        tasks = []
        for image_path in image_paths:
            task = loop.run_in_executor(
                self.executor,
                self._process_single_image,
                image_path
            )
            tasks.append(task)

        results = await asyncio.gather(*tasks, return_exceptions=True)
        return results

    def _process_single_image(self, image_path):
        """Verarbeitet ein einzelnes Bild"""
        with open(image_path, 'rb') as image_file:
            content = image_file.read()

        image = vision.Image(content=content)
        response = self.client.text_detection(image=image)

        return {
            'file': image_path,
            'text': response.text_annotations[0].description if response.text_annotations else "",
            'confidence': response.text_annotations[0].confidence if response.text_annotations else 0
        }

# Verwendung:
async def main():
    processor = AsyncVisionProcessor()
    image_list = ['doc1.jpg', 'doc2.jpg', 'doc3.jpg']
    results = await processor.process_images_async(image_list)

    for result in results:
        if isinstance(result, Exception):
            print(f"Fehler: {result}")
        else:
            print(f"✓ {result['file']}: {len(result['text'])} Zeichen erkannt")

# asyncio.run(main())

import hashlib
import json
from functools import wraps

def cache_vision_results(cache_duration_hours=24):
    """Decorator für Caching von Vision API Ergebnissen"""
    def decorator(func):
        @wraps(func)
        def wrapper(image_path, *args, **kwargs):
            # Cache-Key aus Bildhash erstellen
            cache_key = hashlib.md5(
                open(image_path, 'rb').read()
            ).hexdigest()

            cache_file = f"cache/{cache_key}.json"

            # Cache prüfen
            if os.path.exists(cache_file):
                with open(cache_file, 'r') as f:
                    cached_result = json.load(f)

                # Cache-Alter prüfen
                cache_time = datetime.fromisoformat(cached_result['timestamp'])
                if datetime.now() - cache_time < timedelta(hours=cache_duration_hours):
                    return cached_result['result']

            # API-Call durchführen
            result = func(image_path, *args, **kwargs)

            # Ergebnis cachen
            os.makedirs('cache', exist_ok=True)
            with open(cache_file, 'w') as f:
                json.dump({
                    'timestamp': datetime.now().isoformat(),
                    'result': result
                }, f)

            return result
        return wrapper
    return decorator

@cache_vision_results(cache_duration_hours=48)
def cached_text_detection(image_path):
    """Cached Text Detection"""
    client = vision.ImageAnnotatorClient()

    with open(image_path, 'rb') as image_file:
        content = image_file.read()

    image = vision.Image(content=content)
    response = client.text_detection(image=image)

    return {
        'text': response.text_annotations[0].description if response.text_annotations else "",
        'confidence': response.text_annotations[0].confidence if response.text_annotations else 0
    }

import logging
import time
from google.api_core import retry, exceptions

class RobustVisionAPI:
    def __init__(self):
        self.client = vision.ImageAnnotatorClient()
        self.setup_logging()

    def setup_logging(self):
        """Konfiguriert Logging für API-Aufrufe"""
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('vision_api.log'),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)

    @retry.Retry(
        predicate=retry.if_exception_type(exceptions.ServiceUnavailable),
        deadline=300.0
    )
    def robust_text_detection(self, image_path, max_retries=3):
        """Text Detection mit robuster Fehlerbehandlung"""
        for attempt in range(max_retries):
            try:
                self.logger.info(f"Verarbeite {image_path} (Versuch {attempt + 1})")

                with open(image_path, 'rb') as image_file:
                    content = image_file.read()

                # Bildgröße prüfen
                if len(content) > 20 * 1024 * 1024:  # 20MB Limit
                    raise ValueError("Bild zu groß (>20MB)")

                image = vision.Image(content=content)
                response = self.client.text_detection(image=image)

                # API-Fehler prüfen
                if response.error.message:
                    raise exceptions.GoogleAPIError(response.error.message)

                self.logger.info(f"✓ {image_path} erfolgreich verarbeitet")

                return {
                    'success': True,
                    'text': response.text_annotations[0].description if response.text_annotations else "",
                    'confidence': response.text_annotations[0].confidence if response.text_annotations else 0,
                    'attempt': attempt + 1
                }

            except exceptions.GoogleAPIError as e:
                self.logger.error(f"API-Fehler bei {image_path}: {e}")
                if attempt == max_retries - 1:
                    return {'success': False, 'error': str(e)}
                time.sleep(2 ** attempt)  # Exponential backoff

            except Exception as e:
                self.logger.error(f"Unerwarteter Fehler bei {image_path}: {e}")
                if attempt == max_retries - 1:
                    return {'success': False, 'error': str(e)}
                time.sleep(1)

        return {'success': False, 'error': 'Max retries exceeded'}

def improve_ocr_accuracy(image_path):
    """Verbessert OCR-Genauigkeit durch Bildvorverarbeitung"""
    from PIL import Image, ImageEnhance, ImageFilter

    # Bild laden und vorverarbeiten
    img = Image.open(image_path)

    # Grauwerte-Konvertierung
    if img.mode != 'L':
        img = img.convert('L')

    # Kontrast erhöhen
    enhancer = ImageEnhance.Contrast(img)
    img = enhancer.enhance(2.0)

    # Schärfe erhöhen
    img = img.filter(ImageFilter.SHARPEN)

    # Größe anpassen (optimal: 300-600 DPI)
    width, height = img.size
    if width < 1000:
        scale_factor = 1000 / width
        new_size = (int(width * scale_factor), int(height * scale_factor))
        img = img.resize(new_size, Image.Resampling.LANCZOS)

    # Temporäre Datei speichern und verarbeiten
    import tempfile
    with tempfile.NamedTemporaryFile(suffix='.png', delete=False) as tmp_file:
        img.save(tmp_file.name, 'PNG')

        # Vision API mit verbessertem Bild aufrufen
        client = vision.ImageAnnotatorClient()
        with open(tmp_file.name, 'rb') as image_file:
            content = image_file.read()

        image = vision.Image(content=content)
        response = client.document_text_detection(image=image)

        # Temporäre Datei löschen
        os.unlink(tmp_file.name)

        return response

class CostOptimizedVisionAPI:
    def __init__(self, monthly_budget=100):
        self.client = vision.ImageAnnotatorClient()
        self.monthly_budget = monthly_budget
        self.current_usage = 0
        self.requests_count = 0

    def budget_aware_request(self, image_path, features):
        """API-Request mit Budget-Überwachung"""
        # Geschätzte Kosten berechnen
        estimated_cost = len(features) * 0.0015  # ~1,50€ per 1000 requests

        if self.current_usage + estimated_cost > self.monthly_budget:
            return {
                'error': 'Monthly budget exceeded',
                'current_usage': self.current_usage,
                'budget': self.monthly_budget
            }

        # Request durchführen
        with open(image_path, 'rb') as image_file:
            content = image_file.read()

        image = vision.Image(content=content)

        # Nur erforderliche Features verwenden
        feature_list = [vision.Feature(type_=feature) for feature in features]
        request = vision.AnnotateImageRequest(image=image, features=feature_list)

        response = self.client.annotate_image(request=request)

        # Usage tracking
        self.current_usage += estimated_cost
        self.requests_count += 1

        return {
            'result': response,
            'cost': estimated_cost,
            'total_usage': self.current_usage,
            'requests_count': self.requests_count
        }