Seth
Claude Sonnet 5
Show full chat transcript on leads, fix Chat Agent settings losing place on Drive connect
41f3768 from fastapi import FastAPI, UploadFile, File, Depends, HTTPException, Query, Request
from fastapi.responses import FileResponse, StreamingResponse
from fastapi.staticfiles import StaticFiles
from fastapi.middleware.cors import CORSMiddleware
from starlette.middleware.sessions import SessionMiddleware
from pathlib import Path
from sqlalchemy.orm import Session
from sqlalchemy import func, or_
import pandas as pd
import uuid
import os
import csv
import io
import concurrent.futures
from functools import partial
from typing import Any, Dict, List, Optional
import json
import asyncio
import math
import re
import logging
import requests
from urllib.parse import urlparse, unquote
from datetime import datetime, timedelta
from calendar import monthrange
from .database import (
get_db,
SessionLocal,
db_commit_with_retry,
Tenant,
TenantMembership,
Team,
User,
UploadedFile,
Prompt,
GeneratedSequence,
SmartleadRun,
Contact,
Company,
ContactActivity,
CrmLead,
CrmDeal,
UnipileAccount,
LinkedinCampaign,
UnipileHostedAuthState,
OutreachSendReceipt,
ChatAgentConfig,
FileOutreachExecution,
)
from pydantic import ValidationError
from .models import (
UploadResponse,
PromptSaveRequest,
CampaignMetaPatchRequest,
SequenceResponse,
SmartleadPushRequest,
SmartleadRunResponse,
CrmLeadPatchRequest,
BulkLeadIdsRequest,
BulkContactIdsRequest,
ActivityCreateRequest,
CrmDealPatchRequest,
CrmDealCreateRequest,
ContactCreateRequest,
ContactPatchRequest,
CompanyCreateRequest,
CompanyPatchRequest,
BulkCompanyIdsRequest,
CompanyContactCreateRequest,
WonBillingPayload,
UnipileConnectRequest,
UnipileHostedLinkRequest,
UnipileMailboxHostedLinkRequest,
UserLinkedinPrefsPatch,
UserMailboxPrefsPatch,
UnipileLinkedinLimitsPatch,
LinkedinCampaignCreateRequest,
LinkedinCampaignGenerateRequest,
LinkedinCampaignExecuteRequest,
LinkedinCampaignPatchRequest,
)
from .gmail_invite import send_invite_email_via_gmail
from .gpt_service import (
generate_email_sequence,
generate_linkedin_sequence,
enrich_manual_contact_profile,
linkedin_invite_note,
normalize_linkedin_message_format,
_pick_contact_field,
)
from .smartlead_client import SmartleadClient
from .auth_routes import router as auth_router
from .tenant_deps import TenantContext, get_tenant_context
from .tenant_routes import router as tenant_router
from .outreach_routes import router as outreach_router
from .cta_forms_routes import router as cta_forms_router, lead_is_cta_form
from .landing_pages_routes import router as landing_pages_router
from .chat_agent_routes import router as chat_agent_router
from .chat_agent_drive import router as chat_agent_drive_router
from .chat_agent_ingest import reindex_drive as _chat_agent_reindex_drive
from .chat_agent_ingest import reindex_website as _chat_agent_reindex_website
from .crm_lead_contacts import (
MANUAL_CONTACT_FILE_ID,
SMARTLEAD_IMPORT_FILE_ID,
link_lead_to_contact,
)
from .outreach_routes import (
_format_wait_step_label,
parse_tracking_label,
refresh_outreach_execution_stats_for_file,
normalize_smtp_message_id,
)
from .outreach_leads import (
build_outreach_thread,
find_gmail_receipt_by_parent_hints,
is_genuine_prospect_email_reply,
lead_is_outreach_campaign,
outreach_actions_count,
process_inbound_email_reply,
prospect_email_for_reply_match,
prospect_emails_for_reply_match,
resolve_gmail_receipt_by_prospect_emails,
stop_contact_outreach,
upsert_crm_lead_from_outreach_reply,
_is_likely_email_reply,
_reply_sender_email,
)
from .deal_revenue import build_quarterly_board
app = FastAPI()
# Session cookie for Google OAuth (must be before CORS if using credentials across origins)
_session_secret = os.environ.get("SESSION_SECRET", "dev-session-secret-change-in-production")
_is_https = os.environ.get("HTTPS_ONLY_COOKIES", "").strip() in ("1", "true", "yes")
app.add_middleware(
SessionMiddleware,
secret_key=_session_secret,
same_site="lax",
https_only=_is_https,
max_age=14 * 24 * 60 * 60,
)
_cors_raw = os.environ.get(
"CORS_ORIGINS",
"http://localhost:5173,http://127.0.0.1:5173,http://localhost:3000,http://127.0.0.1:3000",
)
_cors_origins = [o.strip() for o in _cors_raw.split(",") if o.strip()]
if not _cors_origins:
_cors_origins = ["*"]
app.add_middleware(
CORSMiddleware,
allow_origins=_cors_origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
app.include_router(auth_router)
app.include_router(tenant_router)
app.include_router(outreach_router)
app.include_router(cta_forms_router)
app.include_router(landing_pages_router)
app.include_router(chat_agent_router)
app.include_router(chat_agent_drive_router)
# Create uploads directory
UPLOAD_DIR = Path(os.getenv("HF_DATA_DIR", "/data")) / "uploads"
UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
# UniPile credentials are read fresh on each request (see _load_unipile_env) so Spaces secret
# updates and stripping whitespace behave reliably. Module-level kept for backwards compat.
UNIPILE_API_BASE = os.getenv("UNIPILE_API_BASE", "").rstrip("/")
UNIPILE_API_KEY = os.getenv("UNIPILE_API_KEY", "")
FRONTEND_ORIGIN = os.getenv("FRONTEND_ORIGIN", "").rstrip("/")
UNIPILE_WEBHOOK_SECRET = os.getenv("UNIPILE_WEBHOOK_SECRET", "").strip()
UNIPILE_WEBHOOK_LOG_EVENTS = os.getenv("UNIPILE_WEBHOOK_LOG_EVENTS", "").strip().lower() in (
"1",
"true",
"yes",
)
_unipile_webhook_log = logging.getLogger("emailout.unipile_webhook")
# ---- API ----
def _safe_str(value):
if value is None:
return ""
if isinstance(value, float):
if math.isnan(value):
return ""
return str(value).strip()
return str(value).strip()
def _fill_if_empty(current: Any, new_val: Any) -> str:
cur = _safe_str(current)
new_s = _safe_str(new_val)
if new_s and not cur:
return new_s
return cur
def _json_safe(value):
if value is None:
return None
if isinstance(value, float):
if math.isnan(value):
return None
return value
# Pandas may return numpy scalar types; use Python-native via json fallback
try:
json.dumps(value)
return value
except TypeError:
return _safe_str(value)
def _pick_field(row_dict: Dict, aliases: List[str]) -> str:
lowered = {str(k).strip().lower(): v for k, v in row_dict.items()}
for alias in aliases:
if alias in lowered:
return _safe_str(lowered[alias])
return ""
def _pick_from_raw(raw_data: Dict, aliases: List[str]) -> str:
if not isinstance(raw_data, dict):
return ""
lowered = {str(k).strip().lower(): v for k, v in raw_data.items()}
for alias in aliases:
if alias in lowered:
return _safe_str(lowered[alias])
return ""
def _pick_linkedin_url(raw_data: Dict) -> str:
return _pick_from_raw(
raw_data,
[
"linkedin",
"linkedin url",
"linkedin profile",
"person linkedin url",
"person linkedin profile url",
"linkedin profile url",
"linkedin public url",
"linkedin_url",
"linkedin profile link",
],
)
def _normalize_linkedin_url(raw: str) -> str:
"""Strip whitespace; upgrade http→https for stable display and parsing."""
s = _safe_str(raw)
if not s:
return ""
if s.startswith("http://"):
s = "https://" + s[len("http://") :]
return s
def _linkedin_public_identifier(raw: str) -> str:
"""
UniPile profile lookup expects the public slug from /in/{slug}, not a full URL.
Accepts plain slugs (e.g. satyanadella) as well.
"""
s = _normalize_linkedin_url(raw).strip()
if not s:
return ""
low = s.lower()
if "linkedin.com" in low:
try:
path = urlparse(s).path or ""
except Exception:
path = ""
path = unquote(path).strip("/")
parts = [p for p in path.split("/") if p]
for i, seg in enumerate(parts):
if seg.lower() == "in" and i + 1 < len(parts):
slug = parts[i + 1].strip()
return slug.split("?")[0].strip()
if seg.lower() == "sales" and i + 2 < len(parts) and parts[i + 1].lower() == "lead":
# /sales/lead/{slug}
return parts[i + 2].split("?")[0].strip()
return ""
# Already a slug / identifier (no URL)
return s.split("?")[0].strip("/").strip()
def _load_unipile_env() -> tuple[str, str]:
"""Read UniPile DSN and API key from the environment (strip whitespace / newlines from secrets)."""
base = (os.getenv("UNIPILE_API_BASE") or "").strip().rstrip("/")
key = (os.getenv("UNIPILE_API_KEY") or "").strip()
return base, key
def _require_unipile_config():
base, key = _load_unipile_env()
if not base or not key:
raise HTTPException(
status_code=400,
detail="UniPile is not configured. Set UNIPILE_API_BASE and UNIPILE_API_KEY in backend env (Hugging Face: Space → Settings → Secrets).",
)
def _unipile_headers():
_, key = _load_unipile_env()
return {
"X-API-KEY": key,
"accept": "application/json",
"content-type": "application/json",
}
def _unipile_call(method: str, path: str, payload: Optional[dict] = None):
"""
Raw UniPile HTTP call. Returns (status_code, parsed_body).
Raises HTTPException only on transport failure (not on 4xx/5xx responses).
"""
_require_unipile_config()
base, _ = _load_unipile_env()
url = f"{base}{path}"
try:
resp = requests.request(
method=method.upper(),
url=url,
headers=_unipile_headers(),
json=payload,
timeout=45,
)
except requests.RequestException as e:
raise HTTPException(status_code=502, detail=f"UniPile request failed: {str(e)}")
data = None
try:
data = resp.json()
except Exception:
data = {"raw": resp.text}
return resp.status_code, data
def _unipile_request(method: str, path: str, payload: Optional[dict] = None):
status, data = _unipile_call(method, path, payload)
if status >= 400:
msg = None
if isinstance(data, dict):
msg = (
data.get("message")
or data.get("detail")
or data.get("error")
or data.get("errors")
)
if isinstance(msg, (dict, list)):
msg = json.dumps(msg)
err_type = _safe_str(data.get("type"))
if status == 401 and err_type == "errors/missing_credentials":
msg = (
"UniPile rejected the request (missing or invalid API credentials). "
"In Hugging Face Space → Settings → Secrets, set UNIPILE_API_KEY to your UniPile API access token "
"(Dashboard → API keys) and UNIPILE_API_BASE to your UniPile API URL (no trailing slash). "
"Remove accidental quotes or spaces when pasting; redeploy/restart the Space after saving."
)
raise HTTPException(
status_code=status,
detail=msg or f"UniPile error ({status}): {data if not isinstance(data, dict) else json.dumps(data)}",
)
return data
def _linkedin_sender_signoff_for_user(user_row: Optional[User]) -> str:
if not user_row:
return ""
pref = _safe_str(getattr(user_row, "linkedin_profile_display_name", None) or "")
if pref and "@" not in pref:
return pref
name = _safe_str(user_row.name or "")
if name and "@" not in name:
return name
return ""
def _prepare_linkedin_outbound_text(
content: str,
*,
first_name: str = "",
user_row: Optional[User] = None,
for_invite: bool = False,
) -> str:
signoff = _linkedin_sender_signoff_for_user(user_row)
text = normalize_linkedin_message_format(
(content or "").strip(),
first_name=first_name,
signoff=signoff,
)
if for_invite:
return linkedin_invite_note(text)
return text
def _send_unipile_dm(account_id: str, provider_id: str, text: str):
"""Open (or reuse) a 1:1 LinkedIn chat and send a message."""
chat = _unipile_request(
"POST",
"/api/v1/chats",
{"account_id": account_id, "attendees": [provider_id]},
)
chat_id = _safe_str(chat.get("id") if isinstance(chat, dict) else "")
if not chat_id:
raise ValueError("UniPile chat id not found")
_unipile_request(
"POST",
f"/api/v1/chats/{requests.utils.quote(chat_id, safe='')}/messages",
{"text": text or ""},
)
def _public_origin_from_request(request: Request) -> str:
"""
Resolve browser-facing origin behind proxies (HF Spaces, ingress).
Falls back to request URL if forwarded headers are absent.
"""
fwd_proto = request.headers.get("x-forwarded-proto")
proto = (fwd_proto.split(",")[0].strip().lower() if fwd_proto else request.url.scheme or "https")
fwd_host = request.headers.get("x-forwarded-host")
host = (fwd_host.split(",")[0].strip() if fwd_host else request.headers.get("host") or request.url.netloc or "")
if host:
return f"{proto}://{host}".rstrip("/")
return str(request.base_url).rstrip("/")
_UNIPILE_ID_UUID_RE = re.compile(
r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$",
re.IGNORECASE,
)
def _is_unipile_nonsense_display_name(val: str, account_id_hint: str = "") -> bool:
"""Hosted-auth notify bodies store the state token in `name`; account ids are UUID-shaped."""
s = (val or "").strip()
if not s:
return True
if _UNIPILE_ID_UUID_RE.match(s):
return True
if account_id_hint and s == account_id_hint.strip():
return True
if _looks_like_email(s):
return True
return False
def _looks_like_email(val: str) -> bool:
s = (val or "").strip()
if not s or "@" not in s:
return False
return bool(re.match(r"^[^@\s]+@[^@\s]+\.[^@\s]+$", s))
def _display_name_from_email_local(email: str) -> str:
"""Best-effort human label when UniPile only exposes the mailbox address (e.g. maya@co.com → Maya)."""
e = (email or "").strip()
if "@" not in e:
return ""
local = e.split("@", 1)[0].strip()
if not local:
return ""
parts = [p for p in re.split(r"[._+\-]+", local) if p and re.search(r"[A-Za-z]", p)]
if not parts:
return ""
return " ".join(p[:1].upper() + p[1:].lower() if len(p) > 1 else p.upper() for p in parts)
def _mailbox_email_from_unipile_meta(meta: Any) -> str:
if not isinstance(meta, dict):
return ""
for layer in (meta.get("account_profile"), meta):
if not isinstance(layer, dict):
continue
cp = layer.get("connection_params")
if isinstance(cp, dict):
mail = cp.get("mail")
if isinstance(mail, dict):
for key in ("username", "imap_user", "smtp_user", "email"):
v = _safe_str(mail.get(key))
if _looks_like_email(v):
return v
for key in ("email", "username"):
v = _safe_str(layer.get(key))
if _looks_like_email(v):
return v
n = _safe_str(layer.get("name"))
if _looks_like_email(n):
return n
return ""
def _name_from_mailbox_label(label: str) -> str:
lab = (label or "").strip()
suffix = "'s mailbox"
if lab.endswith(suffix) and len(lab) > len(suffix):
cand = lab[: -len(suffix)].strip()
if cand and not _looks_like_email(cand) and not _is_unipile_nonsense_display_name(cand):
return cand
return ""
def _try_fetch_mailbox_sender_display_name(account_id: str, mailbox_email: str = "") -> str:
"""
Gmail 'Send mail as' display name often appears on sent messages as from_attendee.display_name.
UniPile account `name` is frequently the mailbox address only.
"""
if not (account_id or "").strip():
return ""
acc_enc = requests.utils.quote(account_id.strip(), safe="")
want = (mailbox_email or "").strip().lower()
folder_candidates = ("SENT", "sent", "Sent", "Sent Items", "INBOX")
for folder in folder_candidates:
folder_enc = requests.utils.quote(folder, safe="")
st, data = _unipile_call(
"GET",
f"/api/v1/emails?account_id={acc_enc}&limit=15&folder={folder_enc}",
)
if st >= 400 or not isinstance(data, dict):
continue
items = data.get("items") or data.get("emails") or data.get("data") or []
if not isinstance(items, list):
continue
for item in items:
if not isinstance(item, dict):
continue
fa = item.get("from_attendee") or item.get("from") or {}
if not isinstance(fa, dict):
continue
dn = _safe_str(fa.get("display_name") or fa.get("name"))
ident = _safe_str(fa.get("identifier") or fa.get("email") or fa.get("address"))
if not dn or _looks_like_email(dn):
continue
if want and ident and ident.lower() != want:
continue
return dn
return ""
def _resolve_mailbox_sender_name(
meta: Any,
*,
unipile_account_id: str = "",
label: str = "",
fetch_sent: bool = True,
) -> str:
"""Human sender name for mailbox UI + email sign-offs — never the raw mailbox address when avoidable."""
mailbox_email = _mailbox_email_from_unipile_meta(meta)
name = _extract_unipile_identity(meta if isinstance(meta, dict) else {})[0].strip()
if name and not _looks_like_email(name):
return name
if fetch_sent and unipile_account_id:
from_sent = _try_fetch_mailbox_sender_display_name(unipile_account_id, mailbox_email)
if from_sent:
return from_sent
from_label = _name_from_mailbox_label(label)
if from_label:
return from_label
if mailbox_email:
derived = _display_name_from_email_local(mailbox_email)
if derived:
return derived
return ""
def _sanitize_sender_signoff_name(val: str) -> str:
s = (val or "").strip()
if not s or _looks_like_email(s):
return ""
return s
def _mailbox_sender_name_for_user(db: Session, user_row: Optional[User]) -> str:
if not user_row:
return ""
pref = _sanitize_sender_signoff_name(getattr(user_row, "mailbox_profile_display_name", None) or "")
if pref:
return pref
ref_id = getattr(user_row, "default_mailbox_unipile_account_ref_id", None)
if ref_id is not None:
row = db.query(UnipileAccount).filter(UnipileAccount.id == int(ref_id)).first()
if row:
resolved = _resolve_mailbox_sender_name(
row.metadata_json or {},
unipile_account_id=row.unipile_account_id or "",
label=row.label or "",
)
if resolved:
return resolved
return _sanitize_sender_signoff_name(user_row.name or "")
def _extract_unipile_identity_from_layer(data: dict, account_id_hint: str = "") -> tuple[str, str]:
"""Read display name + avatar from one object (e.g. GET /accounts/{id} payload)."""
candidates_name = [
data.get("name"),
data.get("display_name"),
data.get("full_name"),
(data.get("user") or {}).get("name") if isinstance(data.get("user"), dict) else None,
(data.get("profile") or {}).get("name") if isinstance(data.get("profile"), dict) else None,
(data.get("account") or {}).get("name") if isinstance(data.get("account"), dict) else None,
]
candidates_avatar = [
data.get("picture"),
data.get("avatar"),
data.get("photo_url"),
data.get("image_url"),
(data.get("user") or {}).get("picture") if isinstance(data.get("user"), dict) else None,
(data.get("profile") or {}).get("picture") if isinstance(data.get("profile"), dict) else None,
(data.get("profile") or {}).get("avatar") if isinstance(data.get("profile"), dict) else None,
]
raw_name = next((_safe_str(x) for x in candidates_name if _safe_str(x)), "")
if _is_unipile_nonsense_display_name(raw_name, account_id_hint):
raw_name = ""
avatar = next((_safe_str(x) for x in candidates_avatar if _safe_str(x)), "")
return raw_name, avatar
def _extract_unipile_identity(data: Optional[dict]) -> tuple[str, str]:
"""
Best-effort extraction of profile display name and avatar URL from UniPile account payloads.
Prefer GET /accounts enrichment under `account_profile`; ignore hosted-auth token in top-level `name`.
"""
if not isinstance(data, dict):
return "", ""
account_id_hint = _safe_str(data.get("account_id"))
layers: list[dict] = []
ap = data.get("account_profile")
if isinstance(ap, dict) and ap:
layers.append(ap)
layers.append(data)
name, avatar = "", ""
for layer in layers:
n, av = _extract_unipile_identity_from_layer(layer, account_id_hint=account_id_hint)
if av and not avatar:
avatar = av
if n and not name:
name = n
if name and avatar:
break
return name, avatar
def _linkedin_account_display_name(row: UnipileAccount) -> str:
"""Human-facing profile name for UI (not the UniPile id or generic label)."""
meta = row.metadata_json if isinstance(row.metadata_json, dict) else {}
n = _extract_unipile_identity(meta)[0].strip()
if n:
return n
lab = (row.label or "").strip()
suffix = "'s LinkedIn account"
if lab.endswith(suffix) and len(lab) > len(suffix):
return lab[: -len(suffix)].strip()
return lab or "LinkedIn account"
_MAILBOX_PROVIDER_SET = ("GOOGLE", "MICROSOFT", "OUTLOOK", "MAIL", "IMAP")
def _mailbox_account_display_name(row: UnipileAccount) -> str:
n = _resolve_mailbox_sender_name(
row.metadata_json if isinstance(row.metadata_json, dict) else {},
unipile_account_id=row.unipile_account_id or "",
label=row.label or "",
)
if n:
return n
return (row.label or "").strip() or "Mailbox"
def _refresh_mailbox_account_identity(
db: Session,
row: UnipileAccount,
user_row: Optional[User],
*,
update_user_pref: bool = False,
) -> None:
"""Re-fetch UniPile profile / sent-mail display name when stored metadata only has the email address."""
account_id = row.unipile_account_id or ""
if not account_id:
return
profile = _try_fetch_unipile_account_profile(account_id)
meta = dict(row.metadata_json) if isinstance(row.metadata_json, dict) else {}
if isinstance(profile, dict) and profile:
meta["account_profile"] = profile
resolved = _resolve_mailbox_sender_name(
meta,
unipile_account_id=account_id,
label=row.label or "",
fetch_sent=True,
)
row.metadata_json = meta
if resolved:
row.label = f"{resolved}'s mailbox"[:255]
row.updated_at = datetime.utcnow()
if update_user_pref and user_row is not None:
cur = _sanitize_sender_signoff_name(getattr(user_row, "mailbox_profile_display_name", None) or "")
if not cur:
user_row.mailbox_profile_display_name = resolved[:200]
def _try_fetch_unipile_account_profile(account_id: str) -> dict:
"""
Optional enrichment call. If it fails, we just keep existing metadata.
"""
try:
out = _unipile_request("GET", f"/api/v1/accounts/{requests.utils.quote(account_id, safe='')}")
return out if isinstance(out, dict) else {}
except Exception:
return {}
def _contact_value(contact: Contact, field: str):
if field == "first_name":
return contact.first_name or _pick_from_raw(contact.raw_data, ["first name", "firstname", "first_name"])
if field == "last_name":
return contact.last_name or _pick_from_raw(contact.raw_data, ["last name", "lastname", "last_name"])
if field == "email":
return contact.email or _pick_from_raw(contact.raw_data, ["email", "work email", "email address", "secondary email", "tertiary email"])
if field == "company":
return contact.company or _pick_from_raw(contact.raw_data, ["company", "company name", "company name for emails", "organization name", "account name"])
if field == "title":
return contact.title or _pick_from_raw(contact.raw_data, ["title", "job title"])
if field == "file_id":
return contact.file_id or ""
if field == "created_at":
return contact.created_at.isoformat() if contact.created_at else ""
return (contact.raw_data or {}).get(field)
def _to_float(val):
s = _safe_str(val).replace(",", "")
if not s:
return None
try:
return float(s)
except Exception:
return None
def _to_datetime(val):
s = _safe_str(val)
if not s:
return None
try:
return datetime.fromisoformat(s.replace("Z", "+00:00"))
except Exception:
return None
CRM_STATUS_ALLOWED = frozenset({
"none",
"new_lead",
"contacted",
"qualified",
"unqualified",
})
DEAL_STAGE_ALLOWED = frozenset({
"new",
"discovery",
"proposal",
"negotiation",
"won",
"lost",
})
DEAL_LOST_REASON_ALLOWED = frozenset({
"price",
"no_decision",
"chose_another",
"didnt_meet_expectation",
"unknown",
})
DEAL_REVENUE_TYPE_ALLOWED = frozenset({"arr", "qrr", "mrr", "one_time"})
DEMO_CONTACTS_FILE_ID = "demo-contacts-crm"
def _contact_matches_apollo_filter(c: Contact, rule: dict) -> bool:
"""Single dynamic-field rule (same semantics as legacy single-field query)."""
field = _safe_str(rule.get("field"))
if not field:
return True
op_local = (_safe_str(rule.get("op") or "contains")).lower()
value = _safe_str(rule.get("value"))
from_value = _safe_str(rule.get("from_value"))
to_value = _safe_str(rule.get("to_value"))
cmp_value = value or from_value
v = _safe_str(cmp_value).lower()
v_from = from_value
v_to = to_value
field_val = _contact_value(c, field)
field_str = _safe_str(field_val).lower()
if op_local == "equals":
return field_str == v
if op_local == "from":
f_num, v_num = _to_float(field_val), _to_float(v_from or value)
if f_num is not None and v_num is not None:
return f_num >= v_num
f_dt, v_dt = _to_datetime(field_val), _to_datetime(v_from or value)
if f_dt and v_dt:
return f_dt >= v_dt
return field_str >= _safe_str(v_from or value).lower()
if op_local == "to":
f_num, v_num = _to_float(field_val), _to_float(v_to or value)
if f_num is not None and v_num is not None:
return f_num <= v_num
f_dt, v_dt = _to_datetime(field_val), _to_datetime(v_to or value)
if f_dt and v_dt:
return f_dt <= v_dt
return field_str <= _safe_str(v_to or value).lower()
if op_local == "between":
f_num, from_num, to_num = _to_float(field_val), _to_float(v_from), _to_float(v_to)
if f_num is not None and from_num is not None and to_num is not None:
return from_num <= f_num <= to_num
f_dt, from_dt, to_dt = _to_datetime(field_val), _to_datetime(v_from), _to_datetime(v_to)
if f_dt and from_dt and to_dt:
return from_dt <= f_dt <= to_dt
low, high = v_from.lower(), v_to.lower()
return low <= field_str <= high
return v in field_str
async def _chat_agent_reindex_loop():
"""Periodically re-index Drive/website sources whose reindex_interval_hours has elapsed."""
chat_agent_log = logging.getLogger("emailout.chat_agent_reindex")
while True:
try:
db = SessionLocal()
due: List[tuple] = []
try:
now = datetime.utcnow()
configs = db.query(ChatAgentConfig).filter(ChatAgentConfig.is_enabled == 1).all()
for cfg in configs:
interval = timedelta(hours=cfg.reindex_interval_hours or 24)
if cfg.drive_refresh_token and cfg.drive_folder_id:
if not cfg.drive_last_indexed_at or (now - cfg.drive_last_indexed_at) >= interval:
due.append(("drive", cfg.tenant_id))
if cfg.website_base_url:
if not cfg.website_last_indexed_at or (now - cfg.website_last_indexed_at) >= interval:
due.append(("website", cfg.tenant_id))
finally:
db.close()
loop = asyncio.get_event_loop()
for kind, tenant_id in due:
try:
fn = _chat_agent_reindex_drive if kind == "drive" else _chat_agent_reindex_website
await loop.run_in_executor(None, fn, tenant_id)
except Exception as e:
chat_agent_log.warning("Scheduled reindex failed (%s, tenant %s): %s", kind, tenant_id, e)
except Exception as e:
chat_agent_log.warning("Reindex loop error: %s", e)
await asyncio.sleep(3600) # check hourly; per-source cadence enforced by reindex_interval_hours
@app.on_event("startup")
async def _start_chat_agent_reindex_loop():
asyncio.create_task(_chat_agent_reindex_loop())
@app.on_event("startup")
async def _start_outreach_delivery_loop():
from .outreach_scheduler import outreach_delivery_loop
asyncio.create_task(outreach_delivery_loop())
@app.on_event("startup")
def _migrate_lead_status_removed_option():
"""Map legacy attempted_to_contact rows to contacted once at startup."""
db = SessionLocal()
try:
db.query(CrmLead).filter(CrmLead.crm_status == "attempted_to_contact").update(
{CrmLead.crm_status: "contacted"},
synchronize_session=False,
)
db.commit()
except Exception:
db.rollback()
finally:
db.close()
def _strip_html_simple(text: str) -> str:
if not text:
return ""
t = re.sub(r"<[^>]+>", " ", text)
return re.sub(r"\s+", " ", t).strip()
def _parse_smartlead_reply_payload(body: dict) -> Optional[dict]:
"""Normalize Smartlead webhook JSON into fields for CrmLead. Returns None if not a reply or missing identity."""
if not isinstance(body, dict):
return None
nest = body.get("data") if isinstance(body.get("data"), dict) else {}
lead = body.get("lead") or nest.get("lead") or {}
if not isinstance(lead, dict):
lead = {}
reply = body.get("reply") or nest.get("reply") or {}
if not isinstance(reply, dict):
reply = {}
event = str(body.get("event") or body.get("event_type") or body.get("type") or "").upper()
has_reply_content = bool(
reply.get("body")
or reply.get("html_body")
or reply.get("text_body")
or body.get("reply_body")
or body.get("message")
or nest.get("reply_body")
)
is_reply_event = "REPLIED" in event or event in ("EMAIL_REPLIED", "LEAD_REPLIED")
if not is_reply_event and not has_reply_content:
return None
campaign_id = _safe_str(
body.get("campaign_id")
or nest.get("campaign_id")
or lead.get("campaign_id")
or body.get("campaignId")
)
sl_id = lead.get("lead_id") or lead.get("id") or body.get("lead_id") or nest.get("lead_id")
sl_id_str = _safe_str(sl_id) if sl_id is not None else ""
email = (
lead.get("email")
or lead.get("email_address")
or body.get("lead_email")
or body.get("email")
or nest.get("email")
)
email = _safe_str(email).lower()
if not email and not sl_id_str:
return None
reply_body = (
reply.get("body")
or reply.get("html_body")
or reply.get("text_body")
or body.get("reply_body")
or body.get("message")
or nest.get("reply_body")
or ""
)
reply_body = _strip_html_simple(_safe_str(reply_body))
subject = _safe_str(reply.get("subject") or body.get("subject") or body.get("reply_subject") or "")
if not reply_body and not subject:
return None
received_raw = (
reply.get("received_at")
or reply.get("created_at")
or body.get("received_at")
or body.get("timestamp")
or nest.get("received_at")
)
received_at = _to_datetime(received_raw) if received_raw else datetime.utcnow()
fn = _safe_str(lead.get("first_name") or body.get("first_name"))
ln = _safe_str(lead.get("last_name") or body.get("last_name"))
company = _safe_str(
lead.get("company_name")
or lead.get("company")
or body.get("company_name")
or nest.get("company_name")
)
title = _safe_str(lead.get("title") or lead.get("job_title") or body.get("title"))
campaign_name = _safe_str(body.get("campaign_name") or nest.get("campaign_name") or "")
return {
"smartlead_lead_id": sl_id_str,
"campaign_id": campaign_id,
"campaign_name": campaign_name,
"email": email,
"first_name": fn,
"last_name": ln,
"company_name": company,
"title": title,
"last_reply_subject": subject,
"last_reply_body": reply_body,
"last_reply_at": received_at,
"raw_webhook": body,
}
def _crm_lead_to_dict(row: CrmLead) -> dict:
return {
"id": row.id,
"smartlead_lead_id": row.smartlead_lead_id,
"campaign_id": row.campaign_id,
"campaign_name": row.campaign_name or "",
"email": row.email or "",
"first_name": row.first_name or "",
"last_name": row.last_name or "",
"company_name": row.company_name or "",
"title": row.title or "",
"phone": row.phone or "",
"last_reply_subject": row.last_reply_subject or "",
"last_reply_body": row.last_reply_body or "",
"last_reply_at": row.last_reply_at.isoformat() if row.last_reply_at else None,
"crm_status": row.crm_status or "new_lead",
"contact_id": row.contact_id,
"owner_user_id": row.owner_user_id,
"created_at": row.created_at.isoformat() if row.created_at else None,
"updated_at": row.updated_at.isoformat() if row.updated_at else None,
"company_details": _lead_company_details_dict(row),
"source": (
(row.raw_webhook or {}).get("source")
if isinstance(row.raw_webhook, dict)
else None
)
or ("cta_form" if lead_is_cta_form(row) else None)
or ("emailout_campaign" if lead_is_outreach_campaign(row) else "smartlead"),
}
def _attach_owner_display(items: List[dict], db: Session) -> None:
"""Mutates each dict (already carrying "owner_user_id") to add owner_display_name and
owner_initials, batching one User query for the whole page. Shared by Leads (no stored
display-name cache, unlike Deals) so the response shape matches what ownerAvatar.jsx expects."""
oid_set = {d["owner_user_id"] for d in items if d.get("owner_user_id")}
user_map: Dict[int, User] = {}
if oid_set:
for u in db.query(User).filter(User.id.in_(oid_set)).all():
user_map[u.id] = u
for d in items:
oid = d.get("owner_user_id")
if oid and oid in user_map:
u = user_map[oid]
d["owner_display_name"] = (u.name or u.email or "").strip() or None
d["owner_initials"] = _user_initials_from_user(u)
else:
d["owner_display_name"] = None
d["owner_initials"] = None
def _move_lead_to_contacts_core(db: Session, lead: CrmLead) -> dict:
return link_lead_to_contact(db, lead)
CONTACTS_TO_LEADS_CAMPAIGN_ID = "contacts_import"
CONTACTS_TO_LEADS_CAMPAIGN_NAME = "Contacts"
def _convert_contact_to_lead_core(db: Session, contact: Contact) -> dict:
"""
Create a CrmLead row copied from a Contact. The contact row is kept; only deleted manually.
Links the lead via ``contact_id`` so CRM stays consistent. Skips if a lead with the same
email already exists.
"""
email = (_contact_value(contact, "email") or "").strip()
if not email:
return {"ok": False, "error": "Contact has no email"}
dup = (
db.query(CrmLead)
.filter(
CrmLead.tenant_id == contact.tenant_id,
func.lower(CrmLead.email) == email.lower(),
)
.first()
)
if dup:
return {"ok": False, "error": "A lead with this email already exists"}
fn = _contact_value(contact, "first_name") or ""
ln = _contact_value(contact, "last_name") or ""
co = _contact_value(contact, "company") or ""
ti = _contact_value(contact, "title") or ""
ph = contact.phone or ""
rd = contact.raw_data if isinstance(contact.raw_data, dict) else {}
company_snapshot = {
"Company Name": _safe_str(co or rd.get("Company Name")),
"Company Name for Emails": _safe_str(rd.get("Company Name for Emails")),
"Industry": _safe_str(rd.get("Industry")),
"# Employees": _safe_str(rd.get("# Employees")),
"Annual Revenue": _safe_str(rd.get("Annual Revenue")),
"Last Raised At": _safe_str(rd.get("Last Raised At")),
"Website": _safe_str(rd.get("Website")),
"City": _safe_str(rd.get("City")),
"State": _safe_str(rd.get("State")),
"Country": _safe_str(rd.get("Country")),
"Category": _safe_str(rd.get("Category")),
"Type": _safe_str(rd.get("Type")),
"Phone": _safe_str(rd.get("Phone")),
}
raw = {
"source": "contacts_convert",
"source_contact_id": contact.id,
"company_details": company_snapshot,
}
lead = CrmLead(
tenant_id=contact.tenant_id,
smartlead_lead_id=f"from-contact-{contact.id}",
campaign_id=CONTACTS_TO_LEADS_CAMPAIGN_ID,
campaign_name=CONTACTS_TO_LEADS_CAMPAIGN_NAME,
email=email,
first_name=fn,
last_name=ln,
company_name=co,
title=ti,
phone=ph,
last_reply_subject="",
last_reply_body="Added from Contacts",
last_reply_at=datetime.utcnow(),
crm_status="new_lead",
contact_id=contact.id,
raw_webhook=raw,
)
db.add(lead)
db.flush()
return {"ok": True, "lead_id": lead.id}
def _deal_name_from_lead(lead: CrmLead) -> str:
person = " ".join(filter(None, [lead.first_name or "", lead.last_name or ""])).strip()
if lead.company_name and person:
return f"{lead.company_name} — {person}"
if lead.company_name:
return lead.company_name or "Deal"
if person:
return person
return (lead.email or "").strip() or "Untitled deal"
def _user_initials_from_user(u: User | None) -> str:
if not u:
return ""
name = (u.name or "").strip()
em = (u.email or "").strip()
if name:
parts = name.split()
if len(parts) >= 2:
return (parts[0][:1] + parts[-1][:1]).upper()[:2]
return (parts[0][:2] or "?").upper()[:2]
if em:
return em[:2].upper()
return "?"
def _tenant_member_ids(db: Session, tenant_id: int) -> set[int]:
rows = (
db.query(TenantMembership.user_id)
.filter(TenantMembership.tenant_id == int(tenant_id))
.all()
)
return {int(r[0]) for r in rows}
def _visible_owner_user_ids(db: Session, t: TenantContext) -> Optional[set[int]]:
"""None = no restriction (admin sees everything). Otherwise the set of user_ids whose own
records the caller may see: themselves, plus — if they manage a team — everyone on it.
Records with no owner (owner_user_id is None) are always visible to everyone regardless of
this set — an unowned record isn't someone else's data."""
if t.role == "admin":
return None
managed = (
db.query(Team)
.filter(Team.tenant_id == t.tenant_id, Team.manager_user_id == t.user_id)
.first()
)
if managed is None:
return {t.user_id}
ids = {
int(r[0])
for r in db.query(TenantMembership.user_id)
.filter(TenantMembership.tenant_id == t.tenant_id, TenantMembership.team_id == managed.id)
.all()
}
ids.add(t.user_id)
return ids
def _deal_member_can_see(row: CrmDeal, visible_ids: Optional[set[int]]) -> bool:
if row.owner_user_id is None:
return True
if visible_ids is None:
return True
return int(row.owner_user_id) in visible_ids
def _deal_access_or_403(row: CrmDeal, t: TenantContext) -> None:
if t.role == "admin":
return
visible_ids = _visible_owner_user_ids(t.db, t)
if not _deal_member_can_see(row, visible_ids):
raise HTTPException(status_code=403, detail="You do not have access to this deal")
def _lead_access_or_403(row: CrmLead, t: TenantContext) -> None:
if t.role == "admin" or row.owner_user_id is None:
return
visible_ids = _visible_owner_user_ids(t.db, t)
if visible_ids is not None and int(row.owner_user_id) not in visible_ids:
raise HTTPException(status_code=403, detail="You do not have access to this lead")
def _contact_access_or_403(row: Contact, t: TenantContext) -> None:
if t.role == "admin" or row.owner_user_id is None:
return
visible_ids = _visible_owner_user_ids(t.db, t)
if visible_ids is not None and int(row.owner_user_id) not in visible_ids:
raise HTTPException(status_code=403, detail="You do not have access to this contact")
def _company_access_or_403(row: Company, t: TenantContext) -> None:
if t.role == "admin" or row.owner_user_id is None:
return
visible_ids = _visible_owner_user_ids(t.db, t)
if visible_ids is not None and int(row.owner_user_id) not in visible_ids:
raise HTTPException(status_code=403, detail="You do not have access to this company")
def _dashboard_period_bounds_utc(period: str) -> Optional[tuple[datetime, datetime]]:
"""Bounds for filtering won deals by won_at (dashboard date presets). None = all time."""
p = (period or "all").strip().lower()
if p in ("", "all"):
return None
now = datetime.utcnow()
if p == "month":
start = datetime(now.year, now.month, 1)
last = monthrange(now.year, now.month)[1]
end = datetime(now.year, now.month, last, 23, 59, 59)
return start, end
if p == "quarter":
qi = (now.month - 1) // 3
sm = qi * 3 + 1
start = datetime(now.year, sm, 1)
em = sm + 2
if em == 12:
end = datetime(now.year, 12, 31, 23, 59, 59)
else:
end = datetime(now.year, em + 1, 1) - timedelta(microseconds=1)
return start, end
if p == "year":
return datetime(now.year, 1, 1), datetime(now.year, 12, 31, 23, 59, 59)
return None
def _previous_period_bounds_utc(period: str) -> Optional[tuple[datetime, datetime]]:
"""Calendar window immediately before the current preset (month → prior month, etc.)."""
p = (period or "all").strip().lower()
if p in ("", "all"):
return None
now = datetime.utcnow()
if p == "month":
if now.month == 1:
y, m = now.year - 1, 12
else:
y, m = now.year, now.month - 1
start = datetime(y, m, 1)
last = monthrange(y, m)[1]
end = datetime(y, m, last, 23, 59, 59)
return start, end
if p == "quarter":
qi = (now.month - 1) // 3
if qi == 0:
y = now.year - 1
sm = 10
else:
y = now.year
sm = (qi - 1) * 3 + 1
start = datetime(y, sm, 1)
em = sm + 2
if em == 12:
end = datetime(y, 12, 31, 23, 59, 59)
else:
end = datetime(y, em + 1, 1) - timedelta(microseconds=1)
return start, end
if p == "year":
py = now.year - 1
return datetime(py, 1, 1), datetime(py, 12, 31, 23, 59, 59)
return None
def _comparison_period_label(period: str) -> str:
p = (period or "all").strip().lower()
if p == "month":
return "Last month"
if p == "quarter":
return "Last quarter"
if p == "year":
return "Last year"
return ""
def _won_at_within_bounds(won_at: Optional[datetime], bounds: Optional[tuple[datetime, datetime]]) -> bool:
if bounds is None:
return True
if won_at is None:
return False
wdt = won_at.replace(tzinfo=None) if getattr(won_at, "tzinfo", None) else won_at
return bounds[0] <= wdt <= bounds[1]
def _deal_to_dict(d: CrmDeal) -> dict:
fv = None
if d.deal_value is not None and d.close_probability is not None:
fv = int(round(d.deal_value * (d.close_probability / 100.0)))
ecd = None
if d.expected_close_date:
try:
ecd = d.expected_close_date.date().isoformat()
except Exception:
ecd = d.expected_close_date.isoformat()
rt = (getattr(d, "revenue_type", None) or "arr").strip().lower()
if rt not in DEAL_REVENUE_TYPE_ALLOWED:
rt = "arr"
return {
"id": d.id,
"name": d.name or "",
"stage": d.stage or "new",
"owner_user_id": d.owner_user_id,
"owner_initials": d.owner_initials or "",
"deal_value": d.deal_value,
"revenue_type": rt,
"contact_display": d.contact_display or "",
"account_name": d.account_name or "",
"expected_close_date": ecd,
"close_probability": d.close_probability if d.close_probability is not None else 10,
"forecast_value": fv,
"last_interaction_at": d.last_interaction_at.isoformat() if d.last_interaction_at else None,
"country": d.country or "",
"source_lead_id": d.source_lead_id,
"source_campaign_name": d.source_campaign_name or "",
"contact_id": d.contact_id,
"created_at": d.created_at.isoformat() if d.created_at else None,
"lost_reason": _safe_str(getattr(d, "lost_reason", None) or ""),
"lost_price_discount_pct": getattr(d, "lost_price_discount_pct", None),
"lost_chose_product_name": _safe_str(getattr(d, "lost_chose_product_name", None) or ""),
"won_po_number": _safe_str(getattr(d, "won_po_number", None) or ""),
"won_customer_legal_name": _safe_str(getattr(d, "won_customer_legal_name", None) or ""),
"won_customer_address": _safe_str(getattr(d, "won_customer_address", None) or ""),
"won_contact_person_name": _safe_str(getattr(d, "won_contact_person_name", None) or ""),
"won_channel_partner_name": _safe_str(getattr(d, "won_channel_partner_name", None) or ""),
"won_note_to_customer": _safe_str(getattr(d, "won_note_to_customer", None) or ""),
"won_note_to_accounts": _safe_str(getattr(d, "won_note_to_accounts", None) or ""),
"won_line_items": _won_line_items_list(d),
"won_at": d.won_at.isoformat() if getattr(d, "won_at", None) else None,
}
def _won_line_items_list(d: CrmDeal) -> list:
raw = getattr(d, "won_line_items", None)
if raw is None:
return []
if isinstance(raw, list):
return raw
if isinstance(raw, str) and raw.strip():
try:
v = json.loads(raw)
return v if isinstance(v, list) else []
except Exception:
return []
return []
def _clear_won_billing_fields(row: CrmDeal) -> None:
row.won_po_number = ""
row.won_customer_legal_name = ""
row.won_customer_address = ""
row.won_contact_person_name = ""
row.won_channel_partner_name = ""
row.won_note_to_customer = ""
row.won_note_to_accounts = ""
row.won_line_items = None
row.won_at = None
def _won_currency_label(raw) -> str:
c = (_safe_str(raw) or "USD").strip().upper()
return "CAD" if c == "CAD" else "USD"
def _fmt_won_money(amount: float, currency_raw) -> str:
"""Plain-text money with symbol + ISO code for won-deal emails."""
lab = _won_currency_label(currency_raw)
sym = "CA$" if lab == "CAD" else "$"
try:
n = float(amount)
except (TypeError, ValueError):
n = 0.0
return f"{sym}{n:,.2f} {lab}"
def _won_billing_interval_label(raw) -> str:
"""Human-readable billing cadence (matches win popover)."""
s = (_safe_str(raw) or "monthly").strip().lower()
if s in ("one_time", "onetime", "one-time", "non-recurring"):
return "One-time"
if s in ("quarterly", "quarter", "every 3 months"):
return "Every 3 months"
if s in ("annual", "year", "yearly", "every year"):
return "Every year"
return "Every month"
async def _notify_tenant_admins_deal_won(
db: Session,
tenant_id: int,
sender: User,
deal_id: int,
deal_name: str,
lines: list,
subtotal: float,
po: str,
legal: str,
addr: str,
contact_person: str,
channel_partner: str,
note_customer: str,
note_accounts: str,
) -> tuple[bool, Optional[str]]:
"""Email each workspace admin using the marking user's Gmail (same as invites)."""
rt = (sender.google_refresh_token or "").strip()
from_email = (sender.email or "").strip()
if not rt or not from_email:
return (
False,
"Gmail is not connected for your account. The deal was saved as Won, but admins were not emailed automatically.",
)
tenant = db.query(Tenant).filter(Tenant.id == int(tenant_id)).first()
workspace_name = (tenant.name or "Workspace").strip() if tenant else "Workspace"
admins = (
db.query(User)
.join(TenantMembership, TenantMembership.user_id == User.id)
.filter(TenantMembership.tenant_id == int(tenant_id), TenantMembership.role == "admin")
.all()
)
recipients: list[str] = []
seen: set[str] = set()
for u in admins:
em = (u.email or "").strip().lower()
if em and em not in seen:
seen.add(em)
recipients.append((u.email or "").strip())
if not recipients:
return False, "No workspace admin email addresses were found."
subject = f'[Won deal] {deal_name or "Deal"} — PO {po[:80]}'
body_lines = [
f"A deal was marked Won in workspace \"{workspace_name}\".",
"",
f"Deal ID: {deal_id}",
f"Deal name: {deal_name}",
f"Marked won by: {(sender.name or '').strip() or sender.email} <{sender.email}>",
"",
f"PO number: {po}",
f"Customer legal name: {legal}",
"Customer address:",
addr,
f"Contact person: {contact_person}",
]
if channel_partner:
body_lines.append(f"Channel partner: {channel_partner}")
body_lines += [
"",
"Note to customer:",
note_customer,
"",
"Note to our accounts:",
note_accounts,
"",
"Products / services (line items):",
]
for i, li in enumerate(lines, 1):
ps = _safe_str(li.get("product_service", ""))
cur = _won_currency_label(li.get("currency"))
try:
qf = float(li.get("qty") or 0)
rf = float(li.get("rate") or 0)
af = float(li.get("amount") or 0)
except (TypeError, ValueError):
qf = rf = af = 0.0
cadence = _won_billing_interval_label(li.get("billing_interval"))
body_lines.append(
f" {i}. {ps} | {cadence} | Qty {qf:g} × {_fmt_won_money(rf, cur)} = {_fmt_won_money(af, cur)}"
)
desc = _safe_str(li.get("description", ""))
if desc:
body_lines.append(f" Description: {desc}")
body_lines.append("")
if lines:
distinct = {_won_currency_label(li.get("currency")) for li in lines}
if len(distinct) == 1:
only = distinct.pop()
body_lines.append(f"Subtotal: {_fmt_won_money(float(subtotal), only)}")
else:
body_lines.append(
"Subtotal (numeric sum of line amounts; USD and CAD lines are not converted between currencies): "
f"{subtotal:,.2f}"
)
else:
body_lines.append(f"Subtotal: {subtotal:,.2f}")
body_text = "\n".join(body_lines)
errs: list[str] = []
ok_any = False
for to_email in recipients:
try:
await send_invite_email_via_gmail(rt, from_email, to_email, subject, body_text)
ok_any = True
except Exception as e:
errs.append(f"{to_email}: {str(e)}")
if ok_any:
return True, ("; ".join(errs) if errs else None)
return False, "; ".join(errs) if errs else "Could not send notification email."
@app.get("/api/health")
def health():
return {"status": "ok"}
@app.get("/api/hello")
def hello():
return {"message": "Hello from FastAPI"}
@app.post("/api/upload-csv")
async def upload_csv(file: UploadFile = File(...), t: TenantContext = Depends(get_tenant_context)):
"""Upload and parse CSV file from Apollo"""
db = t.db
tenant_id = t.tenant_id
try:
# Generate unique file ID
file_id = str(uuid.uuid4())
file_path = UPLOAD_DIR / f"{file_id}.csv"
# Save file
content = await file.read()
with open(file_path, "wb") as f:
f.write(content)
# Parse CSV to count contacts
df = pd.read_csv(file_path)
contact_count = len(df)
# Save upload metadata
db_file = UploadedFile(
tenant_id=tenant_id,
file_id=file_id,
filename=file.filename,
contact_count=contact_count,
file_path=str(file_path)
)
db.add(db_file)
# Persist uploaded contacts for CRM/contacts views
for idx, row in df.iterrows():
row_dict = row.to_dict()
sanitized_raw_data = {str(k): _json_safe(v) for k, v in row_dict.items()}
contact = Contact(
tenant_id=tenant_id,
file_id=file_id,
row_index=idx + 1,
first_name=_pick_field(row_dict, ["first name", "firstname", "first_name"]),
last_name=_pick_field(row_dict, ["last name", "lastname", "last_name"]),
email=_pick_field(row_dict, ["email", "work email", "email address"]),
company=_pick_field(row_dict, ["company", "company name", "company name for emails", "organization name", "account name"]),
title=_pick_field(row_dict, ["title", "job title"]),
source="apollo_csv",
raw_data=sanitized_raw_data,
)
db.add(contact)
db.commit()
return {
"file_id": file_id,
"contact_count": contact_count,
"message": "File uploaded successfully"
}
except Exception as e:
raise HTTPException(status_code=500, detail=f"Error uploading file: {str(e)}")
@app.get("/api/unipile/linkedin/campaign-defaults")
async def unipile_linkedin_campaign_defaults(t: TenantContext = Depends(get_tenant_context)):
"""
Combined prefs + accounts for sequence builder and campaign UI (saved LinkedIn profile name + default account).
"""
db = t.db
user_row = db.query(User).filter(User.id == t.user_id).first()
rows = (
db.query(UnipileAccount)
.filter(
UnipileAccount.tenant_id == t.tenant_id,
UnipileAccount.user_id == t.user_id,
UnipileAccount.provider == "LINKEDIN",
)
.order_by(UnipileAccount.created_at.desc(), UnipileAccount.id.desc())
.all()
)
accounts = []
for r in rows:
ident = _extract_unipile_identity(r.metadata_json or {})
accounts.append(
{
"id": r.id,
"label": r.label or "LinkedIn account",
"display_name": _linkedin_account_display_name(r),
"avatar_url": ident[1] or "",
"provider": r.provider,
"unipile_account_id": r.unipile_account_id,
"status": r.status or "unknown",
"auth_mode": r.auth_mode or "hosted",
"created_at": r.created_at.isoformat() if r.created_at else None,
"linkedin_daily_connection_cap": r.linkedin_daily_connection_cap,
"linkedin_daily_message_cap": r.linkedin_daily_message_cap,
"linkedin_weekly_connection_cap": r.linkedin_weekly_connection_cap,
"linkedin_weekly_message_cap": r.linkedin_weekly_message_cap,
}
)
prof = (user_row.linkedin_profile_display_name or "").strip() if user_row else ""
default_ref = getattr(user_row, "default_unipile_account_ref_id", None) if user_row else None
# Ensure default still exists
valid_default = None
if default_ref is not None:
if any(a["id"] == default_ref for a in accounts):
valid_default = default_ref
return {
"linkedin_profile_display_name": prof,
"default_unipile_account_ref_id": valid_default,
"accounts": accounts,
}
@app.patch("/api/me/linkedin-prefs")
async def patch_me_linkedin_prefs(
body: UserLinkedinPrefsPatch,
t: TenantContext = Depends(get_tenant_context),
):
db = t.db
u = db.query(User).filter(User.id == t.user_id).first()
if not u:
raise HTTPException(status_code=404, detail="User not found")
patch = body.model_dump(exclude_unset=True)
if "linkedin_profile_display_name" in patch:
s = (patch["linkedin_profile_display_name"] or "").strip()
u.linkedin_profile_display_name = s[:200] if s else None
if "default_unipile_account_ref_id" in patch:
rid = patch["default_unipile_account_ref_id"]
if rid is None or rid == 0:
u.default_unipile_account_ref_id = None
else:
acc = (
db.query(UnipileAccount)
.filter(
UnipileAccount.id == rid,
UnipileAccount.tenant_id == t.tenant_id,
UnipileAccount.user_id == t.user_id,
)
.first()
)
if not acc:
raise HTTPException(status_code=404, detail="LinkedIn account not found for this user")
u.default_unipile_account_ref_id = rid
db.commit()
return {
"ok": True,
"linkedin_profile_display_name": (u.linkedin_profile_display_name or ""),
"default_unipile_account_ref_id": u.default_unipile_account_ref_id,
}
@app.patch("/api/unipile/linkedin/accounts/{ref_id}/limits")
async def patch_unipile_linkedin_account_limits(
ref_id: int,
body: UnipileLinkedinLimitsPatch,
t: TenantContext = Depends(get_tenant_context),
):
db = t.db
acc = (
db.query(UnipileAccount)
.filter(
UnipileAccount.id == ref_id,
UnipileAccount.tenant_id == t.tenant_id,
UnipileAccount.user_id == t.user_id,
UnipileAccount.provider == "LINKEDIN",
)
.first()
)
if not acc:
raise HTTPException(status_code=404, detail="LinkedIn account not found for this user")
patch = body.model_dump(exclude_unset=True)
for field in (
"linkedin_daily_connection_cap",
"linkedin_daily_message_cap",
"linkedin_weekly_connection_cap",
"linkedin_weekly_message_cap",
):
if field in patch and patch[field] is not None:
setattr(acc, field, patch[field])
db.commit()
return {
"ok": True,
"id": acc.id,
"linkedin_daily_connection_cap": acc.linkedin_daily_connection_cap,
"linkedin_daily_message_cap": acc.linkedin_daily_message_cap,
"linkedin_weekly_connection_cap": acc.linkedin_weekly_connection_cap,
"linkedin_weekly_message_cap": acc.linkedin_weekly_message_cap,
}
@app.get("/api/unipile/mailbox/campaign-defaults")
async def unipile_mailbox_campaign_defaults(t: TenantContext = Depends(get_tenant_context)):
db = t.db
user_row = db.query(User).filter(User.id == t.user_id).first()
if user_row is not None and _looks_like_email(getattr(user_row, "mailbox_profile_display_name", None) or ""):
user_row.mailbox_profile_display_name = None
rows = (
db.query(UnipileAccount)
.filter(
UnipileAccount.tenant_id == t.tenant_id,
UnipileAccount.user_id == t.user_id,
UnipileAccount.provider.in_(_MAILBOX_PROVIDER_SET),
)
.order_by(UnipileAccount.created_at.desc(), UnipileAccount.id.desc())
.all()
)
default_ref = getattr(user_row, "default_mailbox_unipile_account_ref_id", None) if user_row else None
identity_dirty = False
for r in rows:
meta = r.metadata_json if isinstance(r.metadata_json, dict) else {}
raw_identity = _extract_unipile_identity(meta)[0]
resolved_quick = _resolve_mailbox_sender_name(
meta,
unipile_account_id=r.unipile_account_id or "",
label=r.label or "",
fetch_sent=False,
)
if r.unipile_account_id and (_looks_like_email(raw_identity) or not resolved_quick):
_refresh_mailbox_account_identity(
db,
r,
user_row,
update_user_pref=(
default_ref is not None and int(default_ref) == int(r.id)
or (default_ref is None and rows and int(r.id) == int(rows[0].id))
),
)
identity_dirty = True
if identity_dirty:
db.commit()
if user_row is not None:
db.refresh(user_row)
accounts = []
for r in rows:
ident = _extract_unipile_identity(r.metadata_json or {})
accounts.append(
{
"id": r.id,
"label": r.label or "Mailbox",
"display_name": _mailbox_account_display_name(r),
"avatar_url": ident[1] or "",
"provider": r.provider,
"unipile_account_id": r.unipile_account_id,
"status": r.status or "unknown",
"auth_mode": r.auth_mode or "hosted",
"created_at": r.created_at.isoformat() if r.created_at else None,
}
)
prof = _sanitize_sender_signoff_name(
(getattr(user_row, "mailbox_profile_display_name", None) or "").strip() if user_row else ""
)
if not prof and user_row:
prof = _mailbox_sender_name_for_user(db, user_row)
default_ref = getattr(user_row, "default_mailbox_unipile_account_ref_id", None) if user_row else None
valid_default = None
if default_ref is not None and any(a["id"] == default_ref for a in accounts):
valid_default = default_ref
return {
"mailbox_profile_display_name": prof,
"default_mailbox_unipile_account_ref_id": valid_default,
"accounts": accounts,
}
@app.patch("/api/me/mailbox-prefs")
async def patch_me_mailbox_prefs(
body: UserMailboxPrefsPatch,
t: TenantContext = Depends(get_tenant_context),
):
db = t.db
u = db.query(User).filter(User.id == t.user_id).first()
if not u:
raise HTTPException(status_code=404, detail="User not found")
patch = body.model_dump(exclude_unset=True)
if "mailbox_profile_display_name" in patch:
s = _sanitize_sender_signoff_name((patch["mailbox_profile_display_name"] or "").strip())
u.mailbox_profile_display_name = s[:200] if s else None
if "default_mailbox_unipile_account_ref_id" in patch:
rid = patch["default_mailbox_unipile_account_ref_id"]
if rid is None or rid == 0:
u.default_mailbox_unipile_account_ref_id = None
else:
acc = (
db.query(UnipileAccount)
.filter(
UnipileAccount.id == rid,
UnipileAccount.tenant_id == t.tenant_id,
UnipileAccount.user_id == t.user_id,
UnipileAccount.provider.in_(_MAILBOX_PROVIDER_SET),
)
.first()
)
if not acc:
raise HTTPException(status_code=404, detail="Mailbox account not found for this user")
u.default_mailbox_unipile_account_ref_id = rid
db.commit()
return {
"ok": True,
"mailbox_profile_display_name": (u.mailbox_profile_display_name or ""),
"default_mailbox_unipile_account_ref_id": u.default_mailbox_unipile_account_ref_id,
}
@app.post("/api/unipile/mailbox/hosted-link")
async def create_unipile_mailbox_hosted_link(
body: UnipileMailboxHostedLinkRequest,
request: Request,
t: TenantContext = Depends(get_tenant_context),
):
origin = (FRONTEND_ORIGIN or "").strip().rstrip("/") or _public_origin_from_request(request)
if not origin:
raise HTTPException(status_code=400, detail="Could not determine frontend origin for hosted auth.")
token = str(uuid.uuid4())
expires_at = datetime.utcnow() + timedelta(minutes=20)
u = t.db.query(User).filter(User.id == t.user_id).first()
if u is not None and body.mailbox_profile_display_name is not None:
s = _sanitize_sender_signoff_name((body.mailbox_profile_display_name or "").strip())
if s:
u.mailbox_profile_display_name = s[:200]
state = UnipileHostedAuthState(
tenant_id=t.tenant_id,
user_id=t.user_id,
token=token,
provider="GOOGLE",
label=(body.label or "Mailbox").strip(),
expires_at=expires_at,
)
t.db.add(state)
t.db.commit()
expires_iso = expires_at.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
api_base, _ = _load_unipile_env()
payload = {
"type": "create",
"providers": ["GOOGLE"],
"api_url": api_base,
"expiresOn": expires_iso,
"notify_url": f"{origin}/api/unipile/mailbox/hosted-callback",
"success_redirect_url": f"{origin}/settings?mailbox=connected",
"failure_redirect_url": f"{origin}/settings?mailbox=failed",
"name": token,
}
response = _unipile_request("POST", "/api/v1/hosted/accounts/link", payload)
hosted_url = _safe_str(response.get("url") if isinstance(response, dict) else "")
if not hosted_url:
raise HTTPException(status_code=400, detail="UniPile did not return hosted auth URL")
return {"url": hosted_url}
@app.post("/api/unipile/mailbox/hosted-callback")
async def unipile_mailbox_hosted_callback(request: Request, db: Session = Depends(get_db)):
try:
body = await request.json()
except Exception:
raise HTTPException(status_code=400, detail="Expected JSON body")
name_token = _safe_str((body or {}).get("name"))
unipile_account_id = _safe_str((body or {}).get("account_id"))
status = _safe_str((body or {}).get("status") or "connected")
if not name_token:
raise HTTPException(status_code=400, detail="Missing name token")
if not unipile_account_id:
return {"ok": True, "ignored": True, "reason": "missing_account_id"}
account_profile = _try_fetch_unipile_account_profile(unipile_account_id)
merged_meta = {}
if isinstance(body, dict):
merged_meta.update(body)
if isinstance(account_profile, dict) and account_profile:
merged_meta["account_profile"] = account_profile
state = (
db.query(UnipileHostedAuthState)
.filter(
UnipileHostedAuthState.token == name_token,
UnipileHostedAuthState.provider == "GOOGLE",
)
.first()
)
if not state:
return {"ok": True, "ignored": True, "reason": "unknown_token"}
if state.used_at is not None:
return {"ok": True, "ignored": True, "reason": "token_already_used"}
if state.expires_at and state.expires_at < datetime.utcnow():
return {"ok": True, "ignored": True, "reason": "token_expired"}
existing = (
db.query(UnipileAccount)
.filter(
UnipileAccount.tenant_id == state.tenant_id,
UnipileAccount.user_id == state.user_id,
UnipileAccount.provider.in_(_MAILBOX_PROVIDER_SET),
UnipileAccount.unipile_account_id == unipile_account_id,
)
.first()
)
user_row = db.query(User).filter(User.id == state.user_id).first()
box_name = _resolve_mailbox_sender_name(
merged_meta,
unipile_account_id=unipile_account_id,
label=(state.label or "").strip(),
fetch_sent=True,
)
if user_row is not None and box_name:
user_row.mailbox_profile_display_name = box_name[:200]
elif user_row is not None and _looks_like_email(getattr(user_row, "mailbox_profile_display_name", None) or ""):
user_row.mailbox_profile_display_name = None
nice_label = (state.label or "").strip() or "Mailbox"
if box_name:
nice_label = f"{box_name}'s mailbox"
nice_label = nice_label[:255]
if existing:
existing.label = nice_label
existing.status = status or existing.status
existing.auth_mode = "hosted"
existing.metadata_json = merged_meta or body
existing.updated_at = datetime.utcnow()
if user_row is not None and getattr(user_row, "default_mailbox_unipile_account_ref_id", None) is None:
user_row.default_mailbox_unipile_account_ref_id = existing.id
else:
row = UnipileAccount(
tenant_id=state.tenant_id,
user_id=state.user_id,
label=nice_label,
provider="GOOGLE",
unipile_account_id=unipile_account_id,
status=status or "connected",
auth_mode="hosted",
metadata_json=merged_meta or body,
)
db.add(row)
db.flush()
acc_ref_id = row.id
if user_row is not None and getattr(user_row, "default_mailbox_unipile_account_ref_id", None) is None:
user_row.default_mailbox_unipile_account_ref_id = acc_ref_id
state.used_at = datetime.utcnow()
db.commit()
return {"ok": True}
@app.get("/api/unipile/mailbox/accounts")
async def list_unipile_mailbox_accounts(t: TenantContext = Depends(get_tenant_context)):
db = t.db
user_row = db.query(User).filter(User.id == t.user_id).first()
rows = (
db.query(UnipileAccount)
.filter(
UnipileAccount.tenant_id == t.tenant_id,
UnipileAccount.user_id == t.user_id,
UnipileAccount.provider.in_(_MAILBOX_PROVIDER_SET),
)
.order_by(UnipileAccount.created_at.desc(), UnipileAccount.id.desc())
.all()
)
identity_dirty = False
for r in rows:
meta = r.metadata_json if isinstance(r.metadata_json, dict) else {}
raw_identity = _extract_unipile_identity(meta)[0]
if r.unipile_account_id and _looks_like_email(raw_identity):
_refresh_mailbox_account_identity(db, r, user_row, update_user_pref=False)
identity_dirty = True
if identity_dirty:
db.commit()
out = []
for r in rows:
ident = _extract_unipile_identity(r.metadata_json or {})
out.append(
{
"id": r.id,
"label": r.label or "Mailbox",
"display_name": _mailbox_account_display_name(r),
"avatar_url": ident[1] or "",
"provider": r.provider,
"unipile_account_id": r.unipile_account_id,
"status": r.status or "unknown",
"auth_mode": r.auth_mode or "hosted",
"created_at": r.created_at.isoformat() if r.created_at else None,
}
)
return {"accounts": out}
@app.get("/api/unipile/linkedin/accounts")
async def list_unipile_linkedin_accounts(t: TenantContext = Depends(get_tenant_context)):
rows = (
t.db.query(UnipileAccount)
.filter(
UnipileAccount.tenant_id == t.tenant_id,
UnipileAccount.user_id == t.user_id,
UnipileAccount.provider == "LINKEDIN",
)
.order_by(UnipileAccount.created_at.desc(), UnipileAccount.id.desc())
.all()
)
acc_out = []
for r in rows:
ident = _extract_unipile_identity(r.metadata_json or {})
acc_out.append(
{
"id": r.id,
"label": r.label or "LinkedIn account",
"display_name": _linkedin_account_display_name(r),
"avatar_url": ident[1] or "",
"provider": r.provider,
"unipile_account_id": r.unipile_account_id,
"status": r.status or "unknown",
"auth_mode": r.auth_mode or "hosted",
"created_at": r.created_at.isoformat() if r.created_at else None,
}
)
return {"accounts": acc_out}
@app.post("/api/unipile/linkedin/hosted-link")
async def create_unipile_linkedin_hosted_link(
body: UnipileHostedLinkRequest,
request: Request,
t: TenantContext = Depends(get_tenant_context),
):
origin = (FRONTEND_ORIGIN or "").strip().rstrip("/") or _public_origin_from_request(request)
if not origin:
raise HTTPException(status_code=400, detail="Could not determine frontend origin for hosted auth.")
token = str(uuid.uuid4())
expires_at = datetime.utcnow() + timedelta(minutes=20)
u = t.db.query(User).filter(User.id == t.user_id).first()
if u is not None and body.linkedin_profile_display_name is not None:
s = (body.linkedin_profile_display_name or "").strip()
if s:
u.linkedin_profile_display_name = s[:200]
state = UnipileHostedAuthState(
tenant_id=t.tenant_id,
user_id=t.user_id,
token=token,
provider="LINKEDIN",
label=(body.label or "LinkedIn Account").strip(),
expires_at=expires_at,
)
t.db.add(state)
t.db.commit()
expires_iso = expires_at.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
api_base, _ = _load_unipile_env()
payload = {
"type": "create",
"providers": ["LINKEDIN"],
"api_url": api_base,
"expiresOn": expires_iso,
"notify_url": f"{origin}/api/unipile/linkedin/hosted-callback",
"success_redirect_url": f"{origin}/settings?linkedin=connected",
"failure_redirect_url": f"{origin}/settings?linkedin=failed",
"name": token,
}
response = _unipile_request("POST", "/api/v1/hosted/accounts/link", payload)
hosted_url = _safe_str(response.get("url") if isinstance(response, dict) else "")
if not hosted_url:
raise HTTPException(status_code=400, detail="UniPile did not return hosted auth URL")
return {"url": hosted_url}
@app.post("/api/unipile/linkedin/hosted-callback")
async def unipile_linkedin_hosted_callback(request: Request, db: Session = Depends(get_db)):
"""
UniPile Hosted Auth notify_url callback.
No session is available on this route (server-to-server call), so we map by stored token in `name`.
"""
try:
body = await request.json()
except Exception:
raise HTTPException(status_code=400, detail="Expected JSON body")
name_token = _safe_str((body or {}).get("name"))
unipile_account_id = _safe_str((body or {}).get("account_id"))
status = _safe_str((body or {}).get("status") or "connected")
if not name_token:
raise HTTPException(status_code=400, detail="Missing name token")
if not unipile_account_id:
return {"ok": True, "ignored": True, "reason": "missing_account_id"}
account_profile = _try_fetch_unipile_account_profile(unipile_account_id)
merged_meta = {}
if isinstance(body, dict):
merged_meta.update(body)
if isinstance(account_profile, dict) and account_profile:
merged_meta["account_profile"] = account_profile
state = (
db.query(UnipileHostedAuthState)
.filter(
UnipileHostedAuthState.token == name_token,
UnipileHostedAuthState.provider == "LINKEDIN",
)
.first()
)
if not state:
return {"ok": True, "ignored": True, "reason": "unknown_token"}
if state.used_at is not None:
return {"ok": True, "ignored": True, "reason": "token_already_used"}
if state.expires_at and state.expires_at < datetime.utcnow():
return {"ok": True, "ignored": True, "reason": "token_expired"}
existing = (
db.query(UnipileAccount)
.filter(
UnipileAccount.tenant_id == state.tenant_id,
UnipileAccount.user_id == state.user_id,
UnipileAccount.provider == "LINKEDIN",
UnipileAccount.unipile_account_id == unipile_account_id,
)
.first()
)
user_row = db.query(User).filter(User.id == state.user_id).first()
li_name = _extract_unipile_identity(merged_meta)[0].strip()
if user_row is not None and li_name:
user_row.linkedin_profile_display_name = li_name[:200]
if li_name:
nice_label = f"{li_name}'s LinkedIn account"
else:
nice_label = (state.label or "").strip() or "LinkedIn account"
nice_label = nice_label[:255]
if existing:
existing.label = nice_label
existing.status = status or existing.status
existing.auth_mode = "hosted"
existing.metadata_json = merged_meta or body
existing.updated_at = datetime.utcnow()
if user_row is not None and getattr(user_row, "default_unipile_account_ref_id", None) is None:
user_row.default_unipile_account_ref_id = existing.id
else:
row = UnipileAccount(
tenant_id=state.tenant_id,
user_id=state.user_id,
label=nice_label,
provider="LINKEDIN",
unipile_account_id=unipile_account_id,
status=status or "connected",
auth_mode="hosted",
metadata_json=merged_meta or body,
)
db.add(row)
db.flush()
acc_ref_id = row.id
if user_row is not None and getattr(user_row, "default_unipile_account_ref_id", None) is None:
user_row.default_unipile_account_ref_id = acc_ref_id
state.used_at = datetime.utcnow()
db.commit()
return {"ok": True}
def _parse_unipile_webhook_datetime(raw) -> Optional[datetime]:
if not raw or not isinstance(raw, str):
return None
try:
s = raw.strip().replace("Z", "")
return datetime.fromisoformat(s)
except Exception:
return None
@app.post("/api/webhooks/unipile")
async def unipile_webhook(request: Request, db: Session = Depends(get_db)):
"""
UniPile webhooks (same URL for multiple UniPile registrations).
Supported ``event`` values:
* ``new_relation`` — LinkedIn invite accepted (source: users). Updates pending contacts.
* ``mail_opened``, ``mail_link_clicked`` — email tracking (source: ``email_tracking``).
* ``mail_received`` — mailbox email (source: ``email`` in UniPile API, “Email” / new-mail webhook in the dashboard). Replies matched via ``in_reply_to`` / ``tracking_id`` / ``label`` / RFC ``message_id``.
"""
if UNIPILE_WEBHOOK_SECRET:
auth_h = request.headers.get("Unipile-Auth") or request.headers.get("X-Unipile-Auth") or ""
if auth_h != UNIPILE_WEBHOOK_SECRET:
raise HTTPException(status_code=401, detail="Invalid webhook secret")
try:
body = await request.json()
except Exception:
raise HTTPException(status_code=400, detail="Expected JSON body")
event = _safe_str((body or {}).get("event"))
if UNIPILE_WEBHOOK_LOG_EVENTS and isinstance(body, dict):
ir = body.get("in_reply_to")
_unipile_webhook_log.info(
"UniPile webhook event=%s role=%s keys=%s in_reply_to_keys=%s",
event,
body.get("role"),
sorted(body.keys()),
sorted(ir.keys()) if isinstance(ir, dict) else None,
)
now = datetime.utcnow()
if event == "new_relation":
account_id = _safe_str((body or {}).get("account_id"))
provider_id = _safe_str((body or {}).get("user_provider_id"))
if not account_id or not provider_id:
return {"ok": True, "ignored": True, "reason": "missing_ids"}
ua = (
db.query(UnipileAccount)
.filter(UnipileAccount.unipile_account_id == account_id)
.first()
)
if not ua:
return {"ok": True, "ignored": True, "reason": "unknown_account"}
contacts = (
db.query(Contact)
.filter(
Contact.tenant_id == ua.tenant_id,
Contact.unipile_provider_id == provider_id,
Contact.source.in_(("apollo_csv", "linkedin_campaign")),
)
.all()
)
updated = 0
for ct in contacts:
if ct.linkedin_connection_accepted_at:
continue
ct.linkedin_invite_pending = 0
ct.linkedin_connection_accepted_at = now
updated += 1
db_commit_with_retry(db)
return {"ok": True, "updated_contacts": updated, "event": event}
if event in ("mail_opened", "mail_link_clicked"):
label = _safe_str((body or {}).get("label"))
parsed = parse_tracking_label(label)
if not parsed:
return {"ok": True, "ignored": True, "reason": "no_tracking_label"}
tenant_id, file_id, contact_id, step_slot = parsed
rec = (
db.query(OutreachSendReceipt)
.filter(
OutreachSendReceipt.tenant_id == tenant_id,
OutreachSendReceipt.file_id == file_id,
OutreachSendReceipt.contact_id == contact_id,
OutreachSendReceipt.step_slot == step_slot,
OutreachSendReceipt.channel == "gmail",
)
.first()
)
if not rec:
return {"ok": True, "ignored": True, "reason": "receipt_not_found"}
when = _parse_unipile_webhook_datetime((body or {}).get("date")) or now
if not rec.opened_at:
rec.opened_at = when
try:
rec.open_count = int(rec.open_count or 0) + 1
except (TypeError, ValueError):
rec.open_count = 1
db_commit_with_retry(db)
refresh_outreach_execution_stats_for_file(db, tenant_id, file_id)
db_commit_with_retry(db)
return {"ok": True, "event": event, "updated_receipt_id": rec.id}
if event in ("mail_received", "mail_sent"):
account_id = _safe_str((body or {}).get("account_id"))
role = _safe_str((body or {}).get("role")).lower()
if event == "mail_received" and role != "inbox":
return {"ok": True, "ignored": True, "reason": "not_inbox"}
if event == "mail_sent" and role not in ("sent", "outbox"):
return {"ok": True, "ignored": True, "reason": "not_sent"}
ua = (
db.query(UnipileAccount)
.filter(UnipileAccount.unipile_account_id == account_id)
.first()
)
if not ua:
return {"ok": True, "ignored": True, "reason": "unknown_account"}
tenant_id = int(ua.tenant_id)
in_reply = (body or {}).get("in_reply_to") or {}
parent_id = _safe_str(in_reply.get("id"))
parent_rfc = normalize_smtp_message_id(_safe_str(in_reply.get("message_id")))
trk = _safe_str((body or {}).get("tracking_id"))
label = _safe_str((body or {}).get("label"))
rec = find_gmail_receipt_by_parent_hints(
db, tenant_id, parent_id=parent_id, parent_rfc=parent_rfc
)
if not rec and trk:
rec = (
db.query(OutreachSendReceipt)
.filter(
OutreachSendReceipt.tenant_id == tenant_id,
OutreachSendReceipt.channel == "gmail",
or_(
OutreachSendReceipt.unipile_tracking_id == trk,
OutreachSendReceipt.unipile_message_id == trk,
),
)
.first()
)
if not rec and label:
p2 = parse_tracking_label(label)
if p2:
tid2, fid2, cid2, slot2 = p2
if tid2 == tenant_id:
rec = (
db.query(OutreachSendReceipt)
.filter(
OutreachSendReceipt.tenant_id == tid2,
OutreachSendReceipt.file_id == fid2,
OutreachSendReceipt.contact_id == cid2,
OutreachSendReceipt.step_slot == slot2,
OutreachSendReceipt.channel == "gmail",
)
.first()
)
folders = (body or {}).get("folders") or []
subj_raw = _safe_str((body or {}).get("subject"))
subj = subj_raw.lower()
bounce_like = any("bounce" in str(f).lower() for f in folders if f) or "undeliverable" in subj
likely_reply = _is_likely_email_reply(subj_raw, in_reply if isinstance(in_reply, dict) else None)
if not rec and likely_reply and not bounce_like:
prospect_ems = prospect_emails_for_reply_match(
body if isinstance(body, dict) else {}, event=event
)
rec = resolve_gmail_receipt_by_prospect_emails(db, tenant_id, prospect_ems)
if rec:
when = _parse_unipile_webhook_datetime((body or {}).get("date")) or now
lead_id = None
stopped = False
contact = (
db.query(Contact)
.filter(Contact.tenant_id == rec.tenant_id, Contact.id == rec.contact_id)
.first()
)
prospect_email = _safe_str(contact.email if contact else "")
genuine_reply = (
not bounce_like
and contact is not None
and is_genuine_prospect_email_reply(
db,
tenant_id,
body if isinstance(body, dict) else {},
event=event,
ua=ua,
prospect_email=prospect_email,
)
)
if bounce_like and not rec.bounced_at:
rec.bounced_at = when
elif genuine_reply:
result = process_inbound_email_reply(
db, tenant_id, rec, body=body if isinstance(body, dict) else {}, when=when
)
stopped = result.get("outreach_stopped", False)
lead_id = result.get("lead_id")
elif not bounce_like and likely_reply:
_unipile_webhook_log.info(
"%s ignored: outbound_or_not_prospect_reply from=%s prospect=%s subject=%s",
event,
_reply_sender_email(body if isinstance(body, dict) else {}),
prospect_email,
subj_raw[:120],
)
return {
"ok": True,
"ignored": True,
"reason": "outbound_or_not_prospect_reply",
"event": event,
}
db_commit_with_retry(db)
refresh_outreach_execution_stats_for_file(db, rec.tenant_id, rec.file_id)
db_commit_with_retry(db)
return {
"ok": True,
"event": event,
"updated_receipt_id": rec.id,
"outreach_stopped": stopped,
"lead_id": lead_id,
}
_unipile_webhook_log.info(
"%s ignored: no_matching_receipt prospect=%s subject=%s parent_id=%s trk=%s",
event,
prospect_email_for_reply_match(body if isinstance(body, dict) else {}, event=event),
subj_raw[:120],
parent_id or "",
trk or "",
)
return {"ok": True, "ignored": True, "reason": "no_matching_receipt"}
if event == "message_received":
account_id = _safe_str((body or {}).get("account_id"))
sender = (body or {}).get("sender") if isinstance((body or {}).get("sender"), dict) else {}
account_info = (body or {}).get("account_info") if isinstance((body or {}).get("account_info"), dict) else {}
sender_pid = _safe_str(sender.get("attendee_provider_id"))
our_pid = _safe_str(account_info.get("user_id"))
if sender_pid and our_pid and sender_pid == our_pid:
return {"ok": True, "ignored": True, "reason": "own_message"}
if not account_id or not sender_pid:
return {"ok": True, "ignored": True, "reason": "missing_sender"}
ua = (
db.query(UnipileAccount)
.filter(UnipileAccount.unipile_account_id == account_id)
.first()
)
if not ua:
return {"ok": True, "ignored": True, "reason": "unknown_account"}
tenant_id = int(ua.tenant_id)
contacts = (
db.query(Contact)
.filter(
Contact.tenant_id == tenant_id,
Contact.unipile_provider_id == sender_pid,
Contact.source.in_(("apollo_csv", "linkedin_campaign")),
)
.all()
)
if not contacts:
return {"ok": True, "ignored": True, "reason": "no_matching_contact"}
when = _parse_unipile_webhook_datetime((body or {}).get("timestamp")) or now
msg_text = _safe_str((body or {}).get("message"))
updated: List[dict] = []
for ct in contacts:
if not ct.file_id:
continue
n_actions = outreach_actions_count(db, tenant_id, ct.file_id)
stopped = stop_contact_outreach(
db,
ct,
reason="linkedin_reply",
when=when,
file_id=ct.file_id,
n_actions=n_actions,
)
lead_row = upsert_crm_lead_from_outreach_reply(
db,
tenant_id,
ct,
file_id=ct.file_id,
channel="linkedin",
reply_subject="",
reply_body=msg_text,
reply_at=when,
raw_webhook=body if isinstance(body, dict) else None,
)
refresh_outreach_execution_stats_for_file(db, tenant_id, ct.file_id)
updated.append(
{
"contact_id": ct.id,
"file_id": ct.file_id,
"outreach_stopped": stopped,
"lead_id": lead_row.id,
}
)
db_commit_with_retry(db)
return {"ok": True, "event": event, "updated": updated}
return {"ok": True, "ignored": True, "reason": "unsupported_event", "event": event or ""}
@app.get("/api/unipile/webhook-url-hint")
async def unipile_webhook_url_hint(request: Request, t: TenantContext = Depends(get_tenant_context)):
"""Public URL your UniPile webhooks should POST to (same host as this API)."""
_ = t
root = _public_origin_from_request(request)
post_url = f"{root}/api/webhooks/unipile"
return {
"post_url": post_url,
"auth_header": "Set Unipile-Auth or X-Unipile-Auth to UNIPILE_WEBHOOK_SECRET on each webhook registration.",
"registrations": [
{
"source": "users",
"events": ["new_relation"],
"doc": "https://developer.unipile.com/docs/detecting-accepted-invitations",
},
{
"source": "email_tracking",
"events": ["mail_opened", "mail_link_clicked"],
"doc": "https://developer.unipile.com/docs/tracking-email",
},
{
"source": "email",
"events": ["mail_received", "mail_sent", "mail_moved"],
"doc": "https://developer.unipile.com/docs/new-emails-webhook",
"dashboard_hint": "Create an **Email** (mailbox) webhook in the UniPile UI — not the **Messaging** (chat) webhook. Use the same callback URL.",
"note": "Reply attribution uses mail_received with role inbox and in_reply_to / tracking_id / label; enable mail_received for connected mail accounts.",
},
{
"source": "messaging",
"events": [
"message_received",
"message_read",
"message_reaction",
"message_edited",
"message_deleted",
"message_delivered",
],
"doc": "https://developer.unipile.com/docs/new-messages-webhook",
"note": "LinkedIn replies stop outreach for that contact and create a Lead. Ignores messages sent by your own account.",
},
],
"debug_logging": "Set UNIPILE_WEBHOOK_LOG_EVENTS=1 to log each webhook's event name, role, and JSON keys (no body content).",
"latency_note": "UniPile may deliver new_relation many hours after acceptance; LinkedIn has no real-time connection API.",
}
@app.post("/api/unipile/linkedin/connect")
async def connect_unipile_linkedin(
body: UnipileConnectRequest,
t: TenantContext = Depends(get_tenant_context),
):
payload = {"provider": "LINKEDIN"}
if body.auth_mode == "credentials":
if not (body.username and body.password):
raise HTTPException(status_code=400, detail="username/password are required for credentials auth")
payload["username"] = body.username
payload["password"] = body.password
else:
if not body.access_token:
raise HTTPException(status_code=400, detail="access_token is required for cookie auth")
payload["access_token"] = body.access_token
if body.user_agent:
payload["user_agent"] = body.user_agent
if body.country:
payload["country"] = body.country
if body.ip:
payload["ip"] = body.ip
response = _unipile_request("POST", "/api/v1/accounts", payload)
account_id = _safe_str(response.get("account_id") if isinstance(response, dict) else "") or _safe_str(
response.get("id") if isinstance(response, dict) else ""
)
if not account_id:
# For checkpoint or alt response structures, still persist and expose raw response.
account_id = _safe_str((response or {}).get("account", {}).get("id") if isinstance(response, dict) else "")
if not account_id:
raise HTTPException(status_code=400, detail="UniPile did not return an account_id")
account_profile = _try_fetch_unipile_account_profile(account_id)
merged_meta = response if isinstance(response, dict) else {"raw": str(response)}
if isinstance(merged_meta, dict) and account_profile:
merged_meta["account_profile"] = account_profile
db_row = UnipileAccount(
tenant_id=t.tenant_id,
user_id=t.user_id,
label=(body.label or "LinkedIn account").strip(),
provider="LINKEDIN",
unipile_account_id=account_id,
status=_safe_str(response.get("status") if isinstance(response, dict) else "") or "connected",
auth_mode=body.auth_mode,
metadata_json=merged_meta,
)
t.db.add(db_row)
t.db.commit()
t.db.refresh(db_row)
return {
"message": "LinkedIn account connected",
"account": {
"id": db_row.id,
"label": db_row.label,
"unipile_account_id": db_row.unipile_account_id,
"status": db_row.status,
},
"unipile_response": response,
}
@app.get("/api/linkedin-campaigns")
async def list_linkedin_campaigns(t: TenantContext = Depends(get_tenant_context)):
rows = (
t.db.query(LinkedinCampaign, UnipileAccount)
.join(UnipileAccount, UnipileAccount.id == LinkedinCampaign.unipile_account_ref_id)
.filter(
LinkedinCampaign.tenant_id == t.tenant_id,
LinkedinCampaign.user_id == t.user_id,
UnipileAccount.tenant_id == t.tenant_id,
UnipileAccount.user_id == t.user_id,
)
.order_by(LinkedinCampaign.created_at.desc(), LinkedinCampaign.id.desc())
.all()
)
return {
"campaigns": [
{
"id": c.id,
"name": c.name,
"status": c.status,
"file_id": c.file_id,
"contact_count": c.contact_count or 0,
"prompt_template": c.prompt_template or "",
"unipile_account_ref_id": c.unipile_account_ref_id,
"unipile_account_label": a.label if a else "",
"created_at": c.created_at.isoformat() if c.created_at else None,
"executed_at": c.executed_at.isoformat() if c.executed_at else None,
"execution_result": c.execution_result,
"followup_interval_hours": c.followup_interval_hours if c.followup_interval_hours is not None else 72,
}
for c, a in rows
]
}
@app.patch("/api/linkedin-campaigns/{campaign_id}")
async def patch_linkedin_campaign(
campaign_id: int,
body: LinkedinCampaignPatchRequest,
t: TenantContext = Depends(get_tenant_context),
):
db = t.db
row = (
db.query(LinkedinCampaign)
.filter(
LinkedinCampaign.tenant_id == t.tenant_id,
LinkedinCampaign.user_id == t.user_id,
LinkedinCampaign.id == campaign_id,
)
.first()
)
if not row:
raise HTTPException(status_code=404, detail="Campaign not found")
if body.followup_interval_hours is not None:
row.followup_interval_hours = body.followup_interval_hours
row.updated_at = datetime.utcnow()
db.commit()
return {"message": "Campaign updated", "followup_interval_hours": row.followup_interval_hours}
@app.post("/api/linkedin-campaigns")
async def create_linkedin_campaign(
body: LinkedinCampaignCreateRequest,
t: TenantContext = Depends(get_tenant_context),
):
account = (
t.db.query(UnipileAccount)
.filter(
UnipileAccount.tenant_id == t.tenant_id,
UnipileAccount.user_id == t.user_id,
UnipileAccount.id == body.unipile_account_ref_id,
)
.first()
)
if not account:
raise HTTPException(status_code=404, detail="LinkedIn account not found")
name = (body.name or "").strip()
if not name:
raise HTTPException(status_code=400, detail="Campaign name is required")
row = LinkedinCampaign(
tenant_id=t.tenant_id,
user_id=t.user_id,
name=name,
status="draft",
unipile_account_ref_id=body.unipile_account_ref_id,
)
t.db.add(row)
t.db.commit()
t.db.refresh(row)
return {"campaign_id": row.id, "message": "Campaign created"}
@app.post("/api/linkedin-campaigns/{campaign_id}/upload-csv")
async def upload_linkedin_campaign_csv(
campaign_id: int,
file: UploadFile = File(...),
t: TenantContext = Depends(get_tenant_context),
):
db = t.db
campaign = (
db.query(LinkedinCampaign)
.filter(
LinkedinCampaign.tenant_id == t.tenant_id,
LinkedinCampaign.user_id == t.user_id,
LinkedinCampaign.id == campaign_id,
)
.first()
)
if not campaign:
raise HTTPException(status_code=404, detail="Campaign not found")
file_id = str(uuid.uuid4())
file_path = UPLOAD_DIR / f"{file_id}.csv"
content = await file.read()
with open(file_path, "wb") as f:
f.write(content)
df = pd.read_csv(file_path)
contact_count = len(df)
db_file = UploadedFile(
tenant_id=t.tenant_id,
file_id=file_id,
filename=file.filename,
contact_count=contact_count,
file_path=str(file_path),
)
db.add(db_file)
# Replace prior campaign contacts from old file, if any.
if campaign.file_id:
db.query(Contact).filter(
Contact.tenant_id == t.tenant_id,
Contact.file_id == campaign.file_id,
Contact.source == "linkedin_campaign",
).delete()
db.query(GeneratedSequence).filter(
GeneratedSequence.tenant_id == t.tenant_id,
GeneratedSequence.file_id == campaign.file_id,
GeneratedSequence.channel == "linkedin",
GeneratedSequence.product == campaign.name,
).delete()
linkedin_profiles_detected = 0
for idx, row in df.iterrows():
row_dict = row.to_dict()
sanitized_raw_data = {str(k): _json_safe(v) for k, v in row_dict.items()}
li_raw = _pick_linkedin_url(sanitized_raw_data)
if _linkedin_public_identifier(li_raw):
linkedin_profiles_detected += 1
contact = Contact(
tenant_id=t.tenant_id,
file_id=file_id,
row_index=idx + 1,
first_name=_pick_field(row_dict, ["first name", "firstname", "first_name"]),
last_name=_pick_field(row_dict, ["last name", "lastname", "last_name"]),
email=_pick_field(row_dict, ["email", "work email", "email address"]),
company=_pick_field(row_dict, ["company", "company name", "organization name", "account name"]),
title=_pick_field(row_dict, ["title", "job title"]),
source="linkedin_campaign",
raw_data=sanitized_raw_data,
)
db.add(contact)
campaign.file_id = file_id
campaign.contact_count = contact_count
campaign.status = "draft"
campaign.updated_at = datetime.utcnow()
db.commit()
return {
"file_id": file_id,
"contact_count": contact_count,
"linkedin_profiles_detected": linkedin_profiles_detected,
"message": "Campaign CSV uploaded",
}
@app.post("/api/linkedin-campaigns/{campaign_id}/generate")
async def generate_linkedin_campaign_sequences(
campaign_id: int,
body: LinkedinCampaignGenerateRequest,
t: TenantContext = Depends(get_tenant_context),
):
db = t.db
campaign = (
db.query(LinkedinCampaign)
.filter(
LinkedinCampaign.tenant_id == t.tenant_id,
LinkedinCampaign.user_id == t.user_id,
LinkedinCampaign.id == campaign_id,
)
.first()
)
if not campaign:
raise HTTPException(status_code=404, detail="Campaign not found")
if not campaign.file_id:
raise HTTPException(status_code=400, detail="Upload a campaign CSV first")
db_file = (
db.query(UploadedFile)
.filter(UploadedFile.tenant_id == t.tenant_id, UploadedFile.file_id == campaign.file_id)
.first()
)
if not db_file:
raise HTTPException(status_code=404, detail="Campaign file not found")
df = pd.read_csv(db_file.file_path)
db.query(GeneratedSequence).filter(
GeneratedSequence.tenant_id == t.tenant_id,
GeneratedSequence.file_id == campaign.file_id,
GeneratedSequence.channel == "linkedin",
).delete()
urow = db.query(User).filter(User.id == t.user_id).first()
li_sig = ""
if urow:
li_sig = (getattr(urow, "linkedin_profile_display_name", None) or "").strip() or (urow.name or "").strip()
sequence_id = 1
generated_rows = 0
for _, row in df.iterrows():
contact = row.to_dict()
li_list = generate_linkedin_sequence(
contact, body.prompt_template, campaign.name, sender_linkedin_name=li_sig or None
)
if li_list and li_list[0].get("generation_failed"):
sequence_id += 1
continue
for seq_data in li_list:
db.add(
GeneratedSequence(
tenant_id=t.tenant_id,
file_id=campaign.file_id,
sequence_id=sequence_id,
email_number=seq_data["email_number"],
channel="linkedin",
first_name=seq_data["first_name"],
last_name=seq_data["last_name"],
email=seq_data["email"],
company=seq_data["company"],
title=seq_data.get("title", ""),
product=campaign.name,
subject=seq_data.get("subject") or "",
email_content=seq_data["email_content"],
)
)
generated_rows += 1
sequence_id += 1
campaign.prompt_template = body.prompt_template
campaign.status = "generated"
campaign.updated_at = datetime.utcnow()
db.commit()
return {"message": "LinkedIn sequences generated", "rows": generated_rows, "contacts": len(df)}
@app.get("/api/linkedin-campaigns/{campaign_id}/sequences")
async def get_linkedin_campaign_sequences(campaign_id: int, t: TenantContext = Depends(get_tenant_context)):
db = t.db
campaign = (
db.query(LinkedinCampaign)
.filter(
LinkedinCampaign.tenant_id == t.tenant_id,
LinkedinCampaign.user_id == t.user_id,
LinkedinCampaign.id == campaign_id,
)
.first()
)
if not campaign:
raise HTTPException(status_code=404, detail="Campaign not found")
if not campaign.file_id:
return {"sequences": []}
rows = (
db.query(GeneratedSequence)
.filter(
GeneratedSequence.tenant_id == t.tenant_id,
GeneratedSequence.file_id == campaign.file_id,
GeneratedSequence.channel == "linkedin",
)
.order_by(GeneratedSequence.sequence_id, GeneratedSequence.email_number)
.all()
)
return {
"sequences": [
{
"id": r.id,
"sequence_id": r.sequence_id,
"email_number": r.email_number,
"first_name": r.first_name or "",
"last_name": r.last_name or "",
"email": r.email or "",
"company": r.company or "",
"title": r.title or "",
"subject": r.subject or "",
"message": r.email_content or "",
}
for r in rows
]
}
@app.post("/api/linkedin-campaigns/{campaign_id}/execute")
async def execute_linkedin_campaign(
campaign_id: int,
body: LinkedinCampaignExecuteRequest,
t: TenantContext = Depends(get_tenant_context),
):
"""
Execute LinkedIn campaign via UniPile.
Sends a connection request using POST /users/invite with the first generated step as the
invite note (visible under LinkedIn → My Network → Manage invitations → Sent).
UniPile expects the public /in/{{slug}} identifier plus account_id on profile lookup — not a raw URL path.
If the invite cannot be sent (e.g. already connected), falls back to a 1:1 LinkedIn DM via chats API.
"""
db = t.db
campaign = (
db.query(LinkedinCampaign)
.filter(
LinkedinCampaign.tenant_id == t.tenant_id,
LinkedinCampaign.user_id == t.user_id,
LinkedinCampaign.id == campaign_id,
)
.first()
)
if not campaign:
raise HTTPException(status_code=404, detail="Campaign not found")
account = (
db.query(UnipileAccount)
.filter(
UnipileAccount.tenant_id == t.tenant_id,
UnipileAccount.user_id == t.user_id,
UnipileAccount.id == campaign.unipile_account_ref_id,
)
.first()
)
if not account:
raise HTTPException(status_code=404, detail="Connected LinkedIn account not found")
if not campaign.file_id:
raise HTTPException(status_code=400, detail="Campaign has no uploaded CSV")
contacts = (
db.query(Contact)
.filter(
Contact.tenant_id == t.tenant_id,
Contact.file_id == campaign.file_id,
Contact.source == "linkedin_campaign",
)
.order_by(Contact.row_index.asc(), Contact.id.asc())
.all()
)
seq_rows = (
db.query(GeneratedSequence)
.filter(
GeneratedSequence.tenant_id == t.tenant_id,
GeneratedSequence.file_id == campaign.file_id,
GeneratedSequence.channel == "linkedin",
GeneratedSequence.email_number == 1,
)
.order_by(GeneratedSequence.sequence_id.asc())
.all()
)
if not seq_rows:
raise HTTPException(
status_code=400,
detail="Generate LinkedIn sequences before executing the campaign.",
)
msg_by_seq = {r.sequence_id: r for r in seq_rows}
invite_sent = 0
dm_sent = 0
dry_preview = 0
skipped = 0
failed = 0
errors = []
attempts = []
acc_uid = account.unipile_account_id
user_row = db.query(User).filter(User.id == t.user_id).first()
def _contact_label(cr: Contact) -> str:
parts = [_safe_str(cr.first_name), _safe_str(cr.last_name)]
lab = " ".join(p for p in parts if p).strip()
return lab or "Contact"
for idx, c in enumerate(contacts, start=1):
first_msg = msg_by_seq.get(idx)
name_label = _contact_label(c)
if not first_msg:
skipped += 1
attempts.append(
{
"contact_id": c.id,
"name": name_label,
"status": "skipped",
"reason": "no_generated_message",
}
)
continue
raw_li = _pick_linkedin_url(c.raw_data or {})
url_disp = _normalize_linkedin_url(raw_li) or _safe_str(raw_li)
slug = _linkedin_public_identifier(raw_li)
if not slug:
skipped += 1
errors.append(
{
"contact_id": c.id,
"reason": "missing_linkedin_url",
"message": "No usable LinkedIn profile URL (expected Apollo-style Person Linkedin Url or /in/{{slug}}).",
}
)
attempts.append(
{
"contact_id": c.id,
"name": name_label,
"linkedin_url": url_disp,
"status": "skipped",
"reason": "missing_or_invalid_linkedin_url",
}
)
continue
note = _prepare_linkedin_outbound_text(
first_msg.email_content or "",
first_name=_safe_str(first_msg.first_name or c.first_name),
user_row=user_row,
for_invite=True,
)
if body.dry_run:
dry_preview += 1
attempts.append(
{
"contact_id": c.id,
"name": name_label,
"linkedin_url": url_disp,
"public_identifier": slug,
"status": "dry_run",
"detail": "Dry run: no UniPile calls. Uncheck the Dry run checkbox to send connection invites.",
"invite_note_preview": note[:180],
}
)
continue
slug_enc = requests.utils.quote(slug, safe="")
acc_enc = requests.utils.quote(acc_uid, safe="")
st_prof, profile = _unipile_call("GET", f"/api/v1/users/{slug_enc}?account_id={acc_enc}")
if st_prof >= 400 or not isinstance(profile, dict):
failed += 1
detail = profile if isinstance(profile, dict) else {}
msg = (
detail.get("message")
or detail.get("detail")
or detail.get("error")
or (json.dumps(detail)[:800] if detail else "profile lookup failed")
)
if isinstance(msg, (dict, list)):
msg = json.dumps(msg)
errors.append({"contact_id": c.id, "reason": "profile_lookup_failed", "message": str(msg)})
attempts.append(
{
"contact_id": c.id,
"name": name_label,
"linkedin_url": url_disp,
"public_identifier": slug,
"status": "failed",
"step": "resolve_profile",
"detail": str(msg)[:500],
}
)
continue
provider_id = _safe_str(profile.get("provider_id"))
rel_hint = _safe_str(
profile.get("network_distance") or profile.get("distance") or ""
)
if not rel_hint and isinstance(profile.get("relation"), dict):
rel_hint = _safe_str((profile.get("relation") or {}).get("distance"))
if not provider_id:
failed += 1
errors.append(
{
"contact_id": c.id,
"reason": "missing_provider_id",
"message": "UniPile profile response had no provider_id.",
}
)
attempts.append(
{
"contact_id": c.id,
"name": name_label,
"linkedin_url": url_disp,
"public_identifier": slug,
"status": "failed",
"step": "resolve_profile",
"detail": "provider_id missing from UniPile profile payload",
}
)
continue
st_inv, inv_res = _unipile_call(
"POST",
"/api/v1/users/invite",
{"provider_id": provider_id, "account_id": acc_uid, "message": note},
)
if st_inv < 400:
invite_sent += 1
ts = datetime.utcnow()
c.unipile_provider_id = provider_id
c.linkedin_invite_pending = 1
c.linkedin_invite_sent_at = ts
c.linkedin_followup_next_email_number = 2
attempts.append(
{
"contact_id": c.id,
"name": name_label,
"linkedin_url": url_disp,
"public_identifier": slug,
"provider_id": provider_id,
"status": "invite_sent",
"network_distance": rel_hint or None,
"connection_hint": rel_hint or None,
}
)
continue
inv_err = ""
if isinstance(inv_res, dict):
inv_err = _safe_str(
inv_res.get("message") or inv_res.get("detail") or inv_res.get("error") or ""
)
if isinstance(inv_res.get("errors"), list):
inv_err = inv_err or json.dumps(inv_res.get("errors"))[:400]
try:
dm_text = _prepare_linkedin_outbound_text(
first_msg.email_content or "",
first_name=_safe_str(first_msg.first_name or c.first_name),
user_row=user_row,
)
_send_unipile_dm(acc_uid, provider_id, dm_text)
dm_sent += 1
ts = datetime.utcnow()
c.unipile_provider_id = provider_id
c.linkedin_invite_pending = 0
c.linkedin_connection_accepted_at = ts
c.linkedin_last_followup_sent_at = ts
c.linkedin_followup_next_email_number = 2
attempts.append(
{
"contact_id": c.id,
"name": name_label,
"linkedin_url": url_disp,
"public_identifier": slug,
"provider_id": provider_id,
"status": "message_sent",
"network_distance": rel_hint or None,
"connection_hint": rel_hint or None,
"fallback_from_invite_error": inv_err or None,
"detail": "Connection invite was not sent (see fallback_from_invite_error); sent LinkedIn DM instead.",
}
)
except Exception as e:
failed += 1
errors.append(
{
"contact_id": c.id,
"reason": "invite_and_dm_failed",
"message": f"invite: {inv_err}; dm: {str(e)}",
}
)
attempts.append(
{
"contact_id": c.id,
"name": name_label,
"linkedin_url": url_disp,
"public_identifier": slug,
"status": "failed",
"step": "invite_and_message",
"invite_error": inv_err or None,
"detail": str(e),
}
)
total_ok = dry_preview if body.dry_run else (invite_sent + dm_sent)
result = {
"dry_run": body.dry_run,
"dry_run_preview_count": dry_preview,
"linkedin_invites_sent": invite_sent,
"direct_messages_sent": dm_sent,
"sent": total_ok,
"skipped": skipped,
"failed": failed,
"errors": errors[:200],
"attempts": attempts[:500],
"execution_kind": "linkedin_connection_invite",
"help": (
"Invites appear under LinkedIn → My Network → Manage invitations → Sent. "
"Follow-up steps are sent via POST /api/linkedin-campaigns/{id}/process-followups after acceptance "
"(UniPile users webhook new_relation marks acceptance; can lag hours)."
),
}
campaign.status = "completed" if failed == 0 else "failed"
campaign.execution_result = result
campaign.executed_at = datetime.utcnow()
campaign.updated_at = datetime.utcnow()
db.commit()
return {"message": "Campaign execution finished", **result}
@app.post("/api/linkedin-campaigns/{campaign_id}/process-followups")
async def process_linkedin_campaign_followups(
campaign_id: int,
t: TenantContext = Depends(get_tenant_context),
):
"""
Deliver generated sequence steps 2+ as LinkedIn DMs when due.
* Invite flow: waits for `linkedin_connection_accepted_at` (set by UniPile `new_relation` webhook or DM fallback).
* First follow-up (step 2) after invite: `accepted_at + followup_interval_hours`.
* If execute fell back to DM for step 1: step 2 waits `last_followup_sent_at + interval`.
Run this endpoint on a schedule (e.g. hourly cron); it only sends when the interval has elapsed.
"""
db = t.db
campaign = (
db.query(LinkedinCampaign)
.filter(
LinkedinCampaign.tenant_id == t.tenant_id,
LinkedinCampaign.user_id == t.user_id,
LinkedinCampaign.id == campaign_id,
)
.first()
)
if not campaign:
raise HTTPException(status_code=404, detail="Campaign not found")
account = (
db.query(UnipileAccount)
.filter(
UnipileAccount.tenant_id == t.tenant_id,
UnipileAccount.user_id == t.user_id,
UnipileAccount.id == campaign.unipile_account_ref_id,
)
.first()
)
if not account:
raise HTTPException(status_code=404, detail="Connected LinkedIn account not found")
if not campaign.file_id:
raise HTTPException(status_code=400, detail="Campaign has no uploaded CSV")
interval_h = campaign.followup_interval_hours if campaign.followup_interval_hours else 72
now = datetime.utcnow()
user_row = db.query(User).filter(User.id == t.user_id).first()
contacts = (
db.query(Contact)
.filter(
Contact.tenant_id == t.tenant_id,
Contact.file_id == campaign.file_id,
Contact.source == "linkedin_campaign",
)
.order_by(Contact.row_index.asc(), Contact.id.asc())
.all()
)
sent = 0
skipped = 0
failed = 0
details = []
for c in contacts:
n = c.linkedin_followup_next_email_number
if not n or not c.unipile_provider_id:
skipped += 1
continue
max_step = (
db.query(func.max(GeneratedSequence.email_number))
.filter(
GeneratedSequence.tenant_id == t.tenant_id,
GeneratedSequence.file_id == campaign.file_id,
GeneratedSequence.channel == "linkedin",
GeneratedSequence.sequence_id == c.row_index,
)
.scalar()
) or 0
if max_step < 2:
skipped += 1
continue
if n > max_step:
c.linkedin_followup_next_email_number = None
skipped += 1
continue
if c.linkedin_invite_pending and not c.linkedin_connection_accepted_at:
skipped += 1
continue
if n == 2:
if c.linkedin_last_followup_sent_at is None:
if not c.linkedin_connection_accepted_at:
skipped += 1
continue
if now < c.linkedin_connection_accepted_at + timedelta(hours=interval_h):
skipped += 1
continue
else:
if now < c.linkedin_last_followup_sent_at + timedelta(hours=interval_h):
skipped += 1
continue
else:
if not c.linkedin_last_followup_sent_at:
skipped += 1
continue
if now < c.linkedin_last_followup_sent_at + timedelta(hours=interval_h):
skipped += 1
continue
msg_row = (
db.query(GeneratedSequence)
.filter(
GeneratedSequence.tenant_id == t.tenant_id,
GeneratedSequence.file_id == campaign.file_id,
GeneratedSequence.channel == "linkedin",
GeneratedSequence.sequence_id == c.row_index,
GeneratedSequence.email_number == n,
)
.first()
)
if not msg_row:
skipped += 1
continue
try:
dm_text = _prepare_linkedin_outbound_text(
msg_row.email_content or "",
first_name=_safe_str(msg_row.first_name or c.first_name),
user_row=user_row,
)
_send_unipile_dm(
account.unipile_account_id,
c.unipile_provider_id,
dm_text,
)
c.linkedin_last_followup_sent_at = now
c.linkedin_followup_next_email_number = n + 1 if n < max_step else None
sent += 1
details.append({"contact_id": c.id, "step": n, "status": "sent"})
except Exception as e:
failed += 1
details.append({"contact_id": c.id, "step": n, "status": "failed", "error": str(e)})
campaign.updated_at = datetime.utcnow()
db.commit()
return {
"message": "Follow-up pass finished",
"sent": sent,
"skipped": skipped,
"failed": failed,
"followup_interval_hours": interval_h,
"details": details[:200],
}
@app.get("/api/contact-fields")
async def contact_fields(t: TenantContext = Depends(get_tenant_context)):
"""Return all available contact field names from uploaded Apollo rows."""
db = t.db
contacts = db.query(Contact.raw_data).filter(Contact.tenant_id == t.tenant_id).all()
fields = set(["first_name", "last_name", "email", "company", "title", "file_id", "created_at"])
for item in contacts:
raw = item[0] or {}
if isinstance(raw, dict):
for key in raw.keys():
fields.add(str(key))
return {"fields": sorted(fields)}
@app.get("/api/contacts")
async def list_contacts(
search: str = Query("", description="Search by name/email/company/title"),
filters: str = Query(
"",
description='JSON array of {field, op, value?, from_value?, to_value?}; combined with AND',
),
field: str = Query("", description="Optional field name to filter (legacy single filter)"),
value: str = Query("", description="Optional field value to filter"),
op: str = Query("contains", description="Filter op: contains|equals|from|to|between"),
from_value: str = Query("", description="From value for range filters"),
to_value: str = Query("", description="To value for range filters"),
sort_by: str = Query("created_at", description="Sort field"),
sort_dir: str = Query("desc", description="Sort direction: asc|desc"),
limit: int = Query(200, ge=1, le=1000),
offset: int = Query(0, ge=0),
t: TenantContext = Depends(get_tenant_context),
):
db = t.db
query = db.query(Contact).filter(Contact.tenant_id == t.tenant_id)
visible_ids = _visible_owner_user_ids(db, t)
if visible_ids is not None:
query = query.filter(or_(Contact.owner_user_id.in_(visible_ids), Contact.owner_user_id.is_(None)))
if search:
pattern = f"%{search}%"
query = query.filter(
(Contact.first_name.ilike(pattern))
| (Contact.last_name.ilike(pattern))
| (Contact.email.ilike(pattern))
| (Contact.company.ilike(pattern))
| (Contact.title.ilike(pattern))
)
contacts_all = query.order_by(Contact.created_at.desc(), Contact.id.desc()).all()
# Multiple Apollo-field filters (AND). Prefer over legacy single-field params when non-empty.
if filters and filters.strip():
try:
rules = json.loads(filters)
except json.JSONDecodeError:
raise HTTPException(status_code=400, detail="Invalid filters JSON")
if not isinstance(rules, list):
raise HTTPException(status_code=400, detail="filters must be a JSON array")
active_rules = [r for r in rules if isinstance(r, dict) and _safe_str(r.get("field"))]
if active_rules:
contacts_all = [
c
for c in contacts_all
if all(_contact_matches_apollo_filter(c, r) for r in active_rules)
]
elif field:
rule = {
"field": field,
"op": op,
"value": value,
"from_value": from_value,
"to_value": to_value,
}
contacts_all = [c for c in contacts_all if _contact_matches_apollo_filter(c, rule)]
# Sort by mapped or raw Apollo fields
reverse = (sort_dir or "desc").lower() != "asc"
sort_field = sort_by or "created_at"
def sort_key(c: Contact):
val = _contact_value(c, sort_field)
num = _to_float(val)
if num is not None:
return (0, num)
dt = _to_datetime(val)
if dt is not None:
return (1, dt.timestamp())
return (2, _safe_str(val).lower())
contacts_all = sorted(contacts_all, key=sort_key, reverse=reverse)
total = len(contacts_all)
contacts = contacts_all[offset:offset + limit]
contacts_out = [
{
"id": c.id,
"file_id": c.file_id,
"row_index": c.row_index,
"first_name": c.first_name or _pick_from_raw(c.raw_data, ["first name", "firstname", "first_name"]),
"last_name": c.last_name or _pick_from_raw(c.raw_data, ["last name", "lastname", "last_name"]),
"email": c.email or _pick_from_raw(c.raw_data, ["email", "work email", "email address", "secondary email", "tertiary email"]),
"company": c.company or _pick_from_raw(c.raw_data, ["company", "company name", "company name for emails", "organization name", "account name"]),
"title": c.title or _pick_from_raw(c.raw_data, ["title", "job title"]),
"phone": c.phone or "",
"source": c.source or "apollo_csv",
"owner_user_id": c.owner_user_id,
"created_at": c.created_at.isoformat() if c.created_at else None,
}
for c in contacts
]
_attach_owner_display(contacts_out, db)
return {
"total": total,
"contacts": contacts_out,
}
@app.get("/api/contacts/by-company")
async def get_contacts_by_company(
company: str = Query("", description="Company name to match (case-insensitive)"),
t: TenantContext = Depends(get_tenant_context),
):
"""Associated-contacts lookup used by the Leads/Deals detail panel, mirroring the same
case-insensitive company-name match already used on the Companies page."""
contacts = _contacts_for_company(t.db, t.tenant_id, company)
return {
"contacts": [
{
"id": c.id,
"first_name": c.first_name or "",
"last_name": c.last_name or "",
"email": c.email or "",
"title": c.title or "",
"phone": c.phone or "",
}
for c in contacts
],
}
def _activity_to_dict(a: ContactActivity, user: User | None, contact: Contact | None = None) -> dict:
d = {
"id": a.id,
"contact_id": a.contact_id,
"activity_type": a.activity_type,
"notes": a.notes or "",
"created_at": a.created_at.isoformat() if a.created_at else None,
"user_id": a.user_id,
"user_display_name": (user.name or user.email or "").strip() if user else "",
"user_initials": _user_initials_from_user(user),
}
if contact is not None:
d["contact_name"] = _format_contact_display(contact)
return d
@app.post("/api/contacts/{contact_id}/activities")
async def create_contact_activity(
contact_id: int, body: ActivityCreateRequest, t: TenantContext = Depends(get_tenant_context)
):
"""Log a Call or Visit against a contact, with optional notes on what was discussed."""
db = t.db
contact = (
db.query(Contact)
.filter(Contact.tenant_id == t.tenant_id, Contact.id == contact_id)
.first()
)
if not contact:
raise HTTPException(status_code=404, detail="Contact not found")
activity = ContactActivity(
tenant_id=t.tenant_id,
contact_id=contact_id,
user_id=t.user_id,
activity_type=body.activity_type,
notes=_safe_str(body.notes),
)
db.add(activity)
db.commit()
db.refresh(activity)
user = db.query(User).filter(User.id == t.user_id).first()
return _activity_to_dict(activity, user)
@app.get("/api/contacts/{contact_id}/activities")
async def list_contact_activities(contact_id: int, t: TenantContext = Depends(get_tenant_context)):
"""Activity timeline (Calls/Visits, with notes and author) for a single contact."""
db = t.db
contact = (
db.query(Contact)
.filter(Contact.tenant_id == t.tenant_id, Contact.id == contact_id)
.first()
)
if not contact:
raise HTTPException(status_code=404, detail="Contact not found")
q = db.query(ContactActivity).filter(
ContactActivity.tenant_id == t.tenant_id, ContactActivity.contact_id == contact_id
)
visible_ids = _visible_owner_user_ids(db, t)
if visible_ids is not None:
q = q.filter(ContactActivity.user_id.in_(visible_ids))
rows = q.order_by(ContactActivity.created_at.desc()).all()
user_ids = {r.user_id for r in rows if r.user_id}
users_by_id = {u.id: u for u in db.query(User).filter(User.id.in_(user_ids)).all()} if user_ids else {}
return {"activities": [_activity_to_dict(r, users_by_id.get(r.user_id)) for r in rows]}
@app.get("/api/activities")
async def list_activities(
limit: int = Query(500, ge=1, le=1000),
offset: int = Query(0, ge=0),
t: TenantContext = Depends(get_tenant_context),
):
"""Tenant-wide Call/Visit log (not scoped to one contact) — powers the Calls/Visits stat
boxes on the Sales Dashboard. Same role scoping as the per-contact timeline: everyone sees
their own logged activity, a team manager also sees their team's, admins see everyone's."""
db = t.db
q = db.query(ContactActivity).filter(ContactActivity.tenant_id == t.tenant_id)
visible_ids = _visible_owner_user_ids(db, t)
if visible_ids is not None:
q = q.filter(ContactActivity.user_id.in_(visible_ids))
total = q.count()
rows = q.order_by(ContactActivity.created_at.desc()).offset(offset).limit(limit).all()
user_ids = {r.user_id for r in rows if r.user_id}
users_by_id = {u.id: u for u in db.query(User).filter(User.id.in_(user_ids)).all()} if user_ids else {}
return {
"total": total,
"activities": [_activity_to_dict(r, users_by_id.get(r.user_id)) for r in rows],
}
@app.post("/api/contacts")
async def create_contact(body: ContactCreateRequest, t: TenantContext = Depends(get_tenant_context)):
"""Create a contact manually (inline table add). Email must be unique."""
db = t.db
email = _safe_str(body.email).lower()
if not email:
raise HTTPException(status_code=400, detail="Email is required")
exists = (
db.query(Contact)
.filter(
Contact.tenant_id == t.tenant_id,
func.lower(Contact.email) == email,
)
.first()
)
if exists:
raise HTTPException(status_code=409, detail="A contact with this email already exists")
fn = _safe_str(body.first_name)
ln = _safe_str(body.last_name)
co = _safe_str(body.company)
ti = _safe_str(body.title)
ph = _safe_str(body.phone)
raw = {
"First Name": fn,
"Last Name": ln,
"Email": email,
"Company Name": co,
"Title": ti,
}
contact = Contact(
tenant_id=t.tenant_id,
file_id=MANUAL_CONTACT_FILE_ID,
row_index=0,
first_name=fn,
last_name=ln,
email=email,
company=co,
title=ti,
phone=ph,
source="manual",
raw_data=raw,
)
db.add(contact)
db.commit()
db.refresh(contact)
return {
"id": contact.id,
"file_id": contact.file_id,
"row_index": contact.row_index,
"first_name": contact.first_name,
"last_name": contact.last_name,
"email": contact.email,
"company": contact.company,
"title": contact.title,
"phone": contact.phone or "",
"source": contact.source or "manual",
"created_at": contact.created_at.isoformat() if contact.created_at else None,
}
@app.post("/api/contacts/bulk-delete")
async def bulk_delete_contacts(body: BulkContactIdsRequest, t: TenantContext = Depends(get_tenant_context)):
db = t.db
if not body.contact_ids:
raise HTTPException(status_code=400, detail="contact_ids required")
ids = list({int(x) for x in body.contact_ids})
deleted = (
db.query(Contact)
.filter(Contact.tenant_id == t.tenant_id, Contact.id.in_(ids))
.delete(synchronize_session=False)
)
db.commit()
return {"deleted": deleted}
@app.post("/api/contacts/seed-demo")
async def seed_demo_contacts(t: TenantContext = Depends(get_tenant_context)):
"""
Replace previous demo-seeded contacts and insert a variety of sample rows
(rich Apollo-style raw_data) for testing filters and UI.
"""
db = t.db
tenant_id = t.tenant_id
removed = (
db.query(Contact)
.filter(
Contact.tenant_id == tenant_id,
Contact.file_id == DEMO_CONTACTS_FILE_ID,
)
.delete(synchronize_session=False)
)
specs = [
{
"fn": "Morgan",
"ln": "Lee",
"email": "morgan.lee@blueharbor.demo",
"co": "Blue Harbor Analytics",
"ti": "Chief Revenue Officer",
"raw": {
"Company Name": "Blue Harbor Analytics",
"Company Name for Emails": "BlueHarbor",
"Industry": "Business intelligence & analytics software",
"# Employees": "51-200",
"Annual Revenue": "$8,200,000",
"Last Raised At": "2023-11",
"Website": "https://www.blueharbor-analytics.demo",
"City": "Seattle",
"State": "WA",
"Country": "United States",
},
},
{
"fn": "Priya",
"ln": "Shah",
"email": "priya.shah@lumengrid.demo",
"co": "LumenGrid Energy",
"ti": "Director of Procurement",
"raw": {
"Company Name": "LumenGrid Energy",
"Industry": "Renewable energy infrastructure",
"# Employees": "201-500",
"Annual Revenue": "$45,000,000",
"Website": "https://lumengrid.demo",
"City": "Toronto",
"State": "ON",
"Country": "Canada",
},
},
{
"fn": "Diego",
"ln": "Ramos",
"email": "diego.ramos@vertexledger.demo",
"co": "Vertex Ledger",
"ti": "Controller",
"raw": {
"Company Name": "Vertex Ledger",
"Industry": "Enterprise document and content management software",
"# Employees": "11-50",
"Annual Revenue": "$3,100,000",
"Website": "https://vertexledger.demo",
"City": "Austin",
"State": "TX",
"Country": "United States",
},
},
{
"fn": "Elena",
"ln": "Volkov",
"email": "elena.volkov@northpole.demo",
"co": "Northpole Freight Co.",
"ti": "Operations Manager",
"raw": {
"Company Name": "Northpole Freight Co.",
"Industry": "Logistics & Supply Chain",
"# Employees": "501-1000",
"Annual Revenue": "$120,000,000",
"Website": "https://northpole-freight.demo",
"City": "Hamburg",
"State": "",
"Country": "Germany",
},
},
{
"fn": "James",
"ln": "Okonkwo",
"email": "james.okonkwo@silverfin.demo",
"co": "Silverfin Capital",
"ti": "Investment Analyst",
"raw": {
"Company Name": "Silverfin Capital",
"Industry": "Financial services",
"# Employees": "51-200",
"Annual Revenue": "$22,500,000",
"Website": "https://silverfin.demo",
"City": "London",
"State": "",
"Country": "United Kingdom",
},
},
{
"fn": "Yuki",
"ln": "Tanaka",
"email": "yuki.tanaka@skyforge.demo",
"co": "Skyforge Robotics",
"ti": "Head of Engineering",
"raw": {
"Company Name": "Skyforge Robotics",
"Industry": "Industrial automation",
"# Employees": "201-500",
"Annual Revenue": "$67,000,000",
"Website": "https://skyforge.demo",
"City": "Osaka",
"State": "",
"Country": "Japan",
},
},
{
"fn": "Amira",
"ln": "Hassan",
"email": "amira.hassan@desertwave.demo",
"co": "Desert Wave Telecom",
"ti": "VP Customer Success",
"raw": {
"Company Name": "Desert Wave Telecom",
"Industry": "Telecommunications",
"# Employees": "1001-5000",
"Annual Revenue": "$410,000,000",
"Website": "https://desertwave.demo",
"City": "Dubai",
"State": "",
"Country": "United Arab Emirates",
},
},
{
"fn": "Chris",
"ln": "Martinez",
"email": "chris.martinez@cascade.demo",
"co": "Cascade Health Systems",
"ti": "Clinical Director",
"raw": {
"Company Name": "Cascade Health Systems",
"Industry": "Healthcare IT",
"# Employees": "501-1000",
"Annual Revenue": "$95,000,000",
"Website": "https://cascade-health.demo",
"City": "Denver",
"State": "CO",
"Country": "United States",
},
},
{
"fn": "Sofia",
"ln": "Andersson",
"email": "sofia.andersson@fjord.demo",
"co": "Fjord Maritime AS",
"ti": "Fleet Coordinator",
"raw": {
"Company Name": "Fjord Maritime AS",
"Industry": "Shipping & maritime",
"# Employees": "51-200",
"Annual Revenue": "$18,000,000",
"Website": "https://fjord-maritime.demo",
"City": "Bergen",
"State": "",
"Country": "Norway",
},
},
{
"fn": "Marcus",
"ln": "Webbe",
"email": "marcus.webbe@pixelloft.demo",
"co": "Pixelloft Creative Agency",
"ti": "Creative Director",
"raw": {
"Company Name": "Pixelloft Creative Agency",
"Industry": "Marketing & advertising",
"# Employees": "11-50",
"Annual Revenue": "$1,800,000",
"Website": "https://pixelloft.demo",
"City": "Melbourne",
"State": "VIC",
"Country": "Australia",
},
},
]
for i, s in enumerate(specs):
rd = dict(s["raw"])
rd["Title"] = s["ti"]
rd["Email"] = s["email"]
rd["First Name"] = s["fn"]
rd["Last Name"] = s["ln"]
row = Contact(
tenant_id=tenant_id,
file_id=DEMO_CONTACTS_FILE_ID,
row_index=i,
first_name=s["fn"],
last_name=s["ln"],
email=s["email"],
company=s["co"],
title=s["ti"],
source="demo",
raw_data=rd,
)
db.add(row)
db.commit()
return {
"ok": True,
"removed_previous_demo_rows": removed,
"inserted": len(specs),
}
@app.post("/api/contacts/bulk-convert-to-leads")
async def bulk_convert_contacts_to_leads(body: BulkContactIdsRequest, t: TenantContext = Depends(get_tenant_context)):
"""
For each contact: create a CrmLead (campaign «Contacts») copied from the contact.
Contacts are not removed; they remain until explicitly deleted. Fails per-row if the
contact has no email or a lead with the same email already exists.
"""
db = t.db
if not body.contact_ids:
raise HTTPException(status_code=400, detail="contact_ids required")
converted = 0
errors: List[dict] = []
for cid in body.contact_ids:
contact = (
db.query(Contact)
.filter(Contact.tenant_id == t.tenant_id, Contact.id == int(cid))
.first()
)
if not contact:
errors.append({"contact_id": cid, "error": "not found"})
continue
r = _convert_contact_to_lead_core(db, contact)
if not r["ok"]:
errors.append({"contact_id": cid, "error": r.get("error", "failed")})
else:
converted += 1
db.commit()
return {"converted": converted, "errors": errors}
@app.get("/api/company-names")
async def search_company_names(
q: str = Query("", description="Substring match on Contact.company"),
limit: int = Query(25, ge=1, le=100),
t: TenantContext = Depends(get_tenant_context),
):
"""Distinct company strings from contacts for deal/account linking."""
db = t.db
raw_q = _safe_str(q)
pattern = f"%{raw_q}%" if raw_q else "%"
rows = (
db.query(Contact.company)
.filter(
Contact.tenant_id == t.tenant_id,
Contact.company.isnot(None),
Contact.company != "",
Contact.company.ilike(pattern),
)
.distinct()
.order_by(Contact.company.asc())
.limit(limit * 4)
.all()
)
names: List[str] = []
seen = set()
for (co,) in rows:
s = _safe_str(co)
if not s:
continue
lk = s.lower()
if lk in seen:
continue
seen.add(lk)
names.append(s)
if len(names) >= limit:
break
return {"names": names}
@app.get("/api/contacts/{contact_id}")
async def get_contact(contact_id: int, t: TenantContext = Depends(get_tenant_context)):
db = t.db
contact = (
db.query(Contact)
.filter(Contact.tenant_id == t.tenant_id, Contact.id == contact_id)
.first()
)
if not contact:
raise HTTPException(status_code=404, detail="Contact not found")
_contact_access_or_403(contact, t)
d = {
"id": contact.id,
"file_id": contact.file_id,
"row_index": contact.row_index,
"first_name": contact.first_name or _pick_from_raw(contact.raw_data, ["first name", "firstname", "first_name"]),
"last_name": contact.last_name or _pick_from_raw(contact.raw_data, ["last name", "lastname", "last_name"]),
"email": contact.email or _pick_from_raw(contact.raw_data, ["email", "work email", "email address", "secondary email", "tertiary email"]),
"company": contact.company or _pick_from_raw(contact.raw_data, ["company", "company name", "company name for emails", "organization name", "account name"]),
"title": contact.title or _pick_from_raw(contact.raw_data, ["title", "job title"]),
"phone": contact.phone or "",
"source": contact.source or "apollo_csv",
"owner_user_id": contact.owner_user_id,
"created_at": contact.created_at.isoformat() if contact.created_at else None,
"raw_data": contact.raw_data or {},
}
_attach_owner_display([d], db)
return d
# Merged into Contact.raw_data["Company Name for Emails"], etc. (Apollo CSV keys)
COMPANY_DETAIL_PATCH_FIELDS = {
"company_name_for_emails": "Company Name for Emails",
"industry": "Industry",
"employees": "# Employees",
"annual_revenue": "Annual Revenue",
"last_raised_at": "Last Raised At",
"website": "Website",
"city": "City",
"state": "State",
"country_region": "Country",
"category": "Category",
"type": "Type",
"company_phone": "Phone",
}
def _contact_company_details_dict(c: Contact) -> dict:
rd = c.raw_data or {}
return {
"Company Name": _safe_str(c.company or rd.get("Company Name")),
"Company Name for Emails": _safe_str(rd.get("Company Name for Emails")),
"Industry": _safe_str(rd.get("Industry")),
"# Employees": _safe_str(rd.get("# Employees")),
"Annual Revenue": _safe_str(rd.get("Annual Revenue")),
"Last Raised At": _safe_str(rd.get("Last Raised At")),
"Website": _safe_str(rd.get("Website")),
"City": _safe_str(rd.get("City")),
"State": _safe_str(rd.get("State")),
"Country": _safe_str(rd.get("Country")),
"Category": _safe_str(rd.get("Category")),
"Type": _safe_str(rd.get("Type")),
"Phone": _safe_str(rd.get("Phone")),
}
def _merge_contact_raw_company_details(contact: Contact, patch: dict) -> None:
rd = dict(contact.raw_data or {})
for py_key, raw_key in COMPANY_DETAIL_PATCH_FIELDS.items():
if py_key in patch:
rd[raw_key] = _safe_str(patch[py_key])
contact.raw_data = rd
def _lead_company_details_dict(row: CrmLead) -> dict:
rw = row.raw_webhook if isinstance(row.raw_webhook, dict) else {}
cd = rw.get("company_details") if isinstance(rw.get("company_details"), dict) else {}
return {
"Company Name": _safe_str(row.company_name or cd.get("Company Name")),
"Company Name for Emails": _safe_str(cd.get("Company Name for Emails")),
"Industry": _safe_str(cd.get("Industry")),
"# Employees": _safe_str(cd.get("# Employees")),
"Annual Revenue": _safe_str(cd.get("Annual Revenue")),
"Last Raised At": _safe_str(cd.get("Last Raised At")),
"Website": _safe_str(cd.get("Website")),
"City": _safe_str(cd.get("City")),
"State": _safe_str(cd.get("State")),
"Country": _safe_str(cd.get("Country")),
"Category": _safe_str(cd.get("Category")),
"Type": _safe_str(cd.get("Type")),
"Phone": _safe_str(cd.get("Phone")),
}
def _deal_fallback_company_details(d: CrmDeal) -> dict:
return {
"Company Name": _safe_str(d.account_name),
"Company Name for Emails": "",
"Industry": "",
"# Employees": "",
"Annual Revenue": "",
"Last Raised At": "",
"Website": "",
"City": "",
"State": "",
"Country": _safe_str(d.country),
"Category": "",
"Type": "",
"Phone": "",
}
def _format_contact_display(contact: Contact) -> str:
fn = _safe_str(contact.first_name)
ln = _safe_str(contact.last_name)
person = " ".join(filter(None, [fn, ln])).strip()
return person or _safe_str(contact.email)
def _enrich_deal_response(db: Session, row: CrmDeal) -> dict:
out = _deal_to_dict(row)
if row.owner_user_id:
ou = db.query(User).filter(User.id == row.owner_user_id).first()
if ou:
out["owner_display_name"] = (ou.name or ou.email or "").strip() or None
if not (out.get("owner_initials") or "").strip():
out["owner_initials"] = _user_initials_from_user(ou)
else:
out["owner_display_name"] = None
out["linked_contact"] = None
if row.contact_id:
c = (
db.query(Contact)
.filter(
Contact.tenant_id == row.tenant_id,
Contact.id == row.contact_id,
)
.first()
)
if c:
out["company_details"] = _contact_company_details_dict(c)
out["company_details_contact_id"] = c.id
out["linked_contact"] = {
"id": c.id,
"first_name": c.first_name or "",
"last_name": c.last_name or "",
"email": c.email or "",
"title": c.title or "",
"phone": c.phone or "",
}
else:
out["company_details"] = _deal_fallback_company_details(row)
else:
out["company_details"] = _deal_fallback_company_details(row)
return out
def _sync_contact_raw_from_columns(contact: Contact) -> None:
rd = dict(contact.raw_data or {})
rd["First Name"] = contact.first_name or ""
rd["Last Name"] = contact.last_name or ""
rd["Email"] = (contact.email or "").strip()
rd["Company Name"] = contact.company or ""
rd["Title"] = contact.title or ""
contact.raw_data = rd
@app.patch("/api/contacts/{contact_id}")
async def patch_contact(contact_id: int, body: ContactPatchRequest, t: TenantContext = Depends(get_tenant_context)):
db = t.db
contact = (
db.query(Contact)
.filter(Contact.tenant_id == t.tenant_id, Contact.id == contact_id)
.first()
)
if not contact:
raise HTTPException(status_code=404, detail="Contact not found")
_contact_access_or_403(contact, t)
data = body.model_dump(exclude_unset=True)
if not data:
raise HTTPException(status_code=400, detail="No fields to update")
if "owner_user_id" in data:
val = data.pop("owner_user_id")
mids = _tenant_member_ids(db, t.tenant_id)
if t.role == "admin":
if val is None:
contact.owner_user_id = None
else:
vid = int(val)
if vid not in mids:
raise HTTPException(status_code=400, detail="Owner must be a member of this workspace")
contact.owner_user_id = vid
else:
if val is None:
raise HTTPException(status_code=403, detail="Only admins can set a contact to unassigned")
if int(val) != int(t.user_id):
raise HTTPException(status_code=403, detail="Only admins can assign contacts to other users")
if contact.owner_user_id is not None and int(contact.owner_user_id) != int(t.user_id):
raise HTTPException(status_code=403, detail="You can only take ownership of unassigned contacts")
contact.owner_user_id = t.user_id
if "email" in data:
email = _safe_str(data["email"]).lower()
if not email:
raise HTTPException(status_code=400, detail="Email cannot be empty")
taken = (
db.query(Contact)
.filter(
Contact.tenant_id == t.tenant_id,
Contact.id != contact_id,
func.lower(Contact.email) == email,
)
.first()
)
if taken:
raise HTTPException(status_code=409, detail="Another contact already uses this email")
contact.email = email
if "first_name" in data:
contact.first_name = _safe_str(data["first_name"])
if "last_name" in data:
contact.last_name = _safe_str(data["last_name"])
if "company" in data:
contact.company = _safe_str(data["company"])
if "title" in data:
contact.title = _safe_str(data["title"])
if "phone" in data:
contact.phone = _safe_str(data["phone"])
_sync_contact_raw_from_columns(contact)
_merge_contact_raw_company_details(contact, data)
company_fields = {k: data[k] for k in COMPANY_DETAIL_PATCH_FIELDS if k in data}
if company_fields:
_propagate_company_fields_everywhere(db, t.tenant_id, contact.company, company_fields)
db.commit()
db.refresh(contact)
d = {
"id": contact.id,
"file_id": contact.file_id,
"row_index": contact.row_index,
"first_name": contact.first_name or _pick_from_raw(contact.raw_data, ["first name", "firstname", "first_name"]),
"last_name": contact.last_name or _pick_from_raw(contact.raw_data, ["last name", "lastname", "last_name"]),
"email": contact.email or _pick_from_raw(contact.raw_data, ["email", "work email", "email address", "secondary email", "tertiary email"]),
"company": contact.company or _pick_from_raw(contact.raw_data, ["company", "company name", "company name for emails", "organization name", "account name"]),
"title": contact.title or _pick_from_raw(contact.raw_data, ["title", "job title"]),
"phone": contact.phone or "",
"source": contact.source or "apollo_csv",
"owner_user_id": contact.owner_user_id,
"created_at": contact.created_at.isoformat() if contact.created_at else None,
"raw_data": contact.raw_data or {},
}
_attach_owner_display([d], db)
return d
@app.post("/api/contacts/{contact_id}/enrich")
async def enrich_contact(contact_id: int, t: TenantContext = Depends(get_tenant_context)):
"""
GPT-enrich company/contact fields for manually added contacts only.
Enrichment (gpt_service): multi-page crawl + schema.org JSON-LD on your domain;
Google Custom Search (GOOGLE_API_KEY + GOOGLE_CSE_ID); optional Gemini with
Google Search grounding (GEMINI_API_KEY); DuckDuckGo fallback;
ENRICHMENT_CONTACT_EMAIL helps Wikipedia.
"""
db = t.db
if not os.getenv("OPENAI_API_KEY"):
raise HTTPException(status_code=503, detail="OpenAI API key is not configured")
contact = (
db.query(Contact)
.filter(Contact.tenant_id == t.tenant_id, Contact.id == contact_id)
.first()
)
if not contact:
raise HTTPException(status_code=404, detail="Contact not found")
src = (contact.source or "").strip().lower()
if src not in ("manual", "cta_form", "chat_agent") and contact.file_id != MANUAL_CONTACT_FILE_ID:
raise HTTPException(
status_code=400,
detail="Enrich is only available for manually added or form-submission contacts",
)
seed = {
"first_name": contact.first_name
or _pick_from_raw(contact.raw_data, ["first name", "firstname", "first_name"]),
"last_name": contact.last_name
or _pick_from_raw(contact.raw_data, ["last name", "lastname", "last_name"]),
"email": contact.email
or _pick_from_raw(
contact.raw_data,
["email", "work email", "email address", "secondary email", "tertiary email"],
),
"company": contact.company
or _pick_from_raw(
contact.raw_data,
[
"company",
"company name",
"company name for emails",
"organization name",
"account name",
],
),
"title": contact.title
or _pick_from_raw(contact.raw_data, ["title", "job title"]),
"website_hint": _pick_from_raw(contact.raw_data, ["website", "Website"]),
}
try:
merged, enrichment_report = enrich_manual_contact_profile(seed)
except RuntimeError as e:
raise HTTPException(status_code=503, detail=str(e)) from e
except Exception as e:
raise HTTPException(status_code=502, detail=f"Enrichment failed: {e}") from e
contact.first_name = _fill_if_empty(contact.first_name, merged.get("first_name"))
contact.last_name = _fill_if_empty(contact.last_name, merged.get("last_name"))
contact.title = _fill_if_empty(contact.title, merged.get("title"))
contact.company = _fill_if_empty(contact.company, merged.get("company_name"))
rd = dict(contact.raw_data or {})
rd["First Name"] = contact.first_name
rd["Last Name"] = contact.last_name
rd["Title"] = contact.title
rd["Email"] = _safe_str(contact.email)
if contact.company:
rd["Company Name"] = contact.company
optional_raw = [
("company_name_for_emails", "Company Name for Emails"),
("industry", "Industry"),
("employees", "# Employees"),
("annual_revenue", "Annual Revenue"),
("last_raised_at", "Last Raised At"),
("website", "Website"),
("city", "City"),
("state", "State"),
("country", "Country"),
]
for gpt_key, raw_key in optional_raw:
v = _safe_str(merged.get(gpt_key))
if v:
rd[raw_key] = _fill_if_empty(rd.get(raw_key), v)
contact.raw_data = rd
linked_leads = (
db.query(CrmLead)
.filter(CrmLead.tenant_id == t.tenant_id, CrmLead.contact_id == contact.id)
.all()
)
for lead in linked_leads:
lead.first_name = _fill_if_empty(lead.first_name, merged.get("first_name"))
lead.last_name = _fill_if_empty(lead.last_name, merged.get("last_name"))
lead.title = _fill_if_empty(lead.title, merged.get("title"))
lead.company_name = _fill_if_empty(lead.company_name, merged.get("company_name"))
db.commit()
db.refresh(contact)
return {
"id": contact.id,
"file_id": contact.file_id,
"row_index": contact.row_index,
"first_name": contact.first_name
or _pick_from_raw(contact.raw_data, ["first name", "firstname", "first_name"]),
"last_name": contact.last_name
or _pick_from_raw(contact.raw_data, ["last name", "lastname", "last_name"]),
"email": contact.email
or _pick_from_raw(
contact.raw_data,
["email", "work email", "email address", "secondary email", "tertiary email"],
),
"company": contact.company
or _pick_from_raw(
contact.raw_data,
["company", "company name", "company name for emails", "organization name", "account name"],
),
"title": contact.title or _pick_from_raw(contact.raw_data, ["title", "job title"]),
"phone": contact.phone or "",
"source": contact.source or "manual",
"created_at": contact.created_at.isoformat() if contact.created_at else None,
"raw_data": contact.raw_data or {},
"enrichment_report": enrichment_report,
}
@app.get("/api/contacts/{contact_id}/sequences")
async def get_contact_sequences(contact_id: int, t: TenantContext = Depends(get_tenant_context)):
db = t.db
contact = (
db.query(Contact)
.filter(Contact.tenant_id == t.tenant_id, Contact.id == contact_id)
.first()
)
if not contact:
raise HTTPException(status_code=404, detail="Contact not found")
if not contact.email:
return {"contact_id": contact_id, "sequences": []}
sequences = (
db.query(GeneratedSequence)
.filter(
GeneratedSequence.tenant_id == t.tenant_id,
GeneratedSequence.file_id == contact.file_id,
GeneratedSequence.email == contact.email,
)
.order_by(GeneratedSequence.sequence_id, GeneratedSequence.email_number)
.all()
)
return {
"contact_id": contact_id,
"sequences": [
{
"id": s.id,
"file_id": s.file_id,
"sequence_id": s.sequence_id,
"email_number": s.email_number,
"subject": s.subject or "",
"email_content": s.email_content or "",
"product": s.product or "",
"created_at": s.created_at.isoformat() if s.created_at else None,
}
for s in sequences
],
}
# ---- Companies ----
# Companies are NOT foreign-keyed to Contact — association is a case-insensitive name match, kept
# intentionally soft so this feature is fully additive and never touches existing Contact rows or
# behavior. _sync_companies_from_contacts is one-directional and never overwrites an existing
# Company row, so manual edits (including Category, which only lives on Company/raw_data — never
# reconciled between the two) are always preserved.
def _company_to_dict(c: Company, contact_count: int = None) -> dict:
d = {
"id": c.id,
"name": c.name or "",
"category": c.category or "",
"type": c.type or "",
"company_name_for_emails": c.company_name_for_emails or "",
"industry": c.industry or "",
"employees": c.employees or "",
"annual_revenue": c.annual_revenue or "",
"last_raised_at": c.last_raised_at or "",
"website": c.website or "",
"city": c.city or "",
"state": c.state or "",
"country": c.country or "",
"phone": c.phone or "",
"source": c.source or "manual",
"owner_user_id": c.owner_user_id,
"created_at": c.created_at.isoformat() if c.created_at else None,
"updated_at": c.updated_at.isoformat() if c.updated_at else None,
}
if contact_count is not None:
d["contact_count"] = contact_count
return d
def _contacts_for_company(db: Session, tenant_id: int, company_name: str) -> List[Contact]:
name = _safe_str(company_name).strip()
if not name:
return []
return (
db.query(Contact)
.filter(
Contact.tenant_id == tenant_id,
func.lower(func.trim(Contact.company)) == name.lower(),
)
.order_by(Contact.created_at.desc())
.all()
)
def _sync_companies_from_contacts(db: Session, tenant_id: int) -> int:
"""Creates a Company row for any distinct Contact.company value that doesn't already have a
case-insensitive name match, seeded from the contact with the most non-empty company detail
fields. Also tops up any BLANK field on an already-existing Company row from the richest
matching contact — this is what keeps a Company row from getting stuck with sparse data
forever if it was first synced before richer contact data existed. Never overwrites a field
that already has a value, whether that value came from an earlier sync or a manual edit.
Safe to call on every list load."""
contacts = (
db.query(Contact)
.filter(Contact.tenant_id == tenant_id, Contact.company.isnot(None), Contact.company != "")
.all()
)
if not contacts:
return 0
existing_by_key = {
(c.name or "").strip().lower(): c
for c in db.query(Company).filter(Company.tenant_id == tenant_id).all()
}
richest_by_key: Dict[str, tuple] = {}
for c in contacts:
name = _safe_str(c.company).strip()
if not name:
continue
key = name.lower()
details = _contact_company_details_dict(c)
richness = sum(1 for k, v in details.items() if k != "Company Name" and v)
prev = richest_by_key.get(key)
if prev is None or richness > prev[0]:
richest_by_key[key] = (richness, name, details)
created = 0
touched = 0
for key, (richness, name, details) in richest_by_key.items():
existing = existing_by_key.get(key)
if existing is None:
db.add(
Company(
tenant_id=tenant_id,
name=name,
category=details.get("Category", ""),
type=details.get("Type", ""),
company_name_for_emails=details.get("Company Name for Emails", ""),
industry=details.get("Industry", ""),
employees=details.get("# Employees", ""),
annual_revenue=details.get("Annual Revenue", ""),
last_raised_at=details.get("Last Raised At", ""),
website=details.get("Website", ""),
city=details.get("City", ""),
state=details.get("State", ""),
country=details.get("Country", ""),
phone=details.get("Phone", ""),
source="synced_from_contacts",
)
)
created += 1
continue
changed = False
for col, raw_key in _COMPANY_COLUMN_TO_RAW_KEY.items():
if getattr(existing, col, ""):
continue
value = details.get(raw_key, "")
if value:
setattr(existing, col, value)
changed = True
if changed:
existing.updated_at = datetime.utcnow()
touched += 1
if created or touched:
db_commit_with_retry(db)
return created
# Company row column name for each COMPANY_DETAIL_PATCH_FIELDS py_key — identical except
# "country_region" (the patch-body key, kept for backwards compat with existing request shapes)
# maps to the Company model's "country" column.
_COMPANY_FIELD_TO_COLUMN = {
**{k: k for k in COMPANY_DETAIL_PATCH_FIELDS if k not in ("country_region", "company_phone")},
"country_region": "country",
"company_phone": "phone",
}
# Reverse: Company column name -> Apollo-style raw_data key (e.g. "country" -> "Country"), for
# patch_company, which already uses Company's own column names rather than the country_region-style
# py_keys the Contact/Lead/Deal patch bodies use.
_COMPANY_COLUMN_TO_RAW_KEY = {col: COMPANY_DETAIL_PATCH_FIELDS[k] for k, col in _COMPANY_FIELD_TO_COLUMN.items()}
def _propagate_company_fields_to_contacts_and_leads(
db: Session, tenant_id: int, company_name: str, raw_fields: Dict[str, str]
) -> None:
"""Every company-detail field (industry, employees, website, category, ...) is a
company-level attribute — whenever any of them is set anywhere (the Company page, or a
Contact/Lead/Deal's Company details card), every other record sharing that company name
(case-insensitive) is updated to match, so none of them drift between the four places
they're editable. `raw_fields` is keyed by the Apollo-style raw_data key (e.g. "Industry",
"Category"), not the patch-body py_key. Does NOT touch the Company row itself — callers that
aren't already the Company row use _propagate_company_fields_everywhere for that."""
name = _safe_str(company_name).strip()
if not name or not raw_fields:
return
clean_fields = {k: _safe_str(v) for k, v in raw_fields.items()}
contacts = (
db.query(Contact)
.filter(Contact.tenant_id == tenant_id, func.lower(func.trim(Contact.company)) == name.lower())
.all()
)
for c in contacts:
rd = dict(c.raw_data or {})
rd.update(clean_fields)
c.raw_data = rd
leads = (
db.query(CrmLead)
.filter(
CrmLead.tenant_id == tenant_id,
func.lower(func.trim(CrmLead.company_name)) == name.lower(),
)
.all()
)
for lead in leads:
rw = dict(lead.raw_webhook) if isinstance(lead.raw_webhook, dict) else {}
cd = dict(rw.get("company_details") or {}) if isinstance(rw.get("company_details"), dict) else {}
cd.update(clean_fields)
rw["company_details"] = cd
lead.raw_webhook = rw
lead.updated_at = datetime.utcnow()
db_commit_with_retry(db)
def _propagate_company_fields_everywhere(
db: Session, tenant_id: int, company_name: str, py_fields: Dict[str, str]
) -> None:
"""Like _propagate_company_fields_to_contacts_and_leads, but also upserts the Company row
itself — for callers (Contact/Lead/Deal patches) that aren't already editing the Company row
directly. `py_fields` is keyed by patch-body py_key (e.g. "industry", "category"), matching
COMPANY_DETAIL_PATCH_FIELDS."""
name = _safe_str(company_name).strip()
if not name or not py_fields:
return
company = (
db.query(Company)
.filter(Company.tenant_id == tenant_id, func.lower(Company.name) == name.lower())
.first()
)
if not company:
company = Company(tenant_id=tenant_id, name=name, source="synced_from_contacts")
db.add(company)
for py_key, value in py_fields.items():
col = _COMPANY_FIELD_TO_COLUMN.get(py_key)
if col:
setattr(company, col, _safe_str(value))
company.updated_at = datetime.utcnow()
raw_fields = {COMPANY_DETAIL_PATCH_FIELDS[k]: v for k, v in py_fields.items() if k in COMPANY_DETAIL_PATCH_FIELDS}
_propagate_company_fields_to_contacts_and_leads(db, tenant_id, name, raw_fields)
@app.get("/api/companies")
async def list_companies(
search: str = Query("", description="Search by name/industry/city/country"),
category: str = Query("", description="Filter by category: A|B|C"),
type: str = Query("", description="Filter by type: Existing|New"),
sort_by: str = Query("name", description="Sort field: name|created_at|updated_at"),
sort_dir: str = Query("asc", description="Sort direction: asc|desc"),
limit: int = Query(200, ge=1, le=1000),
offset: int = Query(0, ge=0),
t: TenantContext = Depends(get_tenant_context),
):
db = t.db
_sync_companies_from_contacts(db, t.tenant_id)
query = db.query(Company).filter(Company.tenant_id == t.tenant_id)
visible_ids = _visible_owner_user_ids(db, t)
if visible_ids is not None:
query = query.filter(or_(Company.owner_user_id.in_(visible_ids), Company.owner_user_id.is_(None)))
if search:
pattern = f"%{search}%"
query = query.filter(
(Company.name.ilike(pattern))
| (Company.industry.ilike(pattern))
| (Company.city.ilike(pattern))
| (Company.country.ilike(pattern))
)
if category and category != "all":
query = query.filter(Company.category == category)
if type and type != "all":
query = query.filter(Company.type == type)
sort_col = {
"name": Company.name,
"created_at": Company.created_at,
"updated_at": Company.updated_at,
}.get(sort_by, Company.name)
query = query.order_by(sort_col.desc() if (sort_dir or "asc").lower() == "desc" else sort_col.asc())
all_rows = query.all()
total = len(all_rows)
rows = all_rows[offset : offset + limit]
# One grouped query for every company's contact count rather than N+1 per-row queries.
counts = dict(
db.query(func.lower(func.trim(Contact.company)), func.count(Contact.id))
.filter(Contact.tenant_id == t.tenant_id, Contact.company.isnot(None), Contact.company != "")
.group_by(func.lower(func.trim(Contact.company)))
.all()
)
companies_out = [
_company_to_dict(c, contact_count=counts.get((c.name or "").strip().lower(), 0))
for c in rows
]
_attach_owner_display(companies_out, db)
return {
"total": total,
"companies": companies_out,
}
@app.post("/api/companies")
async def create_company(body: CompanyCreateRequest, t: TenantContext = Depends(get_tenant_context)):
db = t.db
name = _safe_str(body.name).strip()
if not name:
raise HTTPException(status_code=400, detail="Company name is required")
exists = (
db.query(Company)
.filter(Company.tenant_id == t.tenant_id, func.lower(Company.name) == name.lower())
.first()
)
if exists:
raise HTTPException(status_code=409, detail="A company with this name already exists")
company = Company(
tenant_id=t.tenant_id,
name=name,
category=_safe_str(body.category),
type=_safe_str(body.type),
company_name_for_emails=_safe_str(body.company_name_for_emails),
industry=_safe_str(body.industry),
employees=_safe_str(body.employees),
annual_revenue=_safe_str(body.annual_revenue),
last_raised_at=_safe_str(body.last_raised_at),
website=_safe_str(body.website),
city=_safe_str(body.city),
state=_safe_str(body.state),
country=_safe_str(body.country),
phone=_safe_str(body.phone),
source="manual",
)
db.add(company)
db.commit()
db.refresh(company)
return _company_to_dict(company, contact_count=0)
@app.post("/api/companies/bulk-delete")
async def bulk_delete_companies(body: BulkCompanyIdsRequest, t: TenantContext = Depends(get_tenant_context)):
db = t.db
if not body.company_ids:
raise HTTPException(status_code=400, detail="company_ids required")
ids = list({int(x) for x in body.company_ids})
deleted = (
db.query(Company)
.filter(Company.tenant_id == t.tenant_id, Company.id.in_(ids))
.delete(synchronize_session=False)
)
db.commit()
return {"deleted": deleted}
@app.post("/api/companies/sync-from-contacts")
async def sync_companies_endpoint(t: TenantContext = Depends(get_tenant_context)):
"""Explicit re-sync trigger (also runs automatically on every GET /api/companies) — useful
right after adding several contacts and wanting the Companies list to reflect them without
navigating away and back."""
created = _sync_companies_from_contacts(t.db, t.tenant_id)
return {"created": created}
@app.get("/api/companies/{company_id}")
async def get_company(company_id: int, t: TenantContext = Depends(get_tenant_context)):
db = t.db
company = (
db.query(Company)
.filter(Company.tenant_id == t.tenant_id, Company.id == company_id)
.first()
)
if not company:
raise HTTPException(status_code=404, detail="Company not found")
_company_access_or_403(company, t)
contacts = _contacts_for_company(db, t.tenant_id, company.name)
d = _company_to_dict(company, contact_count=len(contacts))
_attach_owner_display([d], db)
d["contacts"] = [
{
"id": c.id,
"first_name": c.first_name or "",
"last_name": c.last_name or "",
"email": c.email or "",
"title": c.title or "",
"phone": c.phone or "",
}
for c in contacts
]
return d
@app.get("/api/companies/{company_id}/activities")
async def list_company_activities(company_id: int, t: TenantContext = Depends(get_tenant_context)):
"""Rolled-up activity timeline for a company — every Call/Visit logged against any contact
sharing this company's name (same case-insensitive match used for associated contacts)."""
db = t.db
company = (
db.query(Company)
.filter(Company.tenant_id == t.tenant_id, Company.id == company_id)
.first()
)
if not company:
raise HTTPException(status_code=404, detail="Company not found")
contacts = _contacts_for_company(db, t.tenant_id, company.name)
contacts_by_id = {c.id: c for c in contacts}
if not contacts_by_id:
return {"activities": []}
q = db.query(ContactActivity).filter(
ContactActivity.tenant_id == t.tenant_id,
ContactActivity.contact_id.in_(contacts_by_id.keys()),
)
visible_ids = _visible_owner_user_ids(db, t)
if visible_ids is not None:
q = q.filter(ContactActivity.user_id.in_(visible_ids))
rows = q.order_by(ContactActivity.created_at.desc()).all()
user_ids = {r.user_id for r in rows if r.user_id}
users_by_id = {u.id: u for u in db.query(User).filter(User.id.in_(user_ids)).all()} if user_ids else {}
return {
"activities": [
_activity_to_dict(r, users_by_id.get(r.user_id), contacts_by_id.get(r.contact_id))
for r in rows
]
}
@app.patch("/api/companies/{company_id}")
async def patch_company(
company_id: int, body: CompanyPatchRequest, t: TenantContext = Depends(get_tenant_context)
):
db = t.db
company = (
db.query(Company)
.filter(Company.tenant_id == t.tenant_id, Company.id == company_id)
.first()
)
if not company:
raise HTTPException(status_code=404, detail="Company not found")
_company_access_or_403(company, t)
data = body.model_dump(exclude_unset=True)
if not data:
raise HTTPException(status_code=400, detail="No fields to update")
if "owner_user_id" in data:
val = data.pop("owner_user_id")
mids = _tenant_member_ids(db, t.tenant_id)
if t.role == "admin":
if val is None:
company.owner_user_id = None
else:
vid = int(val)
if vid not in mids:
raise HTTPException(status_code=400, detail="Owner must be a member of this workspace")
company.owner_user_id = vid
else:
if val is None:
raise HTTPException(status_code=403, detail="Only admins can set a company to unassigned")
if int(val) != int(t.user_id):
raise HTTPException(status_code=403, detail="Only admins can assign companies to other users")
if company.owner_user_id is not None and int(company.owner_user_id) != int(t.user_id):
raise HTTPException(status_code=403, detail="You can only take ownership of unassigned companies")
company.owner_user_id = t.user_id
if "name" in data:
name = _safe_str(data["name"]).strip()
if not name:
raise HTTPException(status_code=400, detail="Company name cannot be empty")
taken = (
db.query(Company)
.filter(
Company.tenant_id == t.tenant_id,
Company.id != company_id,
func.lower(Company.name) == name.lower(),
)
.first()
)
if taken:
raise HTTPException(status_code=409, detail="Another company already uses this name")
company.name = name
company_columns = (
"category",
"type",
"company_name_for_emails",
"industry",
"employees",
"annual_revenue",
"last_raised_at",
"website",
"city",
"state",
"country",
"phone",
)
for field in company_columns:
if field in data:
setattr(company, field, _safe_str(data[field]))
company.updated_at = datetime.utcnow()
raw_fields = {_COMPANY_COLUMN_TO_RAW_KEY[f]: data[f] for f in company_columns if f in data}
if raw_fields:
_propagate_company_fields_to_contacts_and_leads(db, t.tenant_id, company.name, raw_fields)
db.commit()
db.refresh(company)
contacts = _contacts_for_company(db, t.tenant_id, company.name)
d = _company_to_dict(company, contact_count=len(contacts))
_attach_owner_display([d], db)
return d
@app.post("/api/companies/{company_id}/contacts")
async def create_company_contact(
company_id: int, body: CompanyContactCreateRequest, t: TenantContext = Depends(get_tenant_context)
):
"""Add a contact from the Companies page. Reuses the same POST /api/contacts creation rules
(unique email per tenant) — company is forced from this Company row, never from the client."""
db = t.db
company = (
db.query(Company)
.filter(Company.tenant_id == t.tenant_id, Company.id == company_id)
.first()
)
if not company:
raise HTTPException(status_code=404, detail="Company not found")
email = _safe_str(body.email).lower()
if not email:
raise HTTPException(status_code=400, detail="Email is required")
exists = (
db.query(Contact)
.filter(Contact.tenant_id == t.tenant_id, func.lower(Contact.email) == email)
.first()
)
if exists:
raise HTTPException(status_code=409, detail="A contact with this email already exists")
fn = _safe_str(body.first_name)
ln = _safe_str(body.last_name)
ti = _safe_str(body.title)
ph = _safe_str(body.phone)
raw = {
"First Name": fn,
"Last Name": ln,
"Email": email,
"Company Name": company.name,
"Title": ti,
}
contact = Contact(
tenant_id=t.tenant_id,
file_id=MANUAL_CONTACT_FILE_ID,
row_index=0,
first_name=fn,
last_name=ln,
email=email,
company=company.name,
title=ti,
phone=ph,
source="manual",
raw_data=raw,
)
db.add(contact)
db.commit()
db.refresh(contact)
return {
"id": contact.id,
"first_name": contact.first_name,
"last_name": contact.last_name,
"email": contact.email,
"company": contact.company,
"title": contact.title,
"phone": contact.phone or "",
"source": contact.source or "manual",
"created_at": contact.created_at.isoformat() if contact.created_at else None,
}
@app.post("/api/companies/{company_id}/convert-to-leads")
async def convert_company_contacts_to_leads(
company_id: int, t: TenantContext = Depends(get_tenant_context)
):
"""Converts every contact associated with this company into a lead — same per-contact rules
as POST /api/contacts/bulk-convert-to-leads (skips contacts with no email or where a lead
with that email already exists). Contacts are not removed."""
db = t.db
company = (
db.query(Company)
.filter(Company.tenant_id == t.tenant_id, Company.id == company_id)
.first()
)
if not company:
raise HTTPException(status_code=404, detail="Company not found")
contacts = _contacts_for_company(db, t.tenant_id, company.name)
if not contacts:
return {"converted": 0, "errors": []}
converted = 0
errors: List[dict] = []
for contact in contacts:
r = _convert_contact_to_lead_core(db, contact)
if not r["ok"]:
errors.append({"contact_id": contact.id, "error": r.get("error", "failed")})
else:
converted += 1
db.commit()
return {"converted": converted, "errors": errors}
@app.post("/api/save-prompts")
async def save_prompts(request: PromptSaveRequest, t: TenantContext = Depends(get_tenant_context)):
"""Save prompt templates for products"""
db = t.db
try:
# Delete existing prompts for this file
db.query(Prompt).filter(
Prompt.tenant_id == t.tenant_id,
Prompt.file_id == request.file_id,
).delete()
# Save email prompts
for product_name, prompt_template in request.prompts.items():
prompt = Prompt(
tenant_id=t.tenant_id,
file_id=request.file_id,
product_name=product_name,
prompt_template=prompt_template,
prompt_kind="email",
)
db.add(prompt)
# Optional LinkedIn prompts (same product names)
li = request.linkedin_prompts or {}
for product_name, prompt_template in li.items():
if not (prompt_template or "").strip():
continue
db.add(
Prompt(
tenant_id=t.tenant_id,
file_id=request.file_id,
product_name=product_name,
prompt_template=prompt_template,
prompt_kind="linkedin",
)
)
db.commit()
uf = (
db.query(UploadedFile)
.filter(UploadedFile.tenant_id == t.tenant_id, UploadedFile.file_id == request.file_id)
.first()
)
if uf is not None:
if request.sequence_plan is not None:
uf.sequence_plan_json = json.dumps(request.sequence_plan)
if request.campaign_name is not None:
uf.campaign_name = request.campaign_name
if request.status is not None:
uf.status = request.status
if request.email_timezone is not None:
uf.email_timezone = request.email_timezone
if request.email_business_hours_start is not None:
uf.email_business_hours_start = request.email_business_hours_start
if request.email_business_hours_end is not None:
uf.email_business_hours_end = request.email_business_hours_end
if request.email_business_days is not None:
uf.email_business_days = request.email_business_days
if request.email_send_interval_seconds is not None:
uf.email_send_interval_seconds = request.email_send_interval_seconds
uf.updated_at = datetime.utcnow()
db.commit()
return {"message": "Prompts saved successfully"}
except Exception as e:
raise HTTPException(status_code=500, detail=f"Error saving prompts: {str(e)}")
@app.patch("/api/campaign-files/{file_id}/meta")
async def patch_campaign_meta(
file_id: str, body: CampaignMetaPatchRequest, t: TenantContext = Depends(get_tenant_context)
):
"""Lightweight auto-save for campaign name/status — called on every wizard step transition
so the campaign stays discoverable from the backend even if the tab closes before Launch."""
db = t.db
if body.file_id != file_id:
raise HTTPException(status_code=400, detail="file_id mismatch")
uf = (
db.query(UploadedFile)
.filter(UploadedFile.tenant_id == t.tenant_id, UploadedFile.file_id == file_id)
.first()
)
if not uf:
raise HTTPException(status_code=404, detail="Campaign file not found")
if body.campaign_name is not None:
uf.campaign_name = body.campaign_name
if body.status is not None:
uf.status = body.status
uf.updated_at = datetime.utcnow()
db.commit()
return {"ok": True}
@app.post("/api/campaign-files/{file_id}/generation/cancel")
async def cancel_campaign_generation(file_id: str, t: TenantContext = Depends(get_tenant_context)):
"""Request that an in-progress content generation run stop. The SSE generator (generate_sequences)
checks this flag at each per-contact checkpoint and halts within a few seconds; nothing already
generated is deleted, so a later Generate click resumes rather than restarting from scratch."""
db = t.db
uf = (
db.query(UploadedFile)
.filter(UploadedFile.tenant_id == t.tenant_id, UploadedFile.file_id == file_id)
.first()
)
if not uf:
raise HTTPException(status_code=404, detail="Campaign file not found")
uf.generation_cancel_requested = 1
db.commit()
return {"ok": True}
@app.get("/api/campaign-files")
async def list_campaign_files(t: TenantContext = Depends(get_tenant_context)):
"""Tenant-scoped list of campaigns — the backend record of "this campaign exists" that the
frontend campaigns dashboard reads from instead of browser localStorage."""
db = t.db
rows = (
db.query(UploadedFile)
.filter(UploadedFile.tenant_id == t.tenant_id)
.order_by(UploadedFile.created_at.desc())
.all()
)
if not rows:
return {"campaigns": []}
file_ids = [r.file_id for r in rows]
counts = dict(
db.query(GeneratedSequence.file_id, func.count(GeneratedSequence.id))
.filter(GeneratedSequence.tenant_id == t.tenant_id, GeneratedSequence.file_id.in_(file_ids))
.group_by(GeneratedSequence.file_id)
.all()
)
out = []
for r in rows:
plan_pack = _parse_campaign_plan_pack(r)
expected = (r.contact_count or 0) * (plan_pack["total_actions"] if plan_pack else 1)
out.append(
{
"file_id": r.file_id,
"campaign_name": r.campaign_name or r.filename or r.file_id,
"filename": r.filename or "",
"contact_count": r.contact_count or 0,
"status": r.status or "draft",
"created_at": r.created_at.isoformat() if r.created_at else None,
"updated_at": r.updated_at.isoformat() if r.updated_at else None,
"generated_count": counts.get(r.file_id, 0),
"expected_count": expected,
"has_sequence_plan": bool((r.sequence_plan_json or "").strip()),
}
)
return {"campaigns": out}
@app.delete("/api/campaign-files/{file_id}")
async def delete_campaign_file(file_id: str, t: TenantContext = Depends(get_tenant_context)):
"""Delete a campaign and everything generated for it. Cascades manually since these rows
were never modeled with DB-level ON DELETE CASCADE."""
db = t.db
uf = (
db.query(UploadedFile)
.filter(UploadedFile.tenant_id == t.tenant_id, UploadedFile.file_id == file_id)
.first()
)
if not uf:
raise HTTPException(status_code=404, detail="Campaign file not found")
for model in (GeneratedSequence, Prompt, OutreachSendReceipt, FileOutreachExecution):
db.query(model).filter(model.tenant_id == t.tenant_id, model.file_id == file_id).delete(
synchronize_session=False
)
db.delete(uf)
db.commit()
return {"ok": True}
@app.get("/api/campaign-files/{file_id}")
async def get_campaign_file_detail(file_id: str, t: TenantContext = Depends(get_tenant_context)):
"""Full campaign detail — sequence plan + saved prompts — for reopening the wizard on a
campaign whose local browser state was lost (cleared storage, different device/tab)."""
db = t.db
uf = (
db.query(UploadedFile)
.filter(UploadedFile.tenant_id == t.tenant_id, UploadedFile.file_id == file_id)
.first()
)
if not uf:
raise HTTPException(status_code=404, detail="Campaign file not found")
sequence_plan = None
parallel_tracks = False
raw = (uf.sequence_plan_json or "").strip()
if raw:
try:
parsed = json.loads(raw)
if isinstance(parsed, dict) and isinstance(parsed.get("steps"), list):
sequence_plan = parsed["steps"]
parallel_tracks = bool(parsed.get("parallel_tracks"))
elif isinstance(parsed, list):
sequence_plan = parsed
except Exception:
sequence_plan = None
prompt_rows = (
db.query(Prompt)
.filter(Prompt.tenant_id == t.tenant_id, Prompt.file_id == file_id)
.all()
)
prompts = {
p.product_name: p.prompt_template
for p in prompt_rows
if (getattr(p, "prompt_kind", None) or "email") == "email"
}
linkedin_prompts = {
p.product_name: p.prompt_template
for p in prompt_rows
if getattr(p, "prompt_kind", None) == "linkedin"
}
products = sorted(set(prompts.keys()) | set(linkedin_prompts.keys()))
return {
"file_id": uf.file_id,
"campaign_name": uf.campaign_name or uf.filename or uf.file_id,
"filename": uf.filename or "",
"contact_count": uf.contact_count or 0,
"status": uf.status or "draft",
"sequence_plan": sequence_plan,
"parallel_tracks": parallel_tracks,
"prompts": prompts,
"linkedin_prompts": linkedin_prompts,
"products": products,
"email_timezone": uf.email_timezone,
"email_business_hours_start": uf.email_business_hours_start,
"email_business_hours_end": uf.email_business_hours_end,
"email_business_days": uf.email_business_days,
"email_send_interval_seconds": uf.email_send_interval_seconds,
}
def _normalize_campaign_plan_steps(raw_data) -> Optional[List[Dict]]:
if raw_data is None:
return None
if isinstance(raw_data, dict) and isinstance(raw_data.get("steps"), list):
return raw_data["steps"]
if isinstance(raw_data, list):
return raw_data
return None
def _parse_campaign_plan_pack(db_file) -> Optional[Dict[str, Any]]:
"""Build GPT kwargs + per-row step_order alignment from wizard sequence JSON on UploadedFile."""
raw = getattr(db_file, "sequence_plan_json", None)
if not raw or not str(raw).strip():
return None
try:
parsed = json.loads(raw)
except Exception:
return None
steps = _normalize_campaign_plan_steps(parsed)
if not steps:
return None
gmail_specs: List[Dict[str, Any]] = []
li_wizard: List[Dict[str, Any]] = []
wait_lines: List[str] = []
ord_global = 0
for s in steps:
if not isinstance(s, dict):
continue
if s.get("type") == "wait":
wait_lines.append(
f"{_format_wait_step_label(s)} between surrounding sequence touches."
)
continue
if s.get("type") != "action":
continue
ord_global += 1
ch = s.get("channel")
act = s.get("action")
title = (s.get("title") or "").strip() or ("Email" if ch == "gmail" else "LinkedIn touch")
if ch == "gmail":
gmail_specs.append({"step_order": ord_global, "title": title})
elif ch == "linkedin" and act in ("linkedin_connect", "linkedin_dm"):
li_wizard.append({"step_order": ord_global, "action": act, "title": title})
# Preserve the wizard's actual configured order — do not regroup connects before DMs.
li_for_gpt = li_wizard
total_actions = len(gmail_specs) + len(li_wizard)
if total_actions == 0:
return None
email_campaign_kw = None
if gmail_specs:
email_campaign_kw = {
"email_count": len(gmail_specs),
"email_titles": [g["title"] for g in gmail_specs],
"wait_context": "\n".join(wait_lines),
}
linkedin_campaign_kw = None
if li_for_gpt:
linkedin_campaign_kw = {"linkedin_actions": [x["action"] for x in li_for_gpt]}
return {
"gmail_specs": gmail_specs,
"li_wizard": li_wizard,
"li_for_gpt": li_for_gpt,
"total_actions": total_actions,
"email_campaign_kw": email_campaign_kw,
"linkedin_campaign_kw": linkedin_campaign_kw,
}
@app.get("/api/generation-status")
async def generation_status(file_id: str = Query(...), t: TenantContext = Depends(get_tenant_context)):
"""Return progress for sequence generation (so frontend can resume after sleep/reconnect)."""
db = t.db
db_file = (
db.query(UploadedFile)
.filter(
UploadedFile.tenant_id == t.tenant_id,
UploadedFile.file_id == file_id,
)
.first()
)
if not db_file:
raise HTTPException(status_code=404, detail="File not found")
total_contacts = db_file.contact_count or 0
plan_actions = 0
if getattr(db_file, "sequence_plan_json", None):
try:
raw = json.loads(db_file.sequence_plan_json)
if isinstance(raw, list):
plan_actions = sum(1 for s in raw if isinstance(s, dict) and s.get("type") == "action")
elif isinstance(raw, dict) and isinstance(raw.get("steps"), list):
plan_actions = sum(1 for s in raw["steps"] if isinstance(s, dict) and s.get("type") == "action")
except Exception:
plan_actions = 0
has_linkedin = (
db.query(Prompt.id)
.filter(
Prompt.tenant_id == t.tenant_id,
Prompt.file_id == file_id,
Prompt.prompt_kind == "linkedin",
)
.first()
is not None
)
email_done = (
db.query(GeneratedSequence.sequence_id)
.filter(
GeneratedSequence.tenant_id == t.tenant_id,
GeneratedSequence.file_id == file_id,
GeneratedSequence.channel == "email",
)
.distinct()
.count()
)
li_done = (
db.query(GeneratedSequence.sequence_id)
.filter(
GeneratedSequence.tenant_id == t.tenant_id,
GeneratedSequence.file_id == file_id,
GeneratedSequence.channel == "linkedin",
)
.distinct()
.count()
)
if plan_actions > 0:
row_count = (
db.query(GeneratedSequence.id)
.filter(
GeneratedSequence.tenant_id == t.tenant_id,
GeneratedSequence.file_id == file_id,
)
.count()
)
completed_units = row_count
expected_units = total_contacts * plan_actions
# GPT may return fewer rows than plan_actions × contacts; allow up to one row per contact shortfall
# so progress can reach 100% / is_complete when all contacts are effectively done.
complete_threshold = max(1, expected_units - max(0, total_contacts))
else:
completed_units = email_done + (li_done if has_linkedin else 0)
expected_units = total_contacts * (2 if has_linkedin else 1)
complete_threshold = expected_units
return {
"file_id": file_id,
"total_contacts": total_contacts,
"completed_count": completed_units,
"is_complete": total_contacts > 0 and completed_units >= complete_threshold,
"has_linkedin_prompts": has_linkedin,
"campaign_sequence_actions": plan_actions or None,
}
@app.get("/api/sequences")
async def get_sequences(file_id: str = Query(...), t: TenantContext = Depends(get_tenant_context)):
"""Return all generated sequences for a file (for catch-up after reconnect)."""
db = t.db
sequences = (
db.query(GeneratedSequence)
.filter(
GeneratedSequence.tenant_id == t.tenant_id,
GeneratedSequence.file_id == file_id,
)
.order_by(
GeneratedSequence.sequence_id,
func.coalesce(GeneratedSequence.step_order, 999999),
GeneratedSequence.email_number,
)
.all()
)
out = []
for seq in sequences:
out.append({
"id": seq.sequence_id,
"emailNumber": seq.email_number,
"firstName": seq.first_name,
"lastName": seq.last_name,
"email": seq.email,
"company": seq.company,
"title": seq.title or "",
"product": seq.product,
"subject": seq.subject,
"emailContent": seq.email_content,
"channel": getattr(seq, "channel", None) or "email",
"stepOrder": getattr(seq, "step_order", None),
})
return {"sequences": out}
@app.get("/api/generate-sequences")
async def generate_sequences(
file_id: str = Query(...),
reset: bool = Query(True),
preview: bool = Query(False),
contact_limit: Optional[int] = Query(None, ge=1),
t: TenantContext = Depends(get_tenant_context),
):
"""Generate email sequences using GPT with Server-Sent Events streaming.
Use reset=1 for a fresh run (clears existing). Use reset=0 to resume after disconnect/sleep.
``preview=1`` with ``contact_limit=1`` generates only the first contact for wizard preview."""
db = t.db
tenant_id = t.tenant_id
async def event_generator():
try:
db_file = (
db.query(UploadedFile)
.filter(
UploadedFile.tenant_id == tenant_id,
UploadedFile.file_id == file_id,
)
.first()
)
if not db_file:
yield f"data: {json.dumps({'type': 'error', 'error': 'File not found'})}\n\n"
return
def _generation_cancelled() -> bool:
return bool(
db.query(UploadedFile.generation_cancel_requested)
.filter(UploadedFile.tenant_id == tenant_id, UploadedFile.file_id == file_id)
.scalar()
)
df = pd.read_csv(db_file.file_path)
prompts = (
db.query(Prompt)
.filter(Prompt.tenant_id == tenant_id, Prompt.file_id == file_id)
.all()
)
email_prompt_dict = {
p.product_name: p.prompt_template
for p in prompts
if (getattr(p, "prompt_kind", None) or "email") == "email"
}
linkedin_prompt_dict = {
p.product_name: p.prompt_template
for p in prompts
if getattr(p, "prompt_kind", None) == "linkedin"
}
plan_pack = _parse_campaign_plan_pack(db_file)
run_email = True
run_linkedin = len(linkedin_prompt_dict) > 0
email_kw: Optional[Dict[str, Any]] = None
li_kw: Optional[Dict[str, Any]] = None
gmail_specs: List[Dict[str, Any]] = []
li_meta_for_rows: List[Dict[str, Any]] = []
if plan_pack:
run_email = bool(plan_pack.get("email_campaign_kw"))
run_linkedin = bool(plan_pack.get("linkedin_campaign_kw"))
email_kw = plan_pack.get("email_campaign_kw")
li_kw = plan_pack.get("linkedin_campaign_kw")
gmail_specs = plan_pack.get("gmail_specs") or []
li_meta_for_rows = plan_pack.get("li_for_gpt") or []
if run_email and not email_prompt_dict:
yield f"data: {json.dumps({'type': 'error', 'error': 'No prompts found'})}\n\n"
return
if run_linkedin and not linkedin_prompt_dict:
yield f"data: {json.dumps({'type': 'error', 'error': 'No LinkedIn prompts found'})}\n\n"
return
products = list(email_prompt_dict.keys()) if email_prompt_dict else []
li_products = list(linkedin_prompt_dict.keys()) if linkedin_prompt_dict else []
if reset:
db.query(GeneratedSequence).filter(
GeneratedSequence.tenant_id == tenant_id,
GeneratedSequence.file_id == file_id,
).delete()
db.commit()
user_row = db.query(User).filter(User.id == t.user_id).first()
mailbox_sig = _mailbox_sender_name_for_user(db, user_row)
linkedin_sig = ""
if user_row:
linkedin_sig = (
_sanitize_sender_signoff_name(
(getattr(user_row, "linkedin_profile_display_name", None) or "").strip()
)
or _sanitize_sender_signoff_name((user_row.name or "").strip())
)
total_contacts = len(df)
def progress_pct(seq_id: int, *, linkedin_phase: bool) -> float:
if total_contacts <= 0:
return 0.0
if run_linkedin and not linkedin_phase:
return min(100.0, max(0.0, (seq_id / total_contacts) * 50.0))
if run_linkedin and linkedin_phase:
return min(100.0, max(0.0, 50.0 + (seq_id / total_contacts) * 50.0))
return min(100.0, max(0.0, (seq_id / total_contacts) * 100.0))
def plan_progress_pct() -> float:
if not plan_pack or total_contacts <= 0:
return 0.0
cnt = (
db.query(func.count(GeneratedSequence.id))
.filter(
GeneratedSequence.tenant_id == tenant_id,
GeneratedSequence.file_id == file_id,
)
.scalar()
or 0
)
exp = total_contacts * plan_pack["total_actions"]
return min(100.0, max(0.0, (cnt / max(1, exp)) * 100.0))
def orm_to_sequence_response(seq: GeneratedSequence) -> Dict[str, Any]:
ch = getattr(seq, "channel", None) or "email"
return {
"id": seq.sequence_id,
"emailNumber": seq.email_number,
"firstName": seq.first_name,
"lastName": seq.last_name,
"email": seq.email,
"company": seq.company,
"title": seq.title or "",
"product": seq.product,
"subject": (seq.subject or "") if ch == "linkedin" else seq.subject,
"emailContent": seq.email_content,
"channel": ch,
"stepOrder": getattr(seq, "step_order", None),
}
sequence_id = 1
if run_email:
for idx, row in df.iterrows():
if contact_limit is not None and sequence_id > int(contact_limit):
break
if _generation_cancelled():
yield f"data: {json.dumps({'type': 'cancelled'})}\n\n"
db.query(UploadedFile).filter(
UploadedFile.tenant_id == tenant_id, UploadedFile.file_id == file_id
).update({UploadedFile.generation_cancel_requested: 0, UploadedFile.status: "draft"})
db.commit()
return
if plan_pack and gmail_specs:
email_done_n = (
db.query(func.count(GeneratedSequence.id))
.filter(
GeneratedSequence.tenant_id == tenant_id,
GeneratedSequence.file_id == file_id,
GeneratedSequence.sequence_id == sequence_id,
GeneratedSequence.channel == "email",
)
.scalar()
or 0
)
if email_done_n >= len(gmail_specs):
existing_em = (
db.query(GeneratedSequence)
.filter(
GeneratedSequence.tenant_id == tenant_id,
GeneratedSequence.file_id == file_id,
GeneratedSequence.sequence_id == sequence_id,
GeneratedSequence.channel == "email",
)
.order_by(
func.coalesce(GeneratedSequence.step_order, 999999),
GeneratedSequence.email_number,
)
.all()
)
for seq in existing_em:
yield f"data: {json.dumps({'type': 'sequence', 'sequence': orm_to_sequence_response(seq)})}\n\n"
prog = plan_progress_pct() if plan_pack else progress_pct(sequence_id, linkedin_phase=False)
yield f"data: {json.dumps({'type': 'progress', 'progress': prog})}\n\n"
sequence_id += 1
await asyncio.sleep(0.05)
continue
else:
existing = (
db.query(GeneratedSequence)
.filter(
GeneratedSequence.tenant_id == tenant_id,
GeneratedSequence.file_id == file_id,
GeneratedSequence.sequence_id == sequence_id,
GeneratedSequence.channel == "email",
)
.order_by(GeneratedSequence.email_number)
.all()
)
if existing:
for seq in existing:
yield f"data: {json.dumps({'type': 'sequence', 'sequence': orm_to_sequence_response(seq)})}\n\n"
prog = plan_progress_pct() if plan_pack else progress_pct(sequence_id, linkedin_phase=False)
yield f"data: {json.dumps({'type': 'progress', 'progress': prog})}\n\n"
sequence_id += 1
await asyncio.sleep(0.05)
continue
contact = row.to_dict()
# A contact with no email address (LinkedIn-only prospect) can never receive
# email content — skip cleanly rather than calling generate_email_sequence
# (which would raise) and reporting it as a "failure" the user is prompted to
# retry, since retrying can never succeed without an email on file.
if not _pick_contact_field(contact, ["email", "work email", "email address"]):
prog = plan_progress_pct() if plan_pack else progress_pct(sequence_id, linkedin_phase=False)
yield f"data: {json.dumps({'type': 'progress', 'progress': prog})}\n\n"
sequence_id += 1
await asyncio.sleep(0.05)
continue
product_name = products[(sequence_id - 1) % len(products)]
prompt_template = email_prompt_dict[product_name]
loop = asyncio.get_event_loop()
with concurrent.futures.ThreadPoolExecutor() as executor:
if plan_pack and email_kw:
gen_fn = partial(
generate_email_sequence,
campaign_sequence=email_kw,
sender_mailbox_name=mailbox_sig or None,
)
sequence_data_list = await loop.run_in_executor(
executor,
gen_fn,
contact,
prompt_template,
product_name,
)
else:
sequence_data_list = await loop.run_in_executor(
executor,
partial(generate_email_sequence, sender_mailbox_name=mailbox_sig or None),
contact,
prompt_template,
product_name,
)
email_failed = bool(sequence_data_list) and sequence_data_list[0].get("generation_failed")
if email_failed:
yield f"data: {json.dumps({'type': 'contact_error', 'channel': 'email', 'sequence_id': sequence_id, 'email': sequence_data_list[0].get('email', ''), 'name': sequence_data_list[0].get('name', ''), 'error': sequence_data_list[0].get('error', '')})}\n\n"
else:
for i, seq_data in enumerate(sequence_data_list):
step_order_v = None
if plan_pack and gmail_specs and i < len(gmail_specs):
step_order_v = gmail_specs[i]["step_order"]
db_sequence = GeneratedSequence(
tenant_id=tenant_id,
file_id=file_id,
sequence_id=sequence_id,
email_number=seq_data["email_number"],
channel="email",
first_name=seq_data["first_name"],
last_name=seq_data["last_name"],
email=seq_data["email"],
company=seq_data["company"],
title=seq_data.get("title", ""),
product=seq_data["product"],
subject=seq_data["subject"],
email_content=seq_data["email_content"],
step_order=step_order_v,
)
db.add(db_sequence)
db.commit()
for i, seq_data in enumerate(sequence_data_list):
step_order_v = None
if plan_pack and gmail_specs and i < len(gmail_specs):
step_order_v = gmail_specs[i]["step_order"]
sequence_response = {
"id": sequence_id,
"emailNumber": seq_data["email_number"],
"firstName": seq_data["first_name"],
"lastName": seq_data["last_name"],
"email": seq_data["email"],
"company": seq_data["company"],
"title": seq_data.get("title", ""),
"product": seq_data["product"],
"subject": seq_data["subject"],
"emailContent": seq_data["email_content"],
"channel": "email",
"stepOrder": step_order_v,
}
yield f"data: {json.dumps({'type': 'sequence', 'sequence': sequence_response})}\n\n"
prog = plan_progress_pct() if plan_pack else progress_pct(sequence_id, linkedin_phase=False)
yield f"data: {json.dumps({'type': 'progress', 'progress': prog})}\n\n"
sequence_id += 1
await asyncio.sleep(0.1)
if run_linkedin:
yield f"data: {json.dumps({'type': 'phase', 'phase': 'linkedin', 'message': 'Generating LinkedIn sequences'})}\n\n"
sequence_id = 1
for idx, row in df.iterrows():
if contact_limit is not None and sequence_id > int(contact_limit):
break
if _generation_cancelled():
yield f"data: {json.dumps({'type': 'cancelled'})}\n\n"
db.query(UploadedFile).filter(
UploadedFile.tenant_id == tenant_id, UploadedFile.file_id == file_id
).update({UploadedFile.generation_cancel_requested: 0, UploadedFile.status: "draft"})
db.commit()
return
if plan_pack and li_meta_for_rows:
li_done_n = (
db.query(func.count(GeneratedSequence.id))
.filter(
GeneratedSequence.tenant_id == tenant_id,
GeneratedSequence.file_id == file_id,
GeneratedSequence.sequence_id == sequence_id,
GeneratedSequence.channel == "linkedin",
)
.scalar()
or 0
)
if li_done_n >= len(li_meta_for_rows):
existing_li = (
db.query(GeneratedSequence)
.filter(
GeneratedSequence.tenant_id == tenant_id,
GeneratedSequence.file_id == file_id,
GeneratedSequence.sequence_id == sequence_id,
GeneratedSequence.channel == "linkedin",
)
.order_by(
func.coalesce(GeneratedSequence.step_order, 999999),
GeneratedSequence.email_number,
)
.all()
)
for seq in existing_li:
yield f"data: {json.dumps({'type': 'sequence', 'sequence': orm_to_sequence_response(seq)})}\n\n"
prog = plan_progress_pct() if plan_pack else progress_pct(sequence_id, linkedin_phase=True)
yield f"data: {json.dumps({'type': 'progress', 'progress': prog})}\n\n"
sequence_id += 1
await asyncio.sleep(0.05)
continue
else:
existing_li = (
db.query(GeneratedSequence)
.filter(
GeneratedSequence.tenant_id == tenant_id,
GeneratedSequence.file_id == file_id,
GeneratedSequence.sequence_id == sequence_id,
GeneratedSequence.channel == "linkedin",
)
.order_by(GeneratedSequence.email_number)
.all()
)
if existing_li:
for seq in existing_li:
yield f"data: {json.dumps({'type': 'sequence', 'sequence': orm_to_sequence_response(seq)})}\n\n"
prog = plan_progress_pct() if plan_pack else progress_pct(sequence_id, linkedin_phase=True)
yield f"data: {json.dumps({'type': 'progress', 'progress': prog})}\n\n"
sequence_id += 1
await asyncio.sleep(0.05)
continue
contact = row.to_dict()
li_product = li_products[(sequence_id - 1) % len(li_products)]
li_template = linkedin_prompt_dict[li_product]
loop = asyncio.get_event_loop()
with concurrent.futures.ThreadPoolExecutor() as executor:
if plan_pack and li_kw:
gen_li = partial(
generate_linkedin_sequence,
campaign_sequence=li_kw,
sender_linkedin_name=linkedin_sig or None,
)
li_list = await loop.run_in_executor(
executor,
gen_li,
contact,
li_template,
li_product,
)
else:
li_list = await loop.run_in_executor(
executor,
partial(generate_linkedin_sequence, sender_linkedin_name=linkedin_sig or None),
contact,
li_template,
li_product,
)
li_failed = bool(li_list) and li_list[0].get("generation_failed")
if li_failed:
yield f"data: {json.dumps({'type': 'contact_error', 'channel': 'linkedin', 'sequence_id': sequence_id, 'email': li_list[0].get('email', ''), 'name': li_list[0].get('name', ''), 'error': li_list[0].get('error', '')})}\n\n"
else:
def _step_order_for(seq_data):
if not (plan_pack and li_meta_for_rows):
return None
idx = seq_data["email_number"] - 1
if 0 <= idx < len(li_meta_for_rows):
return li_meta_for_rows[idx]["step_order"]
return None
for seq_data in li_list:
db_sequence = GeneratedSequence(
tenant_id=tenant_id,
file_id=file_id,
sequence_id=sequence_id,
email_number=seq_data["email_number"],
channel="linkedin",
first_name=seq_data["first_name"],
last_name=seq_data["last_name"],
email=seq_data["email"],
company=seq_data["company"],
title=seq_data.get("title", ""),
product=seq_data["product"],
subject=seq_data.get("subject") or "",
email_content=seq_data["email_content"],
step_order=_step_order_for(seq_data),
)
db.add(db_sequence)
db.commit()
for seq_data in li_list:
sequence_response = {
"id": sequence_id,
"emailNumber": seq_data["email_number"],
"firstName": seq_data["first_name"],
"lastName": seq_data["last_name"],
"email": seq_data["email"],
"company": seq_data["company"],
"title": seq_data.get("title", ""),
"product": seq_data["product"],
"subject": seq_data.get("subject") or "",
"emailContent": seq_data["email_content"],
"channel": "linkedin",
"stepOrder": _step_order_for(seq_data),
}
yield f"data: {json.dumps({'type': 'sequence', 'sequence': sequence_response})}\n\n"
prog = plan_progress_pct() if plan_pack else progress_pct(sequence_id, linkedin_phase=True)
yield f"data: {json.dumps({'type': 'progress', 'progress': prog})}\n\n"
sequence_id += 1
await asyncio.sleep(0.1)
if preview and contact_limit is not None:
yield f"data: {json.dumps({'type': 'preview_complete'})}\n\n"
else:
yield f"data: {json.dumps({'type': 'complete'})}\n\n"
except Exception as e:
yield f"data: {json.dumps({'type': 'error', 'error': str(e)})}\n\n"
return StreamingResponse(event_generator(), media_type="text/event-stream")
@app.get("/api/download-sequences")
async def download_sequences(file_id: str = Query(...), t: TenantContext = Depends(get_tenant_context)):
"""Download generated sequences as CSV with all subject/body fields"""
db = t.db
try:
# Get all sequences for this file, grouped by contact
sequences = (
db.query(GeneratedSequence)
.filter(
GeneratedSequence.tenant_id == t.tenant_id,
GeneratedSequence.file_id == file_id,
GeneratedSequence.channel == "email",
)
.order_by(GeneratedSequence.sequence_id, GeneratedSequence.email_number)
.all()
)
if not sequences:
raise HTTPException(
status_code=404,
detail="No email sequences found (CSV export is email-only; view LinkedIn in the app).",
)
# Group sequences by contact
contacts = {}
max_email_number = 0 # Track the maximum email number across all contacts
for seq in sequences:
contact_key = f"{seq.sequence_id}"
if contact_key not in contacts:
contacts[contact_key] = {
'first_name': seq.first_name,
'last_name': seq.last_name,
'email': seq.email,
'company': seq.company,
'title': seq.title or '',
'product': seq.product,
'subjects': {},
'bodies': {}
}
contacts[contact_key]['subjects'][seq.email_number] = seq.subject
contacts[contact_key]['bodies'][seq.email_number] = seq.email_content
# Track the maximum email number
if seq.email_number > max_email_number:
max_email_number = seq.email_number
# Create CSV in memory
# Use QUOTE_MINIMAL to properly quote fields with newlines for Smartlead
output = io.StringIO()
writer = csv.writer(output, quoting=csv.QUOTE_MINIMAL)
# Write header – fixed format: subject1, body1, subject2, body2, ... (separate columns, not merged)
header = ['First Name', 'Last Name', 'Email', 'Company', 'Title', 'Product']
for i in range(1, max_email_number + 1):
header.extend([f'subject{i}', f'body{i}'])
writer.writerow(header)
# Write rows (one row per contact with subjects/bodies up to max_email_number)
# Preserve newlines in email bodies for proper formatting in Smartlead
for contact in contacts.values():
row = [
contact['first_name'],
contact['last_name'],
contact['email'],
contact['company'],
contact['title'],
contact['product']
]
for i in range(1, max_email_number + 1):
subject = contact['subjects'].get(i, '') or ''
body = contact['bodies'].get(i, '') or ''
# Convert newlines to HTML <br> tags for Smartlead to properly render line breaks
# Smartlead's rich text editor expects HTML formatting
if body:
# Replace newlines with <br> tags for HTML rendering in Smartlead
body_html = body.replace('\n', '<br>').replace('\r', '')
else:
body_html = ''
row.append(subject)
row.append(body_html)
writer.writerow(row)
output.seek(0)
# Return as downloadable file (fixed format: one row per contact, separate subject/body columns)
return StreamingResponse(
iter([output.getvalue()]),
media_type="text/csv",
headers={
"Content-Disposition": "attachment; filename=email_sequences_fixed.csv"
}
)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Error downloading sequences: {str(e)}")
@app.get("/api/smartlead-campaigns")
async def get_smartlead_campaigns(t: TenantContext = Depends(get_tenant_context)):
"""Get list of campaigns from Smartlead"""
try:
client = SmartleadClient()
campaigns = client.get_campaigns()
# Format campaigns for frontend
formatted_campaigns = []
for campaign in campaigns:
campaign_id = campaign.get('campaign_id') or campaign.get('id') or campaign.get('campaignId')
campaign_name = campaign.get('name') or campaign.get('campaign_name') or campaign.get('title') or 'Unnamed Campaign'
if campaign_id:
formatted_campaigns.append({
"id": str(campaign_id),
"name": campaign_name
})
return {"campaigns": formatted_campaigns}
except Exception as e:
raise HTTPException(status_code=500, detail=f"Failed to fetch campaigns: {str(e)}")
@app.post("/api/push-to-smartlead")
async def push_to_smartlead(request: SmartleadPushRequest, t: TenantContext = Depends(get_tenant_context)):
"""Push generated sequences to Smartlead campaign (add leads to existing campaign)"""
import uuid
db = t.db
try:
# Get all sequences for this file
sequences = (
db.query(GeneratedSequence)
.filter(
GeneratedSequence.tenant_id == t.tenant_id,
GeneratedSequence.file_id == request.file_id,
)
.order_by(GeneratedSequence.sequence_id, GeneratedSequence.email_number)
.all()
)
if not sequences:
raise HTTPException(status_code=404, detail="No sequences found")
# Group sequences by contact
contacts = {}
for seq in sequences:
contact_key = f"{seq.sequence_id}"
if contact_key not in contacts:
contacts[contact_key] = {
'first_name': seq.first_name,
'last_name': seq.last_name,
'email': seq.email,
'company': seq.company,
'title': seq.title or '',
'subjects': {},
'bodies': {}
}
contacts[contact_key]['subjects'][seq.email_number] = seq.subject
contacts[contact_key]['bodies'][seq.email_number] = seq.email_content
# Initialize Smartlead client
try:
client = SmartleadClient()
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
# Get campaign name for logging
campaigns = client.get_campaigns()
campaign_name = 'Unknown Campaign'
for campaign in campaigns:
campaign_id = campaign.get('campaign_id') or campaign.get('id') or campaign.get('campaignId')
if str(campaign_id) == str(request.campaign_id):
campaign_name = campaign.get('name') or campaign.get('campaign_name') or campaign.get('title') or 'Unknown Campaign'
break
# Create run record
run_id = str(uuid.uuid4())
run = SmartleadRun(
tenant_id=t.tenant_id,
run_id=run_id,
file_id=request.file_id,
mode='existing', # Always 'existing' now
campaign_id=request.campaign_id,
campaign_name=campaign_name,
steps_count=0, # Not needed for existing campaigns
dry_run=1 if request.dry_run else 0,
total_leads=len(contacts),
status='pending'
)
db.add(run)
db.commit()
if request.dry_run:
# Dry run - just return what would be sent
return {
"run_id": run_id,
"campaign_id": request.campaign_id,
"campaign_name": campaign_name,
"total": len(contacts),
"added": 0,
"skipped": 0,
"failed": 0,
"errors": [],
"status": "dry_run_completed",
"message": "Dry run completed. No leads were sent to Smartlead."
}
campaign_id = request.campaign_id
# Prepare leads for Smartlead
leads = []
errors = []
added_count = 0
skipped_count = 0
failed_count = 0
for contact in contacts.values():
try:
# Validate required fields
if not contact.get('email'):
errors.append({
"email": contact.get('email', 'unknown'),
"error": "Missing email address"
})
failed_count += 1
continue
# Helper function to safely convert to string and strip
def safe_str(value):
if value is None:
return ""
if isinstance(value, float):
# Handle NaN and other float values
if math.isnan(value):
return ""
return str(value).strip()
return str(value).strip() if value else ""
# Ensure first_name and last_name are not None/empty/float
first_name = safe_str(contact.get('first_name', '')) or 'Contact'
last_name = safe_str(contact.get('last_name', ''))
company = safe_str(contact.get('company', ''))
title = safe_str(contact.get('title', ''))
email = safe_str(contact.get('email', ''))
if not email:
errors.append({
"email": "unknown",
"error": "Missing email address"
})
failed_count += 1
continue
# Build lead object - Smartlead API doesn't allow "company", "title", or custom variables when adding leads
# Custom variables should be set manually in Smartlead via CSV import
lead = {
"email": email,
"first_name": first_name,
"last_name": last_name
}
leads.append(lead)
except Exception as e:
errors.append({
"email": contact.get('email', 'unknown'),
"error": str(e)
})
failed_count += 1
# Add leads to campaign (chunk in batches of 50)
batch_size = 50
for i in range(0, len(leads), batch_size):
batch = leads[i:i + batch_size]
try:
response = client.add_leads_to_campaign(campaign_id, batch)
added_count += len(batch)
except Exception as e:
# Mark batch as failed
for lead in batch:
errors.append({
"email": lead.get('email', 'unknown'),
"error": f"Failed to add lead: {str(e)}"
})
failed_count += 1
# Update run record
run.status = 'completed'
run.added_leads = added_count
run.skipped_leads = skipped_count
run.failed_leads = failed_count
run.error_details = json.dumps(errors) if errors else None
run.completed_at = datetime.utcnow()
db.commit()
# No warning needed - custom variables are now passed as direct fields
return {
"run_id": run_id,
"campaign_id": campaign_id,
"campaign_name": campaign_name,
"total": len(contacts),
"added": added_count,
"skipped": skipped_count,
"failed": failed_count,
"errors": errors[:10], # Return first 10 errors
"status": "completed"
}
except HTTPException:
raise
except Exception as e:
if 'run' in locals():
run.status = 'failed'
run.error_details = str(e)
db.commit()
raise HTTPException(status_code=500, detail=f"Error pushing to Smartlead: {str(e)}")
@app.get("/api/smartlead-runs")
async def get_smartlead_runs(file_id: str = Query(None), t: TenantContext = Depends(get_tenant_context)):
"""Get Smartlead run history"""
db = t.db
try:
query = db.query(SmartleadRun).filter(SmartleadRun.tenant_id == t.tenant_id)
if file_id:
query = query.filter(SmartleadRun.file_id == file_id)
runs = query.order_by(SmartleadRun.created_at.desc()).limit(50).all()
return [
{
"run_id": run.run_id,
"file_id": run.file_id,
"campaign_id": run.campaign_id,
"campaign_name": run.campaign_name,
"mode": run.mode,
"steps_count": run.steps_count,
"dry_run": bool(run.dry_run),
"total_leads": run.total_leads,
"added_leads": run.added_leads,
"skipped_leads": run.skipped_leads,
"failed_leads": run.failed_leads,
"status": run.status,
"created_at": run.created_at.isoformat() if run.created_at else None,
"completed_at": run.completed_at.isoformat() if run.completed_at else None
}
for run in runs
]
except Exception as e:
raise HTTPException(status_code=500, detail=f"Error fetching runs: {str(e)}")
@app.post("/api/webhooks/smartlead")
async def smartlead_webhook(
request: Request,
tenant_id: int = Query(..., description="Workspace ID — add ?tenant_id=<id> to the webhook URL in Smartlead"),
db: Session = Depends(get_db),
):
"""
Smartlead webhook — configure in Smartlead to POST reply events to this URL when a lead replies.
Append **?tenant_id=<your workspace id>** to the webhook URL (same id as in the app URL bar after signing in).
Optional: set SMARTLEAD_WEBHOOK_SECRET and send the same value in header X-Webhook-Token (or Bearer).
"""
if not db.query(Tenant).filter(Tenant.id == tenant_id).first():
raise HTTPException(status_code=404, detail="Unknown workspace")
secret = os.getenv("SMARTLEAD_WEBHOOK_SECRET")
if secret:
token = request.headers.get("X-Webhook-Token") or ""
if not token:
auth = request.headers.get("Authorization") or ""
if auth.lower().startswith("bearer "):
token = auth[7:].strip()
if token != secret:
raise HTTPException(status_code=401, detail="Invalid webhook token")
try:
body = await request.json()
except Exception:
raise HTTPException(status_code=400, detail="Expected JSON body")
parsed = _parse_smartlead_reply_payload(body)
if not parsed:
return {"ok": True, "ignored": True, "reason": "not_a_reply_or_missing_fields"}
q = db.query(CrmLead).filter(CrmLead.tenant_id == tenant_id)
row = None
if parsed["smartlead_lead_id"] and parsed["campaign_id"]:
row = q.filter(
CrmLead.smartlead_lead_id == parsed["smartlead_lead_id"],
CrmLead.campaign_id == parsed["campaign_id"],
).first()
if row is None and parsed["email"] and parsed["campaign_id"]:
row = q.filter(
CrmLead.email == parsed["email"],
CrmLead.campaign_id == parsed["campaign_id"],
).first()
if parsed["email"]:
apollo = (
db.query(Contact)
.filter(
Contact.tenant_id == tenant_id,
func.lower(Contact.email) == parsed["email"].lower(),
)
.first()
)
if apollo:
if not parsed["company_name"] and apollo.company:
parsed["company_name"] = apollo.company
if not parsed["title"] and apollo.title:
parsed["title"] = apollo.title
if not parsed["first_name"] and apollo.first_name:
parsed["first_name"] = apollo.first_name
if not parsed["last_name"] and apollo.last_name:
parsed["last_name"] = apollo.last_name
new_ts = parsed["last_reply_at"]
if row:
old_ts = row.last_reply_at
if not old_ts or (new_ts and new_ts >= old_ts):
row.last_reply_subject = parsed["last_reply_subject"]
row.last_reply_body = parsed["last_reply_body"]
row.last_reply_at = new_ts
row.raw_webhook = parsed["raw_webhook"]
row.email = parsed["email"] or row.email
row.first_name = parsed["first_name"] or row.first_name
row.last_name = parsed["last_name"] or row.last_name
row.company_name = parsed["company_name"] or row.company_name
row.title = parsed["title"] or row.title
row.campaign_name = parsed["campaign_name"] or row.campaign_name
if parsed["smartlead_lead_id"]:
row.smartlead_lead_id = parsed["smartlead_lead_id"]
else:
row = CrmLead(
tenant_id=tenant_id,
smartlead_lead_id=parsed["smartlead_lead_id"] or "",
campaign_id=parsed["campaign_id"] or "",
campaign_name=parsed["campaign_name"] or "",
email=parsed["email"] or "",
first_name=parsed["first_name"] or "",
last_name=parsed["last_name"] or "",
company_name=parsed["company_name"] or "",
title=parsed["title"] or "",
last_reply_subject=parsed["last_reply_subject"] or "",
last_reply_body=parsed["last_reply_body"] or "",
last_reply_at=new_ts,
crm_status="new_lead",
raw_webhook=parsed["raw_webhook"],
)
db.add(row)
db.commit()
db.refresh(row)
return {"ok": True, "lead_id": row.id}
@app.post("/api/leads/seed-demo")
async def seed_demo_leads(t: TenantContext = Depends(get_tenant_context)):
"""
Insert sample leads so the Leads UI can be previewed without a real Smartlead webhook.
Deletes any previous demo rows (emails like demo.lead.*@emailout.local) then inserts fresh ones.
"""
db = t.db
tid = t.tenant_id
demo_email_filter = CrmLead.email.like("demo.lead.%@emailout.local")
removed = (
db.query(CrmLead)
.filter(CrmLead.tenant_id == tid, demo_email_filter)
.delete(synchronize_session=False)
)
campaign_id = "88001"
campaign_name = "Logistics & Supply Chain Outreach"
now = datetime.utcnow()
specs = [
{
"sl_id": "91001",
"key": "jane",
"first_name": "Jane",
"last_name": "Morgan",
"company_name": "ACME Logistics",
"title": "VP of Operations",
"crm_status": "new_lead",
"subject": "Re: Quick question on your freight volumes",
"body": (
"Hi Sarah,\n\nThanks for reaching out. We're evaluating new 3PL partners in Q2. "
"Could you send a one-pager on pricing for mid-market shippers?\n\nBest,\nJane"
),
"days_ago": 0,
},
{
"sl_id": "91002",
"key": "marcus",
"first_name": "Marcus",
"last_name": "Chen",
"company_name": "Pacific Rail Partners",
"title": "Director of Procurement",
"crm_status": "contacted",
"subject": "Re: Partnership",
"body": (
"Not interested right now — we renewed with our incumbent through 2026. "
"Feel free to circle back next year."
),
"days_ago": 1,
},
{
"sl_id": "91003",
"key": "elena",
"first_name": "Elena",
"last_name": "Vasquez",
"company_name": "Harbor Freight Collective",
"title": "Head of Supply Chain",
"crm_status": "contacted",
"subject": "Re: 15 min chat?",
"body": (
"Tuesday 3pm PT works for me. Here's my calendar link: https://example.com/cal/elena — "
"please include an agenda."
),
"days_ago": 2,
},
{
"sl_id": "91004",
"key": "david",
"first_name": "David",
"last_name": "Okonkwo",
"company_name": "EZOFIS",
"title": "CEO",
"crm_status": "qualified",
"subject": "Re: outbound sequences",
"body": (
"This looks promising. Loop in our COO (coo@ezofis.com) and let's do a 30-min scoping call this week."
),
"days_ago": 3,
},
{
"sl_id": "91005",
"key": "priya",
"first_name": "Priya",
"last_name": "Nair",
"company_name": "Summit Cold Storage",
"title": "Operations Manager",
"crm_status": "unqualified",
"subject": "Re: cold storage",
"body": (
"We only work with vendors on our approved vendor list. Please don't follow up on this thread."
),
"days_ago": 4,
},
{
"sl_id": "91006",
"key": "sam",
"first_name": "Sam",
"last_name": "Rivera",
"company_name": "Inland Distribution Co.",
"title": "Logistics Coordinator",
"crm_status": "none",
"subject": "Re: intro",
"body": "Got it — thanks.",
"days_ago": 5,
},
]
for s in specs:
email = f"demo.lead.{s['key']}@emailout.local"
ts = now - timedelta(days=s["days_ago"])
raw = {
"event": "EMAIL_REPLIED",
"demo_seed": True,
"campaign_id": int(campaign_id),
"campaign_name": campaign_name,
"lead_id": int(s["sl_id"]),
"lead": {
"email": email,
"first_name": s["first_name"],
"last_name": s["last_name"],
"company_name": s["company_name"],
},
"reply": {
"subject": s["subject"],
"body": s["body"],
"received_at": ts.isoformat() + "Z",
},
}
db.add(
CrmLead(
tenant_id=tid,
smartlead_lead_id=s["sl_id"],
campaign_id=campaign_id,
campaign_name=campaign_name,
email=email,
first_name=s["first_name"],
last_name=s["last_name"],
company_name=s["company_name"],
title=s["title"],
last_reply_subject=s["subject"],
last_reply_body=s["body"],
last_reply_at=ts,
crm_status=s["crm_status"],
contact_id=None,
raw_webhook=raw,
)
)
db.commit()
return {
"ok": True,
"removed_previous_demo_rows": removed,
"inserted": len(specs),
}
@app.get("/api/leads")
async def list_leads(
search: str = Query("", description="Search email, name, company"),
status: str = Query("", description="crm_status filter"),
sort_by: str = Query("last_reply_at"),
sort_dir: str = Query("desc"),
limit: int = Query(50, ge=1, le=500),
offset: int = Query(0, ge=0),
t: TenantContext = Depends(get_tenant_context),
):
db = t.db
q = db.query(CrmLead).filter(CrmLead.tenant_id == t.tenant_id)
visible_ids = _visible_owner_user_ids(db, t)
if visible_ids is not None:
q = q.filter(or_(CrmLead.owner_user_id.in_(visible_ids), CrmLead.owner_user_id.is_(None)))
if search.strip():
term = f"%{search.strip().lower()}%"
q = q.filter(
or_(
func.lower(CrmLead.email).like(term),
func.lower(func.coalesce(CrmLead.company_name, "")).like(term),
func.lower(func.coalesce(CrmLead.first_name, "")).like(term),
func.lower(func.coalesce(CrmLead.last_name, "")).like(term),
)
)
if status.strip() and status in CRM_STATUS_ALLOWED:
q = q.filter(CrmLead.crm_status == status)
total = q.count()
col_map = {
"last_reply_at": CrmLead.last_reply_at,
"created_at": CrmLead.created_at,
"email": CrmLead.email,
"company_name": CrmLead.company_name,
}
col = col_map.get(sort_by, CrmLead.last_reply_at)
if sort_dir == "asc":
q = q.order_by(col.asc())
else:
q = q.order_by(col.desc())
rows = q.offset(offset).limit(limit).all()
leads_out = [_crm_lead_to_dict(r) for r in rows]
_attach_owner_display(leads_out, db)
return {
"total": total,
"leads": leads_out,
}
@app.get("/api/leads/{lead_id}")
async def get_lead(lead_id: int, t: TenantContext = Depends(get_tenant_context)):
db = t.db
row = (
db.query(CrmLead)
.filter(CrmLead.tenant_id == t.tenant_id, CrmLead.id == lead_id)
.first()
)
if not row:
raise HTTPException(status_code=404, detail="Lead not found")
_lead_access_or_403(row, t)
if lead_is_cta_form(row) and not row.contact_id:
link_lead_to_contact(db, row)
db.commit()
db.refresh(row)
d = _crm_lead_to_dict(row)
_attach_owner_display([d], db)
if row.contact_id:
c = (
db.query(Contact)
.filter(
Contact.tenant_id == t.tenant_id,
Contact.id == row.contact_id,
)
.first()
)
if c:
d["contact"] = {
"id": c.id,
"email": c.email,
"company": c.company,
"title": c.title,
}
d["company_details"] = _contact_company_details_dict(c)
if not _safe_str(d.get("title")) and _safe_str(c.title):
d["title"] = c.title
if not _safe_str(d.get("first_name")) and _safe_str(c.first_name):
d["first_name"] = c.first_name
if not _safe_str(d.get("last_name")) and _safe_str(c.last_name):
d["last_name"] = c.last_name
if not _safe_str(d.get("company_name")) and _safe_str(c.company):
d["company_name"] = c.company
d["company_details"] = _contact_company_details_dict(c)
return d
@app.patch("/api/leads/{lead_id}")
async def patch_lead(lead_id: int, body: CrmLeadPatchRequest, t: TenantContext = Depends(get_tenant_context)):
db = t.db
row = (
db.query(CrmLead)
.filter(CrmLead.tenant_id == t.tenant_id, CrmLead.id == lead_id)
.first()
)
if not row:
raise HTTPException(status_code=404, detail="Lead not found")
_lead_access_or_403(row, t)
data = body.model_dump(exclude_unset=True)
if not data:
raise HTTPException(status_code=400, detail="No fields to update")
if "owner_user_id" in data:
val = data.pop("owner_user_id")
mids = _tenant_member_ids(db, t.tenant_id)
if t.role == "admin":
if val is None:
row.owner_user_id = None
else:
vid = int(val)
if vid not in mids:
raise HTTPException(
status_code=400,
detail="Owner must be a member of this workspace",
)
row.owner_user_id = vid
else:
if val is None:
raise HTTPException(
status_code=403, detail="Only admins can set a lead to unassigned"
)
if int(val) != int(t.user_id):
raise HTTPException(
status_code=403,
detail="Only admins can assign leads to other users",
)
if row.owner_user_id is not None and int(row.owner_user_id) != int(t.user_id):
raise HTTPException(
status_code=403,
detail="You can only take ownership of unassigned leads",
)
row.owner_user_id = t.user_id
if "crm_status" in data:
if data["crm_status"] not in CRM_STATUS_ALLOWED:
raise HTTPException(
status_code=400,
detail=f"crm_status must be one of: {sorted(CRM_STATUS_ALLOWED)}",
)
row.crm_status = data["crm_status"]
if "first_name" in data:
row.first_name = _safe_str(data["first_name"])
if "last_name" in data:
row.last_name = _safe_str(data["last_name"])
if "company_name" in data:
row.company_name = _safe_str(data["company_name"])
if "title" in data:
row.title = _safe_str(data["title"])
if "phone" in data:
row.phone = _safe_str(data["phone"])
if "last_reply_subject" in data:
row.last_reply_subject = _safe_str(data["last_reply_subject"])
if "last_reply_body" in data:
row.last_reply_body = _safe_str(data["last_reply_body"])
if "email" in data:
email = _safe_str(data["email"]).lower()
if email:
taken = (
db.query(CrmLead)
.filter(
CrmLead.tenant_id == row.tenant_id,
CrmLead.id != lead_id,
func.lower(CrmLead.email) == email,
)
.first()
)
if taken:
raise HTTPException(
status_code=409,
detail="Another lead already uses this email",
)
row.email = email
rw = dict(row.raw_webhook) if isinstance(row.raw_webhook, dict) else {}
cd = dict(rw.get("company_details") or {}) if isinstance(rw.get("company_details"), dict) else {}
cd["Company Name"] = row.company_name or ""
for py_key, raw_key in COMPANY_DETAIL_PATCH_FIELDS.items():
if py_key in data:
cd[raw_key] = _safe_str(data[py_key])
rw["company_details"] = cd
row.raw_webhook = rw
row.updated_at = datetime.utcnow()
company_fields = {k: data[k] for k in COMPANY_DETAIL_PATCH_FIELDS if k in data}
if company_fields:
_propagate_company_fields_everywhere(db, t.tenant_id, row.company_name, company_fields)
db.commit()
db.refresh(row)
d = _crm_lead_to_dict(row)
_attach_owner_display([d], db)
return d
@app.post("/api/leads/{lead_id}/move-to-contacts")
async def move_lead_to_contacts(lead_id: int, t: TenantContext = Depends(get_tenant_context)):
db = t.db
lead = (
db.query(CrmLead)
.filter(CrmLead.tenant_id == t.tenant_id, CrmLead.id == lead_id)
.first()
)
if not lead:
raise HTTPException(status_code=404, detail="Lead not found")
r = _move_lead_to_contacts_core(db, lead)
if not r["ok"]:
raise HTTPException(status_code=400, detail=r.get("error", "Move failed"))
db.commit()
return {"contact_id": r["contact_id"], "message": r["message"]}
@app.post("/api/leads/bulk-move-to-contacts")
async def bulk_move_leads_to_contacts(body: BulkLeadIdsRequest, t: TenantContext = Depends(get_tenant_context)):
db = t.db
if not body.lead_ids:
raise HTTPException(status_code=400, detail="lead_ids required")
moved = 0
errors: List[dict] = []
for lid in body.lead_ids:
lead = (
db.query(CrmLead)
.filter(CrmLead.tenant_id == t.tenant_id, CrmLead.id == lid)
.first()
)
if not lead:
errors.append({"lead_id": lid, "error": "not found"})
continue
r = _move_lead_to_contacts_core(db, lead)
if not r["ok"]:
errors.append({"lead_id": lid, "error": r.get("error", "failed")})
else:
moved += 1
db.commit()
return {"moved": moved, "errors": errors}
@app.post("/api/leads/bulk-delete")
async def bulk_delete_leads(body: BulkLeadIdsRequest, t: TenantContext = Depends(get_tenant_context)):
db = t.db
if not body.lead_ids:
raise HTTPException(status_code=400, detail="lead_ids required")
deleted = (
db.query(CrmLead)
.filter(CrmLead.tenant_id == t.tenant_id, CrmLead.id.in_(body.lead_ids))
.delete(synchronize_session=False)
)
db.commit()
return {"deleted": deleted}
@app.post("/api/deals/from-leads")
async def deals_from_leads(body: BulkLeadIdsRequest, t: TenantContext = Depends(get_tenant_context)):
"""Create one deal per selected lead and remove those leads from the Leads table."""
db = t.db
if not body.lead_ids:
raise HTTPException(status_code=400, detail="lead_ids required")
created: List[dict] = []
errors: List[dict] = []
for lid in body.lead_ids:
lead = (
db.query(CrmLead)
.filter(CrmLead.tenant_id == t.tenant_id, CrmLead.id == lid)
.first()
)
if not lead:
errors.append({"lead_id": lid, "error": "not found"})
continue
person = " ".join(filter(None, [lead.first_name or "", lead.last_name or ""])).strip()
inviter = db.query(User).filter(User.id == t.user_id).first()
deal = CrmDeal(
tenant_id=lead.tenant_id,
name=_deal_name_from_lead(lead),
stage="new",
owner_user_id=t.user_id,
owner_initials=_user_initials_from_user(inviter),
deal_value=None,
revenue_type="arr",
contact_display=person or (lead.email or ""),
account_name=lead.company_name or "",
expected_close_date=datetime.utcnow() + timedelta(days=60),
close_probability=25,
country="",
last_interaction_at=lead.last_reply_at or datetime.utcnow(),
source_lead_id=lead.id,
source_campaign_name=lead.campaign_name or lead.campaign_id or "",
contact_id=lead.contact_id,
)
db.add(deal)
db.flush()
dd = _deal_to_dict(deal)
if inviter:
dd["owner_display_name"] = (inviter.name or inviter.email or "").strip() or None
else:
dd["owner_display_name"] = None
created.append(dd)
db.delete(lead)
db.commit()
return {"created": created, "errors": errors}
@app.get("/api/deals")
async def list_deals(
search: str = Query(""),
sort_by: str = Query("created_at"),
sort_dir: str = Query("desc"),
limit: int = Query(100, ge=1, le=500),
offset: int = Query(0, ge=0),
t: TenantContext = Depends(get_tenant_context),
):
db = t.db
q = db.query(CrmDeal).filter(CrmDeal.tenant_id == t.tenant_id)
visible_ids = _visible_owner_user_ids(db, t)
if visible_ids is not None:
q = q.filter(
or_(CrmDeal.owner_user_id.in_(visible_ids), CrmDeal.owner_user_id.is_(None))
)
if search.strip():
term = f"%{search.strip().lower()}%"
q = q.filter(
or_(
func.lower(CrmDeal.name).like(term),
func.lower(func.coalesce(CrmDeal.account_name, "")).like(term),
func.lower(func.coalesce(CrmDeal.contact_display, "")).like(term),
)
)
total = q.count()
col_map = {
"created_at": CrmDeal.created_at,
"name": CrmDeal.name,
"deal_value": CrmDeal.deal_value,
"stage": CrmDeal.stage,
"last_interaction_at": CrmDeal.last_interaction_at,
}
col = col_map.get(sort_by, CrmDeal.created_at)
if sort_dir == "asc":
q = q.order_by(col.asc())
else:
q = q.order_by(col.desc())
rows = q.offset(offset).limit(limit).all()
oid_set = {r.owner_user_id for r in rows if r.owner_user_id}
user_map: Dict[int, User] = {}
if oid_set:
for u in db.query(User).filter(User.id.in_(oid_set)).all():
user_map[u.id] = u
deals_out = []
for r in rows:
dd = _deal_to_dict(r)
if r.owner_user_id and r.owner_user_id in user_map:
u = user_map[r.owner_user_id]
dd["owner_display_name"] = (u.name or u.email or "").strip() or None
if not (dd.get("owner_initials") or "").strip():
dd["owner_initials"] = _user_initials_from_user(u)
else:
dd["owner_display_name"] = None
deals_out.append(dd)
return {"total": total, "deals": deals_out}
@app.get("/api/deals/quarterly-recurring-metrics")
async def quarterly_recurring_metrics(
owner_user_id: Optional[int] = Query(None),
max_quarters: int = Query(12, ge=1, le=24),
period: str = Query("all"),
t: TenantContext = Depends(get_tenant_context),
):
"""
Won deals → ARR (USD) roll-forward by calendar quarter from PO lines (interval per line + one-time lines).
Optional period=all|month|quarter|year filters by won_at.
When period is not all, includes comparison roll-forward for the prior calendar period (last month / quarter / year).
"""
db = t.db
q = db.query(CrmDeal).filter(
CrmDeal.tenant_id == t.tenant_id,
CrmDeal.stage == "won",
)
visible_ids = _visible_owner_user_ids(db, t)
if visible_ids is not None:
q = q.filter(or_(CrmDeal.owner_user_id.in_(visible_ids), CrmDeal.owner_user_id.is_(None)))
elif owner_user_id is not None:
q = q.filter(CrmDeal.owner_user_id == int(owner_user_id))
rows = q.all()
bounds = _dashboard_period_bounds_utc(period)
deals_payload = []
for r in rows:
wa = getattr(r, "won_at", None)
if not _won_at_within_bounds(wa, bounds):
continue
deals_payload.append(
{
"id": r.id,
"name": r.name or "",
"revenue_type": (getattr(r, "revenue_type", None) or "arr"),
"won_line_items": _won_line_items_list(r),
"won_at": wa,
}
)
base = build_quarterly_board(deals_payload, max_quarters=max_quarters)
pnorm = (period or "all").strip().lower()
if pnorm in ("", "all"):
return {**base, "comparison": None}
bounds_prev = _previous_period_bounds_utc(period)
deals_prev: list[dict] = []
if bounds_prev:
for r in rows:
wa = getattr(r, "won_at", None)
if not _won_at_within_bounds(wa, bounds_prev):
continue
deals_prev.append(
{
"id": r.id,
"name": r.name or "",
"revenue_type": (getattr(r, "revenue_type", None) or "arr"),
"won_line_items": _won_line_items_list(r),
"won_at": wa,
}
)
comp = build_quarterly_board(deals_prev, max_quarters=max_quarters)
return {
**base,
"comparison": {
"period_label": _comparison_period_label(period),
"quarters": comp.get("quarters", []),
},
}
@app.post("/api/deals")
async def create_deal(body: CrmDealCreateRequest, t: TenantContext = Depends(get_tenant_context)):
stage = (body.stage or "new").strip().lower() or "new"
if stage not in DEAL_STAGE_ALLOWED:
raise HTTPException(
status_code=400,
detail=f"stage must be one of: {sorted(DEAL_STAGE_ALLOWED)}",
)
if stage == "lost":
raise HTTPException(
status_code=400,
detail="Cannot create a deal in Lost. Create the deal in another stage, then move it to Lost.",
)
if stage == "won":
raise HTTPException(
status_code=400,
detail="Cannot create a deal as Won. Create the deal in another stage, then move it to Won with billing details.",
)
db = t.db
raw_name = _safe_str(body.name) if body.name is not None else ""
name = raw_name or "Untitled deal"
creator = db.query(User).filter(User.id == t.user_id).first()
row = CrmDeal(
tenant_id=t.tenant_id,
name=name,
stage=stage,
owner_user_id=t.user_id,
owner_initials=_user_initials_from_user(creator),
deal_value=None,
revenue_type="arr",
contact_display="",
account_name="",
expected_close_date=None,
close_probability=10,
country="",
last_interaction_at=None,
source_lead_id=None,
source_campaign_name="",
contact_id=None,
)
db.add(row)
db.commit()
db.refresh(row)
return _enrich_deal_response(db, row)
@app.get("/api/deals/{deal_id}")
async def get_deal(deal_id: int, t: TenantContext = Depends(get_tenant_context)):
db = t.db
row = (
db.query(CrmDeal)
.filter(CrmDeal.tenant_id == t.tenant_id, CrmDeal.id == deal_id)
.first()
)
if not row:
raise HTTPException(status_code=404, detail="Deal not found")
_deal_access_or_403(row, t)
return _enrich_deal_response(db, row)
@app.patch("/api/deals/{deal_id}")
async def patch_deal(deal_id: int, body: CrmDealPatchRequest, t: TenantContext = Depends(get_tenant_context)):
db = t.db
row = (
db.query(CrmDeal)
.filter(CrmDeal.tenant_id == t.tenant_id, CrmDeal.id == deal_id)
.first()
)
if not row:
raise HTTPException(status_code=404, detail="Deal not found")
_deal_access_or_403(row, t)
data = body.model_dump(exclude_unset=True)
if not data:
raise HTTPException(status_code=400, detail="No fields to update")
did_mark_won_this_request = False
if "owner_initials" in data:
del data["owner_initials"]
if "owner_user_id" in data:
val = data.pop("owner_user_id")
mids = _tenant_member_ids(db, t.tenant_id)
if t.role == "admin":
if val is None:
row.owner_user_id = None
row.owner_initials = ""
else:
vid = int(val)
if vid not in mids:
raise HTTPException(
status_code=400,
detail="Owner must be a member of this workspace",
)
ou = db.query(User).filter(User.id == vid).first()
if not ou:
raise HTTPException(status_code=404, detail="User not found")
row.owner_user_id = vid
row.owner_initials = _user_initials_from_user(ou)
else:
if val is None:
raise HTTPException(
status_code=403, detail="Only admins can set a deal to unassigned"
)
if int(val) != int(t.user_id):
raise HTTPException(
status_code=403,
detail="Only admins can assign deals to other users",
)
if row.owner_user_id is not None and int(row.owner_user_id) != int(t.user_id):
raise HTTPException(
status_code=403,
detail="You can only take ownership of unassigned deals",
)
claimer = db.query(User).filter(User.id == t.user_id).first()
row.owner_user_id = t.user_id
row.owner_initials = _user_initials_from_user(claimer)
if "contact_id" in data:
cid = data["contact_id"]
if cid is None:
row.contact_id = None
else:
contact = (
db.query(Contact)
.filter(
Contact.tenant_id == t.tenant_id,
Contact.id == int(cid),
)
.first()
)
if not contact:
raise HTTPException(status_code=404, detail="Contact not found")
row.contact_id = contact.id
row.contact_display = _format_contact_display(contact)
if "account_name" not in data and _safe_str(contact.company):
row.account_name = _safe_str(contact.company)
if "name" in data:
row.name = _safe_str(data["name"])
if "contact_display" in data:
row.contact_display = _safe_str(data["contact_display"])
if "account_name" in data:
row.account_name = _safe_str(data["account_name"])
if "stage" in data:
new_stage = data["stage"]
if new_stage not in DEAL_STAGE_ALLOWED:
raise HTTPException(
status_code=400,
detail=f"stage must be one of: {sorted(DEAL_STAGE_ALLOWED)}",
)
prev_stage = (row.stage or "new").strip().lower()
if new_stage == "lost":
_clear_won_billing_fields(row)
lr = _safe_str(data.get("lost_reason", "")).strip().lower()
if not lr or lr not in DEAL_LOST_REASON_ALLOWED:
raise HTTPException(
status_code=400,
detail=(
"When marking a deal as Lost, lost_reason is required. "
f"Use one of: {', '.join(sorted(DEAL_LOST_REASON_ALLOWED))}."
),
)
if lr == "price":
pct = data.get("lost_price_discount_pct")
if pct is None:
raise HTTPException(
status_code=400,
detail="lost_price_discount_pct (0–100) is required when lost_reason is price.",
)
try:
pf = float(pct)
except (TypeError, ValueError):
raise HTTPException(
status_code=400,
detail="lost_price_discount_pct must be a number.",
)
if pf < 0 or pf > 100:
raise HTTPException(
status_code=400,
detail="lost_price_discount_pct must be between 0 and 100.",
)
row.lost_reason = lr
row.lost_price_discount_pct = pf
row.lost_chose_product_name = ""
elif lr == "chose_another":
pn = _safe_str(data.get("lost_chose_product_name") or "")
if not pn:
raise HTTPException(
status_code=400,
detail="lost_chose_product_name is required when lost_reason is chose_another.",
)
row.lost_reason = lr
row.lost_price_discount_pct = None
row.lost_chose_product_name = pn
else:
row.lost_reason = lr
row.lost_price_discount_pct = None
row.lost_chose_product_name = ""
row.stage = "lost"
elif new_stage == "won":
wb_raw = data.pop("won_billing", None)
if wb_raw is None:
raise HTTPException(
status_code=400,
detail="won_billing is required when marking a deal as Won (PO, customer, line items, notes).",
)
try:
wb = WonBillingPayload.model_validate(wb_raw)
except ValidationError as e:
raise HTTPException(status_code=400, detail=str(e)[:3000])
items_out = []
subtotal = 0.0
for li in wb.line_items:
amt = round(float(li.qty) * float(li.rate), 2)
subtotal += amt
items_out.append(
{
"product_service": _safe_str(li.product_service),
"description": _safe_str(li.description or ""),
"qty": float(li.qty),
"rate": float(li.rate),
"amount": amt,
"billing_interval": li.billing_interval,
"currency": li.currency,
}
)
row.won_po_number = _safe_str(wb.po_number).strip()
row.won_customer_legal_name = _safe_str(wb.customer_legal_name).strip()
row.won_customer_address = _safe_str(wb.customer_address).strip()
row.won_contact_person_name = _safe_str(wb.contact_person_name).strip()
row.won_channel_partner_name = _safe_str(wb.channel_partner_name or "").strip()
row.won_note_to_customer = _safe_str(wb.note_to_customer).strip()
row.won_note_to_accounts = _safe_str(wb.note_to_accounts).strip()
row.won_line_items = items_out
# Actual closed value = invoice subtotal from the won popover (USD whole dollars).
row.deal_value = int(round(subtotal))
row.lost_reason = ""
row.lost_price_discount_pct = None
row.lost_chose_product_name = ""
row.stage = "won"
if prev_stage != "won":
row.won_at = datetime.utcnow()
did_mark_won_this_request = True
else:
if prev_stage == "lost":
row.lost_reason = ""
row.lost_price_discount_pct = None
row.lost_chose_product_name = ""
if prev_stage == "won":
_clear_won_billing_fields(row)
row.stage = new_stage
data.pop("lost_reason", None)
data.pop("lost_price_discount_pct", None)
data.pop("lost_chose_product_name", None)
data.pop("won_billing", None)
else:
data.pop("lost_reason", None)
data.pop("lost_price_discount_pct", None)
data.pop("lost_chose_product_name", None)
data.pop("won_billing", None)
if "deal_value" in data:
row.deal_value = data["deal_value"]
if "revenue_type" in data:
rt = _safe_str(data["revenue_type"]).lower()
if rt not in DEAL_REVENUE_TYPE_ALLOWED:
raise HTTPException(
status_code=400,
detail=f"revenue_type must be one of: {sorted(DEAL_REVENUE_TYPE_ALLOWED)}",
)
row.revenue_type = rt
if "close_probability" in data:
row.close_probability = max(0, min(100, int(data["close_probability"])))
if "country" in data:
row.country = _safe_str(data["country"])
if "expected_close_date" in data:
ts = _to_datetime(data["expected_close_date"])
row.expected_close_date = ts
if "last_interaction_at" in data:
ts = _to_datetime(data["last_interaction_at"])
row.last_interaction_at = ts
c = None
if row.contact_id:
c = (
db.query(Contact)
.filter(
Contact.tenant_id == t.tenant_id,
Contact.id == row.contact_id,
)
.first()
)
if "account_name" in data and c:
c.company = row.account_name
if c:
_sync_contact_raw_from_columns(c)
if any(k in data for k in COMPANY_DETAIL_PATCH_FIELDS):
_merge_contact_raw_company_details(c, data)
if "country_region" in data:
row.country = _safe_str(data["country_region"])
if "country" in data and "country_region" not in data:
rd = dict(c.raw_data or {})
rd["Country"] = _safe_str(data["country"])
c.raw_data = rd
company_fields = {k: data[k] for k in COMPANY_DETAIL_PATCH_FIELDS if k in data}
if company_fields:
_propagate_company_fields_everywhere(db, t.tenant_id, c.company, company_fields)
db.add(c)
row.updated_at = datetime.utcnow()
db.commit()
db.refresh(row)
out = _enrich_deal_response(db, row)
if did_mark_won_this_request:
sender = db.query(User).filter(User.id == int(t.user_id)).first()
if sender:
lines = _won_line_items_list(row)
st = sum(float(li.get("amount") or 0) for li in lines) if lines else 0.0
ok, err = await _notify_tenant_admins_deal_won(
db,
int(t.tenant_id),
sender,
int(row.id),
row.name or "",
lines,
st,
row.won_po_number or "",
row.won_customer_legal_name or "",
row.won_customer_address or "",
row.won_contact_person_name or "",
row.won_channel_partner_name or "",
row.won_note_to_customer or "",
row.won_note_to_accounts or "",
)
out["won_email_to_admins_sent"] = ok
if err:
out["won_email_error"] = err
return out
@app.delete("/api/deals/{deal_id}")
async def delete_deal(deal_id: int, t: TenantContext = Depends(get_tenant_context)):
db = t.db
row = (
db.query(CrmDeal)
.filter(CrmDeal.tenant_id == t.tenant_id, CrmDeal.id == deal_id)
.first()
)
if not row:
raise HTTPException(status_code=404, detail="Deal not found")
_deal_access_or_403(row, t)
db.delete(row)
db.commit()
return {"ok": True, "id": deal_id}
@app.post("/api/deals/seed-demo")
async def seed_demo_deals(t: TenantContext = Depends(get_tenant_context)):
db = t.db
tid = t.tenant_id
removed = (
db.query(CrmDeal)
.filter(CrmDeal.tenant_id == tid, CrmDeal.name.like("DEMO: %"))
.delete(synchronize_session=False)
)
now = datetime.utcnow()
samples = [
{
"name": "DEMO: Ramco — Aus",
"stage": "discovery",
"owner_initials": "",
"deal_value": 10000,
"contact_display": "KENNETH WON",
"account_name": "Quanticsit",
"close_probability": 10,
"close_in_days": 120,
"country": "Australia",
},
{
"name": "DEMO: HT Pilot",
"stage": "proposal",
"owner_initials": "",
"deal_value": 1600,
"contact_display": "SARAH LEE",
"account_name": "HT Logistics",
"close_probability": 60,
"close_in_days": 45,
"country": "Malaysia",
},
{
"name": "DEMO: Northwind rollout",
"stage": "negotiation",
"owner_initials": "",
"deal_value": 45000,
"contact_display": "JAMES DIAZ",
"account_name": "Northwind Trading",
"close_probability": 40,
"close_in_days": 60,
"country": "USA",
},
{
"name": "DEMO: Maple cold chain",
"stage": "new",
"owner_initials": "",
"deal_value": None,
"contact_display": "ALEX RUIZ",
"account_name": "Maple Cold",
"close_probability": 15,
"close_in_days": 90,
"country": "Canada",
},
]
for s in samples:
db.add(
CrmDeal(
tenant_id=tid,
name=s["name"],
stage=s["stage"],
owner_user_id=None,
owner_initials=s["owner_initials"],
deal_value=s["deal_value"],
contact_display=s["contact_display"],
account_name=s["account_name"],
expected_close_date=now + timedelta(days=s["close_in_days"]),
close_probability=s["close_probability"],
country=s["country"],
last_interaction_at=now - timedelta(days=1),
source_lead_id=None,
source_campaign_name="Demo",
)
)
db.commit()
return {"ok": True, "removed_previous_demo_rows": removed, "inserted": len(samples)}
@app.get("/api/leads/{lead_id}/outreach-thread")
async def lead_outreach_thread(lead_id: int, t: TenantContext = Depends(get_tenant_context)):
"""Full sent + received thread for EMAILOUT wizard campaign leads."""
db = t.db
row = (
db.query(CrmLead)
.filter(CrmLead.tenant_id == t.tenant_id, CrmLead.id == lead_id)
.first()
)
if not row:
raise HTTPException(status_code=404, detail="Lead not found")
if not lead_is_outreach_campaign(row):
raise HTTPException(status_code=400, detail="Lead is not from an EMAILOUT campaign")
if not row.contact_id or not row.campaign_id:
raise HTTPException(status_code=400, detail="Lead missing contact or campaign file id")
contact = (
db.query(Contact)
.filter(Contact.tenant_id == t.tenant_id, Contact.id == row.contact_id)
.first()
)
if not contact:
raise HTTPException(status_code=404, detail="Linked contact not found")
messages = build_outreach_thread(db, t.tenant_id, contact, row.campaign_id)
return {
"source": "emailout_campaign",
"campaign_id": row.campaign_id,
"campaign_name": row.campaign_name or "",
"contact_id": row.contact_id,
"messages": messages,
}
@app.get("/api/leads/{lead_id}/chat-agent-thread")
async def lead_chat_agent_thread(lead_id: int, t: TenantContext = Depends(get_tenant_context)):
"""Full website-chat transcript for a chat-agent lead. Stored inline on raw_webhook.messages
(refreshed on every inbound message, see chat_agent_routes.py) rather than fetched live from
anywhere, so this is just a read — no external call, unlike the Smartlead thread route."""
db = t.db
row = (
db.query(CrmLead)
.filter(CrmLead.tenant_id == t.tenant_id, CrmLead.id == lead_id)
.first()
)
if not row:
raise HTTPException(status_code=404, detail="Lead not found")
rw = row.raw_webhook if isinstance(row.raw_webhook, dict) else {}
if rw.get("source") != "chat_agent":
raise HTTPException(status_code=400, detail="Lead is not from the chat agent")
return {"messages": rw.get("messages") or []}
@app.get("/api/leads/{lead_id}/smartlead-thread")
async def lead_smartlead_thread(lead_id: int, t: TenantContext = Depends(get_tenant_context)):
"""Fetch full thread from Smartlead API (Admin API key required)."""
db = t.db
row = (
db.query(CrmLead)
.filter(CrmLead.tenant_id == t.tenant_id, CrmLead.id == lead_id)
.first()
)
if not row:
raise HTTPException(status_code=404, detail="Lead not found")
if lead_is_outreach_campaign(row):
raise HTTPException(
status_code=400,
detail="Use /api/leads/{id}/outreach-thread for EMAILOUT campaign leads",
)
if not row.smartlead_lead_id or not row.campaign_id:
raise HTTPException(status_code=400, detail="Lead missing Smartlead campaign/lead id")
try:
lid = int(row.smartlead_lead_id)
except (TypeError, ValueError):
raise HTTPException(status_code=400, detail="Invalid smartlead_lead_id")
try:
client = SmartleadClient()
hist = client.get_message_history(str(row.campaign_id), lid)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
except Exception as e:
raise HTTPException(status_code=502, detail=str(e))
return {"history": hist}
# ---- Frontend static serving ----
FRONTEND_DIST = Path(__file__).resolve().parents[2] / "frontend" / "dist"
INDEX_FILE = FRONTEND_DIST / "index.html"
FRONTEND_ASSETS = FRONTEND_DIST / "assets"
# Prevent stale index.html in browsers/CDNs referencing old hashed bundles (→ blank page when *.js 404s).
_SPA_HTML_HEADERS = {
"Cache-Control": "no-store, no-cache, must-revalidate, max-age=0",
"Pragma": "no-cache",
}
def _spa_index_response() -> FileResponse:
return FileResponse(str(INDEX_FILE), headers=_SPA_HTML_HEADERS)
if FRONTEND_DIST.exists() and INDEX_FILE.is_file():
# Mount only /assets so client-side routes (/contacts, /deals, …) are not swallowed by
# StaticFiles (which would 404 for missing files instead of serving the SPA shell).
if FRONTEND_ASSETS.is_dir():
app.mount("/assets", StaticFiles(directory=str(FRONTEND_ASSETS)), name="frontend_assets")
@app.get("/")
def spa_index():
return _spa_index_response()
@app.get("/{full_path:path}")
def spa_fallback(full_path: str):
if full_path.startswith("api"):
raise HTTPException(status_code=404, detail="Not Found")
return _spa_index_response()
from fastapi import FastAPI, UploadFile, File, Depends, HTTPException, Query, Request
from fastapi.responses import FileResponse, StreamingResponse
from fastapi.staticfiles import StaticFiles
from fastapi.middleware.cors import CORSMiddleware
from starlette.middleware.sessions import SessionMiddleware
from pathlib import Path
from sqlalchemy.orm import Session
from sqlalchemy import func, or_
import pandas as pd
import uuid
import os
import csv
import io
import concurrent.futures
from functools import partial
from typing import Any, Dict, List, Optional
import json
import asyncio
import math
import re
import logging
import requests
from urllib.parse import urlparse, unquote
from datetime import datetime, timedelta
from calendar import monthrange
from .database import (
get_db,
SessionLocal,
db_commit_with_retry,
Tenant,
TenantMembership,
Team,
User,
UploadedFile,
Prompt,
GeneratedSequence,
SmartleadRun,
Contact,
Company,
ContactActivity,
CrmLead,
CrmDeal,
UnipileAccount,
LinkedinCampaign,
UnipileHostedAuthState,
OutreachSendReceipt,
ChatAgentConfig,
FileOutreachExecution,
)
from pydantic import ValidationError
from .models import (
UploadResponse,
PromptSaveRequest,
CampaignMetaPatchRequest,
SequenceResponse,
SmartleadPushRequest,
SmartleadRunResponse,
CrmLeadPatchRequest,
BulkLeadIdsRequest,
BulkContactIdsRequest,
ActivityCreateRequest,
CrmDealPatchRequest,
CrmDealCreateRequest,
ContactCreateRequest,
ContactPatchRequest,
CompanyCreateRequest,
CompanyPatchRequest,
BulkCompanyIdsRequest,
CompanyContactCreateRequest,
WonBillingPayload,
UnipileConnectRequest,
UnipileHostedLinkRequest,
UnipileMailboxHostedLinkRequest,
UserLinkedinPrefsPatch,
UserMailboxPrefsPatch,
UnipileLinkedinLimitsPatch,
LinkedinCampaignCreateRequest,
LinkedinCampaignGenerateRequest,
LinkedinCampaignExecuteRequest,
LinkedinCampaignPatchRequest,
)
from .gmail_invite import send_invite_email_via_gmail
from .gpt_service import (
generate_email_sequence,
generate_linkedin_sequence,
enrich_manual_contact_profile,
linkedin_invite_note,
normalize_linkedin_message_format,
_pick_contact_field,
)
from .smartlead_client import SmartleadClient
from .auth_routes import router as auth_router
from .tenant_deps import TenantContext, get_tenant_context
from .tenant_routes import router as tenant_router
from .outreach_routes import router as outreach_router
from .cta_forms_routes import router as cta_forms_router, lead_is_cta_form
from .landing_pages_routes import router as landing_pages_router
from .chat_agent_routes import router as chat_agent_router