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

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."

_client = None


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},
            ],
        }
    ]
    common = dict(
        model=config.ANTHROPIC_MODEL,
        max_tokens=2000,
        system=SYSTEM,
        messages=messages,
        output_config={
            "format": {"type": "json_schema", "schema": SCHEMA},
            "effort": "low",  # reines Ablesen, keine Denkarbeit
        },
    )
    try:
        response = await client.beta.messages.create(
            betas=["server-side-fallback-2026-07-01"],
            fallbacks="default",
            **common,
        )
    except anthropic.BadRequestError:
        # Der Fallback-Parameter haengt am Modell. Ohne ihn laeuft die Anfrage
        # weiter, sie bricht bei einer Ablehnung nur ohne Ersatzmodell ab.
        log.warning("Server-side fallback abgelehnt, Anfrage ohne Fallback")
        response = await client.messages.create(**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, rotated_jpeg: bytes = None) -> 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.
    """
    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) and rotated_jpeg is not None:
        try:
            retry = await _ask(rotated_jpeg)
            if _has_content(retry):
                retry["_rotated"] = True
                return retry
        except Exception:
            log.exception("Zweiter Versuch (gedreht) fehlgeschlagen")
    return result


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

    async def one(upright: bytes, rotated: bytes) -> dict:
        async with limit:
            return await extract(upright, rotated)

    return await asyncio.gather(*(one(u, r) for u, r in pairs))