business-card-scanner/app/extract.py

"""Kontaktdaten aus einem Kartenbild lesen.

Das Bild geht direkt an das Vision-Modell: Layout, Schriftgroesse und
Gruppierung sind genau die Information, die die Feldzuordnung moeglich macht
- ein vorgeschalteter OCR-Schritt wuerde sie wegwerfen.
"""
import asyncio
import base64
import json
import logging

import anthropic

from . import config, imaging

log = logging.getLogger(__name__)

FIELDS = [
    "full_name", "first_name", "last_name", "job_title", "company", "department",
    "email", "phone", "mobile", "fax", "website",
    "street", "postal_code", "city", "country",
]

SCHEMA = {
    "type": "object",
    "properties": {
        "is_business_card": {
            "type": "boolean",
            "description": "False, wenn der Ausschnitt keine Visitenkarte zeigt.",
        },
        **{name: {"type": ["string", "null"]} for name in FIELDS},
    },
    "required": ["is_business_card", *FIELDS],
    "additionalProperties": False,
}

SYSTEM = """Du liest Visitenkarten aus und gibst die Kontaktdaten strukturiert zurueck.

Regeln:
- Uebernimm nur, was auf der Karte steht. Nichts ergaenzen, nichts raten.
- Fehlt ein Feld auf der Karte, gib null zurueck.
- Telefonnummern im internationalen Format, wenn die Vorwahl erkennbar ist
  (z. B. +49 511 123456), sonst unveraendert.
- phone ist die Festnetznummer, mobile die Mobilnummer. Ist nur eine Nummer
  ohne Kennzeichnung angegeben, gehoert sie in phone.
- company ist der Firmenname, job_title die Funktion der Person.
- full_name ist der vollstaendige Name inklusive Titeln wie "Dr.",
  first_name und last_name ohne Titel.
- website ohne Protokollpraefix.
- Zeigt der Ausschnitt keine Visitenkarte (Tischkante, Notizzettel, leere
  Flaeche), setze is_business_card auf false und alle Felder auf null."""

PROMPT = "Lies diese Visitenkarte aus."

# Nicht jedes Modell nimmt jeden Parameter an: die kleinen Modelle kennen
# output_config.effort nicht (400), der server-seitige Fallback gilt nur fuer
# die groessten.
_NO_EFFORT = ("claude-haiku-", "claude-sonnet-4-5")
_WITH_FALLBACK = ("claude-opus-5", "claude-fable-")

_client = None


def _output_config(with_effort: bool) -> dict:
    settings = {"format": {"type": "json_schema", "schema": SCHEMA}}
    if with_effort:
        settings["effort"] = "low"  # reines Ablesen, keine Denkarbeit
    return settings


def available() -> bool:
    return bool(config.ANTHROPIC_API_KEY)


def _get_client() -> anthropic.AsyncAnthropic:
    global _client
    if _client is None:
        _client = anthropic.AsyncAnthropic(api_key=config.ANTHROPIC_API_KEY)
    return _client


def _empty(reason: str) -> dict:
    return {"is_business_card": False, "_error": reason, **{f: None for f in FIELDS}}


def _has_content(result: dict) -> bool:
    return any(result.get(field) for field in FIELDS)


async def _ask(image_jpeg: bytes) -> dict:
    client = _get_client()
    messages = [
        {
            "role": "user",
            "content": [
                {
                    "type": "image",
                    "source": {
                        "type": "base64",
                        "media_type": "image/jpeg",
                        "data": base64.standard_b64encode(image_jpeg).decode(),
                    },
                },
                {"type": "text", "text": PROMPT},
            ],
        }
    ]
    model = config.ANTHROPIC_MODEL
    common = dict(model=model, max_tokens=2000, system=SYSTEM, messages=messages)
    effort = not model.startswith(_NO_EFFORT)
    try:
        if model.startswith(_WITH_FALLBACK):
            response = await client.beta.messages.create(
                betas=["server-side-fallback-2026-07-01"],
                fallbacks="default",
                output_config=_output_config(effort),
                **common,
            )
        else:
            response = await client.messages.create(
                output_config=_output_config(effort), **common
            )
    except anthropic.BadRequestError as exc:
        # Ein Modell mit engerem Parametersatz, als die Listen oben kennen.
        # Einmal ohne Zusatzparameter nachfassen, statt die Karte zu verlieren.
        log.warning("Anfrage abgelehnt (%s) - Wiederholung ohne Zusatzparameter", exc)
        response = await client.messages.create(
            output_config=_output_config(with_effort=False), **common
        )

    if response.stop_reason == "refusal":
        return _empty("Vom Modell abgelehnt")

    text = next((b.text for b in response.content if b.type == "text"), None)
    if not text:
        return _empty("Leere Antwort")
    return json.loads(text)


async def extract(image_jpeg: bytes) -> dict:
    """Ein Kartenbild auslesen.

    Bleibt das Ergebnis leer, wird einmal um 180 Grad gedreht nachgefasst:
    wie herum die Karte auf dem Tisch lag, ist geometrisch nicht bestimmbar.
    Die gedrehte Fassung entsteht erst hier, nicht auf Vorrat fuer jede Karte.
    """
    if not available():
        return _empty("Kein ANTHROPIC_API_KEY gesetzt")
    try:
        result = await _ask(image_jpeg)
    except Exception as exc:  # Netzfehler, Rate Limit, ungueltiges JSON
        log.exception("Extraktion fehlgeschlagen")
        return _empty(f"{type(exc).__name__}: {exc}")

    if not _has_content(result):
        try:
            rotated = await asyncio.to_thread(imaging.rotate_jpeg_180, image_jpeg)
            retry = await _ask(rotated)
            if _has_content(retry):
                retry["_rotated"] = True
                return retry
        except Exception:
            log.exception("Zweiter Versuch (gedreht) fehlgeschlagen")
    return result


async def extract_all(images: list) -> list:
    """Alle Karten eines Fotos parallel auslesen."""
    limit = asyncio.Semaphore(max(1, config.EXTRACT_CONCURRENCY))

    async def one(image_jpeg: bytes) -> dict:
        async with limit:
            return await extract(image_jpeg)

    return await asyncio.gather(*(one(image) for image in images))