sagnik/Project_Velocity
2
1
Fork 2
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
6c93e317416af3e20645094d3337305512a5e42a
Project_Velocity/backend/api/routes_crm_imports.py
Sayan Datta 6c93e31741
All checks were successful
Production Readiness / backend-contracts (pull_request) Successful in 3m19s
Details
Production Readiness / webos-typecheck (pull_request) Successful in 2m38s
Details
Production Readiness / ipad-parse (pull_request) Successful in 1m44s
Details
feat: Ipad app production readiness, Colony orchestration, Social posting
2026-05-03 18:28:04 +05:30

2030 lines
77 KiB
Python
Raw Blame History

"""
backend/api/routes_crm_imports.py
CRM Import Route Family + Client 360 Routes
Implements the canonical CRM API surface as specified in Doc 09 (Root API Spec):
POST /api/crm/imports — upload CSV batch
GET /api/crm/imports — list import batches
GET /api/crm/imports/{id} — get batch detail + proposals
PUT /api/crm/imports/{id}/approve-proposal — approve a single proposal
POST /api/crm/imports/{id}/commit — commit approved proposals to canonical
GET /api/crm/contacts — canonical contact list (with QD summary)
GET /api/crm/contacts/{id} — canonical contact detail
GET /api/crm/client-360/{id} — Client 360 aggregated snapshot
GET /api/crm/opportunities — opportunity pipeline list
PATCH /api/crm/opportunities/{id} — opportunity/deal update
GET /api/crm/tasks — reminder/task list
PATCH /api/crm/tasks/{id} — reminder/task lifecycle update
GET /api/crm/kanban — kanban board (canonical leads)
PATCH /api/crm/leads/{id}/stage — canonical lead stage transition
Uses canonical crm_*, intel_*, workflow_* tables.
"""
from __future__ import annotations
import json
import logging
import uuid
from datetime import datetime, timezone
from typing import Any
from fastapi import APIRouter, Depends, HTTPException, Query, Request, UploadFile, File, status
from pydantic import BaseModel, Field
from backend.auth.dependencies import UserPrincipal, get_current_user
from backend.crm.canonical_schema import ensure_canonical_crm_schema
from backend.services.imports.ingest_service import (
parse_csv_content,
infer_column_mapping,
build_normalized_proposals,
create_import_batch_record,
persist_import_batch,
persist_proposals_as_workflow_actions,
)
from backend.services.client_graph.aggregation_service import (
get_client_360,
get_contact_list,
)
logger = logging.getLogger("velocity.api.crm_imports")
# Canonical CRM surfaces are shared with the authenticated WebOS session model.
# Keep the entire router behind Velocity bearer auth so we do not reintroduce
# an unauthenticated read/write path while the canonical tenant model evolves.
router = APIRouter(dependencies=[Depends(get_current_user)])
CANONICAL_LEAD_STAGES = (
"new",
"contacted",
"qualified",
"site_visit_scheduled",
"site_visited",
"negotiation",
"booking_initiated",
"booked",
"lost",
"dormant",
)
CANONICAL_TASK_STATUSES = ("pending", "confirmed", "done", "snoozed", "cancelled")
CANONICAL_OPPORTUNITY_STAGES = (
"prospect",
"qualified",
"proposal",
"site_visit",
"negotiation",
"booking",
"agreement",
"closed_won",
"closed_lost",
)
IMPORT_DUPLICATE_POLICIES = ("create_new", "update_existing", "skip_duplicate")
CANONICAL_URGENCY_VALUES = ("low", "medium", "high", "critical")
CANONICAL_TASK_PRIORITIES = ("low", "normal", "high", "urgent")
CANONICAL_BUYER_TYPES = (
"end_user",
"hni_end_user",
"nri_investor",
"family_office",
"founder_buyer",
"broker_referral",
"investor",
)
DREAM_WEAVER_ROOM_TYPES = (
("bedroom", "Bedroom", "bed.double"),
("living_room", "Living Room", "sofa"),
("bathroom", "Bathroom", "drop"),
("kitchen", "Kitchen", "refrigerator"),
("dining_room", "Dining Room", "fork.knife"),
("home_office", "Office", "desktopcomputer"),
("hallway", "Hallway", "door.left.hand.open"),
("balcony", "Balcony", "sun.max"),
)
def _label_for_vocab(value: str) -> str:
return value.replace("_", " ").title()
def _vocab_options(values: tuple[str, ...], descriptions: dict[str, str] | None = None) -> list[dict[str, str]]:
descriptions = descriptions or {}
return [
{
"value": value,
"label": _label_for_vocab(value),
"description": descriptions.get(value, ""),
}
for value in values
]
def _room_type_options() -> list[dict[str, str]]:
return [
{
"value": value,
"label": label,
"description": "",
"icon": icon,
}
for value, label, icon in DREAM_WEAVER_ROOM_TYPES
]
def _now() -> str:
return datetime.now(timezone.utc).isoformat()
async def _get_pool(request: Request):
await ensure_canonical_crm_schema(request.app)
pool = getattr(request.app.state, "db_pool", None)
if pool is None:
raise HTTPException(status_code=503, detail="Database unavailable.")
return pool
def _tenant_scope(user: UserPrincipal) -> str:
return user.tenant_id
def _parse_optional_datetime(value: str | None, *, field_name: str) -> datetime | None:
if value is None:
return None
normalized = value.strip()
if not normalized:
return None
try:
parsed = datetime.fromisoformat(normalized.replace("Z", "+00:00"))
except ValueError as exc:
raise HTTPException(status_code=422, detail=f"Invalid ISO-8601 timestamp for '{field_name}'.") from exc
if parsed.tzinfo is None:
parsed = parsed.replace(tzinfo=timezone.utc)
return parsed.astimezone(timezone.utc)
def _parse_optional_date(value: str | None, *, field_name: str):
if value is None:
return None
normalized = value.strip()
if not normalized:
return None
try:
return datetime.fromisoformat(normalized.replace("Z", "+00:00")).date()
except ValueError as exc:
raise HTTPException(status_code=422, detail=f"Invalid ISO-8601 date for '{field_name}'.") from exc
def _optional_uuid(value: str | None) -> str | None:
if not value:
return None
try:
return str(uuid.UUID(value))
except ValueError:
return None
def _normalize_choice(value: str, *, allowed: tuple[str, ...], field_name: str) -> str:
normalized = value.strip().lower()
if normalized not in allowed:
allowed_values = ", ".join(allowed)
raise HTTPException(status_code=422, detail=f"Unsupported {field_name}. Expected one of: {allowed_values}.")
return normalized
def _fields_set(model: BaseModel) -> set[str]:
fields = getattr(model, "model_fields_set", None)
if fields is not None:
return set(fields)
return set(getattr(model, "__fields_set__", set()))
def _row_value(row, key: str, default: Any = None) -> Any:
try:
return row[key]
except (KeyError, IndexError):
return default
def _opportunity_payload(row) -> dict[str, Any]:
value = _row_value(row, "value")
expected_close_date = _row_value(row, "expected_close_date")
return {
"opportunity_id": str(row["opportunity_id"]),
"stage": row["stage"],
"value": float(value) if value is not None else None,
"probability": _row_value(row, "probability"),
"expected_close_date": expected_close_date.isoformat() if expected_close_date else None,
"next_action": _row_value(row, "next_action"),
"notes": _row_value(row, "notes"),
"person_id": str(row["person_id"]),
"client_name": row["full_name"],
"client_phone": row["primary_phone"],
"project_name": row["project_name"],
}
@router.get("/crm/vocabularies", tags=["CRM Vocabularies"])
async def get_crm_vocabularies(
request: Request,
user: UserPrincipal = Depends(get_current_user),
) -> dict[str, Any]:
"""
Canonical business vocabularies for native clients.
The iPad must not own CRM funnel semantics, duplicate merge policy values,
task priorities, buyer personas, or Dream Weaver room vocabularies. This
endpoint gives authenticated clients a single backend-owned contract.
"""
await _get_pool(request)
return {
"status": "ok",
"data": {
"lead_statuses": _vocab_options(CANONICAL_LEAD_STAGES),
"urgencies": _vocab_options(CANONICAL_URGENCY_VALUES),
"buyer_types": _vocab_options(CANONICAL_BUYER_TYPES),
"task_priorities": _vocab_options(CANONICAL_TASK_PRIORITIES),
"lead_stages": _vocab_options(CANONICAL_LEAD_STAGES),
"opportunity_stages": _vocab_options(CANONICAL_OPPORTUNITY_STAGES),
"import_duplicate_policies": _vocab_options(
IMPORT_DUPLICATE_POLICIES,
{
"create_new": "Create a new canonical CRM person from the approved row.",
"update_existing": "Merge approved fields into the strongest duplicate candidate.",
"skip_duplicate": "Skip canonical create/update for this approved row.",
},
),
"dream_weaver_room_types": _room_type_options(),
},
"meta": {
"tenant_id": _tenant_scope(user),
"source_of_truth": "backend_canonical_crm_vocabulary",
},
}
# ── Models ────────────────────────────────────────────────────────────────────
class ProposalApprovalRequest(BaseModel):
proposal_id: str
decision: str = Field(..., pattern="^(approved|rejected|needs_more_info)$")
notes: str = Field(default="", max_length=2000)
field_overrides: dict[str, Any] = Field(default_factory=dict)
duplicate_policy: str = Field(default="create_new", pattern="^(create_new|update_existing|skip_duplicate)$")
class CreatePersonRequest(BaseModel):
full_name: str = Field(..., min_length=1, max_length=200)
primary_email: str | None = None
primary_phone: str | None = None
buyer_type: str | None = None
budget_band: str | None = None
project_name: str | None = None
source_system: str = "manual"
notes: str | None = None
metadata_json: dict[str, Any] = Field(default_factory=dict)
class CreateLeadRequest(BaseModel):
person_id: str
status: str = "new"
budget_band: str | None = None
urgency: str | None = None
financing_posture: str | None = None
assigned_user_id: str | None = None
metadata_json: dict[str, Any] = Field(default_factory=dict)
class CreateReminderRequest(BaseModel):
person_id: str
lead_id: str | None = None
reminder_type: str = "follow_up"
title: str = Field(..., min_length=1, max_length=500)
notes: str | None = None
due_at: str | None = None
priority: str = "normal"
class ClientDataPatchRequest(BaseModel):
full_name: str | None = Field(default=None, max_length=256)
primary_email: str | None = Field(default=None, max_length=256)
primary_phone: str | None = Field(default=None, max_length=64)
buyer_type: str | None = Field(default=None, max_length=128)
communication_preference: str | None = Field(default=None, max_length=64)
best_contact_time: str | None = Field(default=None, max_length=128)
lead_status: str | None = Field(default=None, max_length=64)
budget_band: str | None = Field(default=None, max_length=128)
urgency: str | None = Field(default=None, max_length=64)
IMPORT_VALIDATION_FIELDS = (
"full_name",
"primary_email",
"primary_phone",
"buyer_type",
"budget_band",
"project_name",
)
def _clean_import_value(value: Any) -> str | None:
if value is None:
return None
text = str(value).strip()
return text or None
def _digits_only(value: str | None) -> str:
return "".join(ch for ch in (value or "") if ch.isdigit())
def _validate_import_canonical(canonical: dict[str, Any]) -> list[dict[str, Any]]:
issues: list[dict[str, Any]] = []
full_name = _clean_import_value(canonical.get("full_name"))
email = _clean_import_value(canonical.get("primary_email"))
phone = _clean_import_value(canonical.get("primary_phone"))
if not full_name:
issues.append({
"field": "full_name",
"severity": "error",
"message": "Full name is required before commit.",
})
elif len(full_name) < 2:
issues.append({
"field": "full_name",
"severity": "warning",
"message": "Full name is unusually short.",
})
if email and ("@" not in email or "." not in email.split("@")[-1]):
issues.append({
"field": "primary_email",
"severity": "error",
"message": "Email must look like a valid address.",
})
if phone and len(_digits_only(phone)) < 7:
issues.append({
"field": "primary_phone",
"severity": "error",
"message": "Phone number must contain at least 7 digits.",
})
if not email and not phone:
issues.append({
"field": "primary_phone",
"severity": "warning",
"message": "No phone or email is present; future dedupe and outreach quality will be limited.",
})
return issues
def _crm_person_candidate(row: Any) -> dict[str, Any]:
return {
"person_id": str(row["person_id"]),
"full_name": row["full_name"],
"primary_email": row["primary_email"],
"primary_phone": row["primary_phone"],
"buyer_type": row["buyer_type"],
"source_confidence": float(row["source_confidence"]) if row["source_confidence"] is not None else None,
"created_at": row["created_at"].isoformat() if row["created_at"] else None,
"updated_at": row["updated_at"].isoformat() if row["updated_at"] else None,
"match_reason": row["match_reason"],
"match_score": int(row["match_score"]),
}
def _proposal_field_diff(canonical: dict[str, Any], existing: dict[str, Any] | None) -> list[dict[str, Any]]:
fields = sorted(set(IMPORT_VALIDATION_FIELDS).union(canonical.keys()))
diffs: list[dict[str, Any]] = []
for field in fields:
proposed = _clean_import_value(canonical.get(field))
current = _clean_import_value(existing.get(field)) if existing else None
if proposed is None and current is None:
continue
diffs.append({
"field": field,
"proposed": proposed,
"existing": current,
"changed": proposed != current,
})
return diffs
async def _find_duplicate_person(conn, tenant_id: str, canonical: dict[str, Any]) -> Any | None:
email = _clean_import_value(canonical.get("primary_email"))
phone = _clean_import_value(canonical.get("primary_phone"))
full_name = _clean_import_value(canonical.get("full_name"))
return await conn.fetchrow(
"""
SELECT p.person_id, p.full_name, p.primary_email, p.primary_phone, p.buyer_type,
p.source_confidence, p.created_at, p.updated_at,
CASE
WHEN $2::text IS NOT NULL AND LOWER(p.primary_email) = LOWER($2::text) THEN 'email'
WHEN $3::text IS NOT NULL AND regexp_replace(COALESCE(p.primary_phone, ''), '\\D', '', 'g') = regexp_replace($3::text, '\\D', '', 'g') THEN 'phone'
WHEN $4::text IS NOT NULL AND LOWER(p.full_name) = LOWER($4::text) THEN 'name'
ELSE 'fuzzy'
END AS match_reason,
CASE
WHEN $2::text IS NOT NULL AND LOWER(p.primary_email) = LOWER($2::text) THEN 100
WHEN $3::text IS NOT NULL AND regexp_replace(COALESCE(p.primary_phone, ''), '\\D', '', 'g') = regexp_replace($3::text, '\\D', '', 'g') THEN 95
WHEN $4::text IS NOT NULL AND LOWER(p.full_name) = LOWER($4::text) THEN 70
ELSE 50
END AS match_score
FROM crm_people p
WHERE p.tenant_id = $1
AND (
($2::text IS NOT NULL AND LOWER(p.primary_email) = LOWER($2::text))
OR ($3::text IS NOT NULL AND regexp_replace(COALESCE(p.primary_phone, ''), '\\D', '', 'g') = regexp_replace($3::text, '\\D', '', 'g'))
OR ($4::text IS NOT NULL AND LOWER(p.full_name) = LOWER($4::text))
)
ORDER BY match_score DESC, p.updated_at DESC NULLS LAST
LIMIT 1
""",
tenant_id,
email,
phone,
full_name,
)
class UpdateReminderRequest(BaseModel):
status: str = Field(..., min_length=1, max_length=32)
due_at: str | None = None
notes: str | None = Field(default=None, max_length=4000)
class UpdateLeadStageRequest(BaseModel):
status: str = Field(..., min_length=1, max_length=64)
notes: str | None = Field(default=None, max_length=2000)
class UpdateOpportunityRequest(BaseModel):
stage: str | None = Field(default=None, max_length=64)
value: float | None = Field(default=None, ge=0)
probability: int | None = Field(default=None, ge=0, le=100)
expected_close_date: str | None = None
next_action: str | None = Field(default=None, max_length=1000)
notes: str | None = Field(default=None, max_length=4000)
# ── Import Endpoints ──────────────────────────────────────────────────────────
@router.post("/crm/imports", status_code=201, tags=["CRM Imports"])
async def upload_crm_import(
request: Request,
file: UploadFile = File(...),
source_system: str = Query(default="csv_upload"),
user: UserPrincipal = Depends(get_current_user),
) -> dict[str, Any]:
"""
Upload a CSV file to start a CRM import batch.
Parses headers, infers column mapping, and creates workflow_actions proposals.
"""
pool = await _get_pool(request)
content_bytes = await file.read()
try:
content = content_bytes.decode("utf-8")
except UnicodeDecodeError:
content = content_bytes.decode("latin-1")
parsed = parse_csv_content(content)
mapping = infer_column_mapping(parsed["headers"])
batch = create_import_batch_record(
filename=file.filename or "upload.csv",
row_count=parsed["row_count"],
mapping_manifest=mapping,
source_system=source_system,
tenant_id=_tenant_scope(user),
)
proposals = build_normalized_proposals(
rows=parsed["rows"],
mapping=mapping["mapped"],
batch_id=batch["batch_id"],
source_system=source_system,
)
async with pool.acquire() as conn:
await persist_import_batch(conn, batch)
inserted = await persist_proposals_as_workflow_actions(conn, proposals, _tenant_scope(user))
logger.info("Import batch %s: %d rows, %d proposals", batch["batch_id"], parsed["row_count"], inserted)
return {
"status": "ok",
"data": {
"batch_id": batch["batch_id"],
"row_count": parsed["row_count"],
"mapped_columns": mapping["mapped_count"],
"unmapped_columns": mapping["unmapped_count"],
"mapping_confidence": mapping["confidence"],
"proposals_created": inserted,
"parse_errors": parsed["parse_errors"],
"lifecycle": "parsed",
"message": f"Import batch created. {inserted} proposals queued for review.",
},
}
@router.get("/crm/imports", tags=["CRM Imports"])
async def list_import_batches(
request: Request,
lifecycle: str | None = None,
limit: int = Query(default=20, ge=1, le=100),
offset: int = Query(default=0, ge=0),
user: UserPrincipal = Depends(get_current_user),
) -> dict[str, Any]:
"""List all CRM import batches with lifecycle status."""
pool = await _get_pool(request)
clauses = ["tenant_id = $1"]
params: list[Any] = [_tenant_scope(user)]
if lifecycle:
params.append(lifecycle)
clauses.append(f"lifecycle = ${len(params)}::import_lifecycle")
params.extend([limit, offset])
where = " AND ".join(clauses)
async with pool.acquire() as conn:
rows = await conn.fetch(
f"""
SELECT batch_id, source_system, uploaded_filename, row_count,
mapped_count, unresolved_count, lifecycle, created_at, updated_at
FROM workflow_import_batches
WHERE {where}
ORDER BY created_at DESC
LIMIT ${len(params) - 1} OFFSET ${len(params)}
""",
*params,
)
total = await conn.fetchval(
f"SELECT COUNT(*) FROM workflow_import_batches WHERE {where}",
*params[:-2],
)
batches = [
{
"batch_id": str(r["batch_id"]),
"source_system": r["source_system"],
"filename": r["uploaded_filename"],
"row_count": r["row_count"],
"mapped_count": r["mapped_count"],
"unresolved_count": r["unresolved_count"],
"lifecycle": r["lifecycle"],
"created_at": r["created_at"].isoformat() if r["created_at"] else None,
}
for r in rows
]
return {"status": "ok", "data": batches, "meta": {"total": total, "limit": limit, "offset": offset}}
@router.get("/crm/imports/{batch_id}", tags=["CRM Imports"])
async def get_import_batch(
request: Request,
batch_id: str,
user: UserPrincipal = Depends(get_current_user),
) -> dict[str, Any]:
"""Get import batch detail including pending proposals."""
pool = await _get_pool(request)
async with pool.acquire() as conn:
batch_row = await conn.fetchrow(
"SELECT * FROM workflow_import_batches WHERE batch_id = $1::uuid AND tenant_id = $2",
batch_id,
_tenant_scope(user),
)
if not batch_row:
raise HTTPException(status_code=404, detail=f"Import batch '{batch_id}' not found.")
proposal_rows = await conn.fetch(
"""
SELECT action_id, proposal_payload, confidence, status, approval_required, created_at
FROM workflow_actions
WHERE tenant_id = $1
AND action_type = 'import_proposal'
AND proposal_payload->>'batch_id' = $2
ORDER BY (proposal_payload->>'row_number')::int ASC
LIMIT 200
""",
_tenant_scope(user),
batch_id,
)
proposals = [
{
"proposal_id": str(r["action_id"]),
"payload": r["proposal_payload"],
"confidence": float(r["confidence"]) if r["confidence"] else 0.0,
"status": r["status"],
"review_required": r["approval_required"],
}
for r in proposal_rows
]
return {
"status": "ok",
"data": {
"batch_id": str(batch_row["batch_id"]),
"source_system": batch_row["source_system"],
"filename": batch_row["uploaded_filename"],
"row_count": batch_row["row_count"],
"mapping_manifest": batch_row["mapping_manifest"],
"lifecycle": batch_row["lifecycle"],
"proposals": proposals,
"proposal_count": len(proposals),
},
}
@router.get("/crm/imports/{batch_id}/workbench", tags=["CRM Imports"])
async def get_import_workbench(
request: Request,
batch_id: str,
user: UserPrincipal = Depends(get_current_user),
) -> dict[str, Any]:
"""
Return enterprise review diagnostics for an import batch.
This endpoint is intentionally read-only. It lets iPad/WebOS operators see
per-field validation, duplicate candidates, and row-level diffs before
committing approved proposals into canonical CRM tables.
"""
pool = await _get_pool(request)
tenant_id = _tenant_scope(user)
async with pool.acquire() as conn:
batch_exists = await conn.fetchval(
"SELECT EXISTS (SELECT 1 FROM workflow_import_batches WHERE batch_id = $1::uuid AND tenant_id = $2)",
batch_id,
tenant_id,
)
if not batch_exists:
raise HTTPException(status_code=404, detail=f"Import batch '{batch_id}' not found.")
proposal_rows = await conn.fetch(
"""
SELECT action_id, proposal_payload, confidence, status, approval_required, created_at
FROM workflow_actions
WHERE tenant_id = $1
AND action_type = 'import_proposal'
AND proposal_payload->>'batch_id' = $2
ORDER BY (proposal_payload->>'row_number')::int ASC
LIMIT 200
""",
tenant_id,
batch_id,
)
rows: list[dict[str, Any]] = []
duplicate_count = 0
validation_error_count = 0
validation_warning_count = 0
for proposal in proposal_rows:
payload = dict(proposal["proposal_payload"] or {})
canonical = dict(payload.get("canonical_payload") or {})
email = _clean_import_value(canonical.get("primary_email"))
phone = _clean_import_value(canonical.get("primary_phone"))
full_name = _clean_import_value(canonical.get("full_name"))
duplicate_candidates = await conn.fetch(
"""
SELECT p.person_id, p.full_name, p.primary_email, p.primary_phone, p.buyer_type,
p.source_confidence, p.created_at, p.updated_at,
CASE
WHEN $2::text IS NOT NULL AND LOWER(p.primary_email) = LOWER($2::text) THEN 'email'
WHEN $3::text IS NOT NULL AND regexp_replace(COALESCE(p.primary_phone, ''), '\\D', '', 'g') = regexp_replace($3::text, '\\D', '', 'g') THEN 'phone'
WHEN $4::text IS NOT NULL AND LOWER(p.full_name) = LOWER($4::text) THEN 'name'
ELSE 'fuzzy'
END AS match_reason,
CASE
WHEN $2::text IS NOT NULL AND LOWER(p.primary_email) = LOWER($2::text) THEN 100
WHEN $3::text IS NOT NULL AND regexp_replace(COALESCE(p.primary_phone, ''), '\\D', '', 'g') = regexp_replace($3::text, '\\D', '', 'g') THEN 95
WHEN $4::text IS NOT NULL AND LOWER(p.full_name) = LOWER($4::text) THEN 70
ELSE 50
END AS match_score
FROM crm_people p
WHERE p.tenant_id = $1
AND (
($2::text IS NOT NULL AND LOWER(p.primary_email) = LOWER($2::text))
OR ($3::text IS NOT NULL AND regexp_replace(COALESCE(p.primary_phone, ''), '\\D', '', 'g') = regexp_replace($3::text, '\\D', '', 'g'))
OR ($4::text IS NOT NULL AND LOWER(p.full_name) = LOWER($4::text))
)
ORDER BY match_score DESC, p.updated_at DESC NULLS LAST
LIMIT 5
""",
tenant_id,
email,
phone,
full_name,
)
candidates = [_crm_person_candidate(row) for row in duplicate_candidates]
existing = candidates[0] if candidates else None
validation = _validate_import_canonical(canonical)
validation_error_count += sum(1 for issue in validation if issue["severity"] == "error")
validation_warning_count += sum(1 for issue in validation if issue["severity"] == "warning")
if candidates:
duplicate_count += 1
rows.append({
"proposal_id": str(proposal["action_id"]),
"row_number": payload.get("row_number"),
"status": proposal["status"],
"confidence": float(proposal["confidence"]) if proposal["confidence"] else 0.0,
"validation": validation,
"duplicate_candidates": candidates,
"duplicate_policy": payload.get("duplicate_policy") or ("update_existing" if candidates else "create_new"),
"field_diffs": _proposal_field_diff(canonical, existing),
})
return {
"status": "ok",
"data": {
"batch_id": batch_id,
"summary": {
"proposal_count": len(rows),
"duplicate_count": duplicate_count,
"validation_error_count": validation_error_count,
"validation_warning_count": validation_warning_count,
},
"rows": rows,
},
}
@router.put("/crm/imports/{batch_id}/review-proposal", tags=["CRM Imports"])
async def review_proposal(
request: Request,
batch_id: str,
body: ProposalApprovalRequest,
user: UserPrincipal = Depends(get_current_user),
) -> dict[str, Any]:
"""
Human review of a single import proposal.
Creates a workflow_approvals record and updates the action status.
Approved actions with high confidence may be auto-staged for commit.
"""
pool = await _get_pool(request)
async with pool.acquire() as conn:
action = await conn.fetchrow(
"""
SELECT action_id, confidence, approval_required, proposal_payload
FROM workflow_actions
WHERE action_id = $1::uuid
AND tenant_id = $2
AND proposal_payload->>'batch_id' = $3
""",
body.proposal_id,
_tenant_scope(user),
batch_id,
)
if not action:
raise HTTPException(status_code=404, detail="Proposal not found.")
decision_id = str(uuid.uuid4())
new_status = {
"approved": "approved",
"rejected": "rejected",
"needs_more_info": "review_required",
}[body.decision]
proposal_payload = dict(action["proposal_payload"] or {})
if body.field_overrides:
canonical_payload = dict(proposal_payload.get("canonical_payload") or {})
cleaned_overrides = {
key: value
for key, value in body.field_overrides.items()
if key and value is not None and str(value).strip()
}
canonical_payload.update(cleaned_overrides)
if cleaned_overrides:
missing_required = [
field for field in proposal_payload.get("missing_required", [])
if field not in cleaned_overrides
]
unresolved_fields = [
field for field in proposal_payload.get("unresolved_fields", [])
if field not in cleaned_overrides
]
proposal_payload["canonical_payload"] = canonical_payload
proposal_payload["missing_required"] = missing_required
proposal_payload["unresolved_fields"] = unresolved_fields
proposal_payload["remediation"] = {
"source": "ipad_import_workbench",
"field_overrides": cleaned_overrides,
"updated_at": _now(),
"updated_by": user.user_id,
}
proposal_payload["duplicate_policy"] = body.duplicate_policy
proposal_payload["duplicate_policy_updated_at"] = _now()
proposal_payload["duplicate_policy_updated_by"] = user.user_id
await conn.execute(
"""
INSERT INTO workflow_approvals (decision_id, tenant_id, action_id, decision, decision_notes, decided_at)
VALUES ($1::uuid, $2, $3::uuid, $4, $5, NOW())
""",
decision_id,
_tenant_scope(user),
body.proposal_id,
body.decision,
body.notes,
)
await conn.execute(
"""
UPDATE workflow_actions
SET status = $1::wf_status, proposal_payload = $2::jsonb, updated_at = NOW()
WHERE action_id = $3::uuid
AND tenant_id = $4
""",
new_status,
json.dumps(proposal_payload),
body.proposal_id,
_tenant_scope(user),
)
return {
"status": "ok",
"data": {
"decision_id": decision_id,
"proposal_id": body.proposal_id,
"decision": body.decision,
"message": f"Proposal {body.decision}.",
},
}
@router.post("/crm/imports/{batch_id}/commit", tags=["CRM Imports"])
async def commit_approved_proposals(
request: Request,
batch_id: str,
user: UserPrincipal = Depends(get_current_user),
) -> dict[str, Any]:
"""
Commit all approved proposals for a batch into canonical crm_people + crm_leads tables.
Only approved proposals are committed. Rejected/pending are skipped.
This implements the writeback flow from Doc 07 and Doc 09.
"""
pool = await _get_pool(request)
committed = 0
skipped = 0
errors: list[str] = []
async with pool.acquire() as conn:
approved_rows = await conn.fetch(
"""
SELECT action_id, proposal_payload
FROM workflow_actions
WHERE tenant_id = $1
AND action_type = 'import_proposal'
AND proposal_payload->>'batch_id' = $2
AND status = 'approved'
""",
_tenant_scope(user),
batch_id,
)
for row in approved_rows:
try:
payload = row["proposal_payload"]
canonical = payload.get("canonical_payload", {})
validation_errors = [
issue for issue in _validate_import_canonical(canonical)
if issue["severity"] == "error"
]
if validation_errors:
skipped += 1
errors.append(
f"Proposal {row['action_id']}: "
+ "; ".join(issue["message"] for issue in validation_errors)
)
continue
duplicate_policy = payload.get("duplicate_policy") or "create_new"
if duplicate_policy not in IMPORT_DUPLICATE_POLICIES:
duplicate_policy = "create_new"
duplicate_person = await _find_duplicate_person(conn, _tenant_scope(user), canonical)
if duplicate_person and duplicate_policy == "skip_duplicate":
skipped += 1
await conn.execute(
"""
UPDATE workflow_actions
SET status = 'executed'::wf_status, updated_at = NOW()
WHERE action_id = $1::uuid
AND tenant_id = $2
""",
row["action_id"],
_tenant_scope(user),
)
continue
if duplicate_person and duplicate_policy == "update_existing":
person_id = str(duplicate_person["person_id"])
await conn.execute(
"""
UPDATE crm_people
SET full_name = COALESCE($3, full_name),
primary_email = COALESCE($4, primary_email),
primary_phone = COALESCE($5, primary_phone),
buyer_type = COALESCE($6, buyer_type),
source_confidence = GREATEST(COALESCE(source_confidence, 0), $7),
metadata_json = COALESCE(metadata_json, '{}'::jsonb) || $8::jsonb,
updated_at = NOW()
WHERE person_id = $1::uuid
AND tenant_id = $2
""",
person_id,
_tenant_scope(user),
canonical.get("full_name"),
canonical.get("primary_email"),
canonical.get("primary_phone"),
canonical.get("buyer_type"),
payload.get("confidence", 0.5),
json.dumps({
"source_batch": batch_id,
"import_row": payload.get("row_number"),
"duplicate_policy": duplicate_policy,
"merged_from_import": True,
}),
)
else:
person_id = str(uuid.uuid4())
await conn.execute(
"""
INSERT INTO crm_people (
person_id, tenant_id, full_name, primary_email, primary_phone,
buyer_type, source_confidence, metadata_json, created_at, updated_at
) VALUES (
$1::uuid, $2, $3, $4, $5, $6, $7, $8::jsonb, NOW(), NOW()
)
ON CONFLICT DO NOTHING
""",
person_id,
_tenant_scope(user),
canonical.get("full_name"),
canonical.get("primary_email"),
canonical.get("primary_phone"),
canonical.get("buyer_type"),
payload.get("confidence", 0.5),
json.dumps({
"source_batch": batch_id,
"import_row": payload.get("row_number"),
"duplicate_policy": duplicate_policy,
}),
)
if canonical.get("status") or canonical.get("budget_band"):
lead_id = str(uuid.uuid4())
await conn.execute(
"""
INSERT INTO crm_leads (
lead_id, tenant_id, person_id, source_system, status, budget_band,
metadata_json, created_at, updated_at
) VALUES (
$1::uuid, $2, $3::uuid, $4, 'new'::crm_lead_status, $5, $6::jsonb, NOW(), NOW()
)
ON CONFLICT DO NOTHING
""",
lead_id,
_tenant_scope(user),
person_id,
payload.get("source_system", "csv_upload"),
canonical.get("budget_band"),
json.dumps({"import_batch": batch_id}),
)
# Stage property interest if project_name present
if canonical.get("project_name"):
await conn.execute(
"""
INSERT INTO crm_property_interests (
interest_id, tenant_id, person_id, lead_id, project_name, created_at
) VALUES (
$1::uuid, $2, $3::uuid, $4::uuid, $5, NOW()
)
ON CONFLICT DO NOTHING
""",
str(uuid.uuid4()),
_tenant_scope(user),
person_id,
lead_id,
canonical.get("project_name"),
)
# Mark action as executed
await conn.execute(
"""
UPDATE workflow_actions
SET status = 'executed'::wf_status, updated_at = NOW()
WHERE action_id = $1::uuid
AND tenant_id = $2
""",
row["action_id"],
_tenant_scope(user),
)
committed += 1
except Exception as e:
errors.append(f"Proposal {row['action_id']}: {str(e)}")
skipped += 1
# Update batch lifecycle
await conn.execute(
"""
UPDATE workflow_import_batches
SET lifecycle = 'committed'::import_lifecycle, canonical_count = $1, updated_at = NOW()
WHERE batch_id = $2
AND tenant_id = $3
""",
committed,
batch_id,
_tenant_scope(user),
)
return {
"status": "ok",
"data": {
"batch_id": batch_id,
"committed": committed,
"skipped": skipped,
"errors": errors,
"lifecycle": "committed",
},
}
# ── Contact / Person Endpoints ──────────────────────────────────────────────
@router.get("/crm/contacts", tags=["CRM Contacts"])
async def list_contacts(
request: Request,
search: str | None = Query(default=None),
buyer_type: str | None = Query(default=None),
status: str | None = Query(default=None),
limit: int = Query(default=50, ge=1, le=200),
offset: int = Query(default=0, ge=0),
user: UserPrincipal = Depends(get_current_user),
) -> dict[str, Any]:
"""Canonical contact list with QD summary, interaction count, and lead status."""
pool = await _get_pool(request)
async with pool.acquire() as conn:
result = await get_contact_list(conn, _tenant_scope(user), search, buyer_type, status, limit, offset)
return {"status": "ok", "data": result}
@router.post("/crm/contacts", status_code=201, tags=["CRM Contacts"])
async def create_contact(
request: Request,
body: CreatePersonRequest,
user: UserPrincipal = Depends(get_current_user),
) -> dict[str, Any]:
"""Create a new canonical person record manually."""
pool = await _get_pool(request)
person_id = str(uuid.uuid4())
async with pool.acquire() as conn:
await conn.execute(
"""
INSERT INTO crm_people (
person_id, tenant_id, full_name, primary_email, primary_phone,
buyer_type, source_confidence, metadata_json, created_at, updated_at
) VALUES ($1::uuid, $2, $3, $4, $5, $6, 1.0, $7::jsonb, NOW(), NOW())
""",
person_id,
_tenant_scope(user),
body.full_name,
body.primary_email,
body.primary_phone,
body.buyer_type,
json.dumps({**body.metadata_json, "manual_entry": True}),
)
if body.project_name or body.budget_band:
lead_id = str(uuid.uuid4())
await conn.execute(
"""
INSERT INTO crm_leads (
lead_id, tenant_id, person_id, source_system, status, budget_band,
metadata_json, created_at, updated_at
) VALUES ($1::uuid, $2, $3::uuid, $4, 'new'::crm_lead_status, $5, '{}'::jsonb, NOW(), NOW())
""",
lead_id,
_tenant_scope(user),
person_id,
body.source_system,
body.budget_band,
)
if body.project_name:
await conn.execute(
"""
INSERT INTO crm_property_interests (interest_id, tenant_id, person_id, lead_id, project_name, created_at)
VALUES ($1::uuid, $2, $3::uuid, $4::uuid, $5, NOW())
""",
str(uuid.uuid4()),
_tenant_scope(user),
person_id,
lead_id,
body.project_name,
)
return {"status": "ok", "data": {"person_id": person_id, "full_name": body.full_name}}
@router.get("/crm/contacts/{person_id}", tags=["CRM Contacts"])
async def get_contact(
request: Request,
person_id: str,
user: UserPrincipal = Depends(get_current_user),
) -> dict[str, Any]:
"""Get a single canonical contact record."""
pool = await _get_pool(request)
async with pool.acquire() as conn:
row = await conn.fetchrow(
"""
SELECT person_id, full_name, primary_email, primary_phone, secondary_phone,
buyer_type, persona_labels, source_confidence, created_at, updated_at
FROM crm_people
WHERE person_id = $1::uuid
AND tenant_id = $2
""",
person_id,
_tenant_scope(user),
)
if not row:
raise HTTPException(status_code=404, detail=f"Contact '{person_id}' not found.")
return {
"status": "ok",
"data": {
"person_id": str(row["person_id"]),
"full_name": row["full_name"],
"primary_email": row["primary_email"],
"primary_phone": row["primary_phone"],
"buyer_type": row["buyer_type"],
"persona_labels": row["persona_labels"] or [],
"source_confidence": float(row["source_confidence"] or 0.0),
},
}
# ── Client 360 Endpoint ────────────────────────────────────────────────────
@router.get("/crm/client-360/{person_id}", tags=["CRM Client 360"])
async def client_360(
request: Request,
person_id: str,
user: UserPrincipal = Depends(get_current_user),
) -> dict[str, Any]:
"""
Aggregated Client360 dossier — identity, opportunities, interactions,
property interests, tasks, QD overview, risk flags, and next actions.
Derived read model — not primary truth.
"""
pool = await _get_pool(request)
async with pool.acquire() as conn:
snapshot = await get_client_360(conn, _tenant_scope(user), person_id)
if not snapshot:
raise HTTPException(status_code=404, detail=f"Client '{person_id}' not found.")
return {"status": "ok", "data": snapshot}
# ── Opportunities Endpoint ─────────────────────────────────────────────────
@router.get("/crm/opportunities", tags=["CRM Opportunities"])
async def list_opportunities(
request: Request,
stage: str | None = None,
limit: int = Query(default=50, ge=1, le=200),
offset: int = Query(default=0, ge=0),
user: UserPrincipal = Depends(get_current_user),
) -> dict[str, Any]:
"""Canonical opportunity pipeline list."""
pool = await _get_pool(request)
clauses = ["co.tenant_id = $1", "cl.tenant_id = $1", "p.tenant_id = $1"]
params: list[Any] = [_tenant_scope(user)]
if stage:
params.append(stage)
clauses.append(f"co.stage = ${len(params)}::crm_opportunity_stage")
params.extend([limit, offset])
where = " AND ".join(clauses)
async with pool.acquire() as conn:
rows = await conn.fetch(
f"""
SELECT co.opportunity_id, co.stage, co.value, co.probability,
co.expected_close_date, co.next_action,
p.person_id, p.full_name, p.primary_phone,
ip.project_name,
co.created_at, co.updated_at
FROM crm_opportunities co
INNER JOIN crm_leads cl ON cl.lead_id = co.lead_id
INNER JOIN crm_people p ON p.person_id = cl.person_id
LEFT JOIN inventory_projects ip ON ip.project_id = co.project_id
WHERE {where}
ORDER BY co.updated_at DESC
LIMIT ${len(params) - 1} OFFSET ${len(params)}
""",
*params,
)
opportunities = [
_opportunity_payload(r)
for r in rows
]
return {"status": "ok", "data": opportunities, "meta": {"count": len(opportunities)}}
@router.patch("/crm/opportunities/{opportunity_id}", tags=["CRM Opportunities"])
async def update_opportunity(
request: Request,
opportunity_id: str,
body: UpdateOpportunityRequest,
user: UserPrincipal = Depends(get_current_user),
) -> dict[str, Any]:
"""Update canonical opportunity/deal fields without leaving tenant scope."""
pool = await _get_pool(request)
updated_fields = _fields_set(body)
if not updated_fields:
raise HTTPException(status_code=422, detail="At least one opportunity field must be provided.")
next_stage = None
if "stage" in updated_fields:
if body.stage is None:
raise HTTPException(status_code=422, detail="Opportunity stage cannot be null.")
next_stage = _normalize_choice(
body.stage,
allowed=CANONICAL_OPPORTUNITY_STAGES,
field_name="opportunity stage",
)
expected_close_date = None
if "expected_close_date" in updated_fields:
expected_close_date = _parse_optional_date(body.expected_close_date, field_name="expected_close_date")
async with pool.acquire() as conn:
existing = await conn.fetchrow(
"""
SELECT co.opportunity_id, co.stage, co.value, co.probability,
co.expected_close_date, co.next_action, co.notes,
p.person_id, p.full_name, p.primary_phone,
ip.project_name
FROM crm_opportunities co
INNER JOIN crm_leads cl
ON cl.lead_id = co.lead_id
AND cl.tenant_id = co.tenant_id
INNER JOIN crm_people p
ON p.person_id = cl.person_id
AND p.tenant_id = co.tenant_id
LEFT JOIN inventory_projects ip ON ip.project_id = co.project_id
WHERE co.opportunity_id = $1::uuid
AND co.tenant_id = $2
""",
opportunity_id,
_tenant_scope(user),
)
if not existing:
raise HTTPException(status_code=404, detail=f"Opportunity '{opportunity_id}' not found.")
row = await conn.fetchrow(
"""
WITH updated AS (
UPDATE crm_opportunities co
SET stage = COALESCE($3::crm_opportunity_stage, co.stage),
value = CASE WHEN $4::boolean THEN $5::numeric ELSE co.value END,
probability = CASE WHEN $6::boolean THEN $7::int ELSE co.probability END,
expected_close_date = CASE WHEN $8::boolean THEN $9::date ELSE co.expected_close_date END,
next_action = CASE WHEN $10::boolean THEN $11 ELSE co.next_action END,
notes = CASE WHEN $12::boolean THEN $13 ELSE co.notes END,
updated_at = NOW()
FROM crm_leads cl
WHERE co.opportunity_id = $1::uuid
AND co.tenant_id = $2
AND cl.lead_id = co.lead_id
AND cl.tenant_id = co.tenant_id
RETURNING co.opportunity_id, co.lead_id, co.stage, co.value, co.probability,
co.expected_close_date, co.next_action, co.notes, co.project_id, co.updated_at
)
SELECT updated.opportunity_id, updated.stage, updated.value, updated.probability,
updated.expected_close_date, updated.next_action, updated.notes,
p.person_id, p.full_name, p.primary_phone,
ip.project_name
FROM updated
INNER JOIN crm_leads cl
ON cl.lead_id = updated.lead_id
AND cl.tenant_id = $2
INNER JOIN crm_people p
ON p.person_id = cl.person_id
AND p.tenant_id = $2
LEFT JOIN inventory_projects ip ON ip.project_id = updated.project_id
""",
opportunity_id,
_tenant_scope(user),
next_stage,
"value" in updated_fields,
body.value,
"probability" in updated_fields,
body.probability,
"expected_close_date" in updated_fields,
expected_close_date,
"next_action" in updated_fields,
body.next_action,
"notes" in updated_fields,
body.notes,
)
return {
"status": "ok",
"data": _opportunity_payload(row),
"meta": {
"previous_stage": existing["stage"],
"changed": _opportunity_payload(existing) != _opportunity_payload(row),
},
}
# ── Tasks / Reminders Endpoint ─────────────────────────────────────────────
@router.get("/crm/tasks", tags=["CRM Tasks"])
async def list_tasks(
request: Request,
status_filter: str | None = Query(default="pending", alias="status"),
assigned_to: str | None = None,
limit: int = Query(default=50, ge=1, le=200),
user: UserPrincipal = Depends(get_current_user),
) -> dict[str, Any]:
"""Reminder / task inbox for the CRM operator."""
pool = await _get_pool(request)
clauses = ["ir.tenant_id = $1"]
params: list[Any] = [_tenant_scope(user)]
if status_filter and status_filter.lower() != "all":
params.append(status_filter)
clauses.append(f"ir.status = ${len(params)}")
if assigned_to:
params.append(assigned_to)
clauses.append(f"ir.assigned_to = ${len(params)}::uuid")
params.append(limit)
where = " AND ".join(clauses)
async with pool.acquire() as conn:
rows = await conn.fetch(
f"""
SELECT ir.reminder_id, ir.reminder_type, ir.title, ir.notes,
ir.due_at, ir.status, ir.priority,
p.person_id, p.full_name, p.primary_phone
FROM intel_reminders ir
INNER JOIN crm_people p
ON p.person_id = ir.person_id
AND p.tenant_id = ir.tenant_id
WHERE {where}
ORDER BY ir.due_at ASC NULLS LAST, ir.created_at DESC
LIMIT ${len(params)}
""",
*params,
)
tasks = [
{
"reminder_id": str(r["reminder_id"]),
"reminder_type": r["reminder_type"],
"title": r["title"],
"notes": r["notes"],
"due_at": r["due_at"].isoformat() if r["due_at"] else None,
"status": r["status"],
"priority": r["priority"],
"person_id": str(r["person_id"]),
"client_name": r["full_name"],
"client_phone": r["primary_phone"],
}
for r in rows
]
return {"status": "ok", "data": tasks, "meta": {"count": len(tasks)}}
@router.post("/crm/tasks", status_code=201, tags=["CRM Tasks"])
async def create_task(
request: Request,
body: CreateReminderRequest,
user: UserPrincipal = Depends(get_current_user),
) -> dict[str, Any]:
"""Create a reminder / follow-up task."""
pool = await _get_pool(request)
reminder_id = str(uuid.uuid4())
next_priority = _normalize_choice(
body.priority,
allowed=CANONICAL_TASK_PRIORITIES,
field_name="task priority",
)
async with pool.acquire() as conn:
due_dt = _parse_optional_datetime(body.due_at, field_name="due_at")
person = await conn.fetchrow(
"""
SELECT person_id
FROM crm_people
WHERE person_id = $1::uuid
AND tenant_id = $2
""",
body.person_id,
_tenant_scope(user),
)
if not person:
raise HTTPException(status_code=404, detail=f"Contact '{body.person_id}' not found.")
if body.lead_id:
lead = await conn.fetchrow(
"""
SELECT lead_id
FROM crm_leads
WHERE lead_id = $1::uuid
AND person_id = $2::uuid
AND tenant_id = $3
""",
body.lead_id,
body.person_id,
_tenant_scope(user),
)
if not lead:
raise HTTPException(status_code=404, detail=f"Lead '{body.lead_id}' not found.")
await conn.execute(
"""
INSERT INTO intel_reminders (
reminder_id, tenant_id, person_id, lead_id, reminder_type, title,
notes, due_at, status, priority, created_by_type, created_at
) VALUES (
$1::uuid, $2, $3::uuid, $4::uuid, $5, $6, $7, $8, 'pending', $9, 'human', NOW()
)
""",
reminder_id,
_tenant_scope(user),
body.person_id,
body.lead_id,
body.reminder_type,
body.title,
body.notes,
due_dt,
next_priority,
)
return {"status": "ok", "data": {"reminder_id": reminder_id, "title": body.title}}
@router.patch("/crm/tasks/{reminder_id}", tags=["CRM Tasks"])
async def update_task(
request: Request,
reminder_id: str,
body: UpdateReminderRequest,
user: UserPrincipal = Depends(get_current_user),
) -> dict[str, Any]:
"""Update canonical reminder lifecycle state for an operator-visible task."""
pool = await _get_pool(request)
next_status = _normalize_choice(
body.status,
allowed=CANONICAL_TASK_STATUSES,
field_name="task status",
)
due_dt = _parse_optional_datetime(body.due_at, field_name="due_at")
now = datetime.now(timezone.utc)
if next_status == "snoozed":
if due_dt is None:
raise HTTPException(status_code=422, detail="Snoozed reminders require a future 'due_at' timestamp.")
if due_dt <= now:
raise HTTPException(status_code=422, detail="Snoozed reminders must be assigned a future 'due_at' timestamp.")
async with pool.acquire() as conn:
existing = await conn.fetchrow(
"""
SELECT reminder_id, status, due_at
FROM intel_reminders
WHERE reminder_id = $1::uuid
AND tenant_id = $2
""",
reminder_id,
_tenant_scope(user),
)
if not existing:
raise HTTPException(status_code=404, detail=f"Task '{reminder_id}' not found.")
row = await conn.fetchrow(
"""
UPDATE intel_reminders ir
SET status = $3,
due_at = CASE
WHEN $4::timestamptz IS NULL THEN ir.due_at
ELSE $4::timestamptz
END,
notes = COALESCE($5, ir.notes),
completed_at = CASE
WHEN $3 = 'done' THEN COALESCE(ir.completed_at, NOW())
WHEN $3 IN ('pending', 'confirmed', 'snoozed', 'cancelled') THEN NULL
ELSE ir.completed_at
END
FROM crm_people p
WHERE ir.reminder_id = $1::uuid
AND ir.tenant_id = $2
AND p.person_id = ir.person_id
AND p.tenant_id = ir.tenant_id
RETURNING ir.reminder_id,
ir.reminder_type,
ir.title,
ir.notes,
ir.due_at,
ir.status,
ir.priority,
p.person_id,
p.full_name,
p.primary_phone
""",
reminder_id,
_tenant_scope(user),
next_status,
due_dt,
body.notes,
)
return {
"status": "ok",
"data": {
"reminder_id": str(row["reminder_id"]),
"reminder_type": row["reminder_type"],
"title": row["title"],
"notes": row["notes"],
"due_at": row["due_at"].isoformat() if row["due_at"] else None,
"status": row["status"],
"priority": row["priority"],
"person_id": str(row["person_id"]),
"client_name": row["full_name"],
"client_phone": row["primary_phone"],
},
"meta": {
"previous_status": existing["status"],
"changed": existing["status"] != row["status"],
},
}
# ── Canonical Kanban (from crm_leads) ─────────────────────────────────────
@router.get("/crm/kanban", tags=["CRM Kanban"])
async def get_canonical_kanban(
request: Request,
user: UserPrincipal = Depends(get_current_user),
) -> dict[str, Any]:
"""
Canonical Kanban board from crm_leads table.
Groups clients by lead status with QD summary.
"""
pool = await _get_pool(request)
async with pool.acquire() as conn:
rows = await conn.fetch(
"""
SELECT
cl.lead_id, cl.status, cl.budget_band, cl.urgency,
p.person_id, p.full_name, p.primary_phone, p.buyer_type,
COALESCE(qs.intent_value, 0.0) AS intent_score
FROM crm_leads cl
INNER JOIN crm_people p
ON p.person_id = cl.person_id
AND p.tenant_id = cl.tenant_id
LEFT JOIN LATERAL (
SELECT MAX(CASE WHEN score_type = 'intent_score' THEN current_value END) AS intent_value
FROM intel_qd_scores
WHERE tenant_id = cl.tenant_id
AND person_id = p.person_id
) qs ON TRUE
WHERE cl.tenant_id = $1
ORDER BY qs.intent_value DESC NULLS LAST, cl.updated_at DESC
""",
_tenant_scope(user),
)
grouped: dict[str, list[dict]] = {s: [] for s in CANONICAL_LEAD_STAGES}
for r in rows:
s = r["status"] or "new"
grouped.setdefault(s, []).append({
"lead_id": str(r["lead_id"]),
"person_id": str(r["person_id"]),
"client_name": r["full_name"],
"client_phone": r["primary_phone"],
"buyer_type": r["buyer_type"],
"budget_band": r["budget_band"],
"urgency": r["urgency"],
"intent_score": float(r["intent_score"]),
})
board = [
{
"status": s,
"label": s.replace("_", " ").title(),
"count": len(grouped.get(s, [])),
"items": grouped.get(s, []),
}
for s in CANONICAL_LEAD_STAGES
]
return {"status": "ok", "data": board}
@router.patch("/crm/leads/{lead_id}/stage", tags=["CRM Kanban"])
async def update_lead_stage(
request: Request,
lead_id: str,
body: UpdateLeadStageRequest,
user: UserPrincipal = Depends(get_current_user),
) -> dict[str, Any]:
"""Move a canonical CRM lead between funnel stages and persist audit history."""
pool = await _get_pool(request)
next_status = _normalize_choice(
body.status,
allowed=CANONICAL_LEAD_STAGES,
field_name="lead stage",
)
changed_by = _optional_uuid(user.user_id)
async with pool.acquire() as conn:
existing = await conn.fetchrow(
"""
SELECT cl.lead_id,
cl.person_id,
cl.status,
cl.budget_band,
cl.urgency,
p.full_name,
p.primary_phone
FROM crm_leads cl
INNER JOIN crm_people p
ON p.person_id = cl.person_id
AND p.tenant_id = cl.tenant_id
WHERE cl.lead_id = $1::uuid
AND cl.tenant_id = $2
""",
lead_id,
_tenant_scope(user),
)
if not existing:
raise HTTPException(status_code=404, detail=f"Lead '{lead_id}' not found.")
changed = existing["status"] != next_status
if changed:
updated = await conn.fetchrow(
"""
UPDATE crm_leads
SET status = $3::crm_lead_status,
updated_at = NOW()
WHERE lead_id = $1::uuid
AND tenant_id = $2
RETURNING lead_id, person_id, status, budget_band, urgency
""",
lead_id,
_tenant_scope(user),
next_status,
)
await conn.execute(
"""
INSERT INTO crm_stage_history (
history_id, lead_id, from_status, to_status, changed_by,
changed_by_type, notes, happened_at
) VALUES (
$1::uuid, $2::uuid, $3, $4, $5::uuid, 'human', $6, NOW()
)
""",
str(uuid.uuid4()),
lead_id,
existing["status"],
next_status,
changed_by,
body.notes,
)
else:
updated = existing
return {
"status": "ok",
"data": {
"lead_id": str(updated["lead_id"]),
"person_id": str(updated["person_id"]),
"status": updated["status"],
"budget_band": updated["budget_band"],
"urgency": updated["urgency"],
"client_name": existing["full_name"],
"client_phone": existing["primary_phone"],
},
"meta": {
"previous_status": existing["status"],
"changed": changed,
},
}
# ── QD Score Access ────────────────────────────────────────────────────────
@router.get("/crm/qd/{person_id}", tags=["CRM QD"])
async def get_qd_score(
request: Request,
person_id: str,
user: UserPrincipal = Depends(get_current_user),
) -> dict[str, Any]:
"""QD score summary and recent timeseries for a client."""
pool = await _get_pool(request)
async with pool.acquire() as conn:
scores = await conn.fetch(
"""
SELECT score_type, current_value, computed_at, reasoning
FROM intel_qd_scores
WHERE person_id = $1::uuid
AND tenant_id = $2
""",
person_id,
_tenant_scope(user),
)
timeseries = await conn.fetch(
"""
SELECT score_type, value, timestamp, signal_source, delta
FROM intel_qd_timeseries
WHERE person_id = $1::uuid
AND tenant_id = $2
ORDER BY timestamp DESC
LIMIT 50
""",
person_id,
_tenant_scope(user),
)
if not scores:
raise HTTPException(status_code=404, detail=f"No QD scores for client '{person_id}'.")
return {
"status": "ok",
"data": {
"person_id": person_id,
"scores": {
r["score_type"]: {
"current_value": float(r["current_value"]),
"computed_at": r["computed_at"].isoformat() if r["computed_at"] else None,
"reasoning": r["reasoning"],
}
for r in scores
},
"timeseries": [
{
"score_type": r["score_type"],
"value": float(r["value"]),
"timestamp": r["timestamp"].isoformat() if r["timestamp"] else None,
"signal_source": r["signal_source"],
"delta": float(r["delta"]) if r["delta"] else None,
}
for r in timeseries
],
},
}
# ── Oracle Client Data Lens ───────────────────────────────────────────────────
@router.get("/crm/client-data", tags=["CRM Client Data"])
async def list_client_data(
request: Request,
search: str | None = Query(default=None),
limit: int = Query(default=50, ge=1, le=200),
offset: int = Query(default=0, ge=0),
) -> dict[str, Any]:
pool = await _get_pool(request)
filter_params: list[Any] = []
where = "1=1"
if search:
filter_params.append(f"%{search.lower()}%")
where = f"(lower(p.full_name) LIKE ${len(filter_params)} OR lower(COALESCE(p.primary_phone,'')) LIKE ${len(filter_params)} OR lower(COALESCE(p.primary_email,'')) LIKE ${len(filter_params)} OR lower(COALESCE(pi.projects,'')) LIKE ${len(filter_params)})"
row_params = [*filter_params, limit, offset]
async with pool.acquire() as conn:
total_count = await conn.fetchval(
f"""
WITH interests AS (
SELECT person_id, string_agg(DISTINCT project_name, ', ') AS projects
FROM crm_property_interests GROUP BY person_id
)
SELECT COUNT(*)::int
FROM crm_people p
LEFT JOIN interests pi ON pi.person_id = p.person_id
WHERE {where}
""",
*filter_params,
)
rows = await conn.fetch(
f"""
WITH interests AS (
SELECT person_id, string_agg(DISTINCT project_name, ', ') AS projects, COUNT(*)::int AS interest_count
FROM crm_property_interests GROUP BY person_id
),
qd AS (
SELECT DISTINCT ON (person_id) person_id, current_value
FROM intel_qd_scores
ORDER BY person_id, current_value DESC, computed_at DESC
)
SELECT p.person_id::text, p.full_name, p.primary_email, p.primary_phone, p.buyer_type,
p.broker_name, p.communication_preference, p.best_contact_time,
l.lead_id::text, l.status::text AS lead_status, l.budget_band, l.urgency,
COALESCE(qd.current_value, p.engagement_score, 0)::float AS qd_score,
lc.last_contact_at, lc.last_channel, lc.days_since_contact,
nba.recommended_action AS next_best_action, nba.priority AS next_action_priority,
COALESCE(pi.projects, '') AS projects, COALESCE(pi.interest_count, 0)::int AS interest_count
FROM crm_people p
LEFT JOIN LATERAL (
SELECT * FROM crm_leads l WHERE l.person_id = p.person_id ORDER BY l.updated_at DESC LIMIT 1
) l ON TRUE
LEFT JOIN interests pi ON pi.person_id = p.person_id
LEFT JOIN qd ON qd.person_id = p.person_id
LEFT JOIN read_last_contacted lc ON lc.person_id = p.person_id
LEFT JOIN read_next_best_action nba ON nba.person_id = p.person_id
WHERE {where}
ORDER BY lc.last_contact_at DESC NULLS LAST, qd_score DESC, p.full_name ASC
LIMIT ${len(row_params)-1} OFFSET ${len(row_params)}
""",
*row_params,
)
return {
"status": "ok",
"data": [dict(r) for r in rows],
"meta": {
"count": len(rows),
"total_count": total_count or 0,
"has_more": offset + len(rows) < (total_count or 0),
"limit": limit,
"offset": offset,
},
}
@router.get("/crm/client-data/{person_id}", tags=["CRM Client Data"])
async def get_client_data(request: Request, person_id: str) -> dict[str, Any]:
pool = await _get_pool(request)
async with pool.acquire() as conn:
base = await conn.fetchrow(
"""
SELECT p.*, l.lead_id::text, l.status::text AS lead_status, l.budget_band, l.urgency,
l.financing_posture, l.timeline_to_decision, l.broker_team,
lc.last_contact_at, lc.last_channel, lc.days_since_contact,
nba.recommended_action, nba.priority AS next_action_priority, nba.rationale,
nba.suggested_channel, nba.due_within_days
FROM crm_people p
LEFT JOIN LATERAL (
SELECT * FROM crm_leads l WHERE l.person_id = p.person_id ORDER BY l.updated_at DESC LIMIT 1
) l ON TRUE
LEFT JOIN read_last_contacted lc ON lc.person_id = p.person_id
LEFT JOIN read_next_best_action nba ON nba.person_id = p.person_id
WHERE p.person_id = $1::uuid
""",
person_id,
)
if not base:
raise HTTPException(status_code=404, detail=f"Client '{person_id}' not found.")
interests = await conn.fetch("SELECT * FROM crm_property_interests WHERE person_id = $1::uuid ORDER BY priority ASC, created_at DESC", person_id)
opportunities = await conn.fetch(
"""
SELECT o.*, ip.project_name FROM crm_opportunities o
JOIN crm_leads l ON l.lead_id = o.lead_id
LEFT JOIN inventory_projects ip ON ip.project_id = o.project_id
WHERE l.person_id = $1::uuid ORDER BY o.updated_at DESC
""",
person_id,
)
facts = await conn.fetch("SELECT * FROM intel_extracted_facts WHERE person_id = $1::uuid ORDER BY extracted_at DESC LIMIT 50", person_id)
qd = await conn.fetch("SELECT * FROM intel_qd_scores WHERE person_id = $1::uuid ORDER BY current_value DESC", person_id)
timeline = await _client_timeline(conn, person_id, 60)
return {
"status": "ok",
"data": {
"profile": dict(base),
"property_interests": [dict(r) for r in interests],
"opportunities": [dict(r) for r in opportunities],
"extracted_facts": [dict(r) for r in facts],
"qd_scores": [dict(r) for r in qd],
"timeline": timeline,
},
}
@router.patch("/crm/client-data/{person_id}", tags=["CRM Client Data"])
async def patch_client_data(
request: Request,
person_id: str,
body: ClientDataPatchRequest,
user: UserPrincipal = Depends(get_current_user),
) -> dict[str, Any]:
pool = await _get_pool(request)
payload = body.model_dump(exclude_unset=True)
tenant_id = _tenant_scope(user)
if "lead_status" in payload and payload["lead_status"] is not None:
payload["lead_status"] = _normalize_choice(
payload["lead_status"],
allowed=CANONICAL_LEAD_STAGES,
field_name="lead status",
)
if "urgency" in payload and payload["urgency"] is not None:
payload["urgency"] = _normalize_choice(
payload["urgency"],
allowed=CANONICAL_URGENCY_VALUES,
field_name="urgency",
)
person_fields = {k: payload[k] for k in ("full_name", "primary_email", "primary_phone", "buyer_type", "communication_preference", "best_contact_time") if k in payload}
lead_fields = {k: payload[k] for k in ("budget_band", "urgency") if k in payload}
async with pool.acquire() as conn:
async with conn.transaction():
person_exists = await conn.fetchval(
"SELECT EXISTS (SELECT 1 FROM crm_people WHERE person_id = $1::uuid AND tenant_id = $2)",
person_id,
tenant_id,
)
if not person_exists:
raise HTTPException(status_code=404, detail=f"Client '{person_id}' not found.")
if person_fields:
sets = ", ".join(f"{key} = ${idx}" for idx, key in enumerate(person_fields, start=1))
await conn.execute(
f"UPDATE crm_people SET {sets}, updated_at = NOW() WHERE person_id = ${len(person_fields)+1}::uuid AND tenant_id = ${len(person_fields)+2}",
*person_fields.values(),
person_id,
tenant_id,
)
if lead_fields:
lead_id = await conn.fetchval(
"SELECT lead_id FROM crm_leads WHERE person_id = $1::uuid AND tenant_id = $2 ORDER BY updated_at DESC LIMIT 1",
person_id,
tenant_id,
)
if lead_id:
sets = ", ".join(f"{key} = ${idx}" for idx, key in enumerate(lead_fields, start=1))
await conn.execute(
f"UPDATE crm_leads SET {sets}, updated_at = NOW() WHERE lead_id = ${len(lead_fields)+1}::uuid AND tenant_id = ${len(lead_fields)+2}",
*lead_fields.values(),
str(lead_id),
tenant_id,
)
if "lead_status" in payload:
await conn.execute(
"UPDATE crm_leads SET status = $1::crm_lead_status, updated_at = NOW() WHERE person_id = $2::uuid AND tenant_id = $3",
payload["lead_status"],
person_id,
tenant_id,
)
return {"status": "ok", "data": {"person_id": person_id, "updated": sorted(payload.keys())}}
@router.get("/crm/client-data/{person_id}/timeline", tags=["CRM Client Data"])
async def get_client_data_timeline(request: Request, person_id: str, limit: int = Query(default=80, ge=1, le=200)) -> dict[str, Any]:
pool = await _get_pool(request)
async with pool.acquire() as conn:
rows = await _client_timeline(conn, person_id, limit)
return {"status": "ok", "data": rows}
@router.post("/crm/client-data/{person_id}/tasks", status_code=201, tags=["CRM Client Data"])
async def create_client_data_task(
request: Request,
person_id: str,
body: CreateReminderRequest,
user: UserPrincipal = Depends(get_current_user),
) -> dict[str, Any]:
patched = body.model_copy(update={"person_id": person_id})
return await create_task(request, patched, user)
async def _client_timeline(conn: Any, person_id: str, limit: int) -> list[dict[str, Any]]:
rows = await conn.fetch(
"""
SELECT * FROM (
SELECT i.interaction_id::text AS id, i.channel::text AS type, i.interaction_type AS title,
i.summary, i.happened_at AS date, i.broker_name AS actor
FROM intel_interactions i WHERE i.person_id = $1::uuid
UNION ALL
SELECT m.message_id::text, 'message', m.sender_role, m.message_text, m.delivered_at, m.sender_name
FROM intel_messages m JOIN intel_interactions i ON i.interaction_id = m.interaction_id WHERE i.person_id = $1::uuid
UNION ALL
SELECT r.reminder_id::text, 'reminder', r.title, r.notes, r.due_at, r.priority
FROM intel_reminders r WHERE r.person_id = $1::uuid
UNION ALL
SELECT v.visit_id::text, 'visit', COALESCE(v.outcome_type, 'site_visit'), COALESCE(v.visit_notes, v.broker_notes), v.visited_at, v.broker_name
FROM intel_visits v WHERE v.person_id = $1::uuid
UNION ALL
SELECT q.timeseries_id::text, 'qd', q.score_type, q.signal_source, q.timestamp, q.value::text
FROM intel_qd_timeseries q WHERE q.person_id = $1::uuid
) events
ORDER BY date DESC NULLS LAST
LIMIT $2
""",
person_id,
limit,
)
return [dict(r) for r in rows]
Reference in New Issue View Git Blame Copy Permalink