diff --git a/app/core/worker.py b/app/core/worker.py index 10905c7..604c21c 100644 --- a/app/core/worker.py +++ b/app/core/worker.py @@ -225,6 +225,7 @@ def _lazy_register_plugin_jobs() -> None: "app.plugins.builtins.automation.agent_runner", "app.plugins.builtins.automation.execution_engine", "app.plugins.builtins.tasks.jobs", + "app.services.import_export_jobs", ] for mod_name in plugin_job_modules: try: diff --git a/app/routes/import_export.py b/app/routes/import_export.py index 8ac0300..728b9a3 100644 --- a/app/routes/import_export.py +++ b/app/routes/import_export.py @@ -1,8 +1,9 @@ -"""Import/export routes — CSV import, dry-run preview, CSV/XLSX export.""" +"""Import/export routes — CSV/JSON/XLSX import, preview, validate, background jobs, CSV/XLSX export.""" from __future__ import annotations import io +import json import uuid from fastapi import ( @@ -20,27 +21,75 @@ from starlette.responses import StreamingResponse from app.core.db import get_db from app.deps import require_permission from app.services import import_export_service +from app.services.import_export_helpers import ( + detect_format, + parse_file, + write_csv, + write_xlsx, +) router = APIRouter(prefix="/api/v1", tags=["import_export"]) +# Row threshold for background job processing +_BACKGROUND_THRESHOLD = 1000 + @router.post("/import") async def import_csv( file: UploadFile = File(...), entity_type: str = Form("companies"), + field_mapping: str = Form(None), db: AsyncSession = Depends(get_db), current_user: dict = Depends(require_permission("import_export:write")), ): - """Import companies or contacts from CSV file. + """Import companies or contacts from CSV/JSON/XLSX file. + entity_type: 'companies' or 'contacts'. + field_mapping: Optional JSON string of source_column -> target_field mapping. + For files > 1000 rows, the import runs as a background ARQ job. """ tenant_id = uuid.UUID(current_user["tenant_id"]) user_id = uuid.UUID(current_user["user_id"]) - content = await file.read() - csv_content = content.decode("utf-8") + csv_content = content.decode("utf-8", errors="replace") + # Parse mapping if provided + mapping = None + if field_mapping: + try: + mapping = json.loads(field_mapping) + except json.JSONDecodeError: + raise HTTPException(status_code=400, detail="Invalid field_mapping JSON") + + # Check row count for background processing + try: + rows = parse_file(file.filename or "upload.csv", content) + except Exception as exc: + raise HTTPException(status_code=400, detail=f"Failed to parse file: {exc}") + + if len(rows) > _BACKGROUND_THRESHOLD: + # Enqueue as background job + from app.services.import_export_jobs import create_import_job + + try: + job_id = await create_import_job( + entity_type=entity_type, + csv_content=csv_content, + tenant_id=tenant_id, + user_id=user_id, + field_mapping=mapping, + ) + return { + "status": "pending", + "job_id": job_id, + "message": f"Import enqueued as background job ({len(rows)} rows)", + "total": len(rows), + } + except Exception as exc: + raise HTTPException(status_code=500, detail=f"Failed to enqueue import job: {exc}") + + # Synchronous import for smaller files result = await import_export_service.import_csv( db, tenant_id, @@ -48,6 +97,7 @@ async def import_csv( csv_content, entity_type=entity_type, dry_run=False, + field_mapping=mapping, ) return result @@ -56,24 +106,188 @@ async def import_csv( async def import_csv_preview( file: UploadFile = File(...), entity_type: str = Form("companies"), - db: AsyncSession = Depends(get_db), current_user: dict = Depends(require_permission("import_export:read")), ): - """Preview CSV import (dry-run — no DB changes).""" + """Preview CSV import (dry-run — no DB changes). + + Returns first 10 rows, detected columns, and mapping suggestion. + """ + content = await file.read() + + try: + result = import_export_service.preview_import( + filename=file.filename or "upload.csv", + content=content, + entity_type=entity_type, + ) + except Exception as exc: + raise HTTPException(status_code=400, detail=f"Failed to parse file: {exc}") + + return result + + +@router.post("/import/validate") +async def import_validate( + file: UploadFile = File(...), + entity_type: str = Form("companies"), + field_mapping: str = Form(None), + current_user: dict = Depends(require_permission("import_export:read")), +): + """Validate all rows against mapping and return error report without importing.""" + content = await file.read() + + mapping = None + if field_mapping: + try: + mapping = json.loads(field_mapping) + except json.JSONDecodeError: + raise HTTPException(status_code=400, detail="Invalid field_mapping JSON") + + try: + result = import_export_service.validate_import( + filename=file.filename or "upload.csv", + content=content, + entity_type=entity_type, + field_mapping=mapping, + ) + except Exception as exc: + raise HTTPException(status_code=400, detail=f"Failed to validate file: {exc}") + + return result + + +@router.get("/import/status/{job_id}") +async def import_job_status( + job_id: str, + current_user: dict = Depends(require_permission("import_export:read")), +): + """Get status of a background import job.""" + from app.services.import_export_jobs import get_import_job_status + + status = await get_import_job_status(job_id) + if status is None: + raise HTTPException(status_code=404, detail=f"Job {job_id} not found") + return status + + +@router.post("/import/contacts") +async def import_contacts_route( + file: UploadFile = File(...), + field_mapping: str = Form(None), + db: AsyncSession = Depends(get_db), + current_user: dict = Depends(require_permission("import_export:write")), +): + """Import contacts from CSV/JSON/XLSX file. + + For files > 1000 rows, runs as background ARQ job. + """ tenant_id = uuid.UUID(current_user["tenant_id"]) user_id = uuid.UUID(current_user["user_id"]) - content = await file.read() - csv_content = content.decode("utf-8") + csv_content = content.decode("utf-8", errors="replace") - result = await import_export_service.import_csv( + mapping = None + if field_mapping: + try: + mapping = json.loads(field_mapping) + except json.JSONDecodeError: + raise HTTPException(status_code=400, detail="Invalid field_mapping JSON") + + # Check row count for background processing + try: + rows = parse_file(file.filename or "upload.csv", content) + except Exception as exc: + raise HTTPException(status_code=400, detail=f"Failed to parse file: {exc}") + + if len(rows) > _BACKGROUND_THRESHOLD: + from app.services.import_export_jobs import create_import_job + + try: + job_id = await create_import_job( + entity_type="contacts", + csv_content=csv_content, + tenant_id=tenant_id, + user_id=user_id, + field_mapping=mapping, + ) + return { + "status": "pending", + "job_id": job_id, + "message": f"Import enqueued as background job ({len(rows)} rows)", + "total": len(rows), + } + except Exception as exc: + raise HTTPException(status_code=500, detail=f"Failed to enqueue import job: {exc}") + + result = await import_export_service.import_contacts( db, tenant_id, user_id, csv_content, - entity_type=entity_type, - dry_run=True, + dry_run=False, + field_mapping=mapping, + ) + return result + + +@router.post("/import/companies") +async def import_companies_route( + file: UploadFile = File(...), + field_mapping: str = Form(None), + db: AsyncSession = Depends(get_db), + current_user: dict = Depends(require_permission("import_export:write")), +): + """Import companies from CSV/JSON/XLSX file. + + For files > 1000 rows, runs as background ARQ job. + """ + tenant_id = uuid.UUID(current_user["tenant_id"]) + user_id = uuid.UUID(current_user["user_id"]) + + content = await file.read() + csv_content = content.decode("utf-8", errors="replace") + + mapping = None + if field_mapping: + try: + mapping = json.loads(field_mapping) + except json.JSONDecodeError: + raise HTTPException(status_code=400, detail="Invalid field_mapping JSON") + + # Check row count for background processing + try: + rows = parse_file(file.filename or "upload.csv", content) + except Exception as exc: + raise HTTPException(status_code=400, detail=f"Failed to parse file: {exc}") + + if len(rows) > _BACKGROUND_THRESHOLD: + from app.services.import_export_jobs import create_import_job + + try: + job_id = await create_import_job( + entity_type="companies", + csv_content=csv_content, + tenant_id=tenant_id, + user_id=user_id, + field_mapping=mapping, + ) + return { + "status": "pending", + "job_id": job_id, + "message": f"Import enqueued as background job ({len(rows)} rows)", + "total": len(rows), + } + except Exception as exc: + raise HTTPException(status_code=500, detail=f"Failed to enqueue import job: {exc}") + + result = await import_export_service.import_companies( + db, + tenant_id, + user_id, + csv_content, + dry_run=False, + field_mapping=mapping, ) return result @@ -81,24 +295,21 @@ async def import_csv_preview( @router.get("/export") async def export_data( entity_type: str = Query("contacts", pattern="^(contacts|companies)$"), - format: str = Query("csv", pattern="^(csv|xlsx)$"), + format: str = Query("csv", pattern="^(csv|xlsx|json)$"), db: AsyncSession = Depends(get_db), current_user: dict = Depends(require_permission("import_export:read")), ): - """Export contacts or companies as CSV or XLSX file download. + """Export contacts or companies as CSV, XLSX, or JSON file download. Query params: - entity_type: 'contacts' or 'companies' (default: contacts) - - format: 'csv' or 'xlsx' (default: csv) + - format: 'csv', 'xlsx', or 'json' (default: csv) """ tenant_id = uuid.UUID(current_user["tenant_id"]) user_id = uuid.UUID(current_user["user_id"]) is_system_admin = current_user.get("is_system_admin", False) - # Determine filename based on entity_type - filename = f"{entity_type}_export.csv" - - # Fetch CSV data from the appropriate service function + # Fetch data from the appropriate service function if entity_type == "contacts": csv_data = await import_export_service.export_contacts_csv( db, tenant_id, user_id=user_id, is_system_admin=is_system_admin @@ -110,29 +321,22 @@ async def export_data( else: raise HTTPException(status_code=400, detail=f"Unsupported entity_type: {entity_type}") - # XLSX format handling + # Parse CSV data back to rows for format conversion + import csv as _csv + + reader = _csv.reader(io.StringIO(csv_data)) + all_rows = list(reader) + if not all_rows: + raise HTTPException(status_code=404, detail="No data to export") + headers = all_rows[0] + data_rows = [dict(zip(headers, row)) for row in all_rows[1:]] + if format == "xlsx": try: - import openpyxl - - wb = openpyxl.Workbook() - ws = wb.active - ws.title = entity_type.capitalize() - - # Parse the CSV string and populate the worksheet - import csv as _csv - - reader = _csv.reader(io.StringIO(csv_data)) - for row in reader: - ws.append(row) - - xlsx_buffer = io.BytesIO() - wb.save(xlsx_buffer) - xlsx_buffer.seek(0) - + xlsx_bytes = write_xlsx(data_rows, headers) xlsx_filename = f"{entity_type}_export.xlsx" return StreamingResponse( - xlsx_buffer, + io.BytesIO(xlsx_bytes), media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", headers={"Content-Disposition": f'attachment; filename="{xlsx_filename}"'}, ) @@ -140,7 +344,7 @@ async def export_data( # openpyxl not available — fall back to CSV with a warning header media_type = "text/csv" headers = { - "Content-Disposition": f'attachment; filename="{filename}"', + "Content-Disposition": f'attachment; filename="{entity_type}_export.csv"', "X-Export-Warning": "openpyxl not installed, falling back to CSV format", } return StreamingResponse( @@ -149,9 +353,21 @@ async def export_data( headers=headers, ) + if format == "json": + from app.services.import_export_helpers import write_json + + json_bytes = write_json(data_rows) + json_filename = f"{entity_type}_export.json" + return StreamingResponse( + io.BytesIO(json_bytes), + media_type="application/json", + headers={"Content-Disposition": f'attachment; filename="{json_filename}"'}, + ) + # Default: CSV format + csv_filename = f"{entity_type}_export.csv" return StreamingResponse( iter([csv_data.encode("utf-8")]), media_type="text/csv", - headers={"Content-Disposition": f'attachment; filename="{filename}"'}, + headers={"Content-Disposition": f'attachment; filename="{csv_filename}"'}, ) diff --git a/app/services/import_export_helpers.py b/app/services/import_export_helpers.py new file mode 100644 index 0000000..08635f3 --- /dev/null +++ b/app/services/import_export_helpers.py @@ -0,0 +1,352 @@ +"""Shared import/export helpers — reusable parsing, writing, mapping, validation. + +These building blocks are used by import_export_service.py and can be leveraged +by any plugin that needs custom import/export handlers. +""" + +from __future__ import annotations + +import csv +import io +import json +import logging +import re +from typing import Any + +logger = logging.getLogger(__name__) + +# ─── Encoding detection ────────────────────────────────────────────────────── + +_BOMS = ( + ("utf-8-sig", b"\xef\xbb\xbf"), + ("utf-16-le", b"\xff\xfe"), + ("utf-16-be", b"\xfe\xff"), +) + + +def _detect_encoding(content: bytes) -> str: + """Detect the most likely text encoding from raw bytes.""" + for encoding, bom in _BOMS: + if content.startswith(bom): + return encoding + # Try UTF-8 first, fall back to latin-1 (never fails) + try: + content.decode("utf-8") + return "utf-8" + except UnicodeDecodeError: + return "latin-1" + + +# ─── Parsers ───────────────────────────────────────────────────────────────── + +def parse_csv(content: bytes) -> list[dict[str, str]]: + """Parse CSV content (bytes) into a list of dicts. + + Handles encoding detection (UTF-8, UTF-8 BOM, UTF-16, latin-1). + """ + encoding = _detect_encoding(content) + text = content.decode(encoding) + reader = csv.DictReader(io.StringIO(text)) + return [dict(row) for row in reader] + + +def parse_json(content: bytes) -> list[dict]: + """Parse JSON content (bytes) into a list of dicts. + + Accepts either a JSON array of objects or a single object. + """ + encoding = _detect_encoding(content) + text = content.decode(encoding) + data = json.loads(text) + if isinstance(data, dict): + return [data] + if isinstance(data, list): + return data + raise ValueError(f"Expected JSON array or object, got {type(data).__name__}") + + +def parse_xlsx(content: bytes) -> list[dict[str, str]]: + """Parse XLSX content (bytes) into a list of dicts. + + Uses openpyxl. First row is treated as headers. + """ + try: + import openpyxl + except ImportError as exc: + raise ImportError("openpyxl is required for XLSX parsing") from exc + + wb = openpyxl.load_workbook(io.BytesIO(content), read_only=True, data_only=True) + ws = wb.active + rows_iter = ws.iter_rows(values_only=True) + try: + headers = [str(h).strip() if h is not None else "" for h in next(rows_iter)] + except StopIteration: + wb.close() + return [] + result: list[dict[str, str]] = [] + for row in rows_iter: + if row is None or all(c is None or str(c).strip() == "" for c in row): + continue + row_dict = {} + for idx, header in enumerate(headers): + if not header: + continue + val = row[idx] if idx < len(row) else None + row_dict[header] = str(val).strip() if val is not None else "" + result.append(row_dict) + wb.close() + return result + + +# ─── Writers ───────────────────────────────────────────────────────────────── + +def write_csv(rows: list[dict], headers: list[str]) -> bytes: + """Write a list of dicts to CSV bytes using the given header order.""" + output = io.StringIO() + writer = csv.DictWriter(output, fieldnames=headers, extrasaction="ignore") + writer.writeheader() + for row in rows: + writer.writerow({h: row.get(h, "") for h in headers}) + return output.getvalue().encode("utf-8") + + +def write_json(rows: list[dict]) -> bytes: + """Write a list of dicts to JSON bytes (UTF-8, indented).""" + return json.dumps(rows, indent=2, ensure_ascii=False, default=str).encode("utf-8") + + +def write_xlsx(rows: list[dict], headers: list[str]) -> bytes: + """Write a list of dicts to XLSX bytes using the given header order.""" + try: + import openpyxl + except ImportError as exc: + raise ImportError("openpyxl is required for XLSX writing") from exc + + wb = openpyxl.Workbook() + ws = wb.active + ws.append(headers) + for row in rows: + ws.append([row.get(h, "") for h in headers]) + buf = io.BytesIO() + wb.save(buf) + buf.seek(0) + data = buf.read() + wb.close() + return data + + +# ─── Field mapping ─────────────────────────────────────────────────────────── + +def map_fields(row: dict, mapping: dict[str, str]) -> dict: + """Map source columns to target fields. + + Args: + row: Source row dict with original column names. + mapping: Dict of source_column -> target_field. + If a source column is not in the row, it is skipped. + + Returns: + New dict with target field names as keys. + """ + result: dict[str, Any] = {} + for source_col, target_field in mapping.items(): + if source_col in row: + result[target_field] = row[source_col] + return result + + +def suggest_mapping(source_columns: list[str], target_fields: list[str]) -> dict[str, str]: + """Auto-suggest a mapping from source columns to target fields. + + Uses case-insensitive exact match and common alias matching. + """ + # Common aliases: source column variants -> canonical target + aliases: dict[str, str] = { + "first_name": "firstname", + "last_name": "surname", + "last_name": "surname", + "email": "email", + "email_address": "email", + "phone": "phone", + "mobile": "mobile", + "phone_2": "mobile", + "position": "function", + "title": "function", + "job_title": "function", + "company": "name", + "company_name": "name", + "organization": "name", + "web": "website", + "url": "website", + "homepage": "website", + "zip": "postalcode", + "postal_code": "postalcode", + "zip_code": "postalcode", + "city": "city", + "country": "country", + "department": "department", + "industry": "industry", + } + + target_lower = {f.lower(): f for f in target_fields} + mapping: dict[str, str] = {} + for col in source_columns: + col_lower = col.lower().strip() + # Direct match + if col_lower in target_lower: + mapping[col] = target_lower[col_lower] + continue + # Alias match + alias_target = aliases.get(col_lower) + if alias_target and alias_target in target_lower: + mapping[col] = target_lower[alias_target] + continue + # Fuzzy: target contains source or vice versa + for tl, tf in target_lower.items(): + if col_lower in tl or tl in col_lower: + mapping[col] = tf + break + return mapping + + +# ─── Validation ────────────────────────────────────────────────────────────── + +# Basic email regex (pragmatic, not RFC-perfect) +_EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$") + + +def validate_row( + row: dict, + required: list[str], + validators: dict[str, Any] | None = None, +) -> list[str]: + """Validate a single row against required fields and optional validators. + + Args: + row: The row dict to validate. + required: List of field names that must be non-empty. + validators: Optional dict of field_name -> validator spec. + Supported specs: + - {"type": "email"} — must match email pattern + - {"type": "url"} — must look like a URL + - {"type": "int"} — must be parseable as integer + - {"max_length": N} — string length <= N + - {"min_length": N} — string length >= N + + Returns: + List of error message strings (empty if valid). + """ + errors: list[str] = [] + for field in required: + val = str(row.get(field, "")).strip() + if not val: + errors.append(f"Missing required field: {field}") + + if validators: + for field, spec in validators.items(): + val = str(row.get(field, "")).strip() + if not val: + continue # Skip empty optional fields + vtype = spec.get("type") + if vtype == "email" and not _EMAIL_RE.match(val): + errors.append(f"Invalid email format in field: {field}") + elif vtype == "url" and not (val.startswith("http://") or val.startswith("https://")): + errors.append(f"Invalid URL format in field: {field}") + elif vtype == "int": + try: + int(val) + except ValueError: + errors.append(f"Field {field} must be an integer") + max_len = spec.get("max_length") + if max_len and len(val) > max_len: + errors.append(f"Field {field} exceeds max length {max_len}") + min_len = spec.get("min_length") + if min_len and len(val) < min_len: + errors.append(f"Field {field} below min length {min_len}") + return errors + + +# ─── Error report ──────────────────────────────────────────────────────────── + +def build_error_report(errors: list[dict]) -> dict: + """Build a structured error report from a list of per-row errors. + + Args: + errors: List of dicts, each with keys: + - row: int (1-based row number) + - field: str (optional, which field failed) + - message: str (error description) + + Returns: + Dict with: + - total_errors: int + - errors: list of the input dicts (normalized) + """ + normalized: list[dict] = [] + for e in errors: + normalized.append({ + "row": e.get("row", 0), + "field": e.get("field", ""), + "message": e.get("message", e.get("error", "")), + }) + return { + "total_errors": len(normalized), + "errors": normalized, + } + + +def build_import_result( + total: int, + succeeded: int, + failed: int, + errors: list[dict] | None = None, + created: list[dict] | None = None, + dry_run: bool = False, +) -> dict[str, Any]: + """Build a standardized import result dict with error report.""" + error_report = build_error_report(errors or []) + status = "success" + if failed > 0 and succeeded > 0: + status = "partial_success" + elif failed > 0 and succeeded == 0: + status = "failed" + return { + "total": total, + "succeeded": succeeded, + "failed": failed, + "status": status, + "dry_run": dry_run, + "error_report": error_report, + "created": created or [], + } + + +# ─── Format detection ──────────────────────────────────────────────────────── + +def detect_format(filename: str, content: bytes) -> str: + """Detect file format from filename extension and content magic bytes.""" + name_lower = filename.lower() + if name_lower.endswith(".csv"): + return "csv" + if name_lower.endswith(".json"): + return "json" + if name_lower.endswith(".xlsx") or name_lower.endswith(".xls"): + return "xlsx" + # Magic bytes + if content.startswith(b"PK\x03\x04"): + return "xlsx" # XLSX is a ZIP archive + if content.lstrip().startswith(b"[") or content.lstrip().startswith(b"{"): + return "json" + return "csv" # Default fallback + + +def parse_file(filename: str, content: bytes) -> list[dict[str, str]]: + """Auto-detect format and parse file content into list of dicts.""" + fmt = detect_format(filename, content) + if fmt == "csv": + return parse_csv(content) + if fmt == "json": + return parse_json(content) + if fmt == "xlsx": + return parse_xlsx(content) + raise ValueError(f"Unsupported file format: {fmt}") diff --git a/app/services/import_export_jobs.py b/app/services/import_export_jobs.py new file mode 100644 index 0000000..e98522c --- /dev/null +++ b/app/services/import_export_jobs.py @@ -0,0 +1,165 @@ +"""ARQ background jobs for large import operations. + +When an import file exceeds BACKGROUND_JOB_THRESHOLD rows, the import is +enqueued as an ARQ job instead of running synchronously. Job status is +tracked in Redis with the key pattern 'leocrm:import_job:{job_id}'. +""" + +from __future__ import annotations + +import json +import logging +import uuid +from typing import Any + +from app.core.job_registry import register_job +from app.services import import_export_service + +logger = logging.getLogger(__name__) + +_JOB_KEY_PREFIX = "leocrm:import_job" +_JOB_TTL = 3600 # 1 hour + + +def _job_key(job_id: str) -> str: + return f"{_JOB_KEY_PREFIX}:{job_id}" + + +async def _set_job_status( + redis_client: Any, + job_id: str, + status: str, + **extra: Any, +) -> None: + """Store job status in Redis with TTL.""" + data = {"status": status, **extra} + await redis_client.set(_job_key(job_id), json.dumps(data, default=str), ex=_JOB_TTL) + + +async def _get_job_status(redis_client: Any, job_id: str) -> dict[str, Any] | None: + """Retrieve job status from Redis.""" + raw = await redis_client.get(_job_key(job_id)) + if raw is None: + return None + if isinstance(raw, bytes): + raw = raw.decode("utf-8") + return json.loads(raw) + + +async def import_background_job(ctx: dict[str, Any], job_id: str, params: dict[str, Any]) -> dict[str, Any]: + """ARQ job function: run import in background with status tracking. + + Args: + ctx: ARQ worker context. + job_id: Unique job identifier. + params: Dict with keys: + - entity_type: 'contacts' or 'companies' + - csv_content: File content as string + - tenant_id: Tenant UUID string + - user_id: User UUID string + - field_mapping: Optional dict[str, str] + """ + from app.core.auth import get_redis + from app.core.db import get_worker_session_factory + from app.core.db import set_tenant_context + + redis_client = get_redis() + entity_type = params["entity_type"] + csv_content = params["csv_content"] + tenant_id = uuid.UUID(params["tenant_id"]) + user_id = uuid.UUID(params["user_id"]) + field_mapping = params.get("field_mapping") + + # Set initial status + await _set_job_status(redis_client, job_id, "processing", progress=0, total=0) + + factory = get_worker_session_factory() + try: + async with factory() as db: + # Set tenant context for RLS + await set_tenant_context(db, tenant_id) + + result = await import_export_service.import_csv( + db, + tenant_id, + user_id, + csv_content, + entity_type=entity_type, + dry_run=False, + field_mapping=field_mapping, + ) + + # Determine final status + status = result.get("status", "completed") + if status == "success": + final_status = "completed" + elif status == "partial_success": + final_status = "partial_success" + else: + final_status = "failed" + + await _set_job_status( + redis_client, + job_id, + final_status, + progress=100, + total=result.get("total", 0), + succeeded=result.get("succeeded", 0), + failed=result.get("failed", 0), + error_report=result.get("error_report", {}), + created_count=len(result.get("created", [])), + ) + return result + + except Exception as exc: + logger.error("Import background job %s failed: %s", job_id, exc, exc_info=True) + await _set_job_status( + redis_client, + job_id, + "failed", + progress=0, + total=0, + error=str(exc), + ) + raise + + +# Register the job so it appears in WorkerSettings.functions +register_job("import_background", import_background_job) + + +async def get_import_job_status(job_id: str) -> dict[str, Any] | None: + """Public helper to get job status from Redis.""" + from app.core.auth import get_redis + + redis_client = get_redis() + return await _get_job_status(redis_client, job_id) + + +async def create_import_job( + entity_type: str, + csv_content: str, + tenant_id: uuid.UUID, + user_id: uuid.UUID, + field_mapping: dict[str, str] | None = None, +) -> str: + """Enqueue an import as ARQ background job and return job_id.""" + from app.core.jobs import enqueue_job + + job_id = str(uuid.uuid4()) + params = { + "entity_type": entity_type, + "csv_content": csv_content, + "tenant_id": str(tenant_id), + "user_id": str(user_id), + "field_mapping": field_mapping, + } + + # Pre-set status as pending + from app.core.auth import get_redis + + redis_client = get_redis() + await _set_job_status(redis_client, job_id, "pending", progress=0, total=0) + + await enqueue_job("import_background", job_id, params) + return job_id diff --git a/app/services/import_export_service.py b/app/services/import_export_service.py index 7d9d55a..d38b90c 100644 --- a/app/services/import_export_service.py +++ b/app/services/import_export_service.py @@ -1,4 +1,8 @@ -"""Import/export service — CSV import with dry-run preview, CSV/XLSX export. +"""Import/export service — CSV/JSON/XLSX import with dry-run preview, partial-failure, CSV/XLSX export. + +Uses shared helpers from import_export_helpers.py for parsing, writing, mapping, +and validation. Implements partial-failure semantics: valid rows are committed +while invalid rows are collected into a structured error report. Uses unified Contact model fields: firstname, surname, email_1, phone_1, phone_2, function. Company import creates Contact with type='company'. @@ -6,8 +10,7 @@ Company import creates Contact with type='company'. from __future__ import annotations -import csv -import io +import logging import uuid from typing import Any @@ -17,203 +20,322 @@ from sqlalchemy.ext.asyncio import AsyncSession from app.core.audit import log_audit from app.core.visibility import apply_visibility_filter from app.models.contact import Contact +from app.services import import_export_helpers as helpers from app.services.contact_service import _serialize_contact as _contact_to_dict +logger = logging.getLogger(__name__) + # Expected CSV columns for each entity type # Company import creates Contact with type='company' using name field COMPANY_COLUMNS = ["name", "industry", "phone", "email", "website"] # Contact import uses unified Contact fields CONTACT_COLUMNS = ["firstname", "surname", "email", "phone", "mobile", "function", "department"] +# Target fields for mapping suggestions +CONTACT_TARGET_FIELDS = ["firstname", "surname", "email", "phone", "mobile", "function", "department"] +COMPANY_TARGET_FIELDS = ["name", "industry", "phone", "email", "website"] -def _parse_csv(content: str) -> list[dict[str, str]]: - """Parse CSV content into list of dicts.""" - reader = csv.DictReader(io.StringIO(content)) - return [dict(row) for row in reader] +# Validators for each entity type +CONTACT_VALIDATORS: dict[str, dict] = { + "email": {"type": "email"}, +} +COMPANY_VALIDATORS: dict[str, dict] = { + "email": {"type": "email"}, + "website": {"type": "url"}, +} + +# Threshold for background job processing (rows above this go to ARQ) +BACKGROUND_JOB_THRESHOLD = 1000 + + +def _parse_csv(content: str | bytes) -> list[dict[str, str]]: + """Parse file content into list of dicts (delegates to helpers with format detection). + + Despite the name, this handles CSV, JSON, and XLSX formats by detecting + from content. This maintains backward compatibility with existing callers. + """ + if isinstance(content, str): + content = content.encode("utf-8") + # Detect format from content magic bytes + fmt = helpers.detect_format("upload", content) + if fmt == "json": + return helpers.parse_json(content) + elif fmt == "xlsx": + return helpers.parse_xlsx(content) + return helpers.parse_csv(content) def _validate_row(row: dict[str, str], required: list[str]) -> list[str]: - """Validate a single row. Returns list of error messages (empty if valid).""" - errors = [] - for col in required: - val = row.get(col, "").strip() - if not val: - errors.append(f"Missing required field: {col}") - return errors + """Validate a single row (delegates to helpers for backward compat).""" + return helpers.validate_row(row, required) + + +def _normalize_contact_row(row: dict[str, str]) -> dict[str, str]: + """Normalize contact row to unified field names.""" + firstname = (row.get("firstname") or row.get("first_name") or "").strip() + surname = (row.get("surname") or row.get("last_name") or "").strip() + row["firstname"] = firstname + row["surname"] = surname + # Map old column names to unified ones + if "email" not in row and "email_address" in row: + row["email"] = row["email_address"] + if "mobile" not in row and "phone_2" in row: + row["mobile"] = row["phone_2"] + if "function" not in row and "position" in row: + row["function"] = row["position"] + return row + + +def _normalize_company_row(row: dict[str, str]) -> dict[str, str]: + """Normalize company row to unified field names.""" + name = (row.get("name") or row.get("company") or row.get("company_name") or "").strip() + row["name"] = name + if "email" not in row and "email_address" in row: + row["email"] = row["email_address"] + if "website" not in row and "url" in row: + row["website"] = row["url"] + if "website" not in row and "homepage" in row: + row["website"] = row["homepage"] + return row async def import_companies( db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, - csv_content: str, + csv_content: str | bytes, dry_run: bool = False, + field_mapping: dict[str, str] | None = None, ) -> dict[str, Any]: """Import companies from CSV as Contact with type='company'. Uses unified Contact model: name field for company name, email_1/phone_1 for contact info. + Implements partial-failure: valid rows committed, invalid rows collected in error report. """ rows = _parse_csv(csv_content) total = len(rows) - valid_rows = [] - errors = [] + errors: list[dict] = [] + valid_rows: list[dict] = [] for idx, row in enumerate(rows, start=1): - row_errors = _validate_row(row, ["name"]) + # Apply field mapping if provided + if field_mapping: + row = helpers.map_fields(row, field_mapping) + row = _normalize_company_row(row) + row_errors = helpers.validate_row(row, ["name"], COMPANY_VALIDATORS) if row_errors: for e in row_errors: - errors.append({"row": idx, "error": e}) + errors.append({"row": idx, "field": "", "message": e}) else: valid_rows.append(row) + # Count unique failed rows (a row may have multiple errors) + failed_row_count = len({e["row"] for e in errors}) + if dry_run: - return { - "total": total, - "valid": len(valid_rows), - "invalid": len(errors), - "errors": errors, - "created": [], - "dry_run": True, - } - - created = [] - for row in valid_rows: - contact = Contact( - tenant_id=tenant_id, - type="company", - name=row["name"].strip(), - displayname=row["name"].strip(), - email_1=row.get("email", "").strip() or None, - phone_1=row.get("phone", "").strip() or None, - website=row.get("website", "").strip() or None, - owner_id=user_id, - created_by=user_id, - updated_by=user_id, + return helpers.build_import_result( + total=total, + succeeded=len(valid_rows), + failed=failed_row_count, + errors=errors, + dry_run=True, ) - db.add(contact) - await db.flush() - await log_audit( - db, - tenant_id, - user_id, - "import", - "contact", - contact.id, - changes={"name": contact.name, "type": "company"}, - ) - created.append(_contact_to_dict(contact)) - return { - "total": total, - "valid": len(valid_rows), - "invalid": len(errors), - "errors": errors, - "created": created, - "dry_run": False, - } + created: list[dict] = [] + failed: list[dict] = [] + succeeded = 0 + + for idx, row in enumerate(valid_rows, start=1): + try: + contact = Contact( + tenant_id=tenant_id, + type="company", + name=row["name"].strip(), + displayname=row["name"].strip(), + email_1=row.get("email", "").strip() or None, + phone_1=row.get("phone", "").strip() or None, + website=row.get("website", "").strip() or None, + owner_id=user_id, + created_by=user_id, + updated_by=user_id, + ) + db.add(contact) + await db.flush() + await log_audit( + db, + tenant_id, + user_id, + "import", + "contact", + contact.id, + changes={"name": contact.name, "type": "company"}, + ) + created.append(_contact_to_dict(contact)) + succeeded += 1 + except Exception as exc: + logger.warning("Import companies: row %d failed: %s", idx, exc) + await db.rollback() + failed.append({"row": idx, "field": "", "message": str(exc)}) + + if succeeded > 0: + try: + await db.commit() + except Exception as exc: + logger.error("Import companies: commit failed: %s", exc) + await db.rollback() + return helpers.build_import_result( + total=total, + succeeded=0, + failed=len(valid_rows), + errors=errors + [{"row": 0, "field": "", "message": f"Commit failed: {exc}"}], + created=[], + ) + + # Count unique failed rows from validation errors + runtime failures + all_failed_rows = {e["row"] for e in errors} | {f["row"] for f in failed} + return helpers.build_import_result( + total=total, + succeeded=succeeded, + failed=len(all_failed_rows), + errors=errors + failed, + created=created, + ) async def import_contacts( db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, - csv_content: str, + csv_content: str | bytes, dry_run: bool = False, + field_mapping: dict[str, str] | None = None, ) -> dict[str, Any]: """Import contacts from CSV using unified Contact model fields. CSV columns: firstname, surname, email, phone, mobile, function, department. Maps to Contact fields: firstname, surname, email_1, phone_1, phone_2, function. + Implements partial-failure: valid rows committed, invalid rows collected in error report. """ rows = _parse_csv(csv_content) total = len(rows) - valid_rows = [] - errors = [] + errors: list[dict] = [] + valid_rows: list[dict] = [] for idx, row in enumerate(rows, start=1): - # Accept both old (first_name/last_name) and new (firstname/surname) column names - firstname = (row.get("firstname") or row.get("first_name") or "").strip() - surname = (row.get("surname") or row.get("last_name") or "").strip() - if not firstname and not surname: - errors.append({"row": idx, "error": "Missing required field: firstname or surname"}) + # Apply field mapping if provided + if field_mapping: + row = helpers.map_fields(row, field_mapping) + row = _normalize_contact_row(row) + # Validate: at least one of firstname or surname required + if not row["firstname"] and not row["surname"]: + errors.append({"row": idx, "field": "", "message": "Missing required field: firstname or surname"}) + continue + row_errors = helpers.validate_row(row, [], CONTACT_VALIDATORS) + if row_errors: + for e in row_errors: + errors.append({"row": idx, "field": "", "message": e}) else: - # Normalize row to use unified field names - row["firstname"] = firstname - row["surname"] = surname valid_rows.append(row) + # Count unique failed rows (a row may have multiple errors) + failed_row_count = len({e["row"] for e in errors}) + if dry_run: - return { - "total": total, - "valid": len(valid_rows), - "invalid": len(errors), - "errors": errors, - "created": [], - "dry_run": True, - } - - created = [] - for row in valid_rows: - contact = Contact( - tenant_id=tenant_id, - type="person", - firstname=row["firstname"].strip() or None, - surname=row["surname"].strip() or None, - displayname=f"{row['firstname']} {row['surname']}", - email_1=row.get("email", "").strip() or None, - phone_1=row.get("phone", "").strip() or None, - phone_2=row.get("mobile", "").strip() or None, - owner_id=user_id, - created_by=user_id, - updated_by=user_id, + return helpers.build_import_result( + total=total, + succeeded=len(valid_rows), + failed=failed_row_count, + errors=errors, + dry_run=True, ) - db.add(contact) - await db.flush() - await log_audit( - db, - tenant_id, - user_id, - "import", - "contact", - contact.id, - changes={"firstname": contact.firstname, "surname": contact.surname}, - ) - created.append(_contact_to_dict(contact)) - return { - "total": total, - "valid": len(valid_rows), - "invalid": len(errors), - "errors": errors, - "created": created, - "dry_run": False, - } + created: list[dict] = [] + failed: list[dict] = [] + succeeded = 0 + + for idx, row in enumerate(valid_rows, start=1): + try: + contact = Contact( + tenant_id=tenant_id, + type="person", + firstname=row["firstname"].strip() or None, + surname=row["surname"].strip() or None, + displayname=f"{row['firstname']} {row['surname']}".strip(), + email_1=row.get("email", "").strip() or None, + phone_1=row.get("phone", "").strip() or None, + phone_2=row.get("mobile", "").strip() or None, + owner_id=user_id, + created_by=user_id, + updated_by=user_id, + ) + db.add(contact) + await db.flush() + await log_audit( + db, + tenant_id, + user_id, + "import", + "contact", + contact.id, + changes={"firstname": contact.firstname, "surname": contact.surname}, + ) + created.append(_contact_to_dict(contact)) + succeeded += 1 + except Exception as exc: + logger.warning("Import contacts: row %d failed: %s", idx, exc) + await db.rollback() + failed.append({"row": idx, "field": "", "message": str(exc)}) + + if succeeded > 0: + try: + await db.commit() + except Exception as exc: + logger.error("Import contacts: commit failed: %s", exc) + await db.rollback() + return helpers.build_import_result( + total=total, + succeeded=0, + failed=len(valid_rows), + errors=errors + [{"row": 0, "field": "", "message": f"Commit failed: {exc}"}], + created=[], + ) + + # Count unique failed rows from validation errors + runtime failures + all_failed_rows = {e["row"] for e in errors} | {f["row"] for f in failed} + return helpers.build_import_result( + total=total, + succeeded=succeeded, + failed=len(all_failed_rows), + errors=errors + failed, + created=created, + ) async def import_csv( db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, - csv_content: str, + csv_content: str | bytes, entity_type: str, dry_run: bool = False, + field_mapping: dict[str, str] | None = None, ) -> dict[str, Any]: """Generic CSV import dispatcher based on entity_type ('companies' or 'contacts').""" if entity_type == "companies": - return await import_companies(db, tenant_id, user_id, csv_content, dry_run=dry_run) + return await import_companies(db, tenant_id, user_id, csv_content, dry_run=dry_run, field_mapping=field_mapping) elif entity_type == "contacts": - return await import_contacts(db, tenant_id, user_id, csv_content, dry_run=dry_run) + return await import_contacts(db, tenant_id, user_id, csv_content, dry_run=dry_run, field_mapping=field_mapping) else: - return { - "total": 0, - "valid": 0, - "invalid": 0, - "errors": [{"row": 0, "error": f"Unknown entity_type: {entity_type}"}], - "created": [], - "dry_run": dry_run, - } + return helpers.build_import_result( + total=0, + succeeded=0, + failed=1, + errors=[{"row": 0, "field": "", "message": f"Unknown entity_type: {entity_type}"}], + ) +# ─── Export functions ──────────────────────────────────────────────────────── + async def export_contacts_csv( db: AsyncSession, tenant_id: uuid.UUID, @@ -236,28 +358,24 @@ async def export_contacts_csv( result = await db.execute(q) contacts = result.scalars().all() - output = io.StringIO() - writer = csv.writer(output) - writer.writerow( - ["id", "type", "firstname", "surname", "name", "email", "phone", "mobile", "city", "postalcode", "country"] - ) - for c in contacts: - writer.writerow( - [ - str(c.id), - c.type or "person", - c.firstname or "", - c.surname or "", - c.name or "", - c.email_1 or "", - c.phone_1 or "", - c.phone_2 or "", - c.mailing_city or "", - c.mailing_postalcode or "", - c.mailing_country or "", - ] - ) - return output.getvalue() + headers = ["id", "type", "firstname", "surname", "name", "email", "phone", "mobile", "city", "postalcode", "country"] + rows = [ + { + "id": str(c.id), + "type": c.type or "person", + "firstname": c.firstname or "", + "surname": c.surname or "", + "name": c.name or "", + "email": c.email_1 or "", + "phone": c.phone_1 or "", + "mobile": c.phone_2 or "", + "city": c.mailing_city or "", + "postalcode": c.mailing_postalcode or "", + "country": c.mailing_country or "", + } + for c in contacts + ] + return helpers.write_csv(rows, headers).decode("utf-8") async def export_companies_csv( @@ -283,23 +401,97 @@ async def export_companies_csv( result = await db.execute(q) companies = result.scalars().all() - output = io.StringIO() - writer = csv.writer(output) - writer.writerow( - ["id", "type", "name", "email", "phone", "website", "city", "postalcode", "country"] + headers = ["id", "type", "name", "email", "phone", "website", "city", "postalcode", "country"] + rows = [ + { + "id": str(c.id), + "type": c.type or "company", + "name": c.name or "", + "email": c.email_1 or "", + "phone": c.phone_1 or "", + "website": c.website or "", + "city": c.mailing_city or "", + "postalcode": c.mailing_postalcode or "", + "country": c.mailing_country or "", + } + for c in companies + ] + return helpers.write_csv(rows, headers).decode("utf-8") + + +# ─── Preview & Validate ────────────────────────────────────────────────────── + +def preview_import( + filename: str, + content: bytes, + entity_type: str, +) -> dict[str, Any]: + """Parse a file and return preview data (first 10 rows) + mapping suggestion. + + Does NOT touch the database. + """ + rows = helpers.parse_file(filename, content) + columns = list(rows[0].keys()) if rows else [] + preview_rows = rows[:10] + + if entity_type == "contacts": + target_fields = CONTACT_TARGET_FIELDS + elif entity_type == "companies": + target_fields = COMPANY_TARGET_FIELDS + else: + target_fields = [] + + mapping_suggestion = helpers.suggest_mapping(columns, target_fields) + + return { + "total_rows": len(rows), + "columns": columns, + "preview_rows": preview_rows, + "mapping_suggestion": mapping_suggestion, + "target_fields": target_fields, + } + + +def validate_import( + filename: str, + content: bytes, + entity_type: str, + field_mapping: dict[str, str] | None = None, +) -> dict[str, Any]: + """Validate all rows against mapping and return error report without importing.""" + rows = helpers.parse_file(filename, content) + total = len(rows) + errors: list[dict] = [] + valid_count = 0 + + for idx, row in enumerate(rows, start=1): + if field_mapping: + row = helpers.map_fields(row, field_mapping) + if entity_type == "contacts": + row = _normalize_contact_row(row) + if not row["firstname"] and not row["surname"]: + errors.append({"row": idx, "field": "", "message": "Missing required field: firstname or surname"}) + continue + row_errors = helpers.validate_row(row, [], CONTACT_VALIDATORS) + elif entity_type == "companies": + row = _normalize_company_row(row) + row_errors = helpers.validate_row(row, ["name"], COMPANY_VALIDATORS) + else: + errors.append({"row": idx, "field": "", "message": f"Unknown entity_type: {entity_type}"}) + continue + + if row_errors: + for e in row_errors: + errors.append({"row": idx, "field": "", "message": e}) + else: + valid_count += 1 + + # Count unique failed rows + failed_row_count = len({e["row"] for e in errors}) + return helpers.build_import_result( + total=total, + succeeded=valid_count, + failed=failed_row_count, + errors=errors, + dry_run=True, ) - for c in companies: - writer.writerow( - [ - str(c.id), - c.type or "company", - c.name or "", - c.email_1 or "", - c.phone_1 or "", - c.website or "", - c.mailing_city or "", - c.mailing_postalcode or "", - c.mailing_country or "", - ] - ) - return output.getvalue() diff --git a/docs/plugin-development-guide.md b/docs/plugin-development-guide.md index 06de4ad..a8027ee 100644 --- a/docs/plugin-development-guide.md +++ b/docs/plugin-development-guide.md @@ -2092,4 +2092,113 @@ Plugins deklarieren ihre API-Prefixe im Manifest (`routes.prefix`). Plugin-API- --- +## 31. Import/Export Handler auf Shared Helpers + +LeoCRM stellt wiederverwendbare Bausteine für Import/Export-Funktionalität bereit. Plugins können eigene Import/Export-Handler auf dieser Basis aufsetzen. + +### 31.1 Shared Helpers (`app/services/import_export_helpers.py`) + +Die folgenden Funktionen sind verfügbar und können von jedem Plugin importiert werden: + +```python +from app.services.import_export_helpers import ( + parse_csv, # CSV → list[dict] mit Encoding-Detection + parse_json, # JSON → list[dict] + parse_xlsx, # XLSX → list[dict] (openpyxl) + write_csv, # list[dict] → CSV bytes + write_json, # list[dict] → JSON bytes + write_xlsx, # list[dict] → XLSX bytes + map_fields, # Source-Column → Target-Field Mapping + suggest_mapping, # Auto-Mapping-Vorschlag + validate_row, # Zeilen-Validierung mit required + validators + build_error_report, # Strukturierter Fehler-Report + build_import_result, # Standardisiertes Import-Ergebnis + detect_format, # Format-Erkennung (csv/json/xlsx) + parse_file, # Auto-Detect + Parse +) +``` + +### 31.2 Eigener Import-Handler + +Ein Plugin kann einen eigenen Import-Handler erstellen: + +```python +from app.services.import_export_helpers import parse_file, map_fields, validate_row, build_import_result + +async def import_my_entity(db, tenant_id, user_id, content: bytes, filename: str, field_mapping: dict | None = None): + rows = parse_file(filename, content) + total = len(rows) + errors = [] + valid_rows = [] + + for idx, row in enumerate(rows, start=1): + if field_mapping: + row = map_fields(row, field_mapping) + row_errors = validate_row(row, required=["name"], validators={"email": {"type": "email"}}) + if row_errors: + for e in row_errors: + errors.append({"row": idx, "field": "", "message": e}) + else: + valid_rows.append(row) + + # ... DB-Insert mit Partial-Failure ... + for idx, row in enumerate(valid_rows, start=1): + try: + # Insert entity + pass + except Exception as exc: + errors.append({"row": idx, "field": "", "message": str(exc)}) + + failed_row_count = len({e["row"] for e in errors}) + return build_import_result( + total=total, + succeeded=len(valid_rows), + failed=failed_row_count, + errors=errors, + ) +``` + +### 31.3 Partial-Failure-Semantik + +Import-Handler müssen Partial-Failure implementieren: + +1. **Validierung pro Zeile**: `validate_row()` prüft required fields und validators +2. **Fehlerhafte Zeilen sammeln**: Fehler werden in `errors`-Liste mit `{row, field, message}` gesammelt +3. **Erfolgreiche Zeilen committen**: Gültige Zeilen werden in DB geschrieben, pro Zeile try/except +4. **Status-Klassifizierung**: `build_import_result()` setzt Status auf `success`, `partial_success`, oder `failed` +5. **Fehler-Report**: `build_error_report()` erstellt strukturierten Report mit `total_errors` und `errors`-Liste + +### 31.4 Background Processing für große Imports + +Für Dateien > 1000 Zeilen soll der Import als ARQ-Job laufen: + +```python +from app.services.import_export_jobs import create_import_job, get_import_job_status + +# In Route: +if len(rows) > 1000: + job_id = await create_import_job( + entity_type="my_entity", + csv_content=content.decode("utf-8"), + tenant_id=tenant_id, + user_id=user_id, + field_mapping=mapping, + ) + return {"status": "pending", "job_id": job_id} +``` + +Job-Status wird in Redis gespeichert (`leocrm:import_job:{job_id}`) mit TTL 1h. + +### 31.5 Frontend-Integration + +Das Frontend nutzt die API-Client-Funktionen aus `importExport.ts`: + +- `previewImport(file, entityType)` — Vorschau mit Mapping-Vorschlag +- `validateImport(file, entityType, fieldMapping)` — Validierung ohne Import +- `importCsv(file, entityType, dryRun, fieldMapping)` — Import mit optionalem Mapping +- `getImportJobStatus(jobId)` — Polling für Background-Jobs +- `exportData(entityType, format)` — Export als CSV/XLSX/JSON + +--- + *This document is authoritative for all plugin development at LeoCRM.* diff --git a/frontend/src/api/importExport.ts b/frontend/src/api/importExport.ts index 530a10a..f0906eb 100644 --- a/frontend/src/api/importExport.ts +++ b/frontend/src/api/importExport.ts @@ -2,34 +2,87 @@ import { apiClient } from './client'; // ─── Types ────────────────────────────────────────────────────────────────── +export interface ImportError { + row: number; + field: string; + message: string; +} + +export interface ErrorReport { + total_errors: number; + errors: ImportError[]; +} + export interface ImportResult { total?: number; - created?: number; - updated?: number; + succeeded?: number; + failed?: number; + status?: 'success' | 'partial_success' | 'failed' | 'pending' | 'processing' | 'completed'; + dry_run?: boolean; + error_report?: ErrorReport; + created?: Record[]; errors?: string[]; warnings?: string[]; rows?: Record[]; preview?: boolean; + // Legacy compat + valid?: number; + invalid?: number; + job_id?: string; + message?: string; +} + +export interface PreviewResult { + total_rows: number; + columns: string[]; + preview_rows: Record[]; + mapping_suggestion: Record; + target_fields: string[]; +} + +export interface ValidateResult { + total: number; + succeeded: number; + failed: number; + status: string; + dry_run: boolean; + error_report: ErrorReport; + created: Record[]; +} + +export interface JobStatus { + status: 'pending' | 'processing' | 'completed' | 'partial_success' | 'failed'; + progress?: number; + total?: number; + succeeded?: number; + failed?: number; + error_report?: ErrorReport; + created_count?: number; + error?: string; } export type EntityType = 'companies' | 'contacts'; -export type ExportFormat = 'csv' | 'xlsx'; +export type ExportFormat = 'csv' | 'xlsx' | 'json'; // ─── Import ───────────────────────────────────────────────────────────────── /** - * Import a CSV file. When dryRun is true, a preview is returned without + * Import a CSV/JSON/XLSX file. When dryRun is true, a preview is returned without * writing to the database. */ export async function importCsv( file: File, entityType: string, - dryRun: boolean + dryRun: boolean, + fieldMapping?: Record, ): Promise { const url = dryRun ? '/import/preview' : '/import'; const formData = new FormData(); formData.append('file', file); formData.append('entity_type', entityType); + if (fieldMapping) { + formData.append('field_mapping', JSON.stringify(fieldMapping)); + } const response = await apiClient.post(url, formData, { headers: { 'Content-Type': 'multipart/form-data' }, @@ -37,14 +90,98 @@ export async function importCsv( return response.data; } +/** + * Preview a file: parse, return first 10 rows + columns + mapping suggestion. + */ +export async function previewImport( + file: File, + entityType: string, +): Promise { + const formData = new FormData(); + formData.append('file', file); + formData.append('entity_type', entityType); + + const response = await apiClient.post('/import/preview', formData, { + headers: { 'Content-Type': 'multipart/form-data' }, + }); + return response.data; +} + +/** + * Validate a file against mapping: returns error report without importing. + */ +export async function validateImport( + file: File, + entityType: string, + fieldMapping?: Record, +): Promise { + const formData = new FormData(); + formData.append('file', file); + formData.append('entity_type', entityType); + if (fieldMapping) { + formData.append('field_mapping', JSON.stringify(fieldMapping)); + } + + const response = await apiClient.post('/import/validate', formData, { + headers: { 'Content-Type': 'multipart/form-data' }, + }); + return response.data; +} + +/** + * Import contacts specifically. + */ +export async function importContacts( + file: File, + fieldMapping?: Record, +): Promise { + const formData = new FormData(); + formData.append('file', file); + if (fieldMapping) { + formData.append('field_mapping', JSON.stringify(fieldMapping)); + } + + const response = await apiClient.post('/import/contacts', formData, { + headers: { 'Content-Type': 'multipart/form-data' }, + }); + return response.data; +} + +/** + * Import companies specifically. + */ +export async function importCompanies( + file: File, + fieldMapping?: Record, +): Promise { + const formData = new FormData(); + formData.append('file', file); + if (fieldMapping) { + formData.append('field_mapping', JSON.stringify(fieldMapping)); + } + + const response = await apiClient.post('/import/companies', formData, { + headers: { 'Content-Type': 'multipart/form-data' }, + }); + return response.data; +} + +/** + * Get status of a background import job. + */ +export async function getImportJobStatus(jobId: string): Promise { + const response = await apiClient.get(`/import/status/${jobId}`); + return response.data; +} + // ─── Export ───────────────────────────────────────────────────────────────── /** - * Export data as CSV or XLSX and trigger a browser download. + * Export data as CSV, XLSX, or JSON and trigger a browser download. */ export async function exportData( entityType: string, - format: string + format: string, ): Promise { const url = `/export?entity_type=${encodeURIComponent(entityType)}&format=${encodeURIComponent(format)}`; const response = await apiClient.get(url, { responseType: 'blob' }); @@ -52,7 +189,9 @@ export async function exportData( const blob = new Blob([response.data], { type: format === 'xlsx' ? 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet' - : 'text/csv', + : format === 'json' + ? 'application/json' + : 'text/csv', }); const blobUrl = URL.createObjectURL(blob); diff --git a/frontend/src/components/import-export/ExportPanel.tsx b/frontend/src/components/import-export/ExportPanel.tsx index 88f03af..c1ecb03 100644 --- a/frontend/src/components/import-export/ExportPanel.tsx +++ b/frontend/src/components/import-export/ExportPanel.tsx @@ -18,6 +18,7 @@ const ENTITY_OPTIONS = [ const FORMAT_OPTIONS = [ { value: 'csv', label: 'CSV' }, { value: 'xlsx', label: 'XLSX' }, + { value: 'json', label: 'JSON' }, ]; // ─── Component ────────────────────────────────────────────────────────────── diff --git a/frontend/src/components/import-export/ImportWizard.tsx b/frontend/src/components/import-export/ImportWizard.tsx index c48879d..708e9ff 100644 --- a/frontend/src/components/import-export/ImportWizard.tsx +++ b/frontend/src/components/import-export/ImportWizard.tsx @@ -1,5 +1,6 @@ -import React, { useCallback, useRef, useState } from 'react'; +import React, { useCallback, useEffect, useRef, useState } from 'react'; import { useTranslation } from 'react-i18next'; +import type { TFunction } from 'i18next'; import { AlertCircle, CheckCircle, @@ -10,6 +11,9 @@ import { ArrowLeft, Eye, Play, + ShieldCheck, + Table2, + ArrowRight, } from 'lucide-react'; import clsx from 'clsx'; import { Card } from '@/components/ui/Card'; @@ -17,19 +21,35 @@ import { Button } from '@/components/ui/Button'; import { Select } from '@/components/ui/Select'; import { Badge } from '@/components/ui/Badge'; import { useToast } from '@/components/ui/Toast'; -import { importCsv, type ImportResult } from '@/api/importExport'; +import { + previewImport, + validateImport, + importCsv, + getImportJobStatus, + type PreviewResult, + type ValidateResult, + type ImportResult, + type JobStatus, + type ImportError, +} from '@/api/importExport'; // ─── Constants ────────────────────────────────────────────────────────────── const MAX_FILE_SIZE = 10 * 1024 * 1024; // 10 MB -const ACCEPTED_EXTENSIONS = ['.csv']; +const ACCEPTED_EXTENSIONS = ['.csv', '.json', '.xlsx']; +const ACCEPTED_EXTENSIONS_STR = '.csv,.json,.xlsx'; +const SKIP_VALUE = '(skip)'; +const POLL_INTERVAL_MS = 2000; +const POLL_TERMINAL_STATES: JobStatus['status'][] = ['completed', 'partial_success', 'failed']; const ENTITY_OPTIONS = [ { value: 'contacts', label: 'Kontakte' }, { value: 'companies', label: 'Firmen' }, ]; -type WizardStep = 'upload' | 'preview' | 'review' | 'result'; +type WizardStep = 'upload' | 'preview' | 'validate' | 'review' | 'result'; + +type FieldMapping = Record; // ─── Component ────────────────────────────────────────────────────────────── @@ -40,21 +60,37 @@ export function ImportWizard() { const [step, setStep] = useState('upload'); const [file, setFile] = useState(null); const [entityType, setEntityType] = useState('contacts'); - const [previewResult, setPreviewResult] = useState(null); + const [previewData, setPreviewData] = useState(null); + const [fieldMapping, setFieldMapping] = useState({}); + const [validateResult, setValidateResult] = useState(null); const [importResult, setImportResult] = useState(null); - const [previewConfirmed, setPreviewConfirmed] = useState(false); + const [jobStatus, setJobStatus] = useState(null); + const [importConfirmed, setImportConfirmed] = useState(false); const [isDragOver, setIsDragOver] = useState(false); const [isLoadingPreview, setIsLoadingPreview] = useState(false); + const [isLoadingValidation, setIsLoadingValidation] = useState(false); const [isLoadingImport, setIsLoadingImport] = useState(false); const [fileError, setFileError] = useState(null); const inputRef = useRef(null); + const pollRef = useRef | null>(null); - // ─── File validation ──────────────────────────────────────────────────── + // ─── Job polling cleanup ───────────────────────────────────────────────── + + useEffect(() => { + return () => { + if (pollRef.current) { + clearInterval(pollRef.current); + pollRef.current = null; + } + }; + }, []); + + // ─── File validation ────────────────────────────────────────────────────── const validateFile = useCallback((f: File): string | null => { const ext = f.name.toLowerCase().substring(f.name.lastIndexOf('.')); if (!ACCEPTED_EXTENSIONS.includes(ext)) { - return t('importExport.onlyCsvAllowed', 'Nur CSV-Dateien sind erlaubt'); + return t('importExport.onlyCsvAllowed', 'Nur CSV-, JSON- oder XLSX-Dateien sind erlaubt'); } if (f.size > MAX_FILE_SIZE) { return t('importExport.fileTooLarge', 'Datei ist größer als 10 MB'); @@ -96,75 +132,204 @@ export function ImportWizard() { handleFileSelect(selected); }, [handleFileSelect]); - // ─── Preview ──────────────────────────────────────────────────────────── + // ─── Preview ────────────────────────────────────────────────────────────── const handlePreview = async () => { if (!file) return; setIsLoadingPreview(true); try { - const result = await importCsv(file, entityType, true); - setPreviewResult(result); + const result = await previewImport(file, entityType); + setPreviewData(result); + // Initialize mapping from suggestion, filtering out any '(skip)' values + const initialMapping: FieldMapping = {}; + for (const [source, target] of Object.entries(result.mapping_suggestion)) { + initialMapping[source] = target; + } + // Ensure all columns have a mapping entry + for (const col of result.columns) { + if (!(col in initialMapping)) { + initialMapping[col] = SKIP_VALUE; + } + } + setFieldMapping(initialMapping); setStep('preview'); - } catch (err: any) { - toast.error(err?.message || t('importExport.previewFailed', 'Vorschau fehlgeschlagen')); + } catch (err: unknown) { + const message = err instanceof Error ? err.message : String(err); + toast.error(message || t('importExport.previewFailed', 'Vorschau fehlgeschlagen')); } finally { setIsLoadingPreview(false); } }; - // ─── Import ───────────────────────────────────────────────────────────── + // ─── Mapping change ─────────────────────────────────────────────────────── + + const handleMappingChange = useCallback((sourceCol: string, targetField: string) => { + setFieldMapping((prev) => ({ ...prev, [sourceCol]: targetField })); + }, []); + + // ─── Validation ─────────────────────────────────────────────────────────── + + const handleValidate = async () => { + if (!file) return; + setIsLoadingValidation(true); + try { + // Build clean mapping without skip entries + const cleanMapping: FieldMapping = {}; + for (const [source, target] of Object.entries(fieldMapping)) { + if (target !== SKIP_VALUE) { + cleanMapping[source] = target; + } + } + const result = await validateImport(file, entityType, Object.keys(cleanMapping).length > 0 ? cleanMapping : undefined); + setValidateResult(result); + setStep('validate'); + } catch (err: unknown) { + const message = err instanceof Error ? err.message : String(err); + toast.error(message || t('importExport.validationFailed', 'Validierung fehlgeschlagen')); + } finally { + setIsLoadingValidation(false); + } + }; + + // ─── Import ─────────────────────────────────────────────────────────────── const handleImport = async () => { if (!file) return; setIsLoadingImport(true); try { - const result = await importCsv(file, entityType, false); + const cleanMapping: FieldMapping = {}; + for (const [source, target] of Object.entries(fieldMapping)) { + if (target !== SKIP_VALUE) { + cleanMapping[source] = target; + } + } + const result = await importCsv( + file, + entityType, + false, + Object.keys(cleanMapping).length > 0 ? cleanMapping : undefined, + ); setImportResult(result); + setJobStatus(null); setStep('result'); - toast.success(t('importExport.importSuccess', 'Import erfolgreich abgeschlossen')); - } catch (err: any) { + + // If async job, start polling + if (result.job_id) { + startJobPolling(result.job_id); + } else { + const status = result.status ?? 'success'; + if (status === 'success' || status === 'completed') { + toast.success(t('importExport.importSuccess', 'Import erfolgreich abgeschlossen')); + } else if (status === 'partial_success') { + toast.warning?.(t('importExport.importPartial', 'Import teilweise erfolgreich')); + } + } + } catch (err: unknown) { + const message = err instanceof Error ? err.message : String(err); setImportResult({ - errors: [err?.message || t('importExport.importFailed', 'Import fehlgeschlagen')], + status: 'failed', + errors: [message || t('importExport.importFailed', 'Import fehlgeschlagen')], }); + setJobStatus(null); setStep('result'); } finally { setIsLoadingImport(false); } }; - // ─── Reset ────────────────────────────────────────────────────────────── + // ─── Job polling ────────────────────────────────────────────────────────── + + const startJobPolling = useCallback((jobId: string) => { + if (pollRef.current) { + clearInterval(pollRef.current); + } + + const poll = async () => { + try { + const status = await getImportJobStatus(jobId); + setJobStatus(status); + + if (POLL_TERMINAL_STATES.includes(status.status)) { + if (pollRef.current) { + clearInterval(pollRef.current); + pollRef.current = null; + } + if (status.status === 'completed') { + toast.success(t('importExport.importSuccess', 'Import erfolgreich abgeschlossen')); + } else if (status.status === 'partial_success') { + toast.warning?.(t('importExport.importPartial', 'Import teilweise erfolgreich')); + } else if (status.status === 'failed') { + toast.error(t('importExport.importFailed', 'Import fehlgeschlagen')); + } + } + } catch { + // Silently continue polling on transient errors + } + }; + + // Poll immediately, then at interval + void poll(); + pollRef.current = setInterval(() => void poll(), POLL_INTERVAL_MS); + }, [t, toast]); + + // ─── Reset ──────────────────────────────────────────────────────────────── const handleReset = () => { + if (pollRef.current) { + clearInterval(pollRef.current); + pollRef.current = null; + } setStep('upload'); setFile(null); - setPreviewResult(null); + setPreviewData(null); + setFieldMapping({}); + setValidateResult(null); setImportResult(null); - setPreviewConfirmed(false); + setJobStatus(null); + setImportConfirmed(false); setFileError(null); if (inputRef.current) inputRef.current.value = ''; }; const handleBack = () => { if (step === 'preview') setStep('upload'); - else if (step === 'review') setStep('preview'); + else if (step === 'validate') setStep('preview'); + else if (step === 'review') setStep('validate'); }; - // ─── Step indicator ───────────────────────────────────────────────────── + // ─── Step indicator ─────────────────────────────────────────────────────── const steps: { key: WizardStep; label: string }[] = [ { key: 'upload', label: t('importExport.stepUpload', 'Datei hochladen') }, - { key: 'preview', label: t('importExport.stepPreview', 'Vorschau') }, + { key: 'preview', label: t('importExport.stepPreview', 'Vorschau & Mapping') }, + { key: 'validate', label: t('importExport.stepValidate', 'Validierung') }, { key: 'review', label: t('importExport.stepReview', 'Überprüfung') }, { key: 'result', label: t('importExport.stepResult', 'Ergebnis') }, ]; const currentStepIndex = steps.findIndex((s) => s.key === step); - // ─── Render ───────────────────────────────────────────────────────────── + // ─── Build mapping select options ───────────────────────────────────────── + + const mappingSelectOptions = useCallback((): { value: string; label: string }[] => { + if (!previewData) return []; + const fieldOpts = previewData.target_fields.map((f) => ({ value: f, label: f })); + return [{ value: SKIP_VALUE, label: t('importExport.skipField', '(überspringen)') }, ...fieldOpts]; + }, [previewData, t]); + + // ─── Clean mapping for display ──────────────────────────────────────────── + + const activeMappings = (): { source: string; target: string }[] => { + return Object.entries(fieldMapping) + .filter(([, target]) => target !== SKIP_VALUE) + .map(([source, target]) => ({ source, target })); + }; + + // ─── Render ─────────────────────────────────────────────────────────────── return (
{/* Step indicator */} -
+
{steps.map((s, idx) => (
@@ -173,8 +338,9 @@ export function ImportWizard() { 'flex items-center justify-center w-8 h-8 rounded-full text-sm font-medium transition-colors', idx <= currentStepIndex ? 'bg-primary-600 text-white' - : 'bg-secondary-200 text-secondary-500' + : 'bg-secondary-200 text-secondary-500', )} + aria-current={idx === currentStepIndex ? 'step' : undefined} > {idx < currentStepIndex ? ( @@ -185,7 +351,7 @@ export function ImportWizard() {