"""Import Data - Customers endpoints.""" import io from typing import Optional from fastapi import APIRouter, Request, Query, UploadFile, File, Form, Header from fastapi.responses import JSONResponse, Response import sqlalchemy from pydantic import BaseModel from api.feature.import_data.db import get_cloudsql from api.feature.import_data.parsing import parse_customers_file router = APIRouter() def _property_id(header: Optional[str], query: Optional[str]) -> str: return (query or header or "").strip() class CustomerImportBody(BaseModel): property_id: str customers: list[dict] def _iso(ts) -> Optional[str]: if ts is None: return None return ts.isoformat() if hasattr(ts, "isoformat") else str(ts) @router.get("/") async def list_customers( request: Request, property_id: Optional[str] = Query(None), limit: int = Query(20, le=100), offset: int = Query(0, ge=0), sort_by: str = Query("created_at"), order: str = Query("desc"), customer_type: Optional[str] = Query(None), x_property: Optional[str] = Header(None, alias="x-property"), ): """GET /api/v1/import-data/customers - list + last_import, last_sync.""" pid = _property_id(x_property, property_id) if not pid: return JSONResponse( status_code=400, content={"success": False, "message": "property_id (or x-property) required"}, ) csql = get_cloudsql() order_dir = "DESC" if (order or "desc").lower() == "desc" else "ASC" order_col = sort_by if sort_by in ("created_at", "full_name", "email", "customer_type") else "created_at" where = " property_id = :pid " params = {"pid": pid} if customer_type in ("individual", "company"): where += " AND customer_type = :ct " params["ct"] = customer_type with csql.pool.connect() as conn: total = conn.execute(sqlalchemy.text(f"SELECT COUNT(*) FROM customers WHERE {where}"), params).scalar() or 0 meta = conn.execute( sqlalchemy.text( "SELECT last_import_at, last_sync_at FROM sync_meta WHERE property_id = :pid AND entity = 'customer'" ), {"pid": pid}, ).fetchone() last_import = _iso(meta[0]) if meta else None last_sync = _iso(meta[1]) if meta and len(meta) > 1 else None params["lim"] = limit params["off"] = offset rows = conn.execute( sqlalchemy.text(f""" SELECT id, full_name, title, email, phone, company, customer_type, notes, created_at FROM customers WHERE {where} ORDER BY {order_col} {order_dir} LIMIT :lim OFFSET :off """), params, ).fetchall() data = [ { "id": str(r[0]), "full_name": r[1], "title": r[2], "email": r[3], "phone": r[4], "company": r[5], "customer_type": r[6], "notes": r[7], "created_at": _iso(r[8]), } for r in rows ] return JSONResponse( content={ "success": True, "total": total, "last_import": last_import, "last_sync": last_sync, "limit": limit, "offset": offset, "data": data, } ) @router.post("/import/file") async def import_customers_file( request: Request, file: UploadFile = File(...), property_id: str = Form(...), x_property: Optional[str] = Header(None, alias="x-property"), ): pid = (property_id or (x_property or "")).strip() if not pid: return JSONResponse(status_code=400, content={"success": False, "message": "property_id required"}) content = await file.read() try: rows = parse_customers_file(content, file.filename or "") except Exception as e: return JSONResponse(status_code=400, content={"success": False, "message": f"Parse error: {e}"}) user_id = getattr(request.state, "user_id", None) imported = 0 csql = get_cloudsql() with csql.pool.connect() as conn: for r in rows: fn = (r.get("full_name") or r.get("name") or "").strip() if not fn: continue ct = (r.get("customer_type") or r.get("type") or "individual").strip().lower() if ct not in ("individual", "company"): ct = "individual" conn.execute( sqlalchemy.text(""" INSERT INTO customers (property_id, full_name, title, email, phone, company, customer_type, notes, created_by) VALUES (:pid, :fn, :ti, :em, :ph, :co, :ct, :no, :uid) """), { "pid": pid, "fn": fn, "ti": (r.get("title") or "").strip() or None, "em": (r.get("email") or "").strip() or None, "ph": (r.get("phone") or "").strip() or None, "co": (r.get("company") or "").strip() or None, "ct": ct, "no": (r.get("notes") or r.get("note") or "").strip() or None, "uid": user_id, }, ) imported += 1 conn.execute( sqlalchemy.text(""" INSERT INTO sync_meta (property_id, entity, last_import_at, updated_at) VALUES (:pid, 'customer', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) ON CONFLICT (property_id, entity) DO UPDATE SET last_import_at = CURRENT_TIMESTAMP, updated_at = CURRENT_TIMESTAMP """), {"pid": pid}, ) conn.commit() return JSONResponse(content={"success": True, "message": f"Imported {imported} customer(s)", "imported": imported}) @router.post("/import") async def import_customers_json(request: Request, body: CustomerImportBody): user_id = getattr(request.state, "user_id", None) imported = 0 csql = get_cloudsql() with csql.pool.connect() as conn: for r in body.customers: fn = (r.get("full_name") or r.get("name") or "").strip() if not fn: continue ct = (r.get("customer_type") or r.get("type") or "individual").strip().lower() if ct not in ("individual", "company"): ct = "individual" conn.execute( sqlalchemy.text(""" INSERT INTO customers (property_id, full_name, title, email, phone, company, customer_type, notes, created_by) VALUES (:pid, :fn, :ti, :em, :ph, :co, :ct, :no, :uid) """), { "pid": body.property_id, "fn": fn, "ti": (r.get("title") or "").strip() or None, "em": (r.get("email") or "").strip() or None, "ph": (r.get("phone") or "").strip() or None, "co": (r.get("company") or "").strip() or None, "ct": ct, "no": (r.get("notes") or r.get("note") or "").strip() or None, "uid": user_id, }, ) imported += 1 conn.execute( sqlalchemy.text(""" INSERT INTO sync_meta (property_id, entity, last_import_at, updated_at) VALUES (:pid, 'customer', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) ON CONFLICT (property_id, entity) DO UPDATE SET last_import_at = CURRENT_TIMESTAMP, updated_at = CURRENT_TIMESTAMP """), {"pid": body.property_id}, ) conn.commit() return JSONResponse(content={"success": True, "message": f"Imported {imported} customer(s)", "imported": imported}) _CUSTOMER_CSV = "full_name,title,email,phone,company,customer_type,notes\n" @router.get("/import/example/csv") async def example_customers_csv(): return Response( content=_CUSTOMER_CSV.encode("utf-8"), media_type="text/csv; charset=utf-8", headers={"Content-Disposition": 'attachment; filename="customers-example.csv"'}, ) @router.get("/import/example/xlsx") async def example_customers_xlsx(): try: from openpyxl import Workbook except ImportError: return JSONResponse(status_code=501, content={"success": False, "message": "openpyxl not installed"}) wb = Workbook() ws = wb.active for i, h in enumerate(["full_name", "title", "email", "phone", "company", "customer_type", "notes"], 1): ws.cell(row=1, column=i, value=h) buf = io.BytesIO() wb.save(buf) buf.seek(0) return Response( content=buf.read(), media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", headers={"Content-Disposition": 'attachment; filename="customers-example.xlsx"'}, )