app/services/field_mapping.py

"""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