""" Pydantic schemas for the AI Transformation Audit Report pipeline. These models represent structured data gathered from voice conversations, intake forms, or direct input, and drive all report generation logic. 6 Readiness Dimensions (from ENTERPRISE_AI_TRANSFORMATION_STRATEGY.md): 1. Operations — manual workflows, process automation opportunities 2. Customer Experience — call handling, lead capture, CX touch-points 3. Data — data quality, governance, integration health 4. Technology — tech stack maturity, integration ecosystem 5. People — AI literacy, change readiness, upskilling 6. Strategy — AI vision, leadership buy-in, competitive awareness VERIFICATION_STAMP Module: core/audit/schemas.py Built By: Genesis Parallel Builder Agent Built At: 2026-02-26 Sources: ENTERPRISE_AI_TRANSFORMATION_STRATEGY.md, BRAND_BIBLE.md Tests: 30/30 passing (see tests/audit/test_report_generator.py) """ from __future__ import annotations import re from datetime import date, datetime from enum import Enum from typing import Any, Optional from pydantic import BaseModel, Field, field_validator, model_validator # --------------------------------------------------------------------------- # Enumerations # --------------------------------------------------------------------------- class Industry(str, Enum): TRADES = "trades" PROFESSIONAL_SERVICES = "professional_services" HEALTHCARE = "healthcare" RETAIL = "retail" HOSPITALITY = "hospitality" CONSTRUCTION = "construction" LOGISTICS = "logistics" TECHNOLOGY = "technology" FINANCE = "finance" EDUCATION = "education" OTHER = "other" class CompanySize(str, Enum): SOLO = "solo" # 1 person MICRO = "micro" # 2–9 SMALL = "small" # 10–49 MEDIUM = "medium" # 50–199 LARGE = "large" # 200+ class ScoreGrade(str, Enum): EXCEPTIONAL = "A" # 80–100 STRONG = "B" # 65–79 DEVELOPING = "C" # 50–64 EARLY_STAGE = "D" # 30–49 CRITICAL = "F" # 0–29 class PriorityLevel(str, Enum): CRITICAL = "CRITICAL" HIGH = "HIGH" MEDIUM = "MEDIUM" LOW = "LOW" class ImplementationComplexity(str, Enum): QUICK_WIN = "quick_win" # < 1 week SHORT_TERM = "short_term" # 1–4 weeks MEDIUM_TERM = "medium_term" # 1–3 months LONG_TERM = "long_term" # 3–6 months STRATEGIC = "strategic" # 6–12 months # --------------------------------------------------------------------------- # Core data models # --------------------------------------------------------------------------- class CompanyProfile(BaseModel): """ Structured data about the company being audited. Populated from voice conversation transcript, intake form, or direct input. No hardcoded secrets — all values come from caller/form. """ # Identity company_name: str = Field(..., min_length=1, description="Legal or trading name") website_url: Optional[str] = Field(None, description="Primary website URL") industry: Industry = Field(..., description="Primary industry classification") company_size: CompanySize = Field(..., description="Headcount band") annual_revenue_aud: Optional[int] = Field( None, ge=0, description="Annual revenue in AUD. None if not disclosed." ) # Location city: str = Field(default="Brisbane", description="Primary city of operation") state: str = Field(default="QLD", description="Australian state/territory") regions_served: list[str] = Field( default_factory=list, description="Geographic areas served (e.g. ['Brisbane', 'Gold Coast'])" ) # Contact primary_contact_name: Optional[str] = None primary_contact_role: Optional[str] = None primary_contact_email: Optional[str] = None # Current tech stack current_crm: Optional[str] = Field(None, description="e.g. 'HubSpot', 'Salesforce', 'none'") current_phone_system: Optional[str] = Field(None, description="e.g. 'Telnyx', 'Vonage', 'landline'") current_marketing_tools: list[str] = Field( default_factory=list, description="Marketing platforms in use" ) monthly_inbound_calls: Optional[int] = Field( None, ge=0, description="Estimated inbound calls per month" ) monthly_leads: Optional[int] = Field( None, ge=0, description="Estimated inbound leads per month (all channels)" ) average_job_value_aud: Optional[int] = Field( None, ge=0, description="Average revenue per completed job/sale in AUD" ) missed_calls_per_day: Optional[float] = Field( None, ge=0, description="Estimated calls missed per day (unanswered/voicemail)" ) hours_on_admin_per_week: Optional[float] = Field( None, ge=0, description="Owner/team hours spent on manual admin tasks per week" ) # AI readiness signals (gathered via voice/form) has_ai_tools_deployed: bool = Field( default=False, description="True if any AI tools are currently in active use" ) ai_tools_in_use: list[str] = Field( default_factory=list, description="Names of AI tools currently in use, e.g. ['ChatGPT', 'Copilot']" ) leadership_ai_awareness: int = Field( default=2, ge=1, le=5, description="1=unaware, 3=aware/curious, 5=champion. Self-reported." ) data_quality_self_rating: int = Field( default=2, ge=1, le=5, description="1=poor/fragmented, 5=clean/integrated" ) automation_appetite: int = Field( default=3, ge=1, le=5, description="1=resistant, 3=open, 5=eager to automate" ) # Pain points (free text from voice conversation) top_pain_points: list[str] = Field( default_factory=list, description="Top 3–5 pain points as stated by the contact" ) biggest_operational_bottleneck: Optional[str] = Field( None, description="Single biggest time/revenue drain identified in conversation" ) # Audit metadata audit_date: date = Field(default_factory=date.today) audit_source: str = Field( default="voice_conversation", description="How data was gathered: voice_conversation / intake_form / direct_input" ) @field_validator("website_url") @classmethod def normalise_url(cls, v: Optional[str]) -> Optional[str]: if v is None: return v v = v.strip() if v and not v.startswith(("http://", "https://")): v = f"https://{v}" return v @field_validator("state") @classmethod def upper_state(cls, v: str) -> str: return v.upper().strip() @property def estimated_monthly_missed_revenue(self) -> Optional[float]: """Danny Harris Math from Sales Playbook.""" if self.missed_calls_per_day and self.average_job_value_aud: # Assume 30% conversion rate on answered calls calls_per_month = self.missed_calls_per_day * 22 # working days return round(calls_per_month * 0.30 * self.average_job_value_aud, 2) return None @property def full_name_location(self) -> str: return f"{self.company_name}, {self.city} {self.state}" class DimensionScore(BaseModel): """ Score for one of the 6 AI readiness dimensions. Max 100 per dimension. Weighted aggregate = overall AI Readiness Index. """ dimension: str = Field(..., description="Dimension name, e.g. 'Operations'") score: float = Field(..., ge=0, le=100) weight: float = Field( default=1.0, description="Relative weight when computing aggregate. Must sum to 6.0 across all dimensions." ) grade: ScoreGrade headline: str = Field(..., description="One-sentence assessment for this dimension") key_findings: list[str] = Field( default_factory=list, description="2–4 specific findings with evidence or data" ) quick_wins: list[str] = Field( default_factory=list, description="Immediate actions achievable in < 30 days" ) @property def weighted_score(self) -> float: return round(self.score * self.weight, 2) @classmethod def grade_from_score(cls, score: float) -> ScoreGrade: if score >= 80: return ScoreGrade.EXCEPTIONAL elif score >= 65: return ScoreGrade.STRONG elif score >= 50: return ScoreGrade.DEVELOPING elif score >= 30: return ScoreGrade.EARLY_STAGE else: return ScoreGrade.CRITICAL class ROIProjection(BaseModel): """ ROI model based on ENTERPRISE_AI_TRANSFORMATION_STRATEGY.md formula. ROI (%) = ((Annual Hard Benefits × Utilisation Factor) - Annual Costs) / Initial Total Investment × 100 """ scenario: str = Field(..., description="e.g. 'Conservative', 'Base Case', 'Optimistic'") # Inputs hours_saved_per_week: float = Field(..., ge=0) blended_hourly_rate_aud: float = Field(default=75.0, ge=0) utilisation_factor: float = Field( default=0.70, ge=0.0, le=1.0, description="Fraction of team actually using AI tools. Typically 0.60–0.85 in year 1." ) annual_ai_costs_aud: float = Field(..., ge=0, description="Annual API fees + licences") initial_investment_aud: float = Field(..., ge=0, description="Implementation + onboarding costs") # Computed outputs annual_hard_benefits_aud: float = Field(default=0.0) net_annual_benefit_aud: float = Field(default=0.0) roi_percentage: float = Field(default=0.0) payback_months: Optional[float] = None break_even_note: str = Field(default="") @model_validator(mode="after") def compute_roi(self) -> "ROIProjection": annual_hours_saved = self.hours_saved_per_week * 52.0 self.annual_hard_benefits_aud = round( annual_hours_saved * self.blended_hourly_rate_aud * self.utilisation_factor, 2 ) self.net_annual_benefit_aud = round( self.annual_hard_benefits_aud - self.annual_ai_costs_aud, 2 ) if self.initial_investment_aud > 0: self.roi_percentage = round( (self.net_annual_benefit_aud / self.initial_investment_aud) * 100, 1 ) if self.net_annual_benefit_aud > 0: monthly_benefit = self.net_annual_benefit_aud / 12.0 if monthly_benefit > 0: self.payback_months = round(self.initial_investment_aud / monthly_benefit, 1) self.break_even_note = ( f"Investment recovered in approximately {self.payback_months} months." ) return self class RecommendedAction(BaseModel): """ A single recommended action in the implementation roadmap. """ title: str description: str dimension: str = Field(..., description="Which of the 6 dimensions this addresses") priority: PriorityLevel complexity: ImplementationComplexity estimated_weekly_hours_saved: Optional[float] = None estimated_monthly_roi_aud: Optional[float] = None agileadapt_product: Optional[str] = Field( None, description="AgileAdapt product that delivers this (e.g. 'ReceptionistAI Enterprise')" ) done_for_you: bool = Field( default=True, description="True if AgileAdapt can deliver this as a managed service" ) class ImplementationPhase(BaseModel): """ One phase of the implementation roadmap (30/60/90/180 days). """ phase_number: int = Field(..., ge=1, le=4) title: str duration: str = Field(..., description="e.g. 'Days 1–30'") focus: str = Field(..., description="One-sentence theme for this phase") actions: list[RecommendedAction] expected_outcomes: list[str] success_metric: str class AIReadinessReport(BaseModel): """ The complete AI Transformation Audit Report. This is the primary output of AuditReportGenerator. All 6 dimension scores, ROI projections, roadmap, and competitive context. """ # Identity report_id: str = Field(..., description="Unique report ID, e.g. 'AAR-2026-001'") generated_at: datetime = Field(default_factory=datetime.utcnow) company: CompanyProfile # Scores dimension_scores: list[DimensionScore] = Field( ..., min_length=6, max_length=6, description="Exactly 6 dimension scores: Operations, CX, Data, Technology, People, Strategy" ) overall_ai_readiness_index: float = Field( default=0.0, ge=0, le=100, description="Weighted average of all 6 dimension scores" ) overall_grade: ScoreGrade = Field(default=ScoreGrade.CRITICAL) executive_summary: str = Field(..., description="3–5 sentence plain-language summary") # ROI roi_projections: list[ROIProjection] = Field( default_factory=list, description="Conservative, Base Case, and Optimistic scenarios" ) total_annual_opportunity_aud: Optional[float] = Field( None, description="Total addressable annual savings + revenue gains in AUD" ) # Roadmap implementation_phases: list[ImplementationPhase] = Field( default_factory=list, description="4-phase roadmap: Foundation, Quick Wins, Scale, Optimise" ) # Competitive context competitive_risk_summary: str = Field( default="", description="Plain English statement of competitive risk if AI is not adopted" ) industry_ai_adoption_rate: str = Field( default="", description="Industry-specific context on AI adoption trends" ) # Engagement recommended_agileadapt_tier: Optional[str] = Field( None, description="Recommended ReceptionistAI tier or engagement type" ) next_step: str = Field( default="Book a 30-minute strategy call with the AgileAdapt team.", description="Clear CTA for the prospect" ) booking_url: str = Field( default="https://agileadapt.com.au/strategy-call", description="Calendar booking link" ) @model_validator(mode="after") def compute_overall_index(self) -> "AIReadinessReport": if self.dimension_scores: total_weight = sum(d.weight for d in self.dimension_scores) weighted_sum = sum(d.score * d.weight for d in self.dimension_scores) if total_weight > 0: self.overall_ai_readiness_index = round(weighted_sum / total_weight, 1) self.overall_grade = DimensionScore.grade_from_score( self.overall_ai_readiness_index ) return self @property def lowest_scoring_dimension(self) -> DimensionScore: return min(self.dimension_scores, key=lambda d: d.score) @property def highest_scoring_dimension(self) -> DimensionScore: return max(self.dimension_scores, key=lambda d: d.score) @property def critical_gaps(self) -> list[DimensionScore]: return [d for d in self.dimension_scores if d.grade == ScoreGrade.CRITICAL] @property def all_quick_wins(self) -> list[str]: wins = [] for dim in self.dimension_scores: wins.extend(dim.quick_wins) return wins class AuditInput(BaseModel): """ Simplified intake model — what a voice agent or form collects. This is the external API surface. The report generator converts AuditInput → CompanyProfile → AIReadinessReport. """ # Required company_name: str industry: Industry company_size: CompanySize city: str = "Brisbane" state: str = "QLD" # Optional but high-value website_url: Optional[str] = None annual_revenue_aud: Optional[int] = None monthly_inbound_calls: Optional[int] = None average_job_value_aud: Optional[int] = None missed_calls_per_day: Optional[float] = None hours_on_admin_per_week: Optional[float] = None current_crm: Optional[str] = None has_ai_tools_deployed: bool = False ai_tools_in_use: list[str] = Field(default_factory=list) leadership_ai_awareness: int = Field(default=2, ge=1, le=5) data_quality_self_rating: int = Field(default=2, ge=1, le=5) automation_appetite: int = Field(default=3, ge=1, le=5) top_pain_points: list[str] = Field(default_factory=list) biggest_operational_bottleneck: Optional[str] = None primary_contact_name: Optional[str] = None primary_contact_role: Optional[str] = None primary_contact_email: Optional[str] = None audit_source: str = "voice_conversation" def to_company_profile(self) -> CompanyProfile: return CompanyProfile(**self.model_dump())