Add Universal Connector: POST /v1/predict/smart — auto-map any columns + derive dates, with mapping report

app/routers/predict.py

@@ -14,7 +14,8 @@ from app.dependencies import (
 )
 from app.models.user import User
 from app.models.mlmodel import MLModel
-from app.schemas import PredictionInput, PredictionResponse, SinglePrediction
+from app.schemas import PredictionInput, PredictionResponse, SinglePrediction, SmartPredictInput
+from app.services.field_mapping import normalize_rows
 from app.services.scoring import _churn_factor, _lead_factor
 from app.services.training import predict_with_model
 from app.services.benchmarking import update_benchmarks, compare_to_benchmark
@@ -92,6 +93,51 @@ def _score_with_custom_model(ml_model, data: list, model_type: str) -> list:
     return results
 
 
+@router.post("/smart", response_model=PredictionResponse)
+def predict_smart(
+    body: SmartPredictInput,
+    user: User = Depends(get_current_user),
+    db: Session = Depends(get_db),
+):
+    """Universal connector — send rows with ANY column names. We auto-map them
+    to RevAI's signals (and derive durations from dates), then score. The
+    response includes a `field_mapping` report of what was matched/missed."""
+    if body.model_type not in ("churn", "lead"):
+        raise HTTPException(status_code=400, detail="model_type must be 'churn' or 'lead'")
+
+    rows, report = normalize_rows(body.data, body.mapping, body.model_type)
+
+    n_predictions = len(rows)
+    check_rate_limit(user, db, f"predict/{body.model_type}", n_predictions)
+    check_prediction_quota(user, db, n_predictions)
+
+    if body.model_id:
+        ml_model = db.query(MLModel).filter(
+            MLModel.id == body.model_id, MLModel.user_id == user.id
+        ).first()
+        if not ml_model:
+            raise HTTPException(status_code=404, detail="Model not found")
+        results = _score_with_custom_model(ml_model, rows, body.model_type)
+        model_label = f"custom_ml_{body.model_id[:8]}"
+    else:
+        results = _apply_heuristics(rows, body.model_type)
+        model_label = "heuristic"
+
+    track_usage(user, db, f"predict/{body.model_type}", n_predictions)
+
+    all_scores = [p.score for p in results]
+    update_benchmarks(db, all_scores, body.model_type)
+    benchmark = compare_to_benchmark(all_scores, body.model_type, db)
+
+    return PredictionResponse(
+        predictions=results,
+        model_used=model_label,
+        usage=get_usage_summary(user, db),
+        benchmark=benchmark,
+        field_mapping=report,
+    )
+
+
 @router.post("/csv", response_model=PredictionResponse)
 async def predict_csv(
     file: UploadFile = File(..., description="CSV export of your customers/leads"),

app/schemas/__init__.py

@@ -85,6 +85,16 @@ class PredictionResponse(BaseModel):
     model_used: str  # "heuristic" or "custom_ml_<id>"
     usage: Dict[str, Any]
     benchmark: Optional[Dict[str, Any]] = None  # comparison vs industry
+    field_mapping: Optional[Dict[str, Any]] = None  # set by the universal connector
+
+
+class SmartPredictInput(BaseModel):
+    data: List[Dict[str, Any]] = Field(..., min_length=1, max_length=1000,
+        description="Your rows with WHATEVER column names you already have")
+    mapping: Optional[Dict[str, str]] = Field(None,
+        description="Optional {canonical_signal: your_column} overrides; auto-detected otherwise")
+    model_type: str = Field("churn", description="'churn' or 'lead'")
+    model_id: Optional[str] = Field(None, description="Optional trained model ID")
 
 
 # ── Training ──

app/services/field_mapping.py

@@ -0,0 +1,143 @@
+"""Universal Connector — map ANY customer data onto RevAI's signals.
+
+Real-world exports never use our exact column names (`last_seen` not
+`days_since_last_login`, `signup_date` not `tenure_days`). This layer:
+  1. honors an explicit {canonical: source_column} mapping if given,
+  2. otherwise auto-detects columns by a big alias table,
+  3. derives durations from dates (signup_date -> tenure_days),
+  4. returns a transparency report of what it matched / missed.
+So a customer can send whatever they already have and still get scored.
+"""
+import datetime
+from typing import Any, Dict, List, Optional, Tuple
+
+from dateutil import parser as _dateparser
+
+# canonical signal -> source-column aliases (priority order; canonical name first)
+CHURN_ALIASES: Dict[str, List[str]] = {
+    "customer_id": ["customer_id", "id", "user_id", "account_id", "email", "customer"],
+    "tenure_days": ["tenure_days", "tenure", "account_age_days", "account_age",
+                    "customer_since", "signup_date", "signup", "created_at", "created",
+                    "join_date", "date_joined", "start_date"],
+    "days_since_last_login": ["days_since_last_login", "days_inactive", "last_login",
+                              "last_login_date", "last_seen", "last_seen_date",
+                              "last_active", "last_activity"],
+    "login_frequency_7d": ["login_frequency_7d", "logins_7d", "weekly_logins",
+                           "login_count_7d", "logins_per_week"],
+    "payment_delays_90d": ["payment_delays_90d", "payment_delays", "late_payments",
+                           "failed_payments", "missed_payments", "overdue_count"],
+    "support_tickets_last_30d": ["support_tickets_last_30d", "support_tickets",
+                                 "tickets_30d", "open_tickets", "tickets"],
+    "contract_type": ["contract_type", "plan_interval", "billing_cycle",
+                      "billing_interval", "subscription_type", "plan"],
+    "nps_score": ["nps_score", "nps", "satisfaction", "csat"],
+    "feature_adoption_score": ["feature_adoption_score", "feature_adoption", "adoption",
+                               "adoption_rate", "feature_usage"],
+    "avg_session_minutes": ["avg_session_minutes", "session_length", "avg_session_time",
+                            "avg_session", "session_minutes"],
+    "subscription_status": ["subscription_status", "sub_status", "billing_status", "status"],
+}
+
+LEAD_ALIASES: Dict[str, List[str]] = {
+    "lead_id": ["lead_id", "id", "contact_id", "email", "lead"],
+    "demo_requested": ["demo_requested", "requested_demo", "demo"],
+    "budget_confirmed": ["budget_confirmed", "has_budget", "budget"],
+    "decision_maker_contacted": ["decision_maker_contacted", "dm_contacted",
+                                 "reached_dm", "decision_maker"],
+    "engagement_score": ["engagement_score", "engagement"],
+    "source": ["source", "lead_source", "channel"],
+    "days_in_pipeline": ["days_in_pipeline", "pipeline_days", "age_in_pipeline",
+                         "entered_pipeline", "pipeline_entry"],
+    "previous_conversations": ["previous_conversations", "conversations",
+                               "num_conversations", "touchpoints"],
+    "content_downloads": ["content_downloads", "downloads", "content_downloaded"],
+    "email_opens": ["email_opens", "opens", "email_opened"],
+    "website_visits": ["website_visits", "visits", "page_views", "sessions"],
+}
+
+# canonical fields that should become "days since <date>" when given a date value
+DATE_DERIVED = {"tenure_days", "days_since_last_login", "days_in_pipeline"}
+
+
+def _norm(k: Any) -> str:
+    return str(k).strip().lower().replace(" ", "_").replace("-", "_")
+
+
+def _is_number(v: Any) -> bool:
+    if isinstance(v, (int, float)):
+        return True
+    s = str(v).strip()
+    if not s:
+        return False
+    return s.replace(".", "", 1).replace("-", "", 1).isdigit()
+
+
+def _looks_like_date(v: Any) -> bool:
+    if _is_number(v) or v is None:
+        return False
+    try:
+        _dateparser.parse(str(v))
+        return True
+    except (ValueError, OverflowError, TypeError):
+        return False
+
+
+def _days_since(v: Any) -> Optional[int]:
+    try:
+        dt = _dateparser.parse(str(v))
+    except (ValueError, OverflowError, TypeError):
+        return None
+    now = datetime.datetime.now(dt.tzinfo) if dt.tzinfo else datetime.datetime.now()
+    return max(0, (now - dt).days)
+
+
+def normalize_rows(
+    data: List[Dict[str, Any]],
+    mapping: Optional[Dict[str, str]] = None,
+    model_type: str = "churn",
+) -> Tuple[List[Dict[str, Any]], Dict[str, Any]]:
+    """Return (canonical_rows, report). report shows matched/derived/missing signals."""
+    aliases = CHURN_ALIASES if model_type == "churn" else LEAD_ALIASES
+    explicit = {c: _norm(src) for c, src in (mapping or {}).items()}
+
+    report: Dict[str, Any] = {"matched": {}, "missing": [], "ignored_columns": []}
+    used_source_keys = set()
+    out_rows: List[Dict[str, Any]] = []
+
+    for row in data:
+        norm_row = {_norm(k): v for k, v in row.items()}
+        canon_row: Dict[str, Any] = {}
+
+        for canon, alias_list in aliases.items():
+            src_key = None
+            if canon in explicit and explicit[canon] in norm_row:
+                src_key = explicit[canon]
+            else:
+                for a in alias_list:
+                    if _norm(a) in norm_row:
+                        src_key = _norm(a)
+                        break
+            if src_key is None:
+                continue
+
+            val = norm_row[src_key]
+            how = "direct"
+            if canon in DATE_DERIVED and _looks_like_date(val):
+                d = _days_since(val)
+                if d is not None:
+                    val, how = d, f"derived from date in '{src_key}'"
+
+            canon_row[canon] = val
+            used_source_keys.add(src_key)
+            if canon not in report["matched"]:
+                report["matched"][canon] = {"source_column": src_key, "how": how}
+
+        out_rows.append(canon_row)
+
+    # transparency: which signals never matched, and which columns we ignored
+    report["missing"] = [c for c in aliases if c not in report["matched"]]
+    if data:
+        all_cols = {_norm(k) for k in data[0].keys()}
+        report["ignored_columns"] = sorted(all_cols - used_source_keys)
+
+    return out_rows, report