Visitenkarten-Scanner: Stapelscan, Extraktion, Übersicht, vCard-Export
Selbst gehostete PWA, die Visitenkarten von einem Foto freistellt, ausliest und als Kontakt bereitstellt. - Stapelscan: OpenCV findet die Kartenrechtecke über mehrere Binärmasken, entzerrt sie perspektivisch und schneidet sie einzeln aus. Ohne Fund gilt das ganze Foto als eine Karte. - Extraktion: ein Aufruf je Zuschnitt an das Vision-Modell mit JSON-Schema. Kein vorgeschaltetes OCR - das würde Layout und Schriftgrößen wegwerfen, aus denen die Feldzuordnung entsteht. - Metadaten: Aufnahmezeit und GPS aus den EXIF-Daten des Fotos, Ortsname über Nominatim, Browserstandort nur als Rückfallebene. - Übersicht mit Volltextsuche und Filtern, Detailansicht mit Korrekturmaske. - Notizfeld je Karte, Erinnerungen per Mail inklusive Nachholen verpasster Termine nach einem Neustart. - vCard 3.0 einzeln und als Sammeldatei, Karte gilt danach als exportiert. - Anmeldung über ein Passwort, Sitzung als signiertes Cookie. - Deployment per Tag-Push nach den Konventionen in DEPLOY.md. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Lucas Orth
0
app/__init__.py
Normal file
0
app/__init__.py
Normal file
55
app/config.py
Normal file
55
app/config.py
Normal file
@@ -0,0 +1,55 @@
|
||||
"""Konfiguration. Alle Werte kommen aus der Umgebung (im Betrieb aus .env)."""
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def _s(name: str, default: str = "") -> str:
|
||||
return os.environ.get(name, default).strip()
|
||||
|
||||
|
||||
def _i(name: str, default: int) -> int:
|
||||
try:
|
||||
return int(_s(name) or default)
|
||||
except ValueError:
|
||||
return default
|
||||
|
||||
|
||||
def _b(name: str, default: bool) -> bool:
|
||||
v = _s(name).lower()
|
||||
if not v:
|
||||
return default
|
||||
return v in ("1", "true", "yes", "on")
|
||||
|
||||
|
||||
DATA_DIR = Path(_s("DATA_DIR", "./data"))
|
||||
IMAGE_DIR = DATA_DIR / "images"
|
||||
DB_PATH = DATA_DIR / "cards.db"
|
||||
|
||||
PORT = _i("PORT", 8080)
|
||||
|
||||
# Leeres Passwort schaltet die Anmeldung ab - nur fuer die lokale Entwicklung.
|
||||
APP_PASSWORD = _s("APP_PASSWORD")
|
||||
SECRET_KEY = _s("SECRET_KEY", "dev-only-insecure-secret")
|
||||
SESSION_DAYS = _i("SESSION_DAYS", 30)
|
||||
|
||||
ANTHROPIC_API_KEY = _s("ANTHROPIC_API_KEY")
|
||||
ANTHROPIC_MODEL = _s("ANTHROPIC_MODEL", "claude-opus-5")
|
||||
EXTRACT_CONCURRENCY = _i("EXTRACT_CONCURRENCY", 5)
|
||||
|
||||
SMTP_HOST = _s("SMTP_HOST")
|
||||
SMTP_PORT = _i("SMTP_PORT", 587)
|
||||
SMTP_USER = _s("SMTP_USER")
|
||||
SMTP_PASSWORD = _s("SMTP_PASSWORD")
|
||||
SMTP_FROM = _s("SMTP_FROM") or SMTP_USER
|
||||
SMTP_STARTTLS = _b("SMTP_STARTTLS", True)
|
||||
REMINDER_TO = _s("REMINDER_TO")
|
||||
|
||||
REVERSE_GEOCODE = _b("REVERSE_GEOCODE", True)
|
||||
NOMINATIM_URL = _s("NOMINATIM_URL", "https://nominatim.openstreetmap.org/reverse")
|
||||
USER_AGENT = _s("USER_AGENT", "business-card-scanner/1.0 (self-hosted)")
|
||||
|
||||
PUBLIC_BASE_URL = _s("PUBLIC_BASE_URL").rstrip("/")
|
||||
|
||||
|
||||
def ensure_dirs() -> None:
|
||||
IMAGE_DIR.mkdir(parents=True, exist_ok=True)
|
||||
100
app/db.py
Normal file
100
app/db.py
Normal file
@@ -0,0 +1,100 @@
|
||||
"""SQLite-Zugriff. Eine Verbindung, Schreibzugriffe unter einem Lock."""
|
||||
import sqlite3
|
||||
import threading
|
||||
from typing import Any, Iterable
|
||||
|
||||
from . import config
|
||||
|
||||
_lock = threading.Lock()
|
||||
_conn: sqlite3.Connection | None = None
|
||||
|
||||
SCHEMA = """
|
||||
CREATE TABLE IF NOT EXISTS cards (
|
||||
id TEXT PRIMARY KEY,
|
||||
created_at TEXT NOT NULL,
|
||||
captured_at TEXT,
|
||||
batch_id TEXT,
|
||||
image_file TEXT,
|
||||
source TEXT NOT NULL DEFAULT 'batch',
|
||||
full_name TEXT,
|
||||
first_name TEXT,
|
||||
last_name TEXT,
|
||||
job_title TEXT,
|
||||
company TEXT,
|
||||
department TEXT,
|
||||
email TEXT,
|
||||
phone TEXT,
|
||||
mobile TEXT,
|
||||
fax TEXT,
|
||||
website TEXT,
|
||||
street TEXT,
|
||||
postal_code TEXT,
|
||||
city TEXT,
|
||||
country TEXT,
|
||||
notes TEXT NOT NULL DEFAULT '',
|
||||
lat REAL,
|
||||
lon REAL,
|
||||
place TEXT,
|
||||
geo_source TEXT,
|
||||
extract_status TEXT NOT NULL DEFAULT 'ok',
|
||||
extract_error TEXT,
|
||||
exported_at TEXT
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS reminders (
|
||||
id TEXT PRIMARY KEY,
|
||||
card_id TEXT NOT NULL REFERENCES cards(id) ON DELETE CASCADE,
|
||||
text TEXT NOT NULL,
|
||||
due_at TEXT NOT NULL,
|
||||
created_at TEXT NOT NULL,
|
||||
sent_at TEXT,
|
||||
send_error TEXT
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_cards_created ON cards(created_at DESC);
|
||||
CREATE INDEX IF NOT EXISTS idx_reminders_due ON reminders(due_at) WHERE sent_at IS NULL;
|
||||
|
||||
CREATE TABLE IF NOT EXISTS geocache (
|
||||
key TEXT PRIMARY KEY,
|
||||
place TEXT,
|
||||
fetched_at TEXT NOT NULL
|
||||
);
|
||||
"""
|
||||
|
||||
|
||||
def connect() -> sqlite3.Connection:
|
||||
global _conn
|
||||
if _conn is None:
|
||||
config.ensure_dirs()
|
||||
_conn = sqlite3.connect(config.DB_PATH, check_same_thread=False)
|
||||
_conn.row_factory = sqlite3.Row
|
||||
_conn.execute("PRAGMA journal_mode=WAL")
|
||||
_conn.execute("PRAGMA foreign_keys=ON")
|
||||
_conn.executescript(SCHEMA)
|
||||
_conn.commit()
|
||||
return _conn
|
||||
|
||||
|
||||
def query(sql: str, params: Iterable[Any] = ()) -> list[sqlite3.Row]:
|
||||
with _lock:
|
||||
return connect().execute(sql, tuple(params)).fetchall()
|
||||
|
||||
|
||||
def query_one(sql: str, params: Iterable[Any] = ()) -> sqlite3.Row | None:
|
||||
rows = query(sql, params)
|
||||
return rows[0] if rows else None
|
||||
|
||||
|
||||
def execute(sql: str, params: Iterable[Any] = ()) -> None:
|
||||
with _lock:
|
||||
conn = connect()
|
||||
conn.execute(sql, tuple(params))
|
||||
conn.commit()
|
||||
|
||||
|
||||
def execute_many(statements: list[tuple[str, Iterable[Any]]]) -> None:
|
||||
with _lock:
|
||||
conn = connect()
|
||||
for sql, params in statements:
|
||||
conn.execute(sql, tuple(params))
|
||||
conn.commit()
|
||||
159
app/extract.py
Normal file
159
app/extract.py
Normal file
@@ -0,0 +1,159 @@
|
||||
"""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))
|
||||
52
app/geocode.py
Normal file
52
app/geocode.py
Normal file
@@ -0,0 +1,52 @@
|
||||
"""Reverse-Geocoding ueber Nominatim, Ergebnisse werden dauerhaft gecacht."""
|
||||
import json
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from . import config, db
|
||||
|
||||
|
||||
def _label(address: dict) -> str:
|
||||
city = (
|
||||
address.get("city")
|
||||
or address.get("town")
|
||||
or address.get("village")
|
||||
or address.get("municipality")
|
||||
or address.get("county")
|
||||
)
|
||||
detail = address.get("suburb") or address.get("neighbourhood") or address.get("road")
|
||||
parts = [p for p in (city, detail) if p]
|
||||
if not parts:
|
||||
return address.get("state") or address.get("country") or ""
|
||||
return ", ".join(parts[:2])
|
||||
|
||||
|
||||
def reverse(lat: float | None, lon: float | None) -> str | None:
|
||||
if lat is None or lon is None or not config.REVERSE_GEOCODE:
|
||||
return None
|
||||
|
||||
# Auf ~100 m runden: benachbarte Scans teilen sich denselben Cache-Eintrag.
|
||||
key = f"{lat:.3f},{lon:.3f}"
|
||||
cached = db.query_one("SELECT place FROM geocache WHERE key = ?", (key,))
|
||||
if cached is not None:
|
||||
return cached["place"]
|
||||
|
||||
url = config.NOMINATIM_URL + "?" + urllib.parse.urlencode(
|
||||
{"lat": lat, "lon": lon, "format": "jsonv2", "zoom": "16", "accept-language": "de"}
|
||||
)
|
||||
place = None
|
||||
try:
|
||||
request = urllib.request.Request(url, headers={"User-Agent": config.USER_AGENT})
|
||||
with urllib.request.urlopen(request, timeout=6) as response:
|
||||
data = json.loads(response.read().decode("utf-8"))
|
||||
place = _label(data.get("address") or {}) or data.get("display_name")
|
||||
except Exception:
|
||||
# Ohne Ortsnamen bleiben die Koordinaten - kein Grund, den Scan abzubrechen.
|
||||
return None
|
||||
|
||||
db.execute(
|
||||
"INSERT OR REPLACE INTO geocache (key, place, fetched_at) VALUES (?, ?, ?)",
|
||||
(key, place, datetime.now(timezone.utc).isoformat()),
|
||||
)
|
||||
return place
|
||||
91
app/imaging.py
Normal file
91
app/imaging.py
Normal file
@@ -0,0 +1,91 @@
|
||||
"""EXIF-Auswertung und Bild-Hilfsfunktionen."""
|
||||
import io
|
||||
from datetime import datetime, timezone
|
||||
|
||||
import numpy as np
|
||||
from PIL import Image, ImageOps
|
||||
|
||||
# EXIF-Tags, die uns interessieren (Nummern statt Namen: stabil ueber Pillow-Versionen).
|
||||
_TAG_DATETIME_ORIGINAL = 0x9003
|
||||
_TAG_OFFSET_ORIGINAL = 0x9011
|
||||
_TAG_GPS_IFD = 0x8825
|
||||
_GPS_LAT_REF, _GPS_LAT = 1, 2
|
||||
_GPS_LON_REF, _GPS_LON = 3, 4
|
||||
|
||||
|
||||
def _to_degrees(value) -> float | None:
|
||||
try:
|
||||
d, m, s = (float(x) for x in value)
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
return d + m / 60.0 + s / 3600.0
|
||||
|
||||
|
||||
def read_metadata(raw: bytes) -> dict:
|
||||
"""Aufnahmezeitpunkt und GPS aus den EXIF-Daten des Originalfotos.
|
||||
|
||||
Der Zeitpunkt des Fotos ist verlaesslicher als der des Uploads: Karten
|
||||
werden oft erst Stunden spaeter am Schreibtisch abfotografiert.
|
||||
"""
|
||||
out: dict = {"captured_at": None, "lat": None, "lon": None}
|
||||
try:
|
||||
exif = Image.open(io.BytesIO(raw)).getexif()
|
||||
except Exception:
|
||||
return out
|
||||
if not exif:
|
||||
return out
|
||||
|
||||
taken = exif.get(_TAG_DATETIME_ORIGINAL)
|
||||
if not taken:
|
||||
ifd = exif.get_ifd(0x8769) # Exif-IFD
|
||||
taken = ifd.get(_TAG_DATETIME_ORIGINAL) if ifd else None
|
||||
offset = ifd.get(_TAG_OFFSET_ORIGINAL) if ifd else None
|
||||
else:
|
||||
offset = exif.get(_TAG_OFFSET_ORIGINAL)
|
||||
if taken:
|
||||
try:
|
||||
dt = datetime.strptime(str(taken).strip(), "%Y:%m:%d %H:%M:%S")
|
||||
iso = f"{dt.isoformat()}{offset}" if offset else dt.isoformat()
|
||||
parsed = datetime.fromisoformat(iso)
|
||||
if parsed.tzinfo is None:
|
||||
parsed = parsed.replace(tzinfo=timezone.utc)
|
||||
out["captured_at"] = parsed.astimezone(timezone.utc).isoformat()
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
gps = exif.get_ifd(_TAG_GPS_IFD)
|
||||
if gps:
|
||||
lat = _to_degrees(gps.get(_GPS_LAT))
|
||||
lon = _to_degrees(gps.get(_GPS_LON))
|
||||
if lat is not None and lon is not None:
|
||||
if str(gps.get(_GPS_LAT_REF, "N")).upper().startswith("S"):
|
||||
lat = -lat
|
||||
if str(gps.get(_GPS_LON_REF, "E")).upper().startswith("W"):
|
||||
lon = -lon
|
||||
out["lat"], out["lon"] = round(lat, 6), round(lon, 6)
|
||||
return out
|
||||
|
||||
|
||||
def decode(raw: bytes) -> np.ndarray:
|
||||
"""Bytes zu einem BGR-Array, EXIF-Drehung bereits angewandt."""
|
||||
image = Image.open(io.BytesIO(raw))
|
||||
image = ImageOps.exif_transpose(image).convert("RGB")
|
||||
return np.asarray(image)[:, :, ::-1].copy()
|
||||
|
||||
|
||||
def encode_jpeg(rgb_or_bgr: np.ndarray, quality: int = 88, bgr: bool = True) -> bytes:
|
||||
array = rgb_or_bgr[:, :, ::-1] if bgr else rgb_or_bgr
|
||||
buffer = io.BytesIO()
|
||||
Image.fromarray(array).save(buffer, format="JPEG", quality=quality, optimize=True)
|
||||
return buffer.getvalue()
|
||||
|
||||
|
||||
def fit_within(image: np.ndarray, max_edge: int) -> np.ndarray:
|
||||
height, width = image.shape[:2]
|
||||
longest = max(height, width)
|
||||
if longest <= max_edge:
|
||||
return image
|
||||
scale = max_edge / longest
|
||||
pil = Image.fromarray(image[:, :, ::-1])
|
||||
pil = pil.resize((round(width * scale), round(height * scale)), Image.LANCZOS)
|
||||
return np.asarray(pil)[:, :, ::-1].copy()
|
||||
33
app/mailer.py
Normal file
33
app/mailer.py
Normal file
@@ -0,0 +1,33 @@
|
||||
"""Mailversand ueber das eigene Postfach (SMTP)."""
|
||||
import logging
|
||||
import smtplib
|
||||
from email.message import EmailMessage
|
||||
|
||||
from . import config
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def configured() -> bool:
|
||||
return bool(config.SMTP_HOST and config.SMTP_FROM and config.REMINDER_TO)
|
||||
|
||||
|
||||
def send(subject: str, body: str) -> None:
|
||||
"""Verschickt eine Mail. Wirft bei Fehlern - der Aufrufer entscheidet,
|
||||
ob nachher erneut versucht wird."""
|
||||
if not configured():
|
||||
raise RuntimeError("SMTP ist nicht konfiguriert (SMTP_HOST/SMTP_FROM/REMINDER_TO)")
|
||||
|
||||
message = EmailMessage()
|
||||
message["Subject"] = subject
|
||||
message["From"] = config.SMTP_FROM
|
||||
message["To"] = config.REMINDER_TO
|
||||
message.set_content(body)
|
||||
|
||||
with smtplib.SMTP(config.SMTP_HOST, config.SMTP_PORT, timeout=20) as server:
|
||||
if config.SMTP_STARTTLS:
|
||||
server.starttls()
|
||||
if config.SMTP_USER:
|
||||
server.login(config.SMTP_USER, config.SMTP_PASSWORD)
|
||||
server.send_message(message)
|
||||
log.info("Mail verschickt: %s", subject)
|
||||
394
app/main.py
Normal file
394
app/main.py
Normal file
@@ -0,0 +1,394 @@
|
||||
"""HTTP-Schnittstelle und Auslieferung der PWA."""
|
||||
import asyncio
|
||||
import logging
|
||||
import uuid
|
||||
from contextlib import asynccontextmanager
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
import cv2
|
||||
from fastapi import Depends, FastAPI, Form, HTTPException, Request, Response, UploadFile
|
||||
from fastapi.responses import FileResponse, JSONResponse, PlainTextResponse
|
||||
from fastapi.staticfiles import StaticFiles
|
||||
|
||||
from . import config, db, extract, geocode, imaging, reminders, security, segment, vcard
|
||||
|
||||
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s")
|
||||
log = logging.getLogger("bcs")
|
||||
|
||||
STATIC_DIR = Path(__file__).resolve().parent.parent / "static"
|
||||
MAX_UPLOAD_BYTES = 30 * 1024 * 1024
|
||||
|
||||
EDITABLE = [
|
||||
"full_name", "first_name", "last_name", "job_title", "company", "department",
|
||||
"email", "phone", "mobile", "fax", "website",
|
||||
"street", "postal_code", "city", "country", "notes",
|
||||
]
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def lifespan(app: FastAPI):
|
||||
config.ensure_dirs()
|
||||
db.connect()
|
||||
if not security.enabled():
|
||||
log.warning("APP_PASSWORD ist leer - die App ist ohne Anmeldung erreichbar.")
|
||||
if not extract.available():
|
||||
log.warning("ANTHROPIC_API_KEY fehlt - Karten werden nur zugeschnitten, nicht ausgelesen.")
|
||||
task = asyncio.create_task(reminders.run_forever())
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
task.cancel()
|
||||
|
||||
|
||||
app = FastAPI(title="Business Card Scanner", lifespan=lifespan, docs_url=None, redoc_url=None)
|
||||
auth = Depends(security.require_auth)
|
||||
|
||||
|
||||
def now_iso() -> str:
|
||||
return datetime.now(timezone.utc).isoformat()
|
||||
|
||||
|
||||
def card_dict(row) -> dict:
|
||||
card = dict(row)
|
||||
card["image_url"] = f"/api/cards/{card['id']}/image" if card.get("image_file") else None
|
||||
card["has_contact"] = any(
|
||||
card.get(f) for f in ("full_name", "company", "email", "phone", "mobile")
|
||||
)
|
||||
return card
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- Sitzung
|
||||
|
||||
@app.get("/healthz")
|
||||
def healthz() -> dict:
|
||||
db.query_one("SELECT 1 AS ok")
|
||||
return {"status": "ok"}
|
||||
|
||||
|
||||
@app.get("/api/session")
|
||||
def session(request: Request) -> dict:
|
||||
return {
|
||||
"auth_required": security.enabled(),
|
||||
"authenticated": not security.enabled()
|
||||
or security.valid_token(request.cookies.get(security.COOKIE)),
|
||||
"extraction_available": extract.available(),
|
||||
"mail_configured": bool(config.SMTP_HOST and config.REMINDER_TO),
|
||||
}
|
||||
|
||||
|
||||
@app.post("/api/login")
|
||||
def login(response: Response, password: str = Form(...)) -> dict:
|
||||
if not security.enabled():
|
||||
return {"ok": True}
|
||||
if not security.check_password(password):
|
||||
raise HTTPException(status_code=401, detail="Passwort stimmt nicht")
|
||||
response.set_cookie(
|
||||
security.COOKIE,
|
||||
security.issue_token(),
|
||||
max_age=config.SESSION_DAYS * 86400,
|
||||
httponly=True,
|
||||
samesite="lax",
|
||||
secure=True,
|
||||
path="/",
|
||||
)
|
||||
return {"ok": True}
|
||||
|
||||
|
||||
@app.post("/api/logout")
|
||||
def logout(response: Response) -> dict:
|
||||
response.delete_cookie(security.COOKIE, path="/")
|
||||
return {"ok": True}
|
||||
|
||||
|
||||
# ------------------------------------------------------------------------------ Scan
|
||||
|
||||
def _process_photo(raw: bytes) -> tuple:
|
||||
"""Rechenlastiger Teil: dekodieren, freistellen, als JPEG kodieren."""
|
||||
meta = imaging.read_metadata(raw)
|
||||
image = imaging.decode(raw)
|
||||
crops, fallback = segment.segment(image)
|
||||
encoded = [
|
||||
(
|
||||
imaging.encode_jpeg(crop),
|
||||
imaging.encode_jpeg(cv2.rotate(crop, cv2.ROTATE_180)),
|
||||
)
|
||||
for crop in crops
|
||||
]
|
||||
return meta, encoded, fallback
|
||||
|
||||
|
||||
@app.post("/api/scan", dependencies=[auth])
|
||||
async def scan(
|
||||
file: UploadFile,
|
||||
lat: float = Form(None),
|
||||
lon: float = Form(None),
|
||||
) -> JSONResponse:
|
||||
raw = await file.read()
|
||||
if not raw:
|
||||
raise HTTPException(status_code=400, detail="Leere Datei")
|
||||
if len(raw) > MAX_UPLOAD_BYTES:
|
||||
raise HTTPException(status_code=413, detail="Foto ist groesser als 30 MB")
|
||||
|
||||
try:
|
||||
meta, encoded, fallback = await asyncio.to_thread(_process_photo, raw)
|
||||
except Exception as exc:
|
||||
log.exception("Foto konnte nicht verarbeitet werden")
|
||||
raise HTTPException(status_code=400, detail=f"Bild nicht lesbar: {exc}") from exc
|
||||
|
||||
# Der Ort des Fotos schlaegt den des Uploads: Karten werden oft erst
|
||||
# abends am Schreibtisch abfotografiert, nicht auf der Messe.
|
||||
if meta["lat"] is not None:
|
||||
geo_source = "exif"
|
||||
elif lat is not None and lon is not None:
|
||||
meta["lat"], meta["lon"], geo_source = lat, lon, "browser"
|
||||
else:
|
||||
geo_source = None
|
||||
place = await asyncio.to_thread(geocode.reverse, meta["lat"], meta["lon"])
|
||||
|
||||
results = await extract.extract_all(encoded)
|
||||
|
||||
batch_id = uuid.uuid4().hex
|
||||
created = now_iso()
|
||||
source = "einzel" if fallback else "stapel"
|
||||
cards = []
|
||||
for (upright, _), result in zip(encoded, results):
|
||||
card_id = uuid.uuid4().hex
|
||||
image_file = f"{card_id}.jpg"
|
||||
(config.IMAGE_DIR / image_file).write_bytes(upright)
|
||||
columns = {
|
||||
"id": card_id,
|
||||
"created_at": created,
|
||||
"captured_at": meta["captured_at"] or created,
|
||||
"batch_id": batch_id,
|
||||
"image_file": image_file,
|
||||
"source": source,
|
||||
"notes": "",
|
||||
"lat": meta["lat"],
|
||||
"lon": meta["lon"],
|
||||
"place": place,
|
||||
"geo_source": geo_source,
|
||||
"extract_status": "fehler" if result.get("_error") else "ok",
|
||||
"extract_error": result.get("_error"),
|
||||
}
|
||||
for field in extract.FIELDS:
|
||||
columns[field] = (result.get(field) or None)
|
||||
names = ", ".join(columns)
|
||||
placeholders = ", ".join("?" for _ in columns)
|
||||
db.execute(
|
||||
f"INSERT INTO cards ({names}) VALUES ({placeholders})", list(columns.values())
|
||||
)
|
||||
cards.append(card_dict(db.query_one("SELECT * FROM cards WHERE id = ?", (card_id,))))
|
||||
|
||||
log.info("Scan %s: %d Karte(n), Fallback=%s", batch_id, len(cards), fallback)
|
||||
return JSONResponse(
|
||||
{"batch_id": batch_id, "count": len(cards), "fallback": fallback, "cards": cards}
|
||||
)
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------------- Karten
|
||||
|
||||
@app.get("/api/cards", dependencies=[auth])
|
||||
def list_cards(q: str = "", filter: str = "alle") -> dict:
|
||||
sql = "SELECT * FROM cards"
|
||||
where, params = [], []
|
||||
if q.strip():
|
||||
needle = f"%{q.strip()}%"
|
||||
searchable = [
|
||||
"full_name", "company", "job_title", "email", "phone", "mobile",
|
||||
"website", "city", "place", "notes", "department", "street",
|
||||
]
|
||||
where.append("(" + " OR ".join(f"{c} LIKE ?" for c in searchable) + ")")
|
||||
params += [needle] * len(searchable)
|
||||
if filter == "offen":
|
||||
where.append("exported_at IS NULL")
|
||||
elif filter == "unvollstaendig":
|
||||
where.append(
|
||||
"(extract_status <> 'ok' OR (full_name IS NULL AND company IS NULL))"
|
||||
)
|
||||
elif filter == "erinnerung":
|
||||
where.append(
|
||||
"id IN (SELECT card_id FROM reminders WHERE sent_at IS NULL)"
|
||||
)
|
||||
if where:
|
||||
sql += " WHERE " + " AND ".join(where)
|
||||
sql += " ORDER BY COALESCE(captured_at, created_at) DESC, rowid DESC"
|
||||
|
||||
rows = [card_dict(r) for r in db.query(sql, params)]
|
||||
totals = db.query_one(
|
||||
"""
|
||||
SELECT COUNT(*) AS total,
|
||||
SUM(CASE WHEN exported_at IS NULL THEN 1 ELSE 0 END) AS offen
|
||||
FROM cards
|
||||
"""
|
||||
)
|
||||
pending = db.query_one(
|
||||
"SELECT COUNT(*) AS n FROM reminders WHERE sent_at IS NULL"
|
||||
)
|
||||
return {
|
||||
"cards": rows,
|
||||
"stats": {
|
||||
"total": totals["total"] or 0,
|
||||
"offen": totals["offen"] or 0,
|
||||
"erinnerungen": pending["n"] or 0,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _card_or_404(card_id: str):
|
||||
row = db.query_one("SELECT * FROM cards WHERE id = ?", (card_id,))
|
||||
if row is None:
|
||||
raise HTTPException(status_code=404, detail="Karte nicht gefunden")
|
||||
return row
|
||||
|
||||
|
||||
@app.get("/api/cards/{card_id}", dependencies=[auth])
|
||||
def get_card(card_id: str) -> dict:
|
||||
card = card_dict(_card_or_404(card_id))
|
||||
card["reminders"] = [
|
||||
dict(r)
|
||||
for r in db.query(
|
||||
"SELECT * FROM reminders WHERE card_id = ? ORDER BY due_at", (card_id,)
|
||||
)
|
||||
]
|
||||
return card
|
||||
|
||||
|
||||
@app.patch("/api/cards/{card_id}", dependencies=[auth])
|
||||
async def update_card(card_id: str, request: Request) -> dict:
|
||||
_card_or_404(card_id)
|
||||
payload = await request.json()
|
||||
changes = {k: (payload[k] or None) for k in EDITABLE if k in payload}
|
||||
if "notes" in changes:
|
||||
changes["notes"] = changes["notes"] or ""
|
||||
if not changes:
|
||||
raise HTTPException(status_code=400, detail="Keine bekannten Felder im Aufruf")
|
||||
assignments = ", ".join(f"{k} = ?" for k in changes)
|
||||
db.execute(
|
||||
f"UPDATE cards SET {assignments} WHERE id = ?", [*changes.values(), card_id]
|
||||
)
|
||||
return card_dict(_card_or_404(card_id))
|
||||
|
||||
|
||||
@app.delete("/api/cards/{card_id}", dependencies=[auth])
|
||||
def delete_card(card_id: str) -> dict:
|
||||
row = _card_or_404(card_id)
|
||||
if row["image_file"]:
|
||||
(config.IMAGE_DIR / row["image_file"]).unlink(missing_ok=True)
|
||||
db.execute("DELETE FROM cards WHERE id = ?", (card_id,))
|
||||
return {"ok": True}
|
||||
|
||||
|
||||
@app.get("/api/cards/{card_id}/image", dependencies=[auth])
|
||||
def card_image(card_id: str) -> FileResponse:
|
||||
row = _card_or_404(card_id)
|
||||
path = config.IMAGE_DIR / (row["image_file"] or "")
|
||||
if not row["image_file"] or not path.exists():
|
||||
raise HTTPException(status_code=404, detail="Kein Bild gespeichert")
|
||||
return FileResponse(path, media_type="image/jpeg")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------- Export
|
||||
|
||||
def _mark_exported(ids: list) -> None:
|
||||
stamp = now_iso()
|
||||
db.execute_many(
|
||||
[("UPDATE cards SET exported_at = ? WHERE id = ?", (stamp, i)) for i in ids]
|
||||
)
|
||||
|
||||
|
||||
@app.get("/vcf/{card_id}", dependencies=[auth])
|
||||
def download_vcard(card_id: str) -> Response:
|
||||
card = dict(_card_or_404(card_id))
|
||||
_mark_exported([card_id])
|
||||
return Response(
|
||||
content=vcard.build(card),
|
||||
media_type="text/vcard; charset=utf-8",
|
||||
headers={"Content-Disposition": f'attachment; filename="{vcard.filename(card)}"'},
|
||||
)
|
||||
|
||||
|
||||
@app.get("/vcf", dependencies=[auth])
|
||||
def download_vcards(ids: str = "") -> Response:
|
||||
wanted = [i for i in ids.split(",") if i.strip()]
|
||||
if not wanted:
|
||||
raise HTTPException(status_code=400, detail="Keine Karten ausgewaehlt")
|
||||
placeholders = ", ".join("?" for _ in wanted)
|
||||
rows = db.query(f"SELECT * FROM cards WHERE id IN ({placeholders})", wanted)
|
||||
if not rows:
|
||||
raise HTTPException(status_code=404, detail="Karten nicht gefunden")
|
||||
_mark_exported([r["id"] for r in rows])
|
||||
return Response(
|
||||
content=vcard.build_many([dict(r) for r in rows]),
|
||||
media_type="text/vcard; charset=utf-8",
|
||||
headers={"Content-Disposition": 'attachment; filename="kontakte.vcf"'},
|
||||
)
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------- Erinnerungen
|
||||
|
||||
@app.get("/api/reminders", dependencies=[auth])
|
||||
def list_reminders() -> dict:
|
||||
rows = db.query(
|
||||
"""
|
||||
SELECT r.*, c.full_name, c.company
|
||||
FROM reminders r JOIN cards c ON c.id = r.card_id
|
||||
ORDER BY r.sent_at IS NOT NULL, r.due_at
|
||||
"""
|
||||
)
|
||||
return {"reminders": [dict(r) for r in rows], "mail_configured": bool(config.SMTP_HOST)}
|
||||
|
||||
|
||||
@app.post("/api/cards/{card_id}/reminders", dependencies=[auth])
|
||||
async def create_reminder(card_id: str, request: Request) -> dict:
|
||||
_card_or_404(card_id)
|
||||
payload = await request.json()
|
||||
text = (payload.get("text") or "").strip()
|
||||
due_at = (payload.get("due_at") or "").strip()
|
||||
if not text:
|
||||
raise HTTPException(status_code=400, detail="Erinnerungstext fehlt")
|
||||
try:
|
||||
parsed = datetime.fromisoformat(due_at.replace("Z", "+00:00"))
|
||||
except ValueError as exc:
|
||||
raise HTTPException(status_code=400, detail="Datum nicht lesbar") from exc
|
||||
if parsed.tzinfo is None:
|
||||
parsed = parsed.replace(tzinfo=timezone.utc)
|
||||
|
||||
reminder_id = uuid.uuid4().hex
|
||||
db.execute(
|
||||
"INSERT INTO reminders (id, card_id, text, due_at, created_at) VALUES (?, ?, ?, ?, ?)",
|
||||
(reminder_id, card_id, text, parsed.astimezone(timezone.utc).isoformat(), now_iso()),
|
||||
)
|
||||
return dict(db.query_one("SELECT * FROM reminders WHERE id = ?", (reminder_id,)))
|
||||
|
||||
|
||||
@app.delete("/api/reminders/{reminder_id}", dependencies=[auth])
|
||||
def delete_reminder(reminder_id: str) -> dict:
|
||||
db.execute("DELETE FROM reminders WHERE id = ?", (reminder_id,))
|
||||
return {"ok": True}
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- Statisch
|
||||
|
||||
@app.get("/")
|
||||
def index() -> FileResponse:
|
||||
return FileResponse(STATIC_DIR / "index.html")
|
||||
|
||||
|
||||
@app.get("/sw.js")
|
||||
def service_worker() -> FileResponse:
|
||||
# Muss von der Wurzel kommen, sonst gilt der Service Worker nur fuer /static.
|
||||
return FileResponse(STATIC_DIR / "sw.js", media_type="application/javascript")
|
||||
|
||||
|
||||
@app.get("/manifest.webmanifest")
|
||||
def manifest() -> FileResponse:
|
||||
return FileResponse(STATIC_DIR / "manifest.webmanifest", media_type="application/manifest+json")
|
||||
|
||||
|
||||
@app.get("/robots.txt")
|
||||
def robots() -> PlainTextResponse:
|
||||
return PlainTextResponse("User-agent: *\nDisallow: /\n")
|
||||
|
||||
|
||||
app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
|
||||
90
app/reminders.py
Normal file
90
app/reminders.py
Normal file
@@ -0,0 +1,90 @@
|
||||
"""Faellige Erinnerungen als Mail verschicken.
|
||||
|
||||
Der Scheduler fragt die Datenbank ab, statt Timer im Speicher zu halten:
|
||||
laeuft der Server zum Zeitpunkt einer Erinnerung nicht, wird sie beim
|
||||
naechsten Start nachgeholt.
|
||||
"""
|
||||
import asyncio
|
||||
import logging
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from . import config, db, mailer
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
INTERVAL_SECONDS = 60
|
||||
|
||||
|
||||
def _now() -> str:
|
||||
return datetime.now(timezone.utc).isoformat()
|
||||
|
||||
|
||||
def _body(row) -> str:
|
||||
who = row["full_name"] or row["company"] or "Unbekannter Kontakt"
|
||||
lines = [row["text"], "", "-" * 40, f"Kontakt: {who}"]
|
||||
if row["company"] and row["full_name"]:
|
||||
lines.append(f"Firma: {row['company']}")
|
||||
if row["email"]:
|
||||
lines.append(f"E-Mail: {row['email']}")
|
||||
if row["phone"]:
|
||||
lines.append(f"Telefon: {row['phone']}")
|
||||
if row["mobile"]:
|
||||
lines.append(f"Mobil: {row['mobile']}")
|
||||
if row["place"]:
|
||||
lines.append(f"Kennengelernt in: {row['place']}")
|
||||
if config.PUBLIC_BASE_URL:
|
||||
lines += ["", f"{config.PUBLIC_BASE_URL}/#/karte/{row['card_id']}"]
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def due_reminders() -> list:
|
||||
return db.query(
|
||||
"""
|
||||
SELECT r.id, r.card_id, r.text, r.due_at,
|
||||
c.full_name, c.company, c.email, c.phone, c.mobile, c.place
|
||||
FROM reminders r
|
||||
JOIN cards c ON c.id = r.card_id
|
||||
WHERE r.sent_at IS NULL AND r.due_at <= ?
|
||||
ORDER BY r.due_at
|
||||
""",
|
||||
(_now(),),
|
||||
)
|
||||
|
||||
|
||||
def process_due() -> int:
|
||||
"""Alle faelligen Erinnerungen abarbeiten, Anzahl der Versendeten zurueck."""
|
||||
rows = due_reminders()
|
||||
if not rows:
|
||||
return 0
|
||||
if not mailer.configured():
|
||||
log.warning("%d faellige Erinnerung(en), aber kein SMTP konfiguriert", len(rows))
|
||||
return 0
|
||||
|
||||
sent = 0
|
||||
for row in rows:
|
||||
who = row["full_name"] or row["company"] or "Kontakt"
|
||||
try:
|
||||
mailer.send(f"Erinnerung: {who}", _body(row))
|
||||
except Exception as exc:
|
||||
log.exception("Erinnerung %s konnte nicht verschickt werden", row["id"])
|
||||
db.execute(
|
||||
"UPDATE reminders SET send_error = ? WHERE id = ?",
|
||||
(f"{type(exc).__name__}: {exc}", row["id"]),
|
||||
)
|
||||
continue
|
||||
db.execute(
|
||||
"UPDATE reminders SET sent_at = ?, send_error = NULL WHERE id = ?",
|
||||
(_now(), row["id"]),
|
||||
)
|
||||
sent += 1
|
||||
return sent
|
||||
|
||||
|
||||
async def run_forever() -> None:
|
||||
log.info("Erinnerungs-Scheduler gestartet (Intervall %ds)", INTERVAL_SECONDS)
|
||||
while True:
|
||||
try:
|
||||
await asyncio.to_thread(process_due)
|
||||
except Exception:
|
||||
log.exception("Scheduler-Durchlauf fehlgeschlagen")
|
||||
await asyncio.sleep(INTERVAL_SECONDS)
|
||||
49
app/security.py
Normal file
49
app/security.py
Normal file
@@ -0,0 +1,49 @@
|
||||
"""Anmeldung ueber ein einzelnes Passwort, Sitzung als signiertes Cookie."""
|
||||
import hashlib
|
||||
import hmac
|
||||
import time
|
||||
|
||||
from fastapi import HTTPException, Request
|
||||
|
||||
from . import config
|
||||
|
||||
COOKIE = "bcs_session"
|
||||
|
||||
|
||||
def enabled() -> bool:
|
||||
return bool(config.APP_PASSWORD)
|
||||
|
||||
|
||||
def _sign(payload: str) -> str:
|
||||
return hmac.new(
|
||||
config.SECRET_KEY.encode(), payload.encode(), hashlib.sha256
|
||||
).hexdigest()
|
||||
|
||||
|
||||
def issue_token() -> str:
|
||||
expires = str(int(time.time()) + config.SESSION_DAYS * 86400)
|
||||
return f"{expires}.{_sign(expires)}"
|
||||
|
||||
|
||||
def valid_token(token: str | None) -> bool:
|
||||
if not token or "." not in token:
|
||||
return False
|
||||
expires, signature = token.split(".", 1)
|
||||
if not hmac.compare_digest(signature, _sign(expires)):
|
||||
return False
|
||||
try:
|
||||
return int(expires) > time.time()
|
||||
except ValueError:
|
||||
return False
|
||||
|
||||
|
||||
def check_password(candidate: str) -> bool:
|
||||
return hmac.compare_digest(candidate, config.APP_PASSWORD)
|
||||
|
||||
|
||||
def require_auth(request: Request) -> None:
|
||||
"""FastAPI-Abhaengigkeit fuer alle geschuetzten Endpunkte."""
|
||||
if not enabled():
|
||||
return
|
||||
if not valid_token(request.cookies.get(COOKIE)):
|
||||
raise HTTPException(status_code=401, detail="Nicht angemeldet")
|
||||
154
app/segment.py
Normal file
154
app/segment.py
Normal file
@@ -0,0 +1,154 @@
|
||||
"""Visitenkarten aus einem Foto freistellen.
|
||||
|
||||
Reine Geometrie, kein Modell: Kanten finden, Rechtecke mit Kartenformat
|
||||
behalten, perspektivisch entzerren. Der Inhalt der Karte spielt hier keine
|
||||
Rolle - den liest spaeter das Vision-Modell aus dem Zuschnitt.
|
||||
"""
|
||||
import cv2
|
||||
import numpy as np
|
||||
|
||||
# ISO 7810 ID-1: 85,6 x 54 mm. Toleranz nach unten fuer abweichende Formate.
|
||||
CARD_RATIO = 85.6 / 54.0
|
||||
RATIO_MIN, RATIO_MAX = 1.25, 2.05
|
||||
|
||||
OUT_WIDTH = 1400
|
||||
OUT_HEIGHT = round(OUT_WIDTH / CARD_RATIO)
|
||||
|
||||
DETECT_EDGE = 1600 # Aufloesung, auf der gesucht wird
|
||||
MIN_AREA_FRACTION = 0.004 # kleiner ist Rauschen, kein Kartenfund
|
||||
MAX_AREA_FRACTION = 0.60 # groesser ist der Tisch, nicht die Karte
|
||||
MIN_FILL = 0.80 # Kontur muss ihr eigenes Rechteck fuellen
|
||||
|
||||
|
||||
def _masks(gray: np.ndarray) -> list[np.ndarray]:
|
||||
"""Mehrere Binaerbilder, weil je nach Untergrund ein anderes traegt."""
|
||||
blurred = cv2.GaussianBlur(gray, (5, 5), 0)
|
||||
kernel = cv2.getStructuringElement(cv2.MORPH_RECT, (5, 5))
|
||||
|
||||
edges = cv2.dilate(cv2.Canny(blurred, 40, 120), kernel, iterations=2)
|
||||
edges = cv2.morphologyEx(edges, cv2.MORPH_CLOSE, kernel, iterations=2)
|
||||
|
||||
adaptive = cv2.adaptiveThreshold(
|
||||
blurred, 255, cv2.ADAPTIVE_THRESH_GAUSSIAN_C, cv2.THRESH_BINARY, 51, 10
|
||||
)
|
||||
adaptive = cv2.morphologyEx(adaptive, cv2.MORPH_CLOSE, kernel, iterations=2)
|
||||
|
||||
_, otsu = cv2.threshold(blurred, 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU)
|
||||
|
||||
return [edges, adaptive, otsu, cv2.bitwise_not(otsu)]
|
||||
|
||||
|
||||
def _candidates(mask: np.ndarray, image_area: float) -> list[tuple]:
|
||||
contours, _ = cv2.findContours(mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
|
||||
found = []
|
||||
for contour in contours:
|
||||
rect = cv2.minAreaRect(contour)
|
||||
(_, _), (width, height), _ = rect
|
||||
if width < 24 or height < 24:
|
||||
continue
|
||||
long_side, short_side = max(width, height), min(width, height)
|
||||
if not RATIO_MIN <= long_side / short_side <= RATIO_MAX:
|
||||
continue
|
||||
rect_area = width * height
|
||||
if not MIN_AREA_FRACTION * image_area <= rect_area <= MAX_AREA_FRACTION * image_area:
|
||||
continue
|
||||
fill = cv2.contourArea(contour) / rect_area
|
||||
if fill < MIN_FILL:
|
||||
continue
|
||||
found.append((rect, fill))
|
||||
return found
|
||||
|
||||
|
||||
def _deduplicate(candidates: list[tuple]) -> list[tuple]:
|
||||
"""Dieselbe Karte wird von mehreren Masken gefunden - besten Treffer behalten."""
|
||||
kept: list[tuple] = []
|
||||
for rect, fill in sorted(candidates, key=lambda c: -c[1]):
|
||||
(cx, cy), (width, height), _ = rect
|
||||
radius = min(width, height) * 0.5
|
||||
duplicate = False
|
||||
for other_rect, _ in kept:
|
||||
(ox, oy), _, _ = other_rect
|
||||
if np.hypot(cx - ox, cy - oy) < radius:
|
||||
duplicate = True
|
||||
break
|
||||
if not duplicate:
|
||||
kept.append((rect, fill))
|
||||
return kept
|
||||
|
||||
|
||||
def _order_quad(points: np.ndarray) -> np.ndarray:
|
||||
"""Ecken als oben-links, oben-rechts, unten-rechts, unten-links."""
|
||||
total = points.sum(axis=1)
|
||||
diff = points[:, 1] - points[:, 0]
|
||||
ordered = np.array(
|
||||
[
|
||||
points[np.argmin(total)], # oben links
|
||||
points[np.argmin(diff)], # oben rechts
|
||||
points[np.argmax(total)], # unten rechts
|
||||
points[np.argmax(diff)], # unten links
|
||||
],
|
||||
dtype="float32",
|
||||
)
|
||||
# Hochkant liegende Karte um 90 Grad drehen, damit der Zuschnitt quer ist.
|
||||
width = np.linalg.norm(ordered[1] - ordered[0])
|
||||
height = np.linalg.norm(ordered[3] - ordered[0])
|
||||
if height > width:
|
||||
ordered = np.roll(ordered, -1, axis=0)
|
||||
return ordered
|
||||
|
||||
|
||||
def _reading_order(rects: list) -> list:
|
||||
"""Zeilenweise sortieren, damit die Reihenfolge dem Tisch entspricht."""
|
||||
if not rects:
|
||||
return rects
|
||||
row_height = np.median([min(r[1]) for r in rects]) * 0.7
|
||||
return sorted(rects, key=lambda r: (round(r[0][1] / row_height), r[0][0]))
|
||||
|
||||
|
||||
def segment(image: np.ndarray) -> tuple[list[np.ndarray], bool]:
|
||||
"""Liefert die entzerrten Kartenbilder und ob auf das Gesamtbild
|
||||
zurueckgefallen wurde (kein Kartenrechteck gefunden)."""
|
||||
height, width = image.shape[:2]
|
||||
scale = DETECT_EDGE / max(height, width) if max(height, width) > DETECT_EDGE else 1.0
|
||||
small = (
|
||||
cv2.resize(image, None, fx=scale, fy=scale, interpolation=cv2.INTER_AREA)
|
||||
if scale < 1.0
|
||||
else image
|
||||
)
|
||||
gray = cv2.cvtColor(small, cv2.COLOR_BGR2GRAY)
|
||||
image_area = float(small.shape[0] * small.shape[1])
|
||||
|
||||
candidates: list[tuple] = []
|
||||
for mask in _masks(gray):
|
||||
candidates.extend(_candidates(mask, image_area))
|
||||
|
||||
rects = [rect for rect, _ in _deduplicate(candidates)]
|
||||
if not rects:
|
||||
return [_fit_whole(image)], True
|
||||
|
||||
crops = []
|
||||
for rect in _reading_order(rects):
|
||||
quad = _order_quad(cv2.boxPoints(rect)) / scale # zurueck auf volle Aufloesung
|
||||
target = np.array(
|
||||
[[0, 0], [OUT_WIDTH, 0], [OUT_WIDTH, OUT_HEIGHT], [0, OUT_HEIGHT]],
|
||||
dtype="float32",
|
||||
)
|
||||
matrix = cv2.getPerspectiveTransform(quad, target)
|
||||
crops.append(
|
||||
cv2.warpPerspective(
|
||||
image, matrix, (OUT_WIDTH, OUT_HEIGHT), flags=cv2.INTER_CUBIC
|
||||
)
|
||||
)
|
||||
return crops, False
|
||||
|
||||
|
||||
def _fit_whole(image: np.ndarray) -> np.ndarray:
|
||||
"""Ohne Fund: das ganze Bild als eine Karte behandeln."""
|
||||
height, width = image.shape[:2]
|
||||
if width < height: # Hochformat drehen
|
||||
image = cv2.rotate(image, cv2.ROTATE_90_CLOCKWISE)
|
||||
height, width = image.shape[:2]
|
||||
scale = min(OUT_WIDTH / width, OUT_HEIGHT / height, 1.0)
|
||||
if scale < 1.0:
|
||||
image = cv2.resize(image, None, fx=scale, fy=scale, interpolation=cv2.INTER_AREA)
|
||||
return image
|
||||
90
app/vcard.py
Normal file
90
app/vcard.py
Normal file
@@ -0,0 +1,90 @@
|
||||
"""vCard 3.0 erzeugen - das Format, das iOS beim Antippen zuverlaessig oeffnet."""
|
||||
from datetime import datetime
|
||||
|
||||
_ESCAPES = (("\\", "\\\\"), (";", "\\;"), (",", "\\,"), ("\n", "\\n"))
|
||||
|
||||
|
||||
def _escape(value) -> str:
|
||||
text = str(value)
|
||||
for char, replacement in _ESCAPES:
|
||||
text = text.replace(char, replacement)
|
||||
return text
|
||||
|
||||
|
||||
def _line(name: str, value) -> list:
|
||||
return [name + ":" + _escape(value)] if value else []
|
||||
|
||||
|
||||
def _display_name(card: dict) -> str:
|
||||
if card.get("full_name"):
|
||||
return card["full_name"]
|
||||
joined = " ".join(filter(None, (card.get("first_name"), card.get("last_name"))))
|
||||
return joined or card.get("company") or "Unbenannt"
|
||||
|
||||
|
||||
def _note(card: dict) -> str:
|
||||
parts = []
|
||||
if card.get("notes"):
|
||||
parts.append(card["notes"])
|
||||
|
||||
stamp = card.get("captured_at") or card.get("created_at")
|
||||
scanned = ""
|
||||
if stamp:
|
||||
try:
|
||||
when = datetime.fromisoformat(stamp).strftime("%d.%m.%Y um %H:%M")
|
||||
scanned = "Gescannt am " + when
|
||||
except ValueError:
|
||||
scanned = "Gescannt am " + str(stamp)
|
||||
if card.get("place"):
|
||||
scanned = (scanned + " in " if scanned else "Gescannt in ") + card["place"]
|
||||
if scanned:
|
||||
parts.append(scanned)
|
||||
return "\n".join(parts)
|
||||
|
||||
|
||||
def build(card: dict) -> str:
|
||||
lines = ["BEGIN:VCARD", "VERSION:3.0"]
|
||||
lines.append(
|
||||
"N:{};{};;;".format(
|
||||
_escape(card.get("last_name") or ""), _escape(card.get("first_name") or "")
|
||||
)
|
||||
)
|
||||
lines.append("FN:" + _escape(_display_name(card)))
|
||||
lines += _line("ORG", card.get("company"))
|
||||
lines += _line("TITLE", card.get("job_title"))
|
||||
if card.get("email"):
|
||||
lines.append("EMAIL;TYPE=INTERNET,WORK:" + _escape(card["email"]))
|
||||
if card.get("phone"):
|
||||
lines.append("TEL;TYPE=WORK,VOICE:" + _escape(card["phone"]))
|
||||
if card.get("mobile"):
|
||||
lines.append("TEL;TYPE=CELL:" + _escape(card["mobile"]))
|
||||
if card.get("fax"):
|
||||
lines.append("TEL;TYPE=WORK,FAX:" + _escape(card["fax"]))
|
||||
if card.get("website"):
|
||||
lines.append("URL:" + _escape(card["website"]))
|
||||
if any(card.get(k) for k in ("street", "postal_code", "city", "country")):
|
||||
lines.append(
|
||||
"ADR;TYPE=WORK:;;{};{};;{};{}".format(
|
||||
_escape(card.get("street") or ""),
|
||||
_escape(card.get("city") or ""),
|
||||
_escape(card.get("postal_code") or ""),
|
||||
_escape(card.get("country") or ""),
|
||||
)
|
||||
)
|
||||
if card.get("lat") is not None and card.get("lon") is not None:
|
||||
lines.append("GEO:{};{}".format(card["lat"], card["lon"]))
|
||||
note = _note(card)
|
||||
if note:
|
||||
lines.append("NOTE:" + _escape(note))
|
||||
lines.append("END:VCARD")
|
||||
return "\r\n".join(lines) + "\r\n"
|
||||
|
||||
|
||||
def build_many(cards: list) -> str:
|
||||
return "".join(build(card) for card in cards)
|
||||
|
||||
|
||||
def filename(card: dict) -> str:
|
||||
base = _display_name(card).replace(" ", "-")
|
||||
safe = "".join(c for c in base if c.isalnum() or c in "-_")
|
||||
return (safe or "kontakt") + ".vcf"
|
||||
Reference in New Issue
Block a user