"""Commit a staged import (imports.md §J, rebuild edition): apply the human's decisions, resolve-or-create the entity chains, stamp money, and write the batch + entries. Order preserved from the legacy action: incomplete buckets FIRST (their create chains run before the main steps, so step 1–3 find those entities as existing — B56), then operators -> clients -> projects (B58–B60), then the batch row and the 500-row chunked entry insert (B68). Rebuild deltas (all deliberate; ADR #005): - ONE database transaction. A failed chunk rolls back the entries AND the batch AND every entity created along the way (the legacy "entity debris" wart #2 is gone); the staging row survives so the user retries from the resolver (T-IMP-023 pins the visible half). - raw_operator/raw_client/raw_project store the AS-PARSED text verbatim — the legacy canonical-name rewrite (B66's second half) is moot: display for resolved rows comes from the project FK chain, and the raw_* triple is immutable provenance (02 §3). - Money stamps come from services/money/stamping.py (compensation terms in force + rate_overrides precedence) with per-side currency; NULL = never priced (judgment #19). """ from __future__ import annotations import uuid from datetime import datetime from decimal import Decimal from sqlalchemy import insert from sqlalchemy.orm import Session from app.models import ImportBatch, PendingImport, TimeEntry from app.services.entities import ( EntityGraph, resolve_or_create_client, resolve_or_create_operator, resolve_or_create_project, ) from app.services.imports.errors import ImportsError from app.services.imports.resolver import bucket_signature, compute_resolution from app.services.money.stamping import StampRates, resolve_rates, resolve_worker, stamp_entry CHUNK_SIZE = 500 # imports.md B68 def _bad(summary: str, code: str = "COMMIT_FAILED", status: int = 400) -> ImportsError: return ImportsError(status, code, summary) def _decision_kind(raw: dict | None) -> str: """Absent decision defaults to 'create' (imports.md B47/B55).""" if raw is None: return "create" return raw.get("decision") or "create" def _target_uuid(raw: dict | None) -> uuid.UUID | None: value = (raw or {}).get("target_id") if not value: return None try: return uuid.UUID(str(value)) except ValueError: return None def _parse_naive(value: str | None) -> datetime | None: if not value: return None return datetime.strptime(value, "%Y-%m-%d %H:%M:%S") def _insert_chunk(session: Session, rows: list[dict]) -> None: """Module-level so tests can inject a failing chunk (T-IMP-023).""" session.execute(insert(TimeEntry), rows) def _process_buckets( session: Session, graph: EntityGraph, bucket_infos: list[dict], bucket_decisions: list[dict], ) -> dict[str, uuid.UUID | None]: """B56: map (empty target -> silently skip) / create (all three fields required; chains run resolve-or-create inline) / skip. Returns signature -> target project id (None = rows stay Blocked). A bucket with no decision at all is a skip — legacy parity (the B66 signature lookup simply misses and the rows import Blocked).""" decisions_by_key = { bucket_signature(d.get("operator"), d.get("client"), d.get("project")): d for d in bucket_decisions } targets: dict[str, uuid.UUID | None] = {} for info in bucket_infos: key = info["key"] decision = decisions_by_key.get(key) if decision is None: targets[key] = None continue kind = decision.get("decision") or "create" if kind == "map": target = None raw_target = decision.get("target_project_id") if raw_target: try: target = uuid.UUID(str(raw_target)) except ValueError: target = None if target is not None and target not in graph.project_chain: raise _bad("Mapped project id not found.", "BAD_MAPPING") # empty selection is silently a skip (B56 / T-IMP-020) targets[key] = target continue if kind == "create": op_name = (decision.get("create_operator") or info.get("operator") or "").strip() cl_name = (decision.get("create_client") or info.get("client") or "").strip() pr_name = (decision.get("create_project") or info.get("project") or "").strip() if not (op_name and cl_name and pr_name): raise _bad( "Create-new requires operator + client + project for the" f' bucket "{key}".', "BUCKET_CREATE_INCOMPLETE", ) op_id = resolve_or_create_operator(session, graph, op_name) cl_id = resolve_or_create_client(session, graph, op_id, cl_name) targets[key] = resolve_or_create_project(session, graph, op_id, cl_id, pr_name) continue # skip (or anything else) targets[key] = None return targets def commit_pending( session: Session, actor, pending_id: uuid.UUID, decisions: dict ) -> dict: """The whole commit — one transaction; raises ImportsError (with the pinned message) after rollback on any failure.""" from app.services.imports import staging staging.require_can_manage_imports(actor) pending = staging.get_pending(session, actor, pending_id) if not staging.can_commit(actor, pending): raise _bad("You don't have permission to commit this import.", "NOT_ALLOWED", 403) entries: list[dict] = pending.parsed_entries or [] try: return _commit_inner(session, actor, pending, entries, decisions) except ImportsError: session.rollback() raise except Exception: session.rollback() raise def _commit_inner( session: Session, actor, pending: PendingImport, entries: list[dict], decisions: dict ) -> dict: tenant_id = pending.tenant_id res = compute_resolution(session, tenant_id, entries) graph = res.graph op_decisions = {d.get("name"): d for d in decisions.get("operators", [])} cl_decisions = { (d.get("operator"), d.get("name")): d for d in decisions.get("clients", []) } pr_decisions = { (d.get("operator"), d.get("client"), d.get("name")): d for d in decisions.get("projects", []) } # --- incomplete buckets FIRST (B56) --------------------------------- bucket_targets = _process_buckets( session, graph, res.buckets, decisions.get("buckets", []) ) # --- step 1: operators (B58) ---------------------------------------- op_resolved: dict[str, uuid.UUID] = {} op_names = sorted({e["operator"] for e in entries if e.get("operator")}) for name in op_names: existing = graph.find_operator(name) # existing match WINS over any decision if existing is not None: op_resolved[name] = existing continue decision = op_decisions.get(name) if decision is None: # legacy required an explicit decision for every unknown name raise _bad( f'No decision provided for unknown operator "{name}".' " Please re-submit the resolver.", "MISSING_DECISION", ) kind = _decision_kind(decision) if kind == "map": target = _target_uuid(decision) if target is None or target not in graph.operators: raise _bad("Mapped operator id not found.", "BAD_MAPPING") op_resolved[name] = target else: # create op_resolved[name] = resolve_or_create_operator(session, graph, name) # --- step 2: clients (B59) ------------------------------------------ cl_resolved: dict[tuple[str, str], uuid.UUID] = {} pairs = sorted( { (e["operator"], e["client"]) for e in entries if e.get("operator") and e.get("client") } ) for op_name, cl_name in pairs: op_id = op_resolved.get(op_name) if op_id is None: raise _bad( f'Operator "{op_name}" wasn\'t resolved (internal bug).' " Re-submit the resolver.", "INTERNAL", 500, ) existing = graph.find_client(op_id, cl_name) if existing is not None: cl_resolved[(op_name, cl_name)] = existing continue decision = cl_decisions.get((op_name, cl_name)) if decision is None: raise _bad( f'No decision provided for unknown client "{cl_name}".' " Please re-submit the resolver.", "MISSING_DECISION", ) kind = _decision_kind(decision) if kind == "map": target = _target_uuid(decision) if target is None or target not in graph.client_names: raise _bad("Mapped client id not found.", "BAD_MAPPING") if target not in graph.clients_of.get(op_id, {}).values(): raise _bad( f'Mapped client "{cl_name}" is under a different operator' " than expected.", "BAD_MAPPING", ) cl_resolved[(op_name, cl_name)] = target else: cl_resolved[(op_name, cl_name)] = resolve_or_create_client( session, graph, op_id, cl_name ) # --- step 3: projects (B60) ----------------------------------------- pr_resolved: dict[tuple[str, str, str], uuid.UUID] = {} triples = sorted( { (e["operator"], e["client"], e["project"]) for e in entries if e.get("operator") and e.get("client") and e.get("project") } ) for op_name, cl_name, pr_name in triples: cl_id = cl_resolved.get((op_name, cl_name)) if cl_id is None: raise _bad( f'Client "{cl_name}" wasn\'t resolved (internal bug).' " Re-submit the resolver.", "INTERNAL", 500, ) existing = graph.find_project(cl_id, pr_name) if existing is not None: pr_resolved[(op_name, cl_name, pr_name)] = existing continue decision = pr_decisions.get((op_name, cl_name, pr_name)) if decision is None: raise _bad( f'No decision provided for unknown project "{pr_name}".' " Please re-submit the resolver.", "MISSING_DECISION", ) kind = _decision_kind(decision) if kind == "map": target = _target_uuid(decision) chain = graph.project_chain.get(target) if target is not None else None if chain is None: raise _bad("Mapped project id not found.", "BAD_MAPPING") if chain[2] != cl_id: raise _bad( f'Mapped project "{pr_name}" is under a different client' " than expected.", "BAD_MAPPING", ) pr_resolved[(op_name, cl_name, pr_name)] = target else: op_id = op_resolved[op_name] pr_resolved[(op_name, cl_name, pr_name)] = resolve_or_create_project( session, graph, op_id, cl_id, pr_name ) # --- batch row (B62): row_count = parsed entries, not raw CSV rows -- batch = ImportBatch( tenant_id=tenant_id, source="clockify", filename=pending.filename, name=pending.batch_name, notes=pending.notes, row_count=len(entries), imported_by=actor.user_id, # EFFECTIVE user (ADR #033) ) session.add(batch) session.flush() # --- per-row conversion + stamping (B63–B67, rebuild) ---------------- worker_cache: dict[str, tuple[uuid.UUID, uuid.UUID] | None] = {} rate_cache: dict[tuple[uuid.UUID, uuid.UUID | None], StampRates] = {} rows: list[dict] = [] matched = 0 unknown_emails: set[str] = set() for e in entries: email = (e.get("source_user_email") or "").strip().lower() if email not in worker_cache: worker_cache[email] = resolve_worker(session, tenant_id, email) worker = worker_cache[email] # project resolution: full triple -> step-3 map; partial -> bucket op_name = e.get("operator") or None cl_name = e.get("client") or None pr_name = e.get("project") or None if op_name and cl_name and pr_name: project_id = pr_resolved.get((op_name, cl_name, pr_name)) else: project_id = bucket_targets.get(bucket_signature(op_name, cl_name, pr_name)) if worker is not None: matched += 1 worker_party_id, engagement_id = worker cache_key = (engagement_id, project_id) if cache_key not in rate_cache: rate_cache[cache_key] = resolve_rates( session, tenant_id, engagement_id, worker_party_id, project_id ) stamp = stamp_entry(e.get("duration_seconds"), rate_cache[cache_key]) else: if email: unknown_emails.add(email) worker_party_id, engagement_id = None, None stamp = stamp_entry(None, StampRates(None, None, None, None)) rows.append( { "tenant_id": tenant_id, "project_id": project_id, "worker_party_id": worker_party_id, "via_engagement_id": engagement_id, "source": "clockify", "source_timezone": None, # Clockify CSVs carry no tz (07 #17) "import_id": batch.id, "source_user_name": e.get("source_user_name") or None, "source_user_email": email or None, # the immutable as-parsed payload — never canonicalized "raw_operator": op_name, "raw_client": cl_name, "raw_project": pr_name, "description": e.get("description") or None, "billable": e.get("billable"), "start_at": _parse_naive(e.get("start_at")), "end_at": _parse_naive(e.get("end_at")), "duration_seconds": e.get("duration_seconds"), "source_rate": _as_decimal(e.get("source_rate")), "source_amount": _as_decimal(e.get("source_amount")), "cost_amount": stamp.cost_amount, "cost_currency": stamp.cost_currency, "billout_amount": stamp.billout_amount, "billout_currency": stamp.billout_currency, "invoice_line_id": None, } ) # --- chunked insert (B68) -------------------------------------------- for i in range(0, len(rows), CHUNK_SIZE): chunk = rows[i : i + CHUNK_SIZE] try: _insert_chunk(session, chunk) except ImportsError: raise except Exception as exc: # rollback happens in commit_pending end = min(i + CHUNK_SIZE, len(rows)) raise _bad( f"Insert failed at rows {i + 1}–{end}: {exc}", "INSERT_FAILED", 500 ) from exc # success: consume the staging row, commit everything atomically session.delete(pending) session.commit() return { "batch_id": str(batch.id), "batch_name": batch.name, "row_count": len(rows), "matched": matched, "unmatched": len(rows) - matched, "unknown_emails": sorted(unknown_emails), } def _as_decimal(value) -> Decimal | None: if value is None: return None return Decimal(str(value))