"""
Command Center - SQLAlchemy Models
21 tables based on Lauren Kingsley schema, fully self-hosted.
"""
from typing import Optional
from flask_sqlalchemy import SQLAlchemy
from flask_login import UserMixin
from datetime import datetime, date, timezone
from werkzeug.security import generate_password_hash, check_password_hash
import uuid
db = SQLAlchemy()
def gen_uuid():
"""Generate UUID for Supabase-compatible tables."""
return str(uuid.uuid4())
class User(UserMixin, db.Model):
"""User accounts - replaces Supabase auth.users"""
__tablename__ = 'profiles'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
email = db.Column(db.String(255), unique=True, nullable=False, index=True)
password_hash = db.Column(db.String(255), nullable=False)
full_name = db.Column(db.String(255), default='')
company = db.Column(db.String(255), default='')
phone = db.Column(db.String(50), default='')
role = db.Column(db.String(50), default='user') # super_admin, partner, user
last_login = db.Column(db.DateTime, nullable=True)
is_active = db.Column(db.Boolean, default=True)
failed_login_attempts = db.Column(db.Integer, default=0, nullable=False)
locked_until = db.Column(db.DateTime, nullable=True)
totp_secret = db.Column(db.String(255), nullable=True)
totp_enabled = db.Column(db.Boolean, default=False)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
__table_args__ = (
db.CheckConstraint(
"role IN ('super_admin', 'admin', 'partner', 'user')",
name='ck_profiles_role',
),
)
# Relationships
user_companies = db.relationship('UserCompany', back_populates='user', cascade='all, delete-orphan')
coaching_assignments_given = db.relationship('CoachingAssignment', foreign_keys='CoachingAssignment.coach_id', back_populates='coach')
coaching_assignments_received = db.relationship('CoachingAssignment', foreign_keys='CoachingAssignment.rep_id', back_populates='rep')
user_settings = db.relationship('UserSetting', back_populates='user', cascade='all, delete-orphan')
def set_password(self, password):
self.password_hash = generate_password_hash(password)
def check_password(self, password):
return check_password_hash(self.password_hash, password)
def __repr__(self):
return f'<User {self.email}>'
class Company(db.Model):
"""Home improvement company records"""
__tablename__ = 'companies'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
name = db.Column(db.String(255), nullable=False, unique=True)
industry = db.Column(db.String(100), default='') # roofing, hvac, plumbing, etc.
size = db.Column(db.String(50), default='') # single, multi, enterprise
annual_revenue = db.Column(db.Float, nullable=True)
target_revenue = db.Column(db.Float, nullable=True)
address = db.Column(db.String(500), default='')
city = db.Column(db.String(100), default='')
state = db.Column(db.String(50), default='')
zip_code = db.Column(db.String(20), default='')
website = db.Column(db.String(255), default='')
logo_url = db.Column(db.String(500), default='')
stripe_customer_id = db.Column(db.String(255), default=None, index=True, unique=True)
settings_json = db.Column(db.JSON, default=dict)
is_partner = db.Column(db.Boolean, default=False, nullable=False) # partner org flag (audit M1)
is_deleted = db.Column(db.Boolean, default=False, nullable=False, index=True) # soft delete (audit L4)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
__table_args__ = (
db.CheckConstraint(
"size IN ('single', 'multi', 'enterprise', '')",
name='ck_companies_size',
),
)
# Relationships
users = db.relationship('UserCompany', back_populates='company', cascade='all, delete-orphan')
projects = db.relationship('Project', back_populates='company', cascade='all, delete-orphan')
goals = db.relationship('Goal', back_populates='company', cascade='all, delete-orphan')
kpi_values = db.relationship('KPIValue', back_populates='company', cascade='all, delete-orphan')
forecasts = db.relationship('Forecast', back_populates='company', cascade='all, delete-orphan')
revenue_leaks = db.relationship('RevenueLeak', back_populates='company', cascade='all, delete-orphan')
optimization_moves = db.relationship('OptimizationMove', back_populates='company', cascade='all, delete-orphan')
coaching_assignments = db.relationship('CoachingAssignment', back_populates='company', cascade='all, delete-orphan')
coaching_scorecards = db.relationship('CoachingScorecard', back_populates='company', cascade='all, delete-orphan')
connectors = db.relationship('Connector', back_populates='company', cascade='all, delete-orphan')
connector_logs = db.relationship('ConnectorLog', back_populates='company', cascade='all, delete-orphan')
activity_logs = db.relationship('ActivityLog', back_populates='company', cascade='all, delete-orphan')
notifications = db.relationship('Notification', back_populates='company', cascade='all, delete-orphan')
company_settings = db.relationship('Setting', back_populates='company', cascade='all, delete-orphan')
templates = db.relationship('Template', back_populates='company', cascade='all, delete-orphan')
custom_fields = db.relationship('CustomField', back_populates='company', cascade='all, delete-orphan')
roi_calculations = db.relationship('ROICalculation', back_populates='company', cascade='all, delete-orphan')
angi_leads = db.relationship('AngiLead', back_populates='company', cascade='all, delete-orphan')
sms_messages = db.relationship('SmsMessage', back_populates='company', cascade='all, delete-orphan')
calls = db.relationship('Call', back_populates='company', cascade='all, delete-orphan')
lead_touchpoints = db.relationship('LeadTouchpoint', back_populates='company', cascade='all, delete-orphan')
opt_outs = db.relationship('OptOut', back_populates='company', cascade='all, delete-orphan')
revenue_records = db.relationship('RevenueRecord', back_populates='company', cascade='all, delete-orphan')
def __repr__(self):
return f'<Company {self.name}>'
class UserCompany(db.Model):
"""Many-to-many user↔company"""
__tablename__ = 'users_companies'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
user_id = db.Column(db.String(36), db.ForeignKey('profiles.id', ondelete='CASCADE'), nullable=False, index=True)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
role = db.Column(db.String(50), default='member') # owner, admin, manager, member, viewer
joined_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
__table_args__ = (
db.UniqueConstraint('user_id', 'company_id', name='uq_users_companies_user_company'),
db.CheckConstraint(
"role IN ('owner', 'admin', 'manager', 'member', 'viewer')",
name='ck_users_companies_role',
),
)
user = db.relationship('User', back_populates='user_companies')
company = db.relationship('Company', back_populates='users')
def __repr__(self):
return f'<UserCompany user={self.user_id} company={self.company_id}>'
class Invite(db.Model):
"""Team/company invites — created by owners/admins to add members."""
__tablename__ = 'invites'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
email = db.Column(db.String(255), nullable=False, index=True)
role = db.Column(db.String(50), default='member') # owner, admin, manager, member, viewer
token = db.Column(db.String(128), unique=True, nullable=False, index=True)
status = db.Column(db.String(20), default='pending') # pending, accepted, expired, revoked
expires_at = db.Column(db.DateTime, nullable=False)
created_by = db.Column(db.String(36), db.ForeignKey('profiles.id', ondelete='SET NULL'), nullable=True)
accepted_by = db.Column(db.String(36), db.ForeignKey('profiles.id', ondelete='SET NULL'), nullable=True)
accepted_at = db.Column(db.DateTime, nullable=True)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
__table_args__ = (
db.CheckConstraint(
"role IN ('owner', 'admin', 'manager', 'member', 'viewer')",
name='ck_invites_role',
),
db.CheckConstraint(
"status IN ('pending', 'accepted', 'expired', 'revoked')",
name='ck_invites_status',
),
)
company = db.relationship('Company')
def __repr__(self):
return f'<Invite {self.email} @ {self.company_id}>'
class UserGroup(db.Model):
"""User groups for organizing team members."""
__tablename__ = 'user_groups'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
name = db.Column(db.String(255), nullable=False)
description = db.Column(db.Text, default='')
created_by = db.Column(db.String(36), db.ForeignKey('profiles.id', ondelete='SET NULL'), nullable=True)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
company = db.relationship('Company')
def __repr__(self):
return f'<UserGroup {self.name}>'
class GroupMember(db.Model):
"""Many-to-many user↔group membership."""
__tablename__ = 'group_members'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
group_id = db.Column(db.String(36), db.ForeignKey('user_groups.id', ondelete='CASCADE'), nullable=False, index=True)
user_id = db.Column(db.String(36), db.ForeignKey('profiles.id', ondelete='CASCADE'), nullable=False, index=True)
added_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
__table_args__ = (
db.UniqueConstraint('group_id', 'user_id', name='uq_group_members'),
)
group = db.relationship('UserGroup')
user = db.relationship('User')
def __repr__(self):
return f'<GroupMember group={self.group_id} user={self.user_id}>'
class Project(db.Model):
"""Individual projects within companies"""
__tablename__ = 'projects'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
name = db.Column(db.String(255), nullable=False)
description = db.Column(db.Text, default='')
status = db.Column(db.String(50), default='planning') # planning, active, completed, on_hold, cancelled
budget = db.Column(db.Float, nullable=True)
actual_cost = db.Column(db.Float, nullable=True)
revenue = db.Column(db.Float, nullable=True)
start_date = db.Column(db.DateTime, nullable=True)
end_date = db.Column(db.DateTime, nullable=True)
completed_date = db.Column(db.DateTime, nullable=True)
priority = db.Column(db.String(20), default='medium') # low, medium, high, critical
assigned_to = db.Column(db.String(36), db.ForeignKey('profiles.id', ondelete='SET NULL'), nullable=True)
custom_fields_json = db.Column(db.JSON, default=dict)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
__table_args__ = (
db.CheckConstraint(
"status IN ('planning', 'active', 'completed', 'on_hold', 'cancelled')",
name='ck_projects_status',
),
db.CheckConstraint(
"priority IN ('low', 'medium', 'high', 'critical')",
name='ck_projects_priority',
),
)
company = db.relationship('Company', back_populates='projects')
def __repr__(self):
return f'<Project {self.name}>'
class Goal(db.Model):
"""Cascading goals (org → dept → team → rep)"""
__tablename__ = 'goals'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
parent_goal_id = db.Column(db.String(36), db.ForeignKey('goals.id', ondelete='CASCADE'), nullable=True, index=True)
name = db.Column(db.String(255), nullable=False)
description = db.Column(db.Text, default='')
level = db.Column(db.String(50), default='org') # org, department, team, rep
target_value = db.Column(db.Float, nullable=True)
current_value = db.Column(db.Float, default=0.0)
unit = db.Column(db.String(50), default='$') # $, %, count, etc.
start_date = db.Column(db.DateTime, nullable=True)
end_date = db.Column(db.DateTime, nullable=True)
status = db.Column(db.String(50), default='active') # active, completed, paused, cancelled
assigned_to = db.Column(db.String(36), db.ForeignKey('profiles.id', ondelete='SET NULL'), nullable=True) # user_id for rep-level goals
weight = db.Column(db.Float, default=1.0) # relative importance
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
__table_args__ = (
db.CheckConstraint(
"level IN ('org', 'department', 'team', 'rep')",
name='ck_goals_level',
),
db.CheckConstraint(
"status IN ('active', 'completed', 'paused', 'cancelled')",
name='ck_goals_status',
),
)
company = db.relationship('Company', back_populates='goals')
parent_goal = db.relationship('Goal', remote_side=[id], backref='sub_goals')
metrics = db.relationship('GoalMetric', back_populates='goal', cascade='all, delete-orphan')
def progress_percentage(self):
if self.target_value is None or self.target_value == 0:
return 0.0
return min(100.0, (self.current_value / self.target_value) * 100)
def __repr__(self):
return f'<Goal {self.name}>'
class GoalMetric(db.Model):
"""KPI tracking against goals"""
__tablename__ = 'goal_metrics'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
goal_id = db.Column(db.String(36), db.ForeignKey('goals.id', ondelete='CASCADE'), nullable=False, index=True)
metric_name = db.Column(db.String(255), nullable=False)
target_value = db.Column(db.Float, nullable=True)
actual_value = db.Column(db.Float, default=0.0)
unit = db.Column(db.String(50), default='')
period = db.Column(db.String(20), default='monthly') # daily, weekly, monthly, quarterly
period_start = db.Column(db.DateTime, nullable=True)
period_end = db.Column(db.DateTime, nullable=True)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
__table_args__ = (
db.CheckConstraint(
"period IN ('daily', 'weekly', 'monthly', 'quarterly')",
name='ck_goal_metrics_period',
),
)
goal = db.relationship('Goal', back_populates='metrics')
def __repr__(self):
return f'<GoalMetric {self.metric_name}>'
class KPIValue(db.Model):
"""Time-series KPI measurements"""
__tablename__ = 'kpi_values'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
kpi_name = db.Column(db.String(255), nullable=False, index=True)
value = db.Column(db.Float, nullable=False)
unit = db.Column(db.String(50), default='')
category = db.Column(db.String(100), default='') # revenue, marketing, ops, etc.
period = db.Column(db.String(20), default='monthly')
period_start = db.Column(db.DateTime, nullable=True)
period_end = db.Column(db.DateTime, nullable=True)
metadata_json = db.Column(db.JSON, default=dict)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
company = db.relationship('Company', back_populates='kpi_values')
def __repr__(self):
return f'<KPIValue {self.kpi_name}={self.value}>'
class Forecast(db.Model):
"""Revenue forecasting data"""
__tablename__ = 'forecasts'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
forecast_type = db.Column(db.String(50), default='revenue') # revenue, pipeline, cost
period = db.Column(db.String(20), nullable=False) # monthly, quarterly, annually
period_start = db.Column(db.DateTime, nullable=True)
period_end = db.Column(db.DateTime, nullable=True)
projected_value = db.Column(db.Float, nullable=False)
actual_value = db.Column(db.Float, nullable=True)
confidence = db.Column(db.Float, nullable=True) # 0.0 to 1.0
methodology = db.Column(db.String(100), default='') # historical, weighted_pipeline, etc.
assumptions_json = db.Column(db.JSON, default=dict)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
__table_args__ = (
db.CheckConstraint(
"forecast_type IN ('revenue', 'pipeline', 'cost')",
name='ck_forecasts_forecast_type',
),
)
company = db.relationship('Company', back_populates='forecasts')
@property
def variance(self):
if self.actual_value is None:
return None
return self.actual_value - self.projected_value
@property
def variance_pct(self):
if self.actual_value is None or self.projected_value == 0:
return None
return ((self.actual_value - self.projected_value) / abs(self.projected_value)) * 100
def __repr__(self):
return f'<Forecast {self.forecast_type} {self.period_start}>'
class RevenueLeak(db.Model):
"""Identified revenue leakage points"""
__tablename__ = 'revenue_leaks'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
source = db.Column(db.String(255), nullable=False) # pipeline_stage, channel, product
description = db.Column(db.Text, default='')
estimated_loss = db.Column(db.Float, nullable=True)
severity = db.Column(db.String(20), default='medium') # low, medium, high, critical
detected_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
resolved = db.Column(db.Boolean, default=False)
resolved_at = db.Column(db.DateTime, nullable=True)
resolution_notes = db.Column(db.Text, default='')
metadata_json = db.Column(db.JSON, default=dict)
# Auto-detection fields (Phase 0 — leak detection framework, July 2026)
detection_type = db.Column(db.String(10), nullable=False, default='manual') # 'manual' | 'auto'
detector_id = db.Column(db.String(64), nullable=True, index=True) # e.g. 'qb_overdue_30d'
source_connector = db.Column(db.String(32), nullable=True) # connector service name
rule_params_json = db.Column(db.JSON, nullable=True) # threshold params used for this detection
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
__table_args__ = (
db.CheckConstraint(
"severity IN ('low', 'medium', 'high', 'critical')",
name='ck_revenue_leaks_severity',
),
)
company = db.relationship('Company', back_populates='revenue_leaks')
def __repr__(self):
return f'<RevenueLeak {self.source}>'
class OptimizationMove(db.Model):
"""Recommended budget/action moves"""
__tablename__ = 'optimization_moves'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
move_type = db.Column(db.String(100), default='budget_shift') # budget_shift, channel_change, target_audience
description = db.Column(db.Text, default='')
source_channel = db.Column(db.String(100), default='')
target_channel = db.Column(db.String(100), default='')
current_spend = db.Column(db.Float, nullable=True)
recommended_spend = db.Column(db.Float, nullable=True)
expected_impact = db.Column(db.Float, nullable=True)
confidence = db.Column(db.Float, nullable=True) # 0.0 to 1.0
status = db.Column(db.String(20), default='recommended') # recommended, accepted, rejected, implemented
implemented_at = db.Column(db.DateTime, nullable=True)
actual_result = db.Column(db.Float, nullable=True)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
__table_args__ = (
db.CheckConstraint(
"status IN ('recommended', 'accepted', 'rejected', 'implemented')",
name='ck_optimization_moves_status',
),
)
company = db.relationship('Company', back_populates='optimization_moves')
def __repr__(self):
return f'<OptimizationMove {self.move_type}>'
class CoachingAssignment(db.Model):
"""Coach/rep pairings"""
__tablename__ = 'coaching_assignments'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
coach_id = db.Column(db.String(36), db.ForeignKey('profiles.id', ondelete='CASCADE'), nullable=False, index=True)
rep_id = db.Column(db.String(36), db.ForeignKey('profiles.id', ondelete='CASCADE'), nullable=False, index=True)
focus_area = db.Column(db.String(100), default='') # closing, qualifying, follow-up, etc.
description = db.Column(db.Text, default='')
start_date = db.Column(db.DateTime, nullable=True)
end_date = db.Column(db.DateTime, nullable=True)
status = db.Column(db.String(20), default='active') # active, completed, paused
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
__table_args__ = (
db.CheckConstraint(
"status IN ('active', 'completed', 'paused')",
name='ck_coaching_assignments_status',
),
)
company = db.relationship('Company', back_populates='coaching_assignments')
coach = db.relationship('User', foreign_keys=[coach_id], back_populates='coaching_assignments_given')
rep = db.relationship('User', foreign_keys=[rep_id], back_populates='coaching_assignments_received')
def __repr__(self):
return f'<CoachingAssignment {self.coach_id}→{self.rep_id}>'
class CoachingScorecard(db.Model):
"""Performance scorecards"""
__tablename__ = 'coaching_scorecards'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
assignment_id = db.Column(db.String(36), db.ForeignKey('coaching_assignments.id', ondelete='CASCADE'), nullable=False, index=True)
rep_id = db.Column(db.String(36), db.ForeignKey('profiles.id', ondelete='CASCADE'), nullable=False, index=True)
evaluation_date = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
overall_score = db.Column(db.Float, nullable=True) # 0-100
communication_score = db.Column(db.Float, nullable=True)
technical_score = db.Column(db.Float, nullable=True)
closing_score = db.Column(db.Float, nullable=True)
follow_up_score = db.Column(db.Float, nullable=True)
strengths = db.Column(db.Text, default='')
improvement_areas = db.Column(db.Text, default='')
notes = db.Column(db.Text, default='')
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
company = db.relationship('Company', back_populates='coaching_scorecards')
assignment = db.relationship('CoachingAssignment', backref='scorecards')
def __repr__(self):
return f'<CoachingScorecard {self.rep_id} score={self.overall_score}>'
class Connector(db.Model):
"""Integration connector configs"""
__tablename__ = 'connectors'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
service = db.Column(db.String(100), nullable=False) # servicetitan, hubspot, google_ads, quickbooks
status = db.Column(db.String(20), default='inactive') # inactive, connected, syncing, error
config_encrypted = db.Column(db.Text, nullable=True) # NEW: Fernet-encrypted config
config_json = db.Column(db.JSON, nullable=True) # DEPRECATED: plaintext (kept for migration)
last_sync_at = db.Column(db.DateTime, nullable=True)
sync_frequency = db.Column(db.String(20), default='hourly') # real_time, hourly, daily
error_message = db.Column(db.Text, default='')
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
__table_args__ = (
db.CheckConstraint(
"status IN ('inactive', 'connected', 'syncing', 'error')",
name='ck_connectors_status',
),
)
company = db.relationship('Company', back_populates='connectors')
@property
def config(self) -> dict:
"""Read config — decrypt if encrypted, fall back to plaintext for migration."""
if self.config_encrypted:
from app.utils.encryption import decrypt_config
return decrypt_config(self.config_encrypted)
return self.config_json or {}
@config.setter
def config(self, value: dict) -> None:
"""Write config — always encrypt."""
from app.utils.encryption import encrypt_config
self.config_encrypted = encrypt_config(value)
self.config_json = None # Clear plaintext
def __repr__(self):
return f'<Connector {self.service}>'
class ConnectorLog(db.Model):
"""Sync logs for integrations"""
__tablename__ = 'connector_logs'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
connector_id = db.Column(db.String(36), db.ForeignKey('connectors.id', ondelete='CASCADE'), nullable=False, index=True)
event_type = db.Column(db.String(50), nullable=False) # sync_start, sync_complete, sync_error
record_count = db.Column(db.Integer, default=0)
duration_ms = db.Column(db.Integer, nullable=True)
status = db.Column(db.String(20), default='pending') # pending, success, error
error_message = db.Column(db.Text, default='')
details_json = db.Column(db.JSON, default=dict)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
__table_args__ = (
db.CheckConstraint(
"status IN ('pending', 'success', 'error')",
name='ck_connector_logs_status',
),
)
company = db.relationship('Company', back_populates='connector_logs')
connector = db.relationship('Connector', backref='logs')
def __repr__(self):
return f'<ConnectorLog {self.event_type}>'
def to_dict(self):
return {
'id': self.id,
'company_id': self.company_id,
'connector_id': self.connector_id,
'event_type': self.event_type,
'record_count': self.record_count,
'duration_ms': self.duration_ms,
'status': self.status,
'error_message': self.error_message,
'details_json': self.details_json,
'created_at': self.created_at.isoformat() if self.created_at else None,
}
class ActivityLog(db.Model):
"""User/company activity feed"""
__tablename__ = 'activity_logs'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
user_id = db.Column(db.String(36), db.ForeignKey('profiles.id', ondelete='SET NULL'), nullable=True, index=True)
action = db.Column(db.String(100), nullable=False) # login, create_company, update_goal, etc.
entity_type = db.Column(db.String(50), default='')
entity_id = db.Column(db.String(36), default='')
details = db.Column(db.Text, default='')
ip_address = db.Column(db.String(50), default='')
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
company = db.relationship('Company', back_populates='activity_logs')
user = db.relationship('User', backref='activities')
def __repr__(self):
return f'<ActivityLog {self.action}>'
class AuditLog(db.Model):
"""System audit trail"""
__tablename__ = 'audit_logs'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
user_id = db.Column(db.String(36), db.ForeignKey('profiles.id', ondelete='SET NULL'), nullable=True, index=True)
action = db.Column(db.String(100), nullable=False)
entity_type = db.Column(db.String(50), default='')
entity_id = db.Column(db.String(36), default='')
old_values = db.Column(db.JSON, default=dict)
new_values = db.Column(db.JSON, default=dict)
ip_address = db.Column(db.String(50), default='')
user_agent = db.Column(db.String(500), default='')
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
user = db.relationship('User', backref='audits')
def __repr__(self):
return f'<AuditLog {self.action}>'
class Notification(db.Model):
"""In-app notifications"""
__tablename__ = 'notifications'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
user_id = db.Column(db.String(36), db.ForeignKey('profiles.id', ondelete='CASCADE'), nullable=False, index=True)
type = db.Column(db.String(50), nullable=False) # alert, reminder, update, coaching
title = db.Column(db.String(255), nullable=False)
message = db.Column(db.Text, default='')
link = db.Column(db.String(500), default='')
is_read = db.Column(db.Boolean, default=False)
read_at = db.Column(db.DateTime, nullable=True)
expires_at = db.Column(db.DateTime, nullable=True)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
__table_args__ = (
db.CheckConstraint(
"type IN ('alert', 'reminder', 'update', 'coaching')",
name='ck_notifications_type',
),
)
company = db.relationship('Company', back_populates='notifications')
user = db.relationship('User', backref='notifications')
def __repr__(self):
return f'<Notification {self.type}>'
class Setting(db.Model):
"""Company-level settings"""
__tablename__ = 'settings'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
key = db.Column(db.String(100), nullable=False)
value = db.Column(db.JSON, default=dict)
description = db.Column(db.Text, default='')
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
__table_args__ = (
db.UniqueConstraint('company_id', 'key', name='uq_settings_company_key'),
)
company = db.relationship('Company', back_populates='company_settings')
def __repr__(self):
return f'<Setting {self.key}>'
class UserSetting(db.Model):
"""Per-user settings (notification prefs, appearance, etc.)"""
__tablename__ = 'user_settings'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
user_id = db.Column(db.String(36), db.ForeignKey('profiles.id', ondelete='CASCADE'), nullable=False, index=True)
key = db.Column(db.String(100), nullable=False)
value = db.Column(db.JSON, default=dict)
__table_args__ = (
db.UniqueConstraint('user_id', 'key', name='uq_user_settings_user_key'),
)
user = db.relationship('User', back_populates='user_settings')
def __repr__(self):
return f'<UserSetting {self.user_id}/{self.key}>'
class UserSession(db.Model):
"""Active user sessions tracking"""
__tablename__ = 'user_sessions'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
user_id = db.Column(db.String(36), db.ForeignKey('profiles.id', ondelete='CASCADE'), nullable=False, index=True)
session_id = db.Column(db.String(255), nullable=False, index=True)
ip_address = db.Column(db.String(50), default='')
user_agent = db.Column(db.String(500), default='')
device_info = db.Column(db.String(255), default='')
is_current = db.Column(db.Boolean, default=False)
last_seen_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
expires_at = db.Column(db.DateTime, nullable=True)
def __repr__(self):
return f'<UserSession {self.user_id}/{self.session_id[:8]}>'
class PasswordResetToken(db.Model):
"""One-time password reset tokens (sha256-hashed, 1 hour expiry)."""
__tablename__ = 'password_reset_tokens'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
user_id = db.Column(db.String(36), db.ForeignKey('profiles.id', ondelete='CASCADE'), nullable=False, index=True)
token_hash = db.Column(db.String(64), nullable=False, unique=True, index=True)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
expires_at = db.Column(db.DateTime, nullable=False)
used_at = db.Column(db.DateTime, nullable=True)
@property
def is_valid(self):
"""Token is valid if unused and not expired."""
if self.used_at is not None:
return False
expires = self.expires_at
if expires is not None and expires.tzinfo is None:
expires = expires.replace(tzinfo=timezone.utc)
return expires is not None and expires > datetime.now(timezone.utc)
def __repr__(self):
return f'<PasswordResetToken {self.user_id}>'
class Template(db.Model):
"""Goal/project templates"""
__tablename__ = 'templates'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
name = db.Column(db.String(255), nullable=False)
type = db.Column(db.String(50), default='goal') # goal, project, report
description = db.Column(db.Text, default='')
content_json = db.Column(db.JSON, default=dict)
is_active = db.Column(db.Boolean, default=True)
is_default = db.Column(db.Boolean, default=False)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
company = db.relationship('Company', back_populates='templates')
def __repr__(self):
return f'<Template {self.name}>'
class CustomField(db.Model):
"""Extensible fields per entity"""
__tablename__ = 'custom_fields'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
entity_type = db.Column(db.String(50), nullable=False) # project, goal, etc.
name = db.Column(db.String(255), nullable=False)
field_type = db.Column(db.String(50), default='text') # text, number, date, select, checkbox
options_json = db.Column(db.JSON, default=list)
is_required = db.Column(db.Boolean, default=False)
display_order = db.Column(db.Integer, default=0)
is_active = db.Column(db.Boolean, default=True)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
company = db.relationship('Company', back_populates='custom_fields')
def __repr__(self):
return f'<CustomField {self.entity_type}.{self.name}>'
class ROICalculation(db.Model):
"""ROI computation results"""
__tablename__ = 'roi_calculations'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
product = db.Column(db.String(255), default='')
channel = db.Column(db.String(100), default='')
market = db.Column(db.String(100), default='')
spend = db.Column(db.Float, nullable=True)
revenue = db.Column(db.Float, nullable=True)
roi = db.Column(db.Float, nullable=True) # percentage
ltv = db.Column(db.Float, nullable=True) # lifetime value
cac = db.Column(db.Float, nullable=True) # customer acquisition cost
ltv_cac_ratio = db.Column(db.Float, nullable=True)
period = db.Column(db.String(20), default='monthly')
period_start = db.Column(db.DateTime, nullable=True)
period_end = db.Column(db.DateTime, nullable=True)
metadata_json = db.Column(db.JSON, default=dict)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
company = db.relationship('Company', back_populates='roi_calculations')
@property
def net_revenue(self):
if self.spend is None or self.revenue is None:
return None
return self.revenue - self.spend
def __repr__(self):
return f'<ROICalculation {self.product}/{self.channel} roi={self.roi}>'
# =============================================================================
# CRM models — populated by connector sync (HubSpot, etc.)
# =============================================================================
class CrmContact(db.Model):
"""Contacts synced from CRM (HubSpot)."""
__tablename__ = 'crm_contacts'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
external_id = db.Column(db.String(128), nullable=False, index=True) # HubSpot object ID
email = db.Column(db.String(255), default='')
first_name = db.Column(db.String(255), default='')
last_name = db.Column(db.String(255), default='')
phone = db.Column(db.String(50), default='')
company_name = db.Column(db.String(255), default='')
lifecycle_stage = db.Column(db.String(50), default='') # subscriber, lead, marketingqualifiedlead, etc.
market = db.Column(db.String(100), default='') # geographic/service market
status = db.Column(db.String(50), default='active') # active, inactive, unsubscribed
hubspot_owner_id = db.Column(db.String(64), default='')
properties_json = db.Column(db.JSON, default=dict)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
company = db.relationship('Company', backref='crm_contacts')
__table_args__ = (
db.UniqueConstraint('company_id', 'external_id', name='uq_crm_contact_company_external'),
)
@property
def full_name(self):
parts = [self.first_name, self.last_name]
return ' '.join(p for p in parts if p)
def __repr__(self):
return f'<CrmContact {self.full_name} ({self.external_id})>'
class CrmCompany(db.Model):
"""Companies/accounts synced from CRM (HubSpot)."""
__tablename__ = 'crm_companies'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
external_id = db.Column(db.String(128), nullable=False, index=True) # HubSpot company ID
name = db.Column(db.String(255), default='')
domain = db.Column(db.String(255), default='')
industry = db.Column(db.String(100), default='')
num_employees = db.Column(db.Integer, nullable=True)
properties_json = db.Column(db.JSON, default=dict)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
company = db.relationship('Company', backref='crm_companies')
__table_args__ = (
db.UniqueConstraint('company_id', 'external_id', name='uq_crm_company_company_external'),
)
def __repr__(self):
return f'<CrmCompany {self.name} ({self.external_id})>'
class CrmDeal(db.Model):
"""Deals/opportunities synced from CRM (HubSpot)."""
__tablename__ = 'crm_deals'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
external_id = db.Column(db.String(128), nullable=False, index=True) # HubSpot deal ID
name = db.Column(db.String(255), default='')
amount = db.Column(db.Float, nullable=True)
stage = db.Column(db.String(100), default='') # e.g. 'appointments_set', 'closedwon'
pipeline = db.Column(db.String(100), default='')
probability = db.Column(db.Float, nullable=True) # 0.0 to 1.0
expected_close_date = db.Column(db.DateTime, nullable=True)
contact_id = db.Column(db.String(36), db.ForeignKey('crm_contacts.id', ondelete='SET NULL'), nullable=True, index=True)
deal_owner_id = db.Column(db.String(64), default='')
properties_json = db.Column(db.JSON, default=dict)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
company = db.relationship('Company', backref='crm_deals')
contact = db.relationship('CrmContact')
__table_args__ = (
db.UniqueConstraint('company_id', 'external_id', name='uq_crm_deal_company_external'),
)
def __repr__(self):
return f'<CrmDeal {self.name} ${self.amount} ({self.external_id})>'
# =============================================================================
# Angi lead models — lead lifecycle management for Angi/HomeAdvisor
# =============================================================================
class AngiLead(db.Model):
"""Angi lead — managed from dashboard with full lifecycle tracking."""
__tablename__ = 'angi_leads'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
angi_lead_id = db.Column(db.String(128), nullable=False, index=True) # Angi's lead ID
connector_id = db.Column(db.String(36), nullable=False, index=True) # FK to connector
# Lead contact info
first_name = db.Column(db.String(255), default='')
last_name = db.Column(db.String(255), default='')
email = db.Column(db.String(255), default='')
phone = db.Column(db.String(50), default='')
# Lead project details
project_type = db.Column(db.String(255), default='')
description = db.Column(db.Text, default='')
budget = db.Column(db.String(100), default='')
budget_value = db.Column(db.Float, nullable=True) # Parsed numeric value
timeline = db.Column(db.String(100), default='')
address = db.Column(db.String(500), default='')
city = db.Column(db.String(100), default='')
state = db.Column(db.String(50), default='')
zip_code = db.Column(db.String(20), default='')
# Lead lifecycle
status = db.Column(db.String(50), default='new') # new, responded, contacted, scheduled, won, lost, expired
priority = db.Column(db.String(20), default='medium') # high, medium, low
is_premium = db.Column(db.Boolean, default=False)
# Response tracking
response_time_minutes = db.Column(db.Integer, nullable=True) # Time from lead receipt to first response
responded_at = db.Column(db.DateTime, nullable=True)
last_contacted_at = db.Column(db.DateTime, nullable=True)
# Angi-specific
is_duplicate = db.Column(db.Boolean, default=False)
lead_source = db.Column(db.String(50), default='angi') # angi, homeadvisor
cost_per_lead = db.Column(db.Float, nullable=True) # If available from Angi billing
# Internal notes
internal_notes = db.Column(db.Text, default='')
# SMS/Call integration fields
sms_opted_out = db.Column(db.Boolean, default=False) # TCPA opt-out flag synced from OptOut table
first_contact_at = db.Column(db.DateTime, nullable=True) # Timestamp of first outbound SMS or call
# Angi interview & compliance (P0 — remodeling gap plan, July 2026)
interview_qa = db.Column(db.Text, nullable=True) # Angi interview questionnaire answers (JSON)
tcpa_compliance = db.Column(db.Boolean, default=False) # TCPA consent flag
match_type = db.Column(db.String(50), default="") # e.g. 'basic', 'premium', 'exclusive'
# Timestamps
received_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
company = db.relationship('Company', back_populates='angi_leads')
__table_args__ = (
db.UniqueConstraint('company_id', 'angi_lead_id', name='uq_angi_lead_company_angi_id'),
db.CheckConstraint(
"status IN ('new', 'accepted', 'responded', 'contacted', 'scheduled', 'won', 'lost', 'expired', 'rejected', 'sms_sent', 'replied', 'call_attempted', 'opted_out')",
name='ck_angi_leads_status',
),
db.CheckConstraint(
"priority IN ('high', 'medium', 'low')",
name='ck_angi_leads_priority',
),
)
@property
def full_name(self):
parts = [self.first_name, self.last_name]
return ' '.join(p for p in parts if p)
@property
def full_address(self):
parts = [self.address, self.city]
if self.state:
parts.append(f"{self.state} {self.zip_code}".strip())
else:
if self.zip_code:
parts.append(self.zip_code)
return ', '.join(p for p in parts if p)
def to_dict(self):
return {
'id': self.id,
'company_id': self.company_id,
'angi_lead_id': self.angi_lead_id,
'connector_id': self.connector_id,
'first_name': self.first_name,
'last_name': self.last_name,
'full_name': self.full_name,
'email': self.email,
'phone': self.phone,
'project_type': self.project_type,
'description': self.description,
'budget': self.budget,
'budget_value': self.budget_value,
'timeline': self.timeline,
'address': self.address,
'city': self.city,
'state': self.state,
'zip_code': self.zip_code,
'full_address': self.full_address,
'status': self.status,
'priority': self.priority,
'is_premium': self.is_premium,
'response_time_minutes': self.response_time_minutes,
'responded_at': self.responded_at.isoformat() if self.responded_at else None,
'last_contacted_at': self.last_contacted_at.isoformat() if self.last_contacted_at else None,
'is_duplicate': self.is_duplicate,
'lead_source': self.lead_source,
'cost_per_lead': self.cost_per_lead,
'internal_notes': self.internal_notes,
'sms_opted_out': self.sms_opted_out,
'first_contact_at': self.first_contact_at.isoformat() if self.first_contact_at else None,
'interview_qa': self.interview_qa,
'tcpa_compliance': self.tcpa_compliance,
'match_type': self.match_type,
'received_at': self.received_at.isoformat() if self.received_at else None,
'created_at': self.created_at.isoformat() if self.created_at else None,
'updated_at': self.updated_at.isoformat() if self.updated_at else None,
}
def __repr__(self):
return f'<AngiLead {self.full_name} ({self.angi_lead_id}) {self.status}>'
class AngiLeadAction(db.Model):
"""Audit log of actions taken on Angi leads (responses, status changes, etc.)"""
__tablename__ = 'angi_lead_actions'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
lead_id = db.Column(db.String(36), db.ForeignKey('angi_leads.id', ondelete='CASCADE'), nullable=False, index=True)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
action_type = db.Column(db.String(50), nullable=False) # received, accepted, responded, rejected, status_change, note_added
details = db.Column(db.Text, default='')
message_content = db.Column(db.Text, default='') # For response messages
performed_by = db.Column(db.String(36), nullable=True) # User ID who performed the action
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
lead = db.relationship('AngiLead', backref='actions')
company = db.relationship('Company')
__table_args__ = (
db.CheckConstraint(
"action_type IN ('received', 'accepted', 'responded', 'rejected', 'status_change', 'note_added')",
name='ck_angi_lead_actions_type',
),
)
def to_dict(self):
return {
'id': self.id,
'lead_id': self.lead_id,
'company_id': self.company_id,
'action_type': self.action_type,
'details': self.details,
'message_content': self.message_content,
'performed_by': self.performed_by,
'created_at': self.created_at.isoformat() if self.created_at else None,
}
def __repr__(self):
return f'<AngiLeadAction {self.action_type} on lead {self.lead_id}>'
# =============================================================================
# Accounting models — populated by QuickBooks, etc.
# =============================================================================
class AccountingRecord(db.Model):
"""Generic accounting records synced from QuickBooks and other accounting tools."""
__tablename__ = 'accounting_records'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
external_id = db.Column(db.String(128), nullable=False, index=True)
source_service = db.Column(db.String(50), nullable=False, default='quickbooks') # quickbooks, xero, etc.
record_type = db.Column(db.String(50), nullable=False, index=True) # Invoice, Payment, Customer, Expense
name = db.Column(db.String(255), default='')
amount = db.Column(db.Float, nullable=True)
currency = db.Column(db.String(10), default='USD')
status = db.Column(db.String(50), default='') # Paid, Pending, Draft, etc.
due_date = db.Column(db.DateTime, nullable=True)
transaction_date = db.Column(db.DateTime, nullable=True)
contact_name = db.Column(db.String(255), default='')
contact_email = db.Column(db.String(255), default='')
metadata_json = db.Column(db.JSON, default=dict)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
company = db.relationship('Company', backref='accounting_records')
__table_args__ = (
db.UniqueConstraint('company_id', 'external_id', name='uq_accounting_company_external'),
)
def __repr__(self):
return f'<AccountingRecord {self.record_type} {self.external_id}>'
# =============================================================================
# QuickBooks models — populated by QuickBooks Online connector
# =============================================================================
class QuickbooksInvoice(db.Model):
"""Invoices synced from QuickBooks Online."""
__tablename__ = 'quickbooks_invoices'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
qb_doc_id = db.Column(db.String(100), unique=True, nullable=False, index=True)
qb_realm_id = db.Column(db.String(50), nullable=True)
customer_name = db.Column(db.String(255), default='')
invoice_num = db.Column(db.String(100), default='')
total_amount = db.Column(db.Float, nullable=True)
tax_amount = db.Column(db.Float, default=0.0)
status = db.Column(db.String(20), default='') # Draft, Sent, Paid, Void
due_date = db.Column(db.DateTime, nullable=True)
tx_date = db.Column(db.DateTime, nullable=True)
line_items_json = db.Column(db.JSON, default=dict)
metadata_json = db.Column(db.JSON, default=dict)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
company = db.relationship('Company', backref=db.backref('quickbooks_invoices', cascade='all, delete-orphan'))
def to_dict(self):
return {
'id': self.id,
'company_id': self.company_id,
'qb_doc_id': self.qb_doc_id,
'qb_realm_id': self.qb_realm_id,
'customer_name': self.customer_name,
'invoice_num': self.invoice_num,
'total_amount': self.total_amount,
'tax_amount': self.tax_amount,
'status': self.status,
'due_date': self.due_date.isoformat() if self.due_date else None,
'tx_date': self.tx_date.isoformat() if self.tx_date else None,
'line_items': self.line_items_json,
'metadata': self.metadata_json,
'created_at': self.created_at.isoformat() if self.created_at else None,
'updated_at': self.updated_at.isoformat() if self.updated_at else None,
}
def __repr__(self):
return f'<QuickbooksInvoice {self.invoice_num} {self.qb_doc_id}>'
class QuickbooksTransaction(db.Model):
"""Transactions synced from QuickBooks Online."""
__tablename__ = 'quickbooks_transactions'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
qb_doc_id = db.Column(db.String(100), unique=True, nullable=False, index=True)
qb_realm_id = db.Column(db.String(50), nullable=True)
tx_type = db.Column(db.String(50), default='') # Invoice, CreditMemo, JournalEntry, Payment
tx_date = db.Column(db.DateTime, nullable=True)
amount = db.Column(db.Float, nullable=True)
account_name = db.Column(db.String(255), default='')
account_id = db.Column(db.String(100), default='')
memo = db.Column(db.Text, default='')
class_name = db.Column(db.String(255), default='')
metadata_json = db.Column(db.JSON, default=dict)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
company = db.relationship('Company', backref=db.backref('quickbooks_transactions', cascade='all, delete-orphan'))
def to_dict(self):
return {
'id': self.id,
'company_id': self.company_id,
'qb_doc_id': self.qb_doc_id,
'qb_realm_id': self.qb_realm_id,
'tx_type': self.tx_type,
'tx_date': self.tx_date.isoformat() if self.tx_date else None,
'amount': self.amount,
'account_name': self.account_name,
'account_id': self.account_id,
'memo': self.memo,
'class_name': self.class_name,
'metadata': self.metadata_json,
'created_at': self.created_at.isoformat() if self.created_at else None,
'updated_at': self.updated_at.isoformat() if self.updated_at else None,
}
def __repr__(self):
return f'<QuickbooksTransaction {self.tx_type} {self.qb_doc_id}>'
class QuickbooksCustomer(db.Model):
"""Customers synced from QuickBooks Online."""
__tablename__ = 'quickbooks_customers'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
qb_doc_id = db.Column(db.String(100), nullable=False, index=True)
qb_realm_id = db.Column(db.String(50), nullable=True)
display_name = db.Column(db.String(255), default='')
email = db.Column(db.String(255), default='')
phone = db.Column(db.String(50), default='')
balance = db.Column(db.Float, default=0.0)
billing_address_json = db.Column(db.JSON, default=dict)
metadata_json = db.Column(db.JSON, default=dict)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
company = db.relationship('Company', backref=db.backref('quickbooks_customers', cascade='all, delete-orphan'))
__table_args__ = (
db.UniqueConstraint('company_id', 'qb_doc_id'),
)
def to_dict(self):
return {
'id': self.id,
'company_id': self.company_id,
'qb_doc_id': self.qb_doc_id,
'qb_realm_id': self.qb_realm_id,
'display_name': self.display_name,
'email': self.email,
'phone': self.phone,
'balance': self.balance,
'billing_address': self.billing_address_json,
'metadata': self.metadata_json,
'created_at': self.created_at.isoformat() if self.created_at else None,
'updated_at': self.updated_at.isoformat() if self.updated_at else None,
}
def __repr__(self):
return f'<QuickbooksCustomer {self.display_name} {self.qb_doc_id}>'
class QuickbooksExpense(db.Model):
"""Expenses synced from QuickBooks Online."""
__tablename__ = 'quickbooks_expenses'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
qb_doc_id = db.Column(db.String(100), nullable=False, index=True)
qb_realm_id = db.Column(db.String(50), nullable=True)
account_name = db.Column(db.String(255), default='')
amount = db.Column(db.Float, nullable=True)
vendor_name = db.Column(db.String(255), default='')
tx_date = db.Column(db.DateTime, nullable=True)
memo = db.Column(db.Text, default='')
category = db.Column(db.String(100), default='')
metadata_json = db.Column(db.JSON, default=dict)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
company = db.relationship('Company', backref=db.backref('quickbooks_expenses', cascade='all, delete-orphan'))
__table_args__ = (
db.UniqueConstraint('company_id', 'qb_doc_id'),
)
def to_dict(self):
return {
'id': self.id,
'company_id': self.company_id,
'qb_doc_id': self.qb_doc_id,
'qb_realm_id': self.qb_realm_id,
'account_name': self.account_name,
'amount': self.amount,
'vendor_name': self.vendor_name,
'tx_date': self.tx_date.isoformat() if self.tx_date else None,
'memo': self.memo,
'category': self.category,
'metadata': self.metadata_json,
'created_at': self.created_at.isoformat() if self.created_at else None,
}
def __repr__(self):
return f'<QuickbooksExpense {self.qb_doc_id}>'
# =============================================================================
# Advertising models — populated by Google Ads, Facebook Ads, etc.
# =============================================================================
class AdCampaign(db.Model):
"""Ad campaigns synced from Google Ads, Facebook Ads, etc."""
__tablename__ = 'ad_campaigns'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
external_id = db.Column(db.String(128), nullable=False, index=True)
source_service = db.Column(db.String(50), nullable=False) # google_ads, facebook_ads
name = db.Column(db.String(255), default='')
status = db.Column(db.String(50), default='') # ENABLED, PAUSED, REMOVED, etc.
budget = db.Column(db.Float, nullable=True) # daily or lifetime budget
budget_type = db.Column(db.String(50), default='') # daily, lifetime
start_date = db.Column(db.DateTime, nullable=True)
end_date = db.Column(db.DateTime, nullable=True)
channel_type = db.Column(db.String(50), default='') # SEARCH, DISPLAY, etc.
parent_campaign_id = db.Column(db.String(128), nullable=True)
metadata_json = db.Column(db.JSON, default=dict)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
company = db.relationship('Company', backref='ad_campaigns')
__table_args__ = (
db.UniqueConstraint('company_id', 'external_id', name='uq_ad_campaign_company_external'),
)
def __repr__(self):
return f'<AdCampaign {self.source_service} {self.name}>'
class AdMetric(db.Model):
"""Daily ad performance metrics synced from advertising platforms."""
__tablename__ = 'ad_metrics'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
external_campaign_id = db.Column(db.String(128), nullable=True, index=True)
source_service = db.Column(db.String(50), nullable=False) # google_ads, facebook_ads
metric_date = db.Column(db.Date, nullable=False, index=True)
spend = db.Column(db.Float, default=0.0)
impressions = db.Column(db.BigInteger, default=0)
clicks = db.Column(db.Integer, default=0)
conversions = db.Column(db.Float, default=0.0)
ctr = db.Column(db.Float, nullable=True) # click-through rate
cpc = db.Column(db.Float, nullable=True) # cost per click
cpv = db.Column(db.Float, nullable=True) # cost per view
roas = db.Column(db.Float, nullable=True) # return on ad spend
metadata_json = db.Column(db.JSON, default=dict)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
company = db.relationship('Company', backref='ad_metrics')
def __repr__(self):
return f'<AdMetric {self.source_service} {self.metric_date}>'
# =============================================================================
# Lead Generation Engine — campaign templates, creatives, keywords, attribution
# =============================================================================
class AdTemplate(db.Model):
"""Reusable campaign templates organized by vertical/service type.
Templates define the structure: campaign settings, ad groups, keywords,
and ad copy. Users fill in budget, geo radius, and city → system
generates actual campaigns on Google/Facebook Ads.
"""
__tablename__ = 'ad_templates'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
name = db.Column(db.String(255), nullable=False) # e.g. "Emergency Roof Repair"
slug = db.Column(db.String(100), nullable=False, unique=True, index=True) # "roofing-emergency-repair"
vertical = db.Column(db.String(50), nullable=False, index=True) # roofing, hvac, plumbing
description = db.Column(db.Text, default='')
budget_recommendation = db.Column(db.Float, nullable=True) # monthly minimum
target_cpa = db.Column(db.Float, nullable=True) # recommended cost-per-acquire
bidding_strategy = db.Column(db.String(50), default='target_cpa') # target_cpa, maximize_clicks, manual_cpc
# Template structure (stored as JSON — campaigns, ad groups, keywords, ad copy)
structure_json = db.Column(db.JSON, default=dict)
# Status
is_active = db.Column(db.Boolean, default=True, nullable=False)
# Metadata
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
def __repr__(self):
return f'<AdTemplate {self.vertical}/{self.slug}>'
def to_dict(self):
return {
'id': self.id,
'name': self.name,
'slug': self.slug,
'vertical': self.vertical,
'description': self.description,
'budget_recommendation': self.budget_recommendation,
'target_cpa': self.target_cpa,
'bidding_strategy': self.bidding_strategy,
'structure': self.structure_json,
'is_active': self.is_active,
'created_at': self.created_at.isoformat() if self.created_at else None,
}
class AdCreative(db.Model):
"""Ad copy and creative variants organized by vertical.
Used for A/B testing and populating campaigns from templates.
Each creative tracks which variants won/lost.
"""
__tablename__ = 'ad_creatives'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=True, index=True) # nullable = system-wide templates
vertical = db.Column(db.String(50), nullable=False, index=True)
service_type = db.Column(db.String(100), nullable=False) # emergency-repair, new-install, seasonal-maintenance
ad_type = db.Column(db.String(50), nullable=False) # responsive_search, expanded_text, image, video
# Content
headlines = db.Column(db.JSON, default=list) # array of headline variants (max 15 for RSA)
descriptions = db.Column(db.JSON, default=list) # array of description variants (max 5 for RSA)
display_url = db.Column(db.String(500), default='')
final_url = db.Column(db.String(500), default='')
image_url = db.Column(db.String(500), default='') # for image ads
call_to_action = db.Column(db.String(100), default='')
# Performance tracking
total_impressions = db.Column(db.BigInteger, default=0)
total_clicks = db.Column(db.Integer, default=0)
total_conversions = db.Column(db.Float, default=0.0)
avg_ctr = db.Column(db.Float, nullable=True)
# Status
is_active = db.Column(db.Boolean, default=True, nullable=False)
is_winner = db.Column(db.Boolean, default=False) # auto-set after A/B test
# Metadata
metadata_json = db.Column(db.JSON, default=dict)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
company = db.relationship('Company', backref=db.backref('ad_creatives', cascade='all, delete-orphan'))
def __repr__(self):
return f'<AdCreative {self.vertical}/{self.service_type} {self.id[:8]}>'
class AdKeyword(db.Model):
"""Keywords tracked per campaign/ad group with performance data.
Stores the keyword list used in search campaigns plus match type
and performance metrics for optimization decisions.
"""
__tablename__ = 'ad_keywords'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
source_service = db.Column(db.String(50), nullable=False) # google_ads, facebook_ads
external_campaign_id = db.Column(db.String(128), nullable=False, index=True)
external_ad_group_id = db.Column(db.String(128), nullable=True)
# Keyword
text = db.Column(db.String(255), nullable=False) # the keyword
match_type = db.Column(db.String(20), default='phrase') # broad, phrase, exact
# Bidding
max_cpc = db.Column(db.Float, nullable=True) # max cost-per-click bid
target_cpa = db.Column(db.Float, nullable=True)
# Status
status = db.Column(db.String(20), default='enabled') # enabled, paused, removed
# Performance (updated via sync or attribution)
total_impressions = db.Column(db.BigInteger, default=0)
total_clicks = db.Column(db.Integer, default=0)
total_spend = db.Column(db.Float, default=0.0)
total_conversions = db.Column(db.Float, default=0.0)
avg_position = db.Column(db.Float, nullable=True)
avg_ctr = db.Column(db.Float, nullable=True)
# Research data (populated from Keyword Planner or similar source)
search_volume = db.Column(db.BigInteger, nullable=True) # monthly average
competition_level = db.Column(db.String(20), nullable=True) # low, medium, high
cpc_suggestion = db.Column(db.Float, nullable=True) # suggested bid in cents
trend_data = db.Column(db.JSON, nullable=True) # monthly search volume array
# Metadata
metadata_json = db.Column(db.JSON, default=dict)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
company = db.relationship('Company', backref=db.backref('ad_keywords', cascade='all, delete-orphan'))
__table_args__ = (
db.UniqueConstraint('company_id', 'source_service', 'external_ad_group_id', 'text', 'match_type',
name='uq_ad_keyword_company_service_group_text_match'),
)
def __repr__(self):
return f'<AdKeyword "{self.text}" {self.match_type}>'
class LeadAttribution(db.Model):
"""Maps ad clicks to leads and deals — closes the attribution loop.
When a form is submitted, UTM params from the landing page are captured
and linked to the resulting lead. When a deal closes, we calculate
ROAS: deal_amount / ad_spend.
"""
__tablename__ = 'lead_attribution'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
# Lead reference
external_lead_id = db.Column(db.String(200), nullable=True, index=True) # CRM/Angi lead ID
lead_source = db.Column(db.String(50), default='') # web_form, phone_call, email, angi, hubspot
# UTM tracking params
utm_source = db.Column(db.String(100), default='') # google_ads, facebook_ads, organic
utm_campaign = db.Column(db.String(255), default='', index=True)
utm_ad_group = db.Column(db.String(255), default='')
utm_ad = db.Column(db.String(255), default='')
utm_content = db.Column(db.String(255), default='')
utm_medium = db.Column(db.String(50), default='') # cpc, organic, email
utm_term = db.Column(db.String(255), default='') # keyword
# Timing
click_date = db.Column(db.DateTime, nullable=True)
lead_date = db.Column(db.DateTime, nullable=True)
# Cost
click_cost = db.Column(db.Float, nullable=True) # estimated cost of that click
total_ad_spend = db.Column(db.Float, nullable=True) # total spend on campaign at time of conversion
# Revenue
external_deal_id = db.Column(db.String(200), nullable=True) # CRM deal ID when converted
deal_amount = db.Column(db.Float, nullable=True)
deal_closed_date = db.Column(db.DateTime, nullable=True)
roas = db.Column(db.Float, nullable=True) # deal_amount / total_ad_spend
# Status
attribution_status = db.Column(db.String(20), default='click') # click, lead, deal
attribution_window_days = db.Column(db.Integer, default=30)
# Metadata
metadata_json = db.Column(db.JSON, default=dict)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
company = db.relationship('Company', backref=db.backref('lead_attributions', cascade='all, delete-orphan'))
def __repr__(self):
return f'<LeadAttribution {self.utm_source} {self.utm_campaign[:30] if self.utm_campaign else "unknown"}>'
class OptimizationRule(db.Model):
"""Auto-optimization rules — kill, scale, pause, reallocate.
Configurable thresholds that the optimization engine evaluates daily.
"""
__tablename__ = 'optimization_rules'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
name = db.Column(db.String(255), nullable=False)
rule_type = db.Column(db.String(50), nullable=False, index=True) # kill_high_cpa, scale_low_cpa, pause_zero_conv, kill_low_ctr, budget_reallocation
target_level = db.Column(db.String(20), default='campaign') # campaign, ad_group, ad, keyword
# Rule parameters (JSON — thresholds, conditions, time windows)
params_json = db.Column(db.JSON, default=dict)
# e.g. {"cpa_multiplier": 2.0, "min_spend": 100, "lookback_days": 14, "action": "pause"}
# Status
is_active = db.Column(db.Boolean, default=True, nullable=False)
# Execution
last_run_at = db.Column(db.DateTime, nullable=True)
total_actions = db.Column(db.Integer, default=0)
# Metadata
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
company = db.relationship('Company', backref=db.backref('optimization_rules', cascade='all, delete-orphan'))
def __repr__(self):
return f'<OptimizationRule {self.rule_type} {self.name}>'
class OptimizationLog(db.Model):
"""Audit log of auto-optimization actions taken on campaigns/ads/keywords.
Every change is logged with before/after state for rollback capability.
"""
__tablename__ = 'optimization_logs'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
rule_id = db.Column(db.String(36), db.ForeignKey('optimization_rules.id', ondelete='CASCADE'), nullable=True)
# What was changed
source_service = db.Column(db.String(50), nullable=False) # google_ads, facebook_ads
target_type = db.Column(db.String(20), nullable=False) # campaign, ad_group, ad, keyword, budget
target_id = db.Column(db.String(128), nullable=False) # external ID of the entity
target_name = db.Column(db.String(255), default='')
# Action taken
action = db.Column(db.String(50), nullable=False) # paused, resumed, deleted, scaled, rebudgeted, bid_adjusted
# Before/after state (for rollback)
before_json = db.Column(db.JSON, default=dict)
after_json = db.Column(db.JSON, default=dict)
# Reason
reason = db.Column(db.Text, default='') # e.g. "CPA $147 > $50 target × 2.0 multiplier over 14d, spend $340"
# Result
status = db.Column(db.String(20), default='success') # success, failed, skipped
error_message = db.Column(db.Text, default='')
# Metadata
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
company = db.relationship('Company', backref=db.backref('optimization_logs', cascade='all, delete-orphan'))
rule = db.relationship('OptimizationRule', backref=db.backref('optimization_logs', cascade='all, delete-orphan'))
def __repr__(self):
return f'<OptimizationLog {self.action} {self.target_name[:40]}>'
# =============================================================================
# Revenue Records — actual payment data from POS/accounting connectors
# =============================================================================
class RevenueRecord(db.Model):
"""Ground-truth revenue records synced from POS (Square), accounting (QuickBooks), or CRM invoice payments.
Contrasts with CrmDeal (forecasted/pipeline) — these represent money that actually moved.
The Revenue Leaks engine compares forecasted deal amounts vs actual revenue records.
"""
__tablename__ = 'revenue_records'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
# Source reference
source_service = db.Column(db.String(50), nullable=False, index=True)
external_id = db.Column(db.String(200), nullable=False, index=True)
# Transaction data
transaction_date = db.Column(db.DateTime, nullable=False, index=True)
amount = db.Column(db.Float, nullable=False)
currency = db.Column(db.String(3), default='USD')
status = db.Column(db.String(50)) # completed, pending, refunded, failed
# Location tracking (multi-location companies)
location_id = db.Column(db.String(36), nullable=True, index=True) # FK to locations.id (nullable for backward compat)
location_name = db.Column(db.String(255), nullable=True) # denormalized location name
# Customer info
customer_id = db.Column(db.String(200))
customer_name = db.Column(db.String(500))
customer_email = db.Column(db.String(500))
# Location / branch info (multi-location POS)
location_id = db.Column(db.String(100))
location_name = db.Column(db.String(500))
# Type: payment, invoice, deposit
record_type = db.Column(db.String(50), default='payment')
# Raw metadata for debugging
metadata_json = db.Column(db.JSON, default=dict)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(
db.DateTime, default=lambda: datetime.now(timezone.utc),
onupdate=lambda: datetime.now(timezone.utc)
)
__table_args__ = (
db.UniqueConstraint('company_id', 'source_service', 'external_id',
name='uq_revenue_record_company_source_external'),
)
company = db.relationship('Company', back_populates='revenue_records')
def __repr__(self):
return f'<RevenueRecord {self.source_service} {self.external_id}>'
def to_dict(self):
return {
'id': self.id,
'company_id': self.company_id,
'source_service': self.source_service,
'external_id': self.external_id,
'transaction_date': self.transaction_date.isoformat() if self.transaction_date else None,
'amount': self.amount,
'currency': self.currency,
'status': self.status,
'customer_id': self.customer_id,
'customer_name': self.customer_name,
'customer_email': self.customer_email,
'location_id': self.location_id,
'location_name': self.location_name,
'record_type': self.record_type,
'created_at': self.created_at.isoformat() if self.created_at else None,
}
# =============================================================================
# Location models — multi-location company hierarchy
# =============================================================================
class Location(db.Model):
"""Physical locations/branches for multi-location companies.
Formalizes the location_id/location_name fields that already exist
on RevenueRecord and other denormalized models. Enables proper
rollup analytics and location-level reporting.
"""
__tablename__ = 'locations'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
# Location identity
name = db.Column(db.String(255), nullable=False) # e.g. 'Downtown Store', 'North Branch'
code = db.Column(db.String(50), default='') # Short code: 'DT', 'NB', etc.
# Address
address = db.Column(db.String(500), default='')
city = db.Column(db.String(100), default='')
state = db.Column(db.String(50), default='')
zip_code = db.Column(db.String(20), default='')
country = db.Column(db.String(50), default='US')
# Status
is_active = db.Column(db.Boolean, default=True, nullable=False)
# Metadata
metadata_json = db.Column(db.JSON, default=dict)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
__table_args__ = (
db.UniqueConstraint('company_id', 'name', name='uq_locations_company_name'),
db.UniqueConstraint('company_id', 'code', name='uq_locations_company_code'),
)
company = db.relationship('Company', backref=db.backref('locations', cascade='all, delete-orphan'))
def __repr__(self):
return f'<Location {self.name} @ {self.city or "HQ"}>'
def to_dict(self):
return {
'id': self.id,
'company_id': self.company_id,
'name': self.name,
'code': self.code,
'address': self.address,
'city': self.city,
'state': self.state,
'zip_code': self.zip_code,
'country': self.country,
'is_active': self.is_active,
'metadata': self.metadata_json,
'created_at': self.created_at.isoformat() if self.created_at else None,
'updated_at': self.updated_at.isoformat() if self.updated_at else None,
}
# =============================================================================
# Slack models — workspace and channel data
# =============================================================================
class SlackChannel(db.Model):
"""Slack channels synced from workspace."""
__tablename__ = 'slack_channels'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
external_id = db.Column(db.String(128), nullable=False, index=True) # Slack channel ID
name = db.Column(db.String(255), default='')
is_private = db.Column(db.Boolean, default=False)
is_archived = db.Column(db.Boolean, default=False)
purpose = db.Column(db.Text, default='')
topic = db.Column(db.Text, default='')
created_at_ts = db.Column(db.DateTime, nullable=True)
metadata_json = db.Column(db.JSON, default=dict)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
company = db.relationship('Company', backref='slack_channels')
__table_args__ = (
db.UniqueConstraint('company_id', 'external_id', name='uq_slack_channel_company_external'),
)
def __repr__(self):
return f'<SlackChannel #{self.name}>'
# =============================================================================
# Generic external data — for REST API and other connectors
# =============================================================================
class ExternalSyncRecord(db.Model):
"""Generic records synced from external APIs via Generic REST connector or other integrations."""
__tablename__ = 'external_sync_records'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
external_id = db.Column(db.String(128), nullable=False, index=True)
source_service = db.Column(db.String(50), nullable=False)
entity_type = db.Column(db.String(100), default='') # e.g. 'contact', 'opportunity', 'lead'
name = db.Column(db.String(255), default='')
status = db.Column(db.String(100), default='')
properties_json = db.Column(db.JSON, default=dict)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
company = db.relationship('Company', backref='external_sync_records')
__table_args__ = (
db.UniqueConstraint('company_id', 'external_id', name='uq_external_sync_company_external'),
)
def __repr__(self):
return f'<ExternalSyncRecord {self.source_service}/{self.entity_type} {self.external_id}>'
class DemoRequest(db.Model):
"""Landing page demo request leads"""
__tablename__ = 'demo_requests'
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
name = db.Column(db.String(255), nullable=False)
email = db.Column(db.String(255), nullable=False, index=True)
company = db.Column(db.String(255), nullable=True)
phone = db.Column(db.String(50), nullable=True)
message = db.Column(db.Text, nullable=True)
status = db.Column(db.String(20), default='new') # new, contacted, converted, lost
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
def to_dict(self):
return {
'id': str(self.id),
'name': self.name,
'email': self.email,
'company': self.company or '',
'phone': self.phone,
'message': self.message or '',
'status': self.status,
'source': 'landing_page',
'assignedTo': None,
'createdAt': self.created_at.isoformat() if self.created_at else '',
'notes': self.message or '',
}
def __repr__(self):
return f'<DemoRequest {self.name} ({self.email})>'
class SSOConfig(db.Model):
"""SAML SSO configuration per company"""
__tablename__ = 'sso_configs'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
provider_name = db.Column(db.String(50), default='custom') # okta, azure-ad, one-login, custom
sso_url = db.Column(db.String(500), default='') # IdP SSO URL
sso_binding = db.Column(db.String(10), default='post') # post, get
idp_cert = db.Column(db.Text, default='') # PEM certificate
entity_id = db.Column(db.String(500), default='') # IdP entity ID
acs_url = db.Column(db.String(500), default='') # Assertion Consumer Service URL
sp_metadata_url = db.Column(db.String(500), default='') # SP metadata URL
sp_cert = db.Column(db.Text, default='') # SP certificate (optional, for signed assertions)
enabled = db.Column(db.Boolean, default=False)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
company = db.relationship('Company', backref='sso_config')
def to_dict(self):
return {
'id': self.id,
'company_id': self.company_id,
'provider_name': self.provider_name,
'sso_url': self.sso_url,
'sso_binding': self.sso_binding,
'idp_cert': self.idp_cert,
'entity_id': self.entity_id,
'acs_url': self.acs_url,
'sp_metadata_url': self.sp_metadata_url,
'sp_cert': self.sp_cert,
'enabled': self.enabled,
'created_at': self.created_at.isoformat() if self.created_at else None,
'updated_at': self.updated_at.isoformat() if self.updated_at else None,
}
def __repr__(self):
return f'<SSOConfig {self.provider_name} company={self.company_id}>'
class Subscription(db.Model):
"""Stripe subscription tracking per company"""
__tablename__ = 'subscriptions'
# ─── Plan ID → Stripe Price ID mapping ──────────────────────────────────
# Maps Stripe price IDs (from env) to plan slugs.
# Populated lazily on first access so env vars are guaranteed to be loaded.
# Fallback: any unrecognised price_id → 'launch' (free tier).
_PLAN_MAP: dict = {} # price_id → plan_id; populated in _build_plan_map()
@classmethod
def _build_plan_map(cls) -> None:
"""Build the price_id → plan_id lookup from environment variables."""
import os
cls._PLAN_MAP = {
os.environ.get('STRIPE_PRICE_LAUNCH', 'price_launch'): 'launch',
os.environ.get('STRIPE_PRICE_GROWTH', 'price_growth'): 'growth',
os.environ.get('STRIPE_PRICE_COMMAND', 'price_command'): 'command',
os.environ.get('STRIPE_PRICE_ENTERPRISE', 'price_enterprise'): 'enterprise',
}
# Remove empty-string keys (env var unset → falls through to default)
cls._PLAN_MAP = {k: v for k, v in cls._PLAN_MAP.items() if k}
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True, unique=True)
stripe_subscription_id = db.Column(db.String(255), default='', index=True)
stripe_price_id = db.Column(db.String(255), default='')
stripe_status = db.Column(db.String(50), default='none') # trialing, active, past_due, canceled, incomplete
current_period_start = db.Column(db.DateTime, nullable=True)
current_period_end = db.Column(db.DateTime, nullable=True)
cancel_at_period_end = db.Column(db.Boolean, default=False)
trial_end = db.Column(db.DateTime, nullable=True)
# Idempotency/ordering guard: Stripe `event.created` of the newest
# subscription-scoped webhook event applied to this row. Stale (out-of-order)
# webhook retries with an older timestamp are ignored.
last_event_created = db.Column(db.Integer, nullable=True)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
__table_args__ = (
db.CheckConstraint(
"stripe_status IN ('trialing', 'active', 'past_due', 'canceled', 'incomplete', 'none')",
name='ck_subscriptions_stripe_status',
),
)
company = db.relationship('Company', backref='subscription')
@property
def is_active(self):
return self.stripe_status in ('active', 'trialing')
@property
def plan_id(self) -> str:
"""Map stripe_price_id → plan slug using _PLAN_MAP.
The map is built lazily from environment variables so Stripe price IDs
(STRIPE_PRICE_LAUNCH, STRIPE_PRICE_GROWTH, STRIPE_PRICE_COMMAND,
STRIPE_PRICE_ENTERPRISE) are resolved at runtime rather than import
time. Falls back to 'launch' for free/unrecognised price IDs.
"""
price_id = self.stripe_price_id or ''
if not price_id:
# No paid subscription recorded — treat as free Launch tier
return 'launch'
# Build map lazily (once per process)
if not self.__class__._PLAN_MAP:
self.__class__._build_plan_map()
return self.__class__._PLAN_MAP.get(price_id, 'launch')
def __repr__(self):
return f'<Subscription {self.stripe_status} company={self.company_id}>'
# =============================================================================
# Portfolio models — PE-backed portfolio management (grouping companies)
# =============================================================================
portfolio_companies = db.Table('portfolio_companies',
db.Column('id', db.String(36), primary_key=True, default=gen_uuid),
db.Column('portfolio_id', db.String(36), db.ForeignKey('portfolios.id', ondelete='CASCADE'), nullable=False, index=True),
db.Column('company_id', db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True),
db.Column('added_at', db.DateTime, default=lambda: datetime.now(timezone.utc)),
)
class Portfolio(db.Model):
"""Collections of companies for PE-backed portfolio management"""
__tablename__ = 'portfolios'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
name = db.Column(db.String(255), nullable=False)
description = db.Column(db.Text, default='')
created_by = db.Column(db.String(36), db.ForeignKey('profiles.id'), nullable=False, index=True)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
companies = db.relationship('Company', secondary=portfolio_companies, backref=db.backref('portfolios', lazy='dynamic'))
creator = db.relationship('User')
def to_dict(self):
return {
'id': self.id,
'name': self.name,
'description': self.description or '',
'created_by': self.created_by,
'created_by_name': self.creator.full_name if self.creator else '',
'created_by_email': self.creator.email if self.creator else '',
'company_count': len(self.companies),
'companies': [{
'id': c.id,
'name': c.name,
'industry': c.industry,
'annual_revenue': c.annual_revenue,
} for c in self.companies],
'created_at': self.created_at.isoformat() if self.created_at else None,
'updated_at': self.updated_at.isoformat() if self.updated_at else None,
}
def __repr__(self):
return f'<Portfolio {self.name}>'
# =============================================================================
# Email Campaign models — outbound email delivery tracking
# =============================================================================
class EmailCampaign(db.Model):
"""Outbound email campaign tracking"""
__tablename__ = 'email_campaigns'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
name = db.Column(db.String(255), nullable=False)
status = db.Column(db.String(50), default='draft') # draft, scheduled, sending, sent, completed, paused, failed
subject = db.Column(db.String(500), default='')
body = db.Column(db.Text, default='')
created_by = db.Column(db.String(36), db.ForeignKey('profiles.id'), nullable=False, index=True)
sent_at = db.Column(db.DateTime, nullable=True)
total_sent = db.Column(db.Integer, default=0)
delivered = db.Column(db.Integer, default=0)
opened = db.Column(db.Integer, default=0)
bounced = db.Column(db.Integer, default=0)
failed = db.Column(db.Integer, default=0)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
logs = db.relationship('EmailDeliveryLog', back_populates='campaign', cascade='all, delete-orphan')
creator = db.relationship('User')
def to_dict(self):
return {
'id': self.id,
'name': self.name,
'status': self.status,
'subject': self.subject,
'body': self.body or '',
'created_by': self.created_by,
'created_by_name': self.creator.full_name if self.creator else '',
'created_by_email': self.creator.email if self.creator else '',
'sent_at': self.sent_at.isoformat() if self.sent_at else None,
'total_sent': self.total_sent,
'delivered': self.delivered,
'opened': self.opened,
'bounced': self.bounced,
'failed': self.failed,
'log_count': len(self.logs),
'created_at': self.created_at.isoformat() if self.created_at else None,
'updated_at': self.updated_at.isoformat() if self.updated_at else None,
}
def __repr__(self):
return f'<EmailCampaign {self.name}>'
class EmailDeliveryLog(db.Model):
"""Individual email delivery records for a campaign"""
__tablename__ = 'email_delivery_logs'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
campaign_id = db.Column(db.String(36), db.ForeignKey('email_campaigns.id'), nullable=False, index=True)
recipient = db.Column(db.String(255), nullable=False)
status = db.Column(db.String(50), default='pending') # pending, sent, delivered, opened, bounced, failed
sent_at = db.Column(db.DateTime, nullable=True)
delivered_at = db.Column(db.DateTime, nullable=True)
opened_at = db.Column(db.DateTime, nullable=True)
error_message = db.Column(db.Text, default='')
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
campaign = db.relationship('EmailCampaign', back_populates='logs')
def to_dict(self):
return {
'id': self.id,
'campaign_id': self.campaign_id,
'recipient': self.recipient,
'status': self.status,
'sent_at': self.sent_at.isoformat() if self.sent_at else None,
'delivered_at': self.delivered_at.isoformat() if self.delivered_at else None,
'opened_at': self.opened_at.isoformat() if self.opened_at else None,
'error_message': self.error_message,
'created_at': self.created_at.isoformat() if self.created_at else None,
}
def __repr__(self):
return f'<EmailDeliveryLog {self.recipient} {self.status}>'
class ApiKey(db.Model):
"""API keys for programmatic access"""
__tablename__ = 'api_keys'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
user_id = db.Column(db.String(36), db.ForeignKey('profiles.id', ondelete='CASCADE'), nullable=False, index=True)
name = db.Column(db.String(255), nullable=False)
prefix = db.Column(db.String(10), default='csk_', nullable=False)
hashed_key = db.Column(db.String(255), nullable=False, unique=True)
last_used_at = db.Column(db.DateTime, nullable=True)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
expires_at = db.Column(db.DateTime, nullable=True)
permissions = db.Column(db.JSON, default=dict)
__table_args__ = ()
user = db.relationship('User', backref='api_keys')
def to_dict(self, include_key: bool = False):
result = {
'id': self.id,
'name': self.name,
'prefix': self.prefix,
'last_used_at': self.last_used_at.isoformat() if self.last_used_at else None,
'created_at': self.created_at.isoformat() if self.created_at else None,
'expires_at': self.expires_at.isoformat() if self.expires_at else None,
'permissions': self.permissions or {},
}
if include_key:
result['key'] = self.hashed_key # Only returned once on creation
return result
def __repr__(self):
return f'<ApiKey {self.name}>'
# =============================================================================
# SMS + Call Connector models — telephony integration
# =============================================================================
class SmsMessage(db.Model):
"""SMS messages — inbound and outbound, with delivery tracking."""
__tablename__ = 'sms_messages'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
connector_id = db.Column(db.String(36), nullable=False, index=True) # FK to Connector
lead_id = db.Column(db.String(36), db.ForeignKey('angi_leads.id', ondelete='SET NULL'), nullable=True, index=True)
from_phone = db.Column(db.String(50), nullable=False) # Our business number
to_phone = db.Column(db.String(50), nullable=False) # Lead's phone
direction = db.Column(db.String(10), nullable=False) # 'outbound' | 'inbound'
body = db.Column(db.Text, nullable=False) # Message content
status = db.Column(db.String(20), default='queued') # queued, sent, delivered, failed, rejected
error_code = db.Column(db.String(50), default='') # Provider error code on failure
error_message = db.Column(db.Text, default='') # Human-readable error
provider_message_sid = db.Column(db.String(128), default='') # Twilio/Plivo message SID
keyword_flag = db.Column(db.String(20), default='') # STOP, YES, CALL_ME (for inbound)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
company = db.relationship('Company', back_populates='sms_messages')
lead = db.relationship('AngiLead')
__table_args__ = (
db.CheckConstraint("direction IN ('outbound', 'inbound')", name='ck_sms_direction'),
db.CheckConstraint("status IN ('queued', 'sent', 'delivered', 'received', 'failed', 'rejected')", name='ck_sms_status'),
)
def to_dict(self):
return {
'id': self.id,
'company_id': self.company_id,
'connector_id': self.connector_id,
'lead_id': self.lead_id,
'from_phone': self.from_phone,
'to_phone': self.to_phone,
'direction': self.direction,
'body': self.body,
'status': self.status,
'error_code': self.error_code,
'error_message': self.error_message,
'provider_message_sid': self.provider_message_sid,
'keyword_flag': self.keyword_flag,
'created_at': self.created_at.isoformat() if self.created_at else None,
'updated_at': self.updated_at.isoformat() if self.updated_at else None,
}
def __repr__(self):
return f'<SmsMessage {self.direction} {self.status} to {self.to_phone}>'
class Call(db.Model):
"""Outbound/inbound calls with disposition tracking."""
__tablename__ = 'calls'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
connector_id = db.Column(db.String(36), nullable=False, index=True)
lead_id = db.Column(db.String(36), db.ForeignKey('angi_leads.id', ondelete='SET NULL'), nullable=True, index=True)
triggered_sms_id = db.Column(db.String(36), db.ForeignKey('sms_messages.id', ondelete='SET NULL'), nullable=True) # If triggered by CALL ME
from_phone = db.Column(db.String(50), nullable=False)
to_phone = db.Column(db.String(50), nullable=False)
direction = db.Column(db.String(10), nullable=False) # 'outbound'
disposition = db.Column(db.String(20), default='') # connected, voicemail, no_answer, busy, appointment_set, unknown
duration_seconds = db.Column(db.Integer, nullable=True)
recording_url = db.Column(db.String(500), default='') # Provider recording URL
provider_call_sid = db.Column(db.String(128), default='')
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
started_at = db.Column(db.DateTime, nullable=True)
ended_at = db.Column(db.DateTime, nullable=True)
company = db.relationship('Company', back_populates='calls')
lead = db.relationship('AngiLead')
__table_args__ = (
db.CheckConstraint("direction IN ('outbound', 'inbound')", name='ck_call_direction'),
db.CheckConstraint("disposition IN ('', 'connected', 'voicemail', 'no_answer', 'busy', 'appointment_set', 'unknown')", name='ck_call_disposition'),
)
def to_dict(self):
return {
'id': self.id,
'company_id': self.company_id,
'connector_id': self.connector_id,
'lead_id': self.lead_id,
'triggered_sms_id': self.triggered_sms_id,
'from_phone': self.from_phone,
'to_phone': self.to_phone,
'direction': self.direction,
'disposition': self.disposition,
'duration_seconds': self.duration_seconds,
'recording_url': self.recording_url,
'provider_call_sid': self.provider_call_sid,
'created_at': self.created_at.isoformat() if self.created_at else None,
'started_at': self.started_at.isoformat() if self.started_at else None,
'ended_at': self.ended_at.isoformat() if self.ended_at else None,
}
def __repr__(self):
return f'<Call {self.direction} {self.disposition} to {self.to_phone}>'
class LeadTouchpoint(db.Model):
"""Unified timeline entry — every SMS, call, email, or status change creates one."""
__tablename__ = 'lead_touchpoints'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
lead_id = db.Column(db.String(36), db.ForeignKey('angi_leads.id', ondelete='CASCADE'), nullable=True, index=True)
touchpoint_type = db.Column(db.String(20), nullable=False) # sms_sent, sms_received, sms_delivered, sms_failed, call_outbound, call_connected, call_voicemail, call_no_answer, email_sent, status_change, note_added
direction = db.Column(db.String(10), default='outbound') # outbound | inbound | system
content = db.Column(db.Text, default='') # Summary or message snippet
reference_id = db.Column(db.String(36), default='') # FK to SmsMessage.id, Call.id, etc.
reference_type = db.Column(db.String(20), default='') # sms_message, call, angi_lead_action
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
company = db.relationship('Company', back_populates='lead_touchpoints')
lead = db.relationship('AngiLead')
__table_args__ = (
db.CheckConstraint("touchpoint_type IN ('sms_sent', 'sms_received', 'sms_delivered', 'sms_failed', 'call_outbound', 'call_connected', 'call_voicemail', 'call_no_answer', 'email_sent', 'status_change', 'note_added')", name='ck_touchpoint_type'),
db.CheckConstraint("direction IN ('outbound', 'inbound', 'system')", name='ck_touchpoint_direction'),
)
def to_dict(self):
return {
'id': self.id,
'company_id': self.company_id,
'lead_id': self.lead_id,
'touchpoint_type': self.touchpoint_type,
'direction': self.direction,
'content': self.content,
'reference_id': self.reference_id,
'reference_type': self.reference_type,
'created_at': self.created_at.isoformat() if self.created_at else None,
}
def __repr__(self):
return f'<LeadTouchpoint {self.touchpoint_type} {self.direction}>'
class OptOut(db.Model):
"""Opt-out suppression list — TCPA compliance."""
__tablename__ = 'opt_outs'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
phone = db.Column(db.String(50), nullable=False, index=True)
reason = db.Column(db.String(50), default='stop') # stop, manual, bounced, complaint
source_message_id = db.Column(db.String(36), db.ForeignKey('sms_messages.id', ondelete='SET NULL'), nullable=True)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
company = db.relationship('Company', back_populates='opt_outs')
__table_args__ = (
db.UniqueConstraint('company_id', 'phone', name='uq_optout_company_phone'),
db.CheckConstraint("reason IN ('stop', 'manual', 'bounced', 'complaint')", name='ck_optout_reason'),
)
def to_dict(self):
return {
'id': self.id,
'company_id': self.company_id,
'phone': self.phone,
'reason': self.reason,
'source_message_id': self.source_message_id,
'created_at': self.created_at.isoformat() if self.created_at else None,
}
def __repr__(self):
return f'<OptOut {self.phone} ({self.reason})>'
class StripeWebhookEvent(db.Model):
"""Processed Stripe webhook event IDs — idempotency guard."""
__tablename__ = 'stripe_webhook_events'
id = db.Column(db.String(255), primary_key=True) # Stripe event.id (evt_...)
event_type = db.Column(db.String(100), default='')
processed_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
def __repr__(self):
return f'<StripeWebhookEvent {self.id} ({self.event_type})>'
class PartnerCommission(db.Model):
"""Commission agreement between a partner and a client company.
One row per partner↔company relationship with a revenue-share rate.
The revenue_source field indicates where the company's revenue data
comes from (self-reported, connector-synced, or forecast-based).
"""
__tablename__ = 'partner_commissions'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
partner_id = db.Column(db.String(36), db.ForeignKey('profiles.id', ondelete='CASCADE'), nullable=False, index=True)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
commission_rate = db.Column(db.Float, nullable=False) # e.g. 0.15 for 15%
revenue_source = db.Column(db.String(20), default='self_reported') # self_reported, connector, forecast
effective_date = db.Column(db.DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
terminated_date = db.Column(db.DateTime, nullable=True)
status = db.Column(db.String(20), default='active') # active, terminated, paused
metadata_json = db.Column(db.JSON, default=dict)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
__table_args__ = (
db.UniqueConstraint('partner_id', 'company_id', name='uq_partner_commissions_partner_company'),
db.CheckConstraint(
"revenue_source IN ('self_reported', 'connector', 'forecast')",
name='ck_partner_commissions_revenue_source',
),
db.CheckConstraint(
"status IN ('active', 'terminated', 'paused')",
name='ck_partner_commissions_status',
),
)
# Relationships
partner = db.relationship('User')
company = db.relationship('Company')
commission_records = db.relationship('CommissionRecord', back_populates='partner_commission', cascade='all, delete-orphan')
def __repr__(self):
return f'<PartnerCommission partner={self.partner_id} company={self.company_id} rate={self.commission_rate}>'
class CommissionRecord(db.Model):
"""Monthly commission calculation for a partner↔company agreement.
Revenue amount is sourced from whichever tracking method the company
uses — connector sync, self-reported KPI, or forecast estimate.
The revenue_confidence field tracks data quality.
"""
__tablename__ = 'commission_records'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
partner_commission_id = db.Column(db.String(36), db.ForeignKey('partner_commissions.id', ondelete='CASCADE'), nullable=False, index=True)
period = db.Column(db.String(7), nullable=False, index=True) # YYYY-MM format
period_start = db.Column(db.DateTime, nullable=False)
period_end = db.Column(db.DateTime, nullable=False)
revenue_amount = db.Column(db.Float, nullable=False, default=0.0)
revenue_confidence = db.Column(db.String(20), default='self_reported') # verified, self_reported, estimated
commission_rate = db.Column(db.Float, nullable=False) # Snapshot of rate at time of calculation
commission_amount = db.Column(db.Float, nullable=False, default=0.0)
status = db.Column(db.String(20), default='calculated') # calculated, approved, paid, disputed
payout_id = db.Column(db.String(36), db.ForeignKey('payouts.id', ondelete='SET NULL'), nullable=True, index=True)
notes = db.Column(db.Text, default='')
calculated_at = db.Column(db.DateTime, nullable=True)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
__table_args__ = (
db.UniqueConstraint('partner_commission_id', 'period', name='uq_commission_records_partner_period'),
db.CheckConstraint(
"revenue_confidence IN ('verified', 'self_reported', 'estimated')",
name='ck_commission_records_revenue_confidence',
),
db.CheckConstraint(
"status IN ('calculated', 'approved', 'paid', 'disputed')",
name='ck_commission_records_status',
),
)
# Relationships
partner_commission = db.relationship('PartnerCommission', back_populates='commission_records')
payout = db.relationship('Payout')
def __repr__(self):
return f'<CommissionRecord {self.period} ${self.commission_amount:.2f}>'
class Payout(db.Model):
"""Actual payment to a partner, aggregating multiple commission records.
Payment method is provider-agnostic — Stripe Connect, bank transfer,
check, etc.
"""
__tablename__ = 'payouts'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
partner_id = db.Column(db.String(36), db.ForeignKey('profiles.id', ondelete='CASCADE'), nullable=False, index=True)
amount = db.Column(db.Float, nullable=False, default=0.0)
currency = db.Column(db.String(3), default='USD')
status = db.Column(db.String(20), default='scheduled') # scheduled, processing, completed, failed
scheduled_date = db.Column(db.DateTime, nullable=False)
paid_date = db.Column(db.DateTime, nullable=True)
payment_method = db.Column(db.String(20), default='bank_transfer') # stripe_transfer, bank_transfer, check, crypto, other
transaction_id = db.Column(db.String(255), default='') # External reference (Stripe Transfer ID, etc.)
notes = db.Column(db.Text, default='')
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
__table_args__ = (
db.CheckConstraint(
"status IN ('scheduled', 'processing', 'completed', 'failed')",
name='ck_payouts_status',
),
db.CheckConstraint(
"payment_method IN ('stripe_transfer', 'bank_transfer', 'check', 'crypto', 'other')",
name='ck_payouts_payment_method',
),
)
# Relationships
partner = db.relationship('User')
payout_records = db.relationship('PayoutRecord', back_populates='payout', cascade='all, delete-orphan')
def __repr__(self):
return f'<Payout partner={self.partner_id} ${self.amount:.2f} {self.status}>'
class PayoutRecord(db.Model):
"""Junction table linking commission records to payouts.
One commission record can only belong to one payout.
"""
__tablename__ = 'payout_records'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
payout_id = db.Column(db.String(36), db.ForeignKey('payouts.id', ondelete='CASCADE'), nullable=False, index=True)
commission_record_id = db.Column(db.String(36), db.ForeignKey('commission_records.id', ondelete='CASCADE'), nullable=False, index=True)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
__table_args__ = (
db.UniqueConstraint('payout_id', 'commission_record_id', name='uq_payout_records_payout_commission'),
)
# Relationships
payout = db.relationship('Payout', back_populates='payout_records')
commission_record = db.relationship('CommissionRecord')
def __repr__(self):
return f'<PayoutRecord payout={self.payout_id} commission={self.commission_record_id}>'
# =============================================================================
# Jobber connector models — P2: remodeling sector gap plan
# =============================================================================
class JobberClient(db.Model):
"""Clients synced from Jobber."""
__tablename__ = 'jobber_clients'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
external_id = db.Column(db.String(128), nullable=False, index=True) # Jobber client ID
first_name = db.Column(db.String(255), default='')
last_name = db.Column(db.String(255), default='')
email = db.Column(db.String(255), default='')
phone = db.Column(db.String(50), default='')
company_name = db.Column(db.String(255), default='')
# Address
address_line1 = db.Column(db.String(500), default='')
address_line2 = db.Column(db.String(500), default='')
city = db.Column(db.String(100), default='')
state = db.Column(db.String(100), default='')
postal_code = db.Column(db.String(20), default='')
country = db.Column(db.String(100), default='US')
# Jobber-specific
jobber_timezone = db.Column(db.String(50), default='')
jobber_language = db.Column(db.String(10), default='en')
status = db.Column(db.String(20), default='active') # active, archived
properties_json = db.Column(db.JSON, default=dict)
# Timestamps
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
synced_at = db.Column(db.DateTime, nullable=True)
company = db.relationship('Company', backref='jobber_clients')
__table_args__ = (
db.UniqueConstraint('company_id', 'external_id', name='uq_jobber_client_company_external'),
)
@property
def full_name(self):
parts = [self.first_name, self.last_name]
return ' '.join(p for p in parts if p) or self.company_name
def to_dict(self):
return {
'id': self.id,
'company_id': self.company_id,
'external_id': self.external_id,
'first_name': self.first_name,
'last_name': self.last_name,
'full_name': self.full_name,
'email': self.email,
'phone': self.phone,
'company_name': self.company_name,
'address_line1': self.address_line1,
'address_line2': self.address_line2,
'city': self.city,
'state': self.state,
'postal_code': self.postal_code,
'country': self.country,
'status': self.status,
'properties_json': self.properties_json,
'created_at': self.created_at.isoformat() if self.created_at else None,
'updated_at': self.updated_at.isoformat() if self.updated_at else None,
'synced_at': self.synced_at.isoformat() if self.synced_at else None,
}
def __repr__(self):
return f'<JobberClient {self.full_name} ({self.external_id})>'
class JobberQuote(db.Model):
"""Quotes (estimates) synced from Jobber."""
__tablename__ = 'jobber_quotes'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
external_id = db.Column(db.String(128), nullable=False, index=True) # Jobber quote ID
client_id = db.Column(db.String(36), db.ForeignKey('jobber_clients.id', ondelete='SET NULL'), nullable=True, index=True)
# Quote details
name = db.Column(db.String(255), default='')
description = db.Column(db.Text, default='')
status = db.Column(db.String(50), default='draft') # draft, pending, accepted, rejected, converted
amount = db.Column(db.Float, nullable=True)
currency = db.Column(db.String(3), default='USD')
tax_amount = db.Column(db.Float, nullable=True)
discount_amount = db.Column(db.Float, nullable=True)
# Dates
sent_at = db.Column(db.DateTime, nullable=True)
accepted_at = db.Column(db.DateTime, nullable=True)
rejected_at = db.Column(db.DateTime, nullable=True)
expires_at = db.Column(db.DateTime, nullable=True)
# Linked to Estimate funnel
estimate_id = db.Column(db.String(36), db.ForeignKey('estimates.id', ondelete='SET NULL'), nullable=True)
# Raw data
line_items_json = db.Column(db.JSON, default=list)
properties_json = db.Column(db.JSON, default=dict)
# Timestamps
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
synced_at = db.Column(db.DateTime, nullable=True)
company = db.relationship('Company', backref='jobber_quotes')
client = db.relationship('JobberClient')
estimate = db.relationship('Estimate')
__table_args__ = (
db.UniqueConstraint('company_id', 'external_id', name='uq_jobber_quote_company_external'),
)
def to_dict(self):
return {
'id': self.id,
'company_id': self.company_id,
'external_id': self.external_id,
'client_id': self.client_id,
'name': self.name,
'description': self.description,
'status': self.status,
'amount': self.amount,
'currency': self.currency,
'tax_amount': self.tax_amount,
'discount_amount': self.discount_amount,
'sent_at': self.sent_at.isoformat() if self.sent_at else None,
'accepted_at': self.accepted_at.isoformat() if self.accepted_at else None,
'rejected_at': self.rejected_at.isoformat() if self.rejected_at else None,
'expires_at': self.expires_at.isoformat() if self.expires_at else None,
'estimate_id': self.estimate_id,
'line_items_json': self.line_items_json,
'properties_json': self.properties_json,
'created_at': self.created_at.isoformat() if self.created_at else None,
'updated_at': self.updated_at.isoformat() if self.updated_at else None,
'synced_at': self.synced_at.isoformat() if self.synced_at else None,
}
def __repr__(self):
return f'<JobberQuote {self.name} ${self.amount} {self.status}>'
class JobberJob(db.Model):
"""Jobs (projects) synced from Jobber."""
__tablename__ = 'jobber_jobs'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
external_id = db.Column(db.String(128), nullable=False, index=True) # Jobber job ID
client_id = db.Column(db.String(36), db.ForeignKey('jobber_clients.id', ondelete='SET NULL'), nullable=True, index=True)
# Job details
name = db.Column(db.String(255), default='')
description = db.Column(db.Text, default='')
status = db.Column(db.String(50), default='draft') # draft, scheduled, in_progress, completed, cancelled
project_type = db.Column(db.String(100), default='')
# Financials
estimated_revenue = db.Column(db.Float, nullable=True)
actual_revenue = db.Column(db.Float, nullable=True)
estimated_cost = db.Column(db.Float, nullable=True)
actual_cost = db.Column(db.Float, nullable=True)
# Schedule
start_date = db.Column(db.DateTime, nullable=True)
end_date = db.Column(db.DateTime, nullable=True)
estimated_end_date = db.Column(db.DateTime, nullable=True)
# Address
address_line1 = db.Column(db.String(500), default='')
address_line2 = db.Column(db.String(500), default='')
city = db.Column(db.String(100), default='')
state = db.Column(db.String(100), default='')
postal_code = db.Column(db.String(20), default='')
# Linked to Estimate funnel
estimate_id = db.Column(db.String(36), db.ForeignKey('estimates.id', ondelete='SET NULL'), nullable=True)
quote_id = db.Column(db.String(36), db.ForeignKey('jobber_quotes.id', ondelete='SET NULL'), nullable=True)
# Raw data
properties_json = db.Column(db.JSON, default=dict)
# Timestamps
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
synced_at = db.Column(db.DateTime, nullable=True)
company = db.relationship('Company', backref='jobber_jobs')
client = db.relationship('JobberClient')
estimate = db.relationship('Estimate')
quote = db.relationship('JobberQuote')
__table_args__ = (
db.UniqueConstraint('company_id', 'external_id', name='uq_jobber_job_company_external'),
)
@property
def margin(self) -> Optional[float]:
if self.actual_revenue and self.actual_cost and self.actual_revenue > 0:
return ((self.actual_revenue - self.actual_cost) / self.actual_revenue) * 100
if self.estimated_revenue and self.estimated_cost and self.estimated_revenue > 0:
return ((self.estimated_revenue - self.estimated_cost) / self.estimated_revenue) * 100
return None
@property
def schedule_variance_days(self) -> Optional[int]:
"""Days between estimated end and actual end. Positive = late."""
if self.estimated_end_date and self.end_date:
return (self.end_date - self.estimated_end_date).days
return None
def to_dict(self):
return {
'id': self.id,
'company_id': self.company_id,
'external_id': self.external_id,
'client_id': self.client_id,
'name': self.name,
'description': self.description,
'status': self.status,
'project_type': self.project_type,
'estimated_revenue': self.estimated_revenue,
'actual_revenue': self.actual_revenue,
'estimated_cost': self.estimated_cost,
'actual_cost': self.actual_cost,
'margin': self.margin,
'start_date': self.start_date.isoformat() if self.start_date else None,
'end_date': self.end_date.isoformat() if self.end_date else None,
'estimated_end_date': self.estimated_end_date.isoformat() if self.estimated_end_date else None,
'schedule_variance_days': self.schedule_variance_days,
'address_line1': self.address_line1,
'address_line2': self.address_line2,
'city': self.city,
'state': self.state,
'postal_code': self.postal_code,
'estimate_id': self.estimate_id,
'quote_id': self.quote_id,
'properties_json': self.properties_json,
'created_at': self.created_at.isoformat() if self.created_at else None,
'updated_at': self.updated_at.isoformat() if self.updated_at else None,
'synced_at': self.synced_at.isoformat() if self.synced_at else None,
}
def __repr__(self):
return f'<JobberJob {self.name} {self.status}>'
class JobberInvoice(db.Model):
"""Invoices synced from Jobber."""
__tablename__ = 'jobber_invoices'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
external_id = db.Column(db.String(128), nullable=False, index=True) # Jobber invoice ID
client_id = db.Column(db.String(36), db.ForeignKey('jobber_clients.id', ondelete='SET NULL'), nullable=True, index=True)
job_id = db.Column(db.String(36), db.ForeignKey('jobber_jobs.id', ondelete='SET NULL'), nullable=True, index=True)
# Invoice details
invoice_number = db.Column(db.String(100), default='')
status = db.Column(db.String(50), default='draft') # draft, sent, paid, overdue, cancelled
amount = db.Column(db.Float, nullable=True)
currency = db.Column(db.String(3), default='USD')
tax_amount = db.Column(db.Float, nullable=True)
paid_amount = db.Column(db.Float, nullable=True, default=0.0)
# Dates
issued_at = db.Column(db.DateTime, nullable=True)
due_date = db.Column(db.DateTime, nullable=True)
paid_at = db.Column(db.DateTime, nullable=True)
# Raw data
line_items_json = db.Column(db.JSON, default=list)
properties_json = db.Column(db.JSON, default=dict)
# Timestamps
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
synced_at = db.Column(db.DateTime, nullable=True)
company = db.relationship('Company', backref='jobber_invoices')
client = db.relationship('JobberClient')
job = db.relationship('JobberJob')
__table_args__ = (
db.UniqueConstraint('company_id', 'external_id', name='uq_jobber_invoice_company_external'),
)
@property
def balance(self) -> float:
return (self.amount or 0) - (self.paid_amount or 0)
@property
def is_overdue(self) -> bool:
if self.due_date and self.status not in ('paid', 'cancelled'):
from datetime import datetime, timezone
return datetime.now(timezone.utc) > self.due_date
return False
def to_dict(self):
return {
'id': self.id,
'company_id': self.company_id,
'external_id': self.external_id,
'client_id': self.client_id,
'job_id': self.job_id,
'invoice_number': self.invoice_number,
'status': self.status,
'amount': self.amount,
'currency': self.currency,
'tax_amount': self.tax_amount,
'paid_amount': self.paid_amount,
'balance': self.balance,
'is_overdue': self.is_overdue,
'issued_at': self.issued_at.isoformat() if self.issued_at else None,
'due_date': self.due_date.isoformat() if self.due_date else None,
'paid_at': self.paid_at.isoformat() if self.paid_at else None,
'line_items_json': self.line_items_json,
'properties_json': self.properties_json,
'created_at': self.created_at.isoformat() if self.created_at else None,
'updated_at': self.updated_at.isoformat() if self.updated_at else None,
'synced_at': self.synced_at.isoformat() if self.synced_at else None,
}
def __repr__(self):
return f'<JobberInvoice {self.invoice_number} ${self.amount} {self.status}>'
# =============================================================================
# Estimate models — estimate-to-close funnel tracking for remodeling sector
# =============================================================================
class Estimate(db.Model):
"""Estimate/quote records with funnel stage tracking.
Tracks an estimate from draft through delivery, acceptance/rejection,
scheduling, and final close. Supports multi-source origins (Angi leads,
ServiceTitan, Jobber, manual entry).
Funnel stages: draft → delivered → accepted | rejected | expired
Accepted estimates progress: accepted → scheduled → closed
"""
__tablename__ = 'estimates'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
# Human-readable reference number (auto-generated if not provided)
estimate_number = db.Column(db.String(50), nullable=False, index=True)
# Linkage to existing entities (all nullable — estimates can be standalone)
project_id = db.Column(db.String(36), db.ForeignKey('projects.id', ondelete='SET NULL'), nullable=True, index=True)
customer_id = db.Column(db.String(36), db.ForeignKey('quickbooks_customers.id', ondelete='SET NULL'), nullable=True, index=True)
source_lead_id = db.Column(db.String(36), db.ForeignKey('angi_leads.id', ondelete='SET NULL'), nullable=True, index=True)
# Estimate details
customer_name = db.Column(db.String(255), default='')
customer_email = db.Column(db.String(255), default='')
customer_phone = db.Column(db.String(50), default='')
project_type = db.Column(db.String(100), default='') # roofing, hvac, kitchen, bath, etc.
description = db.Column(db.Text, default='')
address = db.Column(db.String(500), default='')
city = db.Column(db.String(100), default='')
state = db.Column(db.String(50), default='')
zip_code = db.Column(db.String(20), default='')
# Financials
total_value = db.Column(db.Float, nullable=False) # Total estimate amount
estimated_cost = db.Column(db.Float, nullable=True) # Estimated cost to complete
estimated_margin = db.Column(db.Float, nullable=True) # Pre-computed margin %
deposit = db.Column(db.Float, default=0.0)
tax_amount = db.Column(db.Float, default=0.0)
discount_amount = db.Column(db.Float, default=0.0)
line_items_json = db.Column(db.JSON, default=list) # [{description, quantity, unit_price, total}]
# Funnel tracking
stage = db.Column(db.String(50), nullable=False, default='draft', index=True)
# Stage transition audit trail: [{stage, timestamp, notes}]
stage_history = db.Column(db.JSON, default=list)
# Timeline
delivered_at = db.Column(db.DateTime, nullable=True)
accepted_at = db.Column(db.DateTime, nullable=True)
rejected_at = db.Column(db.DateTime, nullable=True)
scheduled_date = db.Column(db.DateTime, nullable=True)
closed_at = db.Column(db.DateTime, nullable=True)
expires_at = db.Column(db.DateTime, nullable=True)
# Outcome details
rejection_reason = db.Column(db.String(200), default='') # price, competitor, timing, etc.
competitor_name = db.Column(db.String(255), default='')
competitor_price = db.Column(db.Float, nullable=True)
notes = db.Column(db.Text, default='')
# Assignment
assigned_to = db.Column(db.String(36), db.ForeignKey('profiles.id', ondelete='SET NULL'), nullable=True)
# Source metadata
source = db.Column(db.String(50), default='manual') # manual, angi, servicetitan, jobber, hubspot, email
source_estimate_id = db.Column(db.String(128), default='') # External system estimate ID
# Timestamps
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
# Relationships
company = db.relationship('Company', backref=db.backref('estimates', cascade='all, delete-orphan'))
project = db.relationship('Project')
customer = db.relationship('QuickbooksCustomer')
source_lead = db.relationship('AngiLead')
__table_args__ = (
db.UniqueConstraint('company_id', 'estimate_number', name='uq_estimate_company_number'),
db.CheckConstraint(
"stage IN ('draft', 'delivered', 'accepted', 'rejected', 'expired', 'scheduled', 'closed')",
name='ck_estimates_stage',
),
db.CheckConstraint(
"source IN ('manual', 'angi', 'servicetitan', 'jobber', 'hubspot', 'email', 'referral', 'website')",
name='ck_estimates_source',
),
)
@staticmethod
def _parse_history(raw):
"""Defensively parse stage_history from SQLite JSON column."""
if isinstance(raw, str):
import json
try:
raw = json.loads(raw)
except (json.JSONDecodeError, TypeError):
return []
if isinstance(raw, list):
return raw
return []
@property
def current_stage_timestamp(self):
"""Timestamp when the estimate entered its current stage (as datetime)."""
history = self._parse_history(self.stage_history)
if history:
ts = history[-1].get('timestamp')
if isinstance(ts, str):
try:
return datetime.fromisoformat(ts)
except (ValueError, TypeError):
return None
if isinstance(ts, datetime):
return ts
return None
@property
def days_in_stage(self):
"""Number of days in the current stage."""
ts = self.current_stage_timestamp
if ts and isinstance(ts, datetime):
now = datetime.now(timezone.utc)
if ts.tzinfo is not None:
ts = ts.replace(tzinfo=None)
if now.tzinfo is not None:
now = now.replace(tzinfo=None)
return (now - ts).days
return 0
@property
def time_to_accept_days(self):
"""Days from delivered to accepted (or None if not applicable)."""
if self.delivered_at and self.accepted_at:
return (self.accepted_at - self.delivered_at).days
return None
@property
def is_active(self):
"""Whether the estimate is in an active (non-terminal) stage."""
return self.stage in ('draft', 'delivered', 'accepted', 'scheduled')
@property
def is_closed(self):
"""Whether the estimate is in a terminal stage."""
return self.stage in ('rejected', 'expired', 'closed')
def transition_to(self, new_stage, notes=''):
"""Transition the estimate to a new stage with audit trail.
Returns True on success, False if the transition is invalid.
"""
valid_transitions = {
'draft': ['delivered'],
'delivered': ['accepted', 'rejected', 'expired'],
'accepted': ['scheduled', 'rejected'],
'scheduled': ['closed', 'rejected'],
'rejected': [],
'expired': [],
'closed': [],
}
if new_stage not in valid_transitions.get(self.stage, []):
return False
now = datetime.now(timezone.utc)
# Record transition
transition_record = {
'from_stage': self.stage,
'to_stage': new_stage,
'timestamp': now.isoformat(),
'notes': notes,
}
# Ensure stage_history is a list (SQLite may return JSON as string)
history = self._parse_history(self.stage_history)
history.append(transition_record)
self.stage_history = history
# Update stage and stage-specific timestamps
old_stage = self.stage
self.stage = new_stage
if new_stage == 'delivered' and not self.delivered_at:
self.delivered_at = now
elif new_stage == 'accepted' and not self.accepted_at:
self.accepted_at = now
elif new_stage == 'rejected':
self.rejected_at = now
elif new_stage == 'expired':
# Set expired_at via updated_at since we don't have a dedicated field
pass
elif new_stage == 'closed' and not self.closed_at:
self.closed_at = now
return True
def to_dict(self):
history = self._parse_history(self.stage_history)
return {
'id': self.id,
'company_id': self.company_id,
'estimate_number': self.estimate_number,
'project_id': self.project_id,
'customer_id': self.customer_id,
'source_lead_id': self.source_lead_id,
'customer_name': self.customer_name,
'customer_email': self.customer_email,
'customer_phone': self.customer_phone,
'project_type': self.project_type,
'description': self.description,
'address': self.address,
'city': self.city,
'state': self.state,
'zip_code': self.zip_code,
'total_value': self.total_value,
'deposit': self.deposit,
'tax_amount': self.tax_amount,
'discount_amount': self.discount_amount,
'line_items': self.line_items_json,
'stage': self.stage,
'stage_history': history,
'delivered_at': self.delivered_at.isoformat() if self.delivered_at else None,
'accepted_at': self.accepted_at.isoformat() if self.accepted_at else None,
'rejected_at': self.rejected_at.isoformat() if self.rejected_at else None,
'scheduled_date': self.scheduled_date.isoformat() if self.scheduled_date else None,
'closed_at': self.closed_at.isoformat() if self.closed_at else None,
'expires_at': self.expires_at.isoformat() if self.expires_at else None,
'rejection_reason': self.rejection_reason,
'competitor_name': self.competitor_name,
'competitor_price': self.competitor_price,
'notes': self.notes,
'assigned_to': self.assigned_to,
'source': self.source,
'source_estimate_id': self.source_estimate_id,
'days_in_stage': self.days_in_stage,
'time_to_accept_days': self.time_to_accept_days,
'is_active': self.is_active,
'is_closed': self.is_closed,
'created_at': self.created_at.isoformat() if self.created_at else None,
'updated_at': self.updated_at.isoformat() if self.updated_at else None,
}
def __repr__(self):
return f'<Estimate {self.estimate_number} ${self.total_value} {self.stage}>'