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 = [
    "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.",
        },
        "rotation": {
            "type": "integer",
            "enum": [0, 90, 180, 270],
            "description": (
                "Grad im Uhrzeigersinn, um die das Bild gedreht werden muss, "
                "damit die Schrift aufrecht steht."
            ),
        },
        **{name: {"type": ["string", "null"]} for name in FIELDS},
    },
    "required": ["is_business_card", "rotation", *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.
- first_name ist der Vorname einschliesslich akademischer Titel ("Dr. Anna"),
  last_name der Nachname.
- website ohne Protokollpraefix.
- Zeigt der Ausschnitt keine Visitenkarte (Tischkante, Notizzettel, leere
  Flaeche), setze is_business_card auf false und alle Felder auf null.
- rotation sagt, um wie viel Grad im Uhrzeigersinn das Bild gedreht werden
  muss, damit die Schrift aufrecht steht. Steht sie schon richtig, ist der
  Wert 0. Steht die Karte auf dem Kopf, ist er 180."""

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, "rotation": 0, "_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, image_jpeg, 180)
            retry = await _ask(rotated)
            if _has_content(retry):
                # Der zweite Anlauf sah das Bild gedreht - dann stimmt auch
                # seine Rotationsangabe relativ zum gedrehten Bild.
                retry["rotation"] = (int(retry.get("rotation") or 0) + 180) % 360
                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))