| import csv |
| import logging |
| import os |
| import tempfile |
|
|
| from fastapi import FastAPI, UploadFile, File, HTTPException |
| from fastapi.middleware.cors import CORSMiddleware |
| from fastapi.responses import FileResponse |
|
|
| from model import extract_contact |
| from vcf import generate_vcf, VCF_DIR |
|
|
| CSV_FIELDS = ["name", "company", "designation", "phone", "email", "website", "address"] |
| CSV_PATH = os.path.join(VCF_DIR, "contacts.csv") |
|
|
|
|
| def append_csv(data): |
| |
| new_file = not os.path.exists(CSV_PATH) |
| with open(CSV_PATH, "a", newline="") as f: |
| w = csv.DictWriter(f, fieldnames=CSV_FIELDS, extrasaction="ignore") |
| if new_file: |
| w.writeheader() |
| w.writerow({k: data.get(k, "") for k in CSV_FIELDS}) |
| return os.path.basename(CSV_PATH) |
|
|
| logging.basicConfig( |
| level=logging.INFO, |
| format="%(asctime)s %(levelname)s: %(message)s", |
| ) |
| log = logging.getLogger("cardextractor") |
|
|
| app = FastAPI(title="Business Card Extractor", version="1.0") |
|
|
| |
| app.add_middleware( |
| CORSMiddleware, |
| allow_origins=["*"], |
| allow_methods=["*"], |
| allow_headers=["*"], |
| ) |
|
|
|
|
| @app.get("/") |
| def home(): |
| return {"message": "Business Card API", "endpoint": "/extract"} |
|
|
|
|
| @app.post("/extract") |
| async def extract(file: UploadFile = File(...)): |
| image_path = None |
| try: |
| suffix = os.path.splitext(file.filename)[1] |
| with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as temp: |
| body = await file.read() |
| temp.write(body) |
| image_path = temp.name |
| log.info("received %s (%d bytes) -> %s", file.filename, len(body), image_path) |
|
|
| data = extract_contact(image_path) |
| log.info("extracted: %s", data) |
|
|
| vcf_path = generate_vcf(data) |
| filename = os.path.basename(vcf_path) |
| csv_name = append_csv(data) |
| log.info("wrote vcf: %s, appended csv: %s", filename, csv_name) |
|
|
| return { |
| "success": True, |
| "contact": data, |
| "vcf": filename, |
| "csv": csv_name, |
| "download": f"/download/{filename}", |
| "csv_download": f"/download/{csv_name}", |
| } |
| except Exception as e: |
| |
| log.exception("extract failed for %s", file.filename) |
| raise HTTPException(500, f"{type(e).__name__}: {e}") |
| finally: |
| if image_path and os.path.exists(image_path): |
| os.remove(image_path) |
|
|
|
|
| @app.get("/download/{filename}") |
| def download(filename: str): |
| |
| path = os.path.join(VCF_DIR, os.path.basename(filename)) |
| if not os.path.exists(path): |
| raise HTTPException(404, "file not found") |
| media = "text/csv" if path.endswith(".csv") else "text/vcard" |
| return FileResponse(path, filename=filename, media_type=media) |
|
|