Hugging Face's logo

Spaces:
Seth0330
/
qualifier
Paused

App Files Community
qualifier / backend /app /models.py
Sethu0330's picture
Sethu0330
Add FTL RFQ Qualifier agent + full SALES CRM app for HF Spaces testing
005df22 unverified
Raw History Blame Contribute Delete
19.4 kB
from pydantic import BaseModel, Field, field_validator
from typing import Optional, List, Dict, Literal, Any, Union
from datetime import datetime
class UploadResponse(BaseModel):
file_id: str
contact_count: int
message: str
class TeamCreateRequest(BaseModel):
name: str = Field(..., min_length=1, max_length=200)
manager_user_id: Optional[int] = None
class TeamPatchRequest(BaseModel):
name: Optional[str] = Field(None, min_length=1, max_length=200)
manager_user_id: Optional[int] = None
class MemberPatchRequest(BaseModel):
team_id: Optional[int] = None
role: Optional[str] = Field(default=None, pattern="^(admin|member)$")
class InvitationPatchRequest(BaseModel):
team_id: Optional[int] = None
role: Optional[str] = Field(default=None, pattern="^(admin|member)$")
class PromptSaveRequest(BaseModel):
file_id: str
prompts: Dict[str, str]
products: List[str]
linkedin_prompts: Dict[str, str] = Field(default_factory=dict)
sequence_plan: Optional[Union[List[Dict], Dict[str, Any]]] = Field(default=None)
campaign_name: Optional[str] = None
status: Optional[str] = None
email_timezone: Optional[str] = None
email_business_hours_start: Optional[int] = None
email_business_hours_end: Optional[int] = None
email_business_days: Optional[str] = None
email_send_interval_seconds: Optional[int] = None
class UnipileLinkedinLimitsPatch(BaseModel):
linkedin_daily_connection_cap: Optional[int] = Field(default=None, ge=1, le=200)
linkedin_daily_message_cap: Optional[int] = Field(default=None, ge=1, le=300)
linkedin_weekly_connection_cap: Optional[int] = Field(default=None, ge=1, le=1000)
linkedin_weekly_message_cap: Optional[int] = Field(default=None, ge=1, le=1500)
class CampaignMetaPatchRequest(BaseModel):
"""Lightweight auto-save for campaign identity, independent of the full prompts/sequence
payload — called on every wizard step transition so an abrupt tab close never strands
already-generated content with no backend pointer back to it."""
file_id: str
campaign_name: Optional[str] = None
status: Optional[str] = None
class SequenceResponse(BaseModel):
id: int
firstName: str
lastName: str
email: str
company: str
title: Optional[str] = None
product: str
subject: str
emailContent: str
class SmartleadPushRequest(BaseModel):
file_id: str
campaign_id: str # Required - campaign ID to add leads to
dry_run: bool = False
class CrmLeadPatchRequest(BaseModel):
crm_status: Optional[str] = None
first_name: Optional[str] = None
last_name: Optional[str] = None
email: Optional[str] = None
company_name: Optional[str] = None
title: Optional[str] = None
phone: Optional[str] = None # the lead's own personal phone — not company-synced
owner_user_id: Optional[int] = None
last_reply_subject: Optional[str] = None
last_reply_body: Optional[str] = None
# Stored under raw_webhook["company_details"] (same Apollo keys as Contacts raw_data)
company_name_for_emails: Optional[str] = None
industry: Optional[str] = None
employees: Optional[str] = None
annual_revenue: Optional[str] = None
last_raised_at: Optional[str] = None
website: Optional[str] = None
city: Optional[str] = None
state: Optional[str] = None
country_region: Optional[str] = None
category: Optional[str] = None
type: Optional[str] = None # raw_webhook.company_details["Type"] — "Existing" | "New" | ""
company_phone: Optional[str] = None # the company's main line — synced, distinct from `phone`
class ContactCreateRequest(BaseModel):
first_name: str = ""
last_name: str = ""
email: str
company: str = ""
title: str = ""
phone: str = ""
class ContactPatchRequest(BaseModel):
first_name: Optional[str] = None
last_name: Optional[str] = None
email: Optional[str] = None
company: Optional[str] = None
title: Optional[str] = None
phone: Optional[str] = None # the contact's own personal phone — not company-synced
# Merged into raw_data (Apollo CSV-style keys)
company_name_for_emails: Optional[str] = None
industry: Optional[str] = None
employees: Optional[str] = None
annual_revenue: Optional[str] = None
last_raised_at: Optional[str] = None
website: Optional[str] = None
city: Optional[str] = None
state: Optional[str] = None
country_region: Optional[str] = None # raw_data["Country"]
category: Optional[str] = None # raw_data["Category"] — "A" | "B" | "C" | ""
type: Optional[str] = None # raw_data["Type"] — "Existing" | "New" | ""
company_phone: Optional[str] = None # raw_data["Phone"] — company's main line, synced
owner_user_id: Optional[int] = None
class BulkLeadIdsRequest(BaseModel):
lead_ids: List[int]
class BulkContactIdsRequest(BaseModel):
contact_ids: List[int]
class ActivityCreateRequest(BaseModel):
activity_type: Literal["call", "visit"]
notes: str = Field(default="", max_length=8000)
class CompanyCreateRequest(BaseModel):
name: str
category: str = ""
type: str = ""
company_name_for_emails: str = ""
industry: str = ""
employees: str = ""
annual_revenue: str = ""
last_raised_at: str = ""
website: str = ""
city: str = ""
state: str = ""
country: str = ""
phone: str = ""
class CompanyPatchRequest(BaseModel):
name: Optional[str] = None
category: Optional[str] = None
type: Optional[str] = None
company_name_for_emails: Optional[str] = None
industry: Optional[str] = None
employees: Optional[str] = None
annual_revenue: Optional[str] = None
last_raised_at: Optional[str] = None
website: Optional[str] = None
city: Optional[str] = None
state: Optional[str] = None
country: Optional[str] = None
phone: Optional[str] = None
owner_user_id: Optional[int] = None
class BulkCompanyIdsRequest(BaseModel):
company_ids: List[int]
class CompanyContactCreateRequest(BaseModel):
"""Add a contact from the Companies page — company name is forced from the URL's company,
not taken from the client, so a contact can never be silently mis-associated."""
first_name: str = ""
last_name: str = ""
email: str
title: str = ""
phone: str = ""
class CrmDealCreateRequest(BaseModel):
"""Create a blank pipeline deal (optionally in a given stage)."""
name: Optional[str] = None
stage: Optional[str] = "new"
class WonLineItemIn(BaseModel):
"""Single invoice line when marking a deal Won."""
product_service: str = Field(..., min_length=1, max_length=600)
description: str = Field(default="", max_length=4000)
qty: float = Field(..., gt=0, le=1_000_000)
rate: float = Field(..., ge=0, le=1_000_000_000)
# Per-line cadence for ARR vs one-time (dashboard uses these; legacy deals may omit).
billing_interval: Literal["monthly", "quarterly", "annual", "one_time"] = "monthly"
currency: Literal["CAD", "USD"] = "USD"
@field_validator("product_service", mode="before")
@classmethod
def strip_product(cls, v):
if not isinstance(v, str) or not v.strip():
raise ValueError("Each line must have a product or service name.")
return v.strip()
@field_validator("description", mode="before")
@classmethod
def strip_desc(cls, v):
if v is None:
return ""
return v.strip() if isinstance(v, str) else v
@field_validator("billing_interval", mode="before")
@classmethod
def norm_interval(cls, v):
if v is None:
return "monthly"
s = v.strip().lower() if isinstance(v, str) else "monthly"
if s in ("one_time", "onetime", "one-time", "non-recurring"):
return "one_time"
if s in ("month", "monthly", "every month", "m"):
return "monthly"
if s in ("quarter", "quarterly", "every 3 months", "q"):
return "quarterly"
if s in ("year", "annual", "yearly", "every year", "a"):
return "annual"
return "monthly"
@field_validator("currency", mode="before")
@classmethod
def norm_currency(cls, v):
if v is None:
return "USD"
s = v.strip().upper() if isinstance(v, str) else "USD"
return s if s in ("CAD", "USD") else "USD"
class WonBillingPayload(BaseModel):
"""Billing / PO details required when setting a deal to Won (invoicing)."""
po_number: str = Field(..., min_length=1, max_length=200)
customer_legal_name: str = Field(..., min_length=1, max_length=500)
customer_address: str = Field(..., min_length=1, max_length=4000)
contact_person_name: str = Field(..., min_length=1, max_length=300)
channel_partner_name: Optional[str] = Field(default=None, max_length=300)
note_to_customer: str = Field(..., min_length=1, max_length=8000)
note_to_accounts: str = Field(..., min_length=1, max_length=8000)
line_items: List[WonLineItemIn] = Field(..., min_length=1, max_length=100)
@field_validator(
"po_number",
"customer_legal_name",
"customer_address",
"contact_person_name",
"note_to_customer",
"note_to_accounts",
mode="before",
)
@classmethod
def strip_required(cls, v):
if v is None:
return v
if isinstance(v, str) and not v.strip():
raise ValueError("must not be empty")
return v.strip() if isinstance(v, str) else v
@field_validator("channel_partner_name", mode="before")
@classmethod
def strip_optional(cls, v):
if v is None:
return None
if isinstance(v, str):
s = v.strip()
return s or None
return v
class CrmDealPatchRequest(BaseModel):
name: Optional[str] = None
stage: Optional[str] = None
owner_initials: Optional[str] = None
# Admin: set to a workspace member's user id, or null for unassigned. Member: may set to own id only to claim an unassigned deal.
owner_user_id: Optional[int] = None
deal_value: Optional[int] = None
revenue_type: Optional[str] = None # arr|qrr|mrr|one_time
close_probability: Optional[int] = None
expected_close_date: Optional[str] = None # ISO date or datetime
country: Optional[str] = None
contact_display: Optional[str] = None
account_name: Optional[str] = None
last_interaction_at: Optional[str] = None # ISO date or datetime
contact_id: Optional[int] = None # set CRM contact on deal; JSON null clears link
# When deal has contact_id, merged into that contact's raw_data (Apollo keys)
company_name_for_emails: Optional[str] = None
industry: Optional[str] = None
employees: Optional[str] = None
annual_revenue: Optional[str] = None
last_raised_at: Optional[str] = None
website: Optional[str] = None
city: Optional[str] = None
state: Optional[str] = None
country_region: Optional[str] = None
category: Optional[str] = None
type: Optional[str] = None
company_phone: Optional[str] = None
# Required (with conditionals) when setting stage to lost
lost_reason: Optional[str] = None # price|no_decision|chose_another|didnt_meet_expectation|unknown
lost_price_discount_pct: Optional[float] = None # 0–100 when lost_reason is price
lost_chose_product_name: Optional[str] = None # when lost_reason is chose_another
# Required when setting stage to won (invoicing / PO)
won_billing: Optional[WonBillingPayload] = None
class SmartleadRunResponse(BaseModel):
run_id: str
campaign_id: Optional[str] = None
campaign_name: Optional[str] = None
steps_count: int
total: int
added: int
skipped: int
failed: int
errors: List[Dict] = []
status: str
class UnipileConnectRequest(BaseModel):
label: str = "LinkedIn Account"
auth_mode: Literal["credentials", "cookie"] = "cookie"
username: Optional[str] = None
password: Optional[str] = None
access_token: Optional[str] = None
user_agent: Optional[str] = None
country: Optional[str] = None
ip: Optional[str] = None
class UnipileHostedLinkRequest(BaseModel):
label: str = "LinkedIn Account"
"""Shown in UniPile and this app as the connection label."""
linkedin_profile_display_name: Optional[str] = None
"""User-facing name for campaigns (saved on your profile when starting hosted auth)."""
class UnipileMailboxHostedLinkRequest(BaseModel):
label: str = "Mailbox"
mailbox_profile_display_name: Optional[str] = None
class UserLinkedinPrefsPatch(BaseModel):
linkedin_profile_display_name: Optional[str] = Field(None, max_length=200)
default_unipile_account_ref_id: Optional[int] = Field(None, ge=0)
"""Use 0 to clear the default account selection."""
class UserMailboxPrefsPatch(BaseModel):
mailbox_profile_display_name: Optional[str] = Field(None, max_length=200)
default_mailbox_unipile_account_ref_id: Optional[int] = Field(None, ge=0)
"""Use 0 to clear the default mailbox account selection."""
class LinkedinCampaignCreateRequest(BaseModel):
name: str
unipile_account_ref_id: int
class LinkedinCampaignGenerateRequest(BaseModel):
prompt_template: str = Field(..., min_length=20)
class LinkedinCampaignExecuteRequest(BaseModel):
dry_run: bool = False
class LinkedinCampaignPatchRequest(BaseModel):
"""Update LinkedIn campaign automation settings."""
followup_interval_hours: Optional[int] = Field(None, ge=1, le=720)
class CtaFormFieldDef(BaseModel):
key: str = Field(..., min_length=1, max_length=64)
label: str = Field(..., min_length=1, max_length=120)
type: str = Field(default="text", pattern="^(text|email|tel|textarea)$")
required: bool = False
map_to: str = Field(
default="custom",
pattern="^(name|email|company_name|title|phone|message|custom)$",
)
class CtaFormCreateRequest(BaseModel):
name: str = Field(..., min_length=1, max_length=200)
fields: List[CtaFormFieldDef] = Field(default_factory=list)
class CtaFormPatchRequest(BaseModel):
name: Optional[str] = Field(None, min_length=1, max_length=200)
fields: Optional[List[CtaFormFieldDef]] = None
is_active: Optional[bool] = None
class CtaFormSubmitRequest(BaseModel):
fields: Dict[str, str] = Field(default_factory=dict)
page_url: Optional[str] = Field(None, max_length=2000)
class LandingPageBenefit(BaseModel):
title: str = Field(default="", max_length=120)
body: str = Field(default="", max_length=800)
class LandingPageContent(BaseModel):
headline: str = Field(default="", max_length=300)
subheadline: str = Field(default="", max_length=600)
benefits: List[LandingPageBenefit] = Field(default_factory=list)
form_heading: str = Field(default="Book a demo", max_length=120)
form_subtext: str = Field(default="", max_length=600)
footer_text: str = Field(default="", max_length=200)
class LandingPageCreateRequest(BaseModel):
name: str = Field(..., min_length=1, max_length=200)
prompt: Optional[str] = Field(None, max_length=4000)
class LandingPagePatchRequest(BaseModel):
name: Optional[str] = Field(None, min_length=1, max_length=200)
prompt: Optional[str] = Field(None, max_length=4000)
content: Optional[LandingPageContent] = None
cta_form_id: Optional[int] = None
is_active: Optional[bool] = None
class LandingPageGenerateRequest(BaseModel):
prompt: str = Field(..., min_length=8, max_length=4000)
class ChatAgentConfigPatch(BaseModel):
is_enabled: Optional[bool] = None
greeting_message: Optional[str] = Field(None, max_length=1000)
chat_model_provider: Optional[Literal["openai", "gemini"]] = None
website_base_url: Optional[str] = Field(None, max_length=2000)
drive_folder_url: Optional[str] = Field(None, max_length=2000)
reindex_interval_hours: Optional[int] = Field(None, ge=1, le=168)
enable_web_search: Optional[bool] = None
class ChatAgentSkillReferenceItem(BaseModel):
title: str = Field("", max_length=300)
content: str = Field("", max_length=20000)
class ChatAgentSkillAssetItem(BaseModel):
name: str = Field("", max_length=300)
content: str = Field("", max_length=20000)
class ChatAgentSkillTemplateItem(BaseModel):
name: str = Field("", max_length=300)
content: str = Field("", max_length=20000)
class ChatAgentSkillScriptItem(BaseModel):
# The live public chat agent never executes this content. An admin can run it explicitly via
# the sandboxed /scripts/{index}/run route (subprocess-isolated, secrets stripped, timed out).
# See SKILLS-SECURITY.md.
name: str = Field("", max_length=300)
description: str = Field("", max_length=2000)
language: str = Field("", max_length=40)
content: str = Field("", max_length=20000)
last_run_status: Optional[str] = None
last_run_output: Optional[str] = None
last_run_at: Optional[str] = None
class ChatAgentSkillEvalItem(BaseModel):
prompt: str = Field("", max_length=2000)
assertion_type: Literal["must_contain", "must_not_contain", "llm_judge"] = "must_contain"
assertion_value: str = Field("", max_length=2000)
last_run_status: Optional[str] = None
last_run_detail: Optional[str] = None
last_run_at: Optional[str] = None
class ChatAgentSkillCreateRequest(BaseModel):
name: str = Field(..., min_length=1, max_length=200)
description: str = Field("", max_length=1000)
instructions: str = Field(..., min_length=1, max_length=14000)
references: List[ChatAgentSkillReferenceItem] = Field(default_factory=list, max_length=20)
assets: List[ChatAgentSkillAssetItem] = Field(default_factory=list, max_length=20)
templates: List[ChatAgentSkillTemplateItem] = Field(default_factory=list, max_length=20)
scripts: List[ChatAgentSkillScriptItem] = Field(default_factory=list, max_length=20)
evals: List[ChatAgentSkillEvalItem] = Field(default_factory=list, max_length=10)
class ChatAgentSkillPatchRequest(BaseModel):
name: Optional[str] = Field(None, min_length=1, max_length=200)
description: Optional[str] = Field(None, max_length=1000)
instructions: Optional[str] = Field(None, min_length=1, max_length=14000)
references: Optional[List[ChatAgentSkillReferenceItem]] = Field(None, max_length=20)
assets: Optional[List[ChatAgentSkillAssetItem]] = Field(None, max_length=20)
templates: Optional[List[ChatAgentSkillTemplateItem]] = Field(None, max_length=20)
scripts: Optional[List[ChatAgentSkillScriptItem]] = Field(None, max_length=20)
evals: Optional[List[ChatAgentSkillEvalItem]] = Field(None, max_length=10)
is_active: Optional[bool] = None
sort_order: Optional[int] = None
class ChatAgentSkillImportRequest(BaseModel):
content: str = Field(..., min_length=1, max_length=100000)
class ChatAgentReindexRequest(BaseModel):
source: Literal["drive", "website", "website_fast", "all"] = "all"
class ChatAgentSessionRequest(BaseModel):
page_url: Optional[str] = Field(None, max_length=2000)
class ChatAgentMessageRequest(BaseModel):
session_id: str = Field(..., min_length=1, max_length=64)
message: str = Field(..., min_length=1, max_length=4000)