MarketingTool/apps/api/app/ai_assistance.py

"""Evidence-bounded AI enrichment with a deterministic, network-free provider.

Remote providers are intentionally only a status/configuration concept here.  No
network client is present, so an unconfigured or unreviewed remote provider fails
closed and cannot accidentally perform outreach or exfiltrate tenant data.
"""
from __future__ import annotations

import hashlib
import json
import os
import re
from typing import Any

MAX_INPUT_ITEMS = 100
MAX_FIELD_CHARS = 500
MAX_OUTPUT_CHARS = 12_000
_SECRET_RE = re.compile(r"(?i)(password|passwd|secret|token|api[_-]?key|authorization|private[_-]?key|credential)\s*[:=]\s*[^\s,;]+")
_SECRET_KEY_RE = re.compile(r"(?i)(password|passwd|secret|token|api[_-]?key|authorization|private[_-]?key|credential)")


def _text(value: Any, limit: int = MAX_FIELD_CHARS) -> str:
    return _SECRET_RE.sub(r"\1: [REDACTED]", "" if value is None else str(value))[:limit]


def redact(value: Any) -> Any:
    if isinstance(value, dict):
        return {str(k)[:80]: ("[REDACTED]" if _SECRET_KEY_RE.search(str(k)) else redact(v)) for k, v in list(value.items())[:MAX_INPUT_ITEMS]}
    if isinstance(value, list): return [redact(v) for v in value[:MAX_INPUT_ITEMS]]
    if isinstance(value, str): return _text(value)
    return value


def _hash(item: Any) -> str:
    return hashlib.sha256(json.dumps(redact(item), sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode()).hexdigest()


def evidence_hashes(evidence: list[dict[str, Any]]) -> list[str]:
    return [_hash(item) for item in evidence[:MAX_INPUT_ITEMS]]


def input_fingerprint(business: dict[str, Any], scans: list[dict[str, Any]], contacts: list[dict[str, Any]], evidence: list[dict[str, Any]], max_items: int = MAX_INPUT_ITEMS) -> str:
    payload = {"business": redact(business), "scans": [redact(x) for x in scans[:max_items]], "contacts": [redact(x) for x in contacts[:max_items]], "evidence": [redact(x) for x in evidence[:max_items]]}
    return _hash(payload)


def _citation(item: dict[str, Any], source_type: str, provenance: str) -> dict[str, Any]:
    return {"source_type": source_type, "source_id": int(item["id"]) if str(item.get("id", "")).isdigit() else None,
            "kind": _text(item.get("kind", source_type), 80), "url": _text(item.get("url", item.get("input_url", item.get("source_url", ""))), 500),
            "provenance": provenance, "hash": _hash(item), "policy": "stored_evidence_only"}


def build_local_suggestions(business: dict[str, Any], scans: list[dict[str, Any]], contacts: list[dict[str, Any]], evidence: list[dict[str, Any]], score_history: list[dict[str, Any]] | None = None) -> dict[str, Any]:
    evidence = [redact(x) for x in evidence[:MAX_INPUT_ITEMS] if _text(x.get("claim", "")).strip()]
    scans = [redact(x) for x in scans[:MAX_INPUT_ITEMS]]
    contacts = [redact(x) for x in contacts[:MAX_INPUT_ITEMS]]
    citations = [_citation(x, "evidence", "discovery_evidence") for x in evidence]
    citations += [_citation(x, "website_scan", "website_scanner") for x in scans]
    citations += [_citation(x, "contact_extraction", "contact_extractor") for x in contacts]
    claims = [_text(x.get("claim", "")).strip() for x in evidence]
    name = _text(business.get("name", "this business"), 200)
    score = business.get("score")
    priority = ("high" if score is not None and int(score) >= 70 else "medium" if score is not None and int(score) >= 40 else "low")
    uncertainty = []
    if not claims: uncertainty.append("no_claim_bearing_evidence")
    if not contacts: uncertainty.append("no_public_contact_extraction")
    conflicts = []
    classifications = [str(x.get("classification", "")).lower() for x in scans if x.get("classification")]
    if len(set(classifications)) > 1: conflicts.append("website_scan_classifications_disagree")
    classification = "business_prospect" if claims or business.get("website") else "insufficient_evidence"
    summary = f"Stored evidence for {name}: " + (" ".join(f"{claim} [evidence:{item['source_id']}]" for claim, item in zip(claims[:5], citations[:5])) if claims else "insufficient claim-bearing evidence")
    output = {
        "classification": classification, "summary": _text(summary, 2000),
        "priority_recommendation": priority, "confidence": round(min(0.95, 0.45 + 0.1 * len(claims) + (0.1 if scans else 0)), 2),
        "uncertainty": uncertainty, "conflicts": conflicts, "citations": citations[:MAX_INPUT_ITEMS],
        "policy": {"claim_policy": "stored_evidence_only", "no_outreach": True, "redacted_inputs": True},
        "suggestions": ([{"type": "summary", "text": _text(summary), "citations": citations[:5]}] if claims else []),
        "grounded": True, "provider": "local", "version": "deterministic-v2",
    }
    encoded = json.dumps(output, sort_keys=True, ensure_ascii=False)
    return output if len(encoded) <= MAX_OUTPUT_CHARS else {**output, "suggestions": output["suggestions"][:1], "citations": citations[:20]}


def provider_name() -> str | None:
    value = os.environ.get("AI_PROVIDER", "").strip().lower()
    return value or None


def provider_status(configured: str | None = None) -> dict[str, Any]:
    provider = configured if configured is not None else provider_name()
    local = provider in {"local", "deterministic"}
    return {"provider": provider or "", "status": "ready" if local else "not_configured", "network_enabled": False, "reviewed": local, "outbound_calls": False}


def generate(business: dict[str, Any], scans: list[dict[str, Any]], contacts: list[dict[str, Any]], evidence: list[dict[str, Any]], score_history: list[dict[str, Any]] | None = None) -> tuple[str, str, str, dict[str, Any]]:
    provider = provider_name()
    metadata = {"input_counts": {"business": 1, "scans": min(len(scans), MAX_INPUT_ITEMS), "contacts": min(len(contacts), MAX_INPUT_ITEMS), "evidence": min(len(evidence), MAX_INPUT_ITEMS)}, "redacted": True, "max_input_items": MAX_INPUT_ITEMS, "max_field_chars": MAX_FIELD_CHARS, "evidence_hashes": evidence_hashes(evidence), "input_fingerprint": input_fingerprint(business, scans, contacts, evidence)}
    if provider not in {"local", "deterministic"}: return "not_configured", provider or "", "", metadata
    return "succeeded", "local", "deterministic-v2", {**metadata, "output": build_local_suggestions(business, scans, contacts, evidence, score_history)}