|
|
"""
|
|
|
Standardized Response Formatter for Kaanta AI.
|
|
|
|
|
|
Provides consistent output formats for WhatsApp, Web, and API responses.
|
|
|
Ensures all tax calculations include proper citations and validation.
|
|
|
"""
|
|
|
|
|
|
from dataclasses import dataclass, field
|
|
|
from typing import Dict, List, Optional, Any, Union
|
|
|
from datetime import date
|
|
|
from enum import Enum
|
|
|
import json
|
|
|
|
|
|
|
|
|
class OutputFormat(Enum):
|
|
|
"""Available output formats."""
|
|
|
WHATSAPP = "whatsapp"
|
|
|
WEB = "web"
|
|
|
API = "api"
|
|
|
REPORT = "report"
|
|
|
|
|
|
|
|
|
@dataclass
|
|
|
class LegalCitation:
|
|
|
"""Legal citation for tax calculations."""
|
|
|
document: str
|
|
|
section: Optional[str] = None
|
|
|
page: Optional[str] = None
|
|
|
|
|
|
def format(self) -> str:
|
|
|
parts = [self.document]
|
|
|
if self.section:
|
|
|
parts.append(f"s.{self.section}")
|
|
|
if self.page:
|
|
|
parts.append(f"p.{self.page}")
|
|
|
return ", ".join(parts)
|
|
|
|
|
|
|
|
|
@dataclass
|
|
|
class KeyPoint:
|
|
|
"""A key point in the response."""
|
|
|
text: str
|
|
|
citation: Optional[LegalCitation] = None
|
|
|
|
|
|
|
|
|
@dataclass
|
|
|
class ActionItem:
|
|
|
"""An action item for the user."""
|
|
|
action: str
|
|
|
priority: str = "normal"
|
|
|
deadline: Optional[str] = None
|
|
|
|
|
|
|
|
|
@dataclass
|
|
|
class StandardResponse:
|
|
|
"""
|
|
|
Standardized response structure for all Kaanta outputs.
|
|
|
|
|
|
This ensures consistency across WhatsApp, Web, and API.
|
|
|
"""
|
|
|
|
|
|
headline: str
|
|
|
summary: str
|
|
|
|
|
|
|
|
|
key_points: List[KeyPoint] = field(default_factory=list)
|
|
|
|
|
|
|
|
|
action_items: List[ActionItem] = field(default_factory=list)
|
|
|
|
|
|
|
|
|
data: Optional[Dict[str, Any]] = None
|
|
|
|
|
|
|
|
|
citations: List[LegalCitation] = field(default_factory=list)
|
|
|
|
|
|
|
|
|
calculation_date: date = field(default_factory=date.today)
|
|
|
regime: str = "NTA 2026"
|
|
|
confidence: float = 1.0
|
|
|
warnings: List[str] = field(default_factory=list)
|
|
|
|
|
|
|
|
|
class ResponseFormatter:
|
|
|
"""
|
|
|
Formats StandardResponse for different output targets.
|
|
|
"""
|
|
|
|
|
|
@staticmethod
|
|
|
def to_whatsapp(response: StandardResponse) -> str:
|
|
|
"""Format for WhatsApp (plain text, clean formatting)."""
|
|
|
lines = []
|
|
|
|
|
|
|
|
|
lines.append(f"*{response.headline}*")
|
|
|
lines.append("")
|
|
|
|
|
|
|
|
|
lines.append(response.summary)
|
|
|
lines.append("")
|
|
|
|
|
|
|
|
|
if response.key_points:
|
|
|
lines.append("*Key Points:*")
|
|
|
for point in response.key_points:
|
|
|
lines.append(f"- {point.text}")
|
|
|
lines.append("")
|
|
|
|
|
|
|
|
|
if response.action_items:
|
|
|
lines.append("*Next Steps:*")
|
|
|
for action in response.action_items:
|
|
|
priority_marker = "[!]" if action.priority == "high" else ""
|
|
|
lines.append(f"- {priority_marker} {action.action}")
|
|
|
lines.append("")
|
|
|
|
|
|
|
|
|
if response.citations:
|
|
|
citation_strs = [c.format() for c in response.citations]
|
|
|
lines.append(f"*Legal Basis:* {'; '.join(citation_strs)}")
|
|
|
lines.append("")
|
|
|
|
|
|
|
|
|
if response.warnings:
|
|
|
lines.append("*Notes:*")
|
|
|
for warning in response.warnings:
|
|
|
lines.append(f"- {warning}")
|
|
|
lines.append("")
|
|
|
|
|
|
|
|
|
lines.append("_Powered by Kaanta_")
|
|
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
@staticmethod
|
|
|
def to_web(response: StandardResponse) -> Dict[str, Any]:
|
|
|
"""Format for Web (structured JSON)."""
|
|
|
return {
|
|
|
"summary": {
|
|
|
"headline": response.headline,
|
|
|
"text": response.summary,
|
|
|
},
|
|
|
"key_points": [
|
|
|
{
|
|
|
"text": p.text,
|
|
|
"citation": p.citation.format() if p.citation else None
|
|
|
}
|
|
|
for p in response.key_points
|
|
|
],
|
|
|
"action_items": [
|
|
|
{
|
|
|
"action": a.action,
|
|
|
"priority": a.priority,
|
|
|
"deadline": a.deadline
|
|
|
}
|
|
|
for a in response.action_items
|
|
|
],
|
|
|
"data": response.data,
|
|
|
"legal": {
|
|
|
"regime": response.regime,
|
|
|
"citations": [c.format() for c in response.citations],
|
|
|
"calculation_date": response.calculation_date.isoformat()
|
|
|
},
|
|
|
"meta": {
|
|
|
"confidence": response.confidence,
|
|
|
"warnings": response.warnings
|
|
|
}
|
|
|
}
|
|
|
|
|
|
@staticmethod
|
|
|
def to_api(response: StandardResponse) -> Dict[str, Any]:
|
|
|
"""Format for API (complete JSON with all details)."""
|
|
|
return {
|
|
|
"status": "success",
|
|
|
"response": {
|
|
|
"headline": response.headline,
|
|
|
"summary": response.summary,
|
|
|
"key_points": [p.text for p in response.key_points],
|
|
|
"action_items": [
|
|
|
{"action": a.action, "priority": a.priority}
|
|
|
for a in response.action_items
|
|
|
],
|
|
|
"data": response.data,
|
|
|
},
|
|
|
"legal": {
|
|
|
"regime": response.regime,
|
|
|
"citations": [
|
|
|
{"document": c.document, "section": c.section, "page": c.page}
|
|
|
for c in response.citations
|
|
|
],
|
|
|
},
|
|
|
"meta": {
|
|
|
"calculation_date": response.calculation_date.isoformat(),
|
|
|
"confidence": response.confidence,
|
|
|
"warnings": response.warnings,
|
|
|
}
|
|
|
}
|
|
|
|
|
|
@staticmethod
|
|
|
def to_report(response: StandardResponse) -> str:
|
|
|
"""Format for PDF/Report (detailed plain text)."""
|
|
|
lines = []
|
|
|
|
|
|
lines.append("=" * 60)
|
|
|
lines.append(response.headline.upper())
|
|
|
lines.append(f"Calculated on: {response.calculation_date.isoformat()}")
|
|
|
lines.append(f"Tax Regime: {response.regime}")
|
|
|
lines.append("=" * 60)
|
|
|
lines.append("")
|
|
|
|
|
|
|
|
|
lines.append("SUMMARY")
|
|
|
lines.append("-" * 40)
|
|
|
lines.append(response.summary)
|
|
|
lines.append("")
|
|
|
|
|
|
|
|
|
if response.key_points:
|
|
|
lines.append("KEY POINTS")
|
|
|
lines.append("-" * 40)
|
|
|
for i, point in enumerate(response.key_points, 1):
|
|
|
lines.append(f"{i}. {point.text}")
|
|
|
if point.citation:
|
|
|
lines.append(f" Reference: {point.citation.format()}")
|
|
|
lines.append("")
|
|
|
|
|
|
|
|
|
if response.action_items:
|
|
|
lines.append("RECOMMENDED ACTIONS")
|
|
|
lines.append("-" * 40)
|
|
|
for i, action in enumerate(response.action_items, 1):
|
|
|
priority_label = f"[{action.priority.upper()}]" if action.priority != "normal" else ""
|
|
|
lines.append(f"{i}. {priority_label} {action.action}")
|
|
|
lines.append("")
|
|
|
|
|
|
|
|
|
lines.append("LEGAL BASIS")
|
|
|
lines.append("-" * 40)
|
|
|
for citation in response.citations:
|
|
|
lines.append(f"- {citation.format()}")
|
|
|
lines.append("")
|
|
|
|
|
|
|
|
|
if response.warnings:
|
|
|
lines.append("IMPORTANT NOTES")
|
|
|
lines.append("-" * 40)
|
|
|
for warning in response.warnings:
|
|
|
lines.append(f"* {warning}")
|
|
|
lines.append("")
|
|
|
|
|
|
lines.append("=" * 60)
|
|
|
lines.append("Prepared by Kaanta AI - Nigerian Tax Assistant")
|
|
|
lines.append("=" * 60)
|
|
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
@classmethod
|
|
|
def format(cls, response: StandardResponse, output_format: OutputFormat) -> Union[str, Dict]:
|
|
|
"""Format response to specified output format."""
|
|
|
formatters = {
|
|
|
OutputFormat.WHATSAPP: cls.to_whatsapp,
|
|
|
OutputFormat.WEB: cls.to_web,
|
|
|
OutputFormat.API: cls.to_api,
|
|
|
OutputFormat.REPORT: cls.to_report,
|
|
|
}
|
|
|
return formatters[output_format](response)
|
|
|
|
|
|
|
|
|
def create_tax_calculation_response(
|
|
|
monthly_tax: float,
|
|
|
monthly_income: float,
|
|
|
monthly_net: float,
|
|
|
effective_rate: float,
|
|
|
deductions: Dict[str, float],
|
|
|
bands: List[Dict[str, Any]],
|
|
|
regime: str = "NTA 2026",
|
|
|
citations: List[str] = None
|
|
|
) -> StandardResponse:
|
|
|
"""
|
|
|
Create a standardized response for tax calculations.
|
|
|
|
|
|
Helper function for common tax calculation outputs.
|
|
|
"""
|
|
|
headline = f"Tax: N{monthly_tax:,.0f}/month on N{monthly_income:,.0f} income"
|
|
|
|
|
|
summary = (
|
|
|
f"Your monthly tax is N{monthly_tax:,.2f} on a gross income of N{monthly_income:,.2f}. "
|
|
|
f"After tax and statutory deductions, your take-home pay is N{monthly_net:,.2f}. "
|
|
|
f"Your effective tax rate is {effective_rate:.1f}%."
|
|
|
)
|
|
|
|
|
|
key_points = [
|
|
|
KeyPoint(text=f"First N800,000 annually is tax-free under {regime}"),
|
|
|
KeyPoint(text=f"Pension contribution (8%) is deducted: N{deductions.get('pension', 0):,.0f}"),
|
|
|
]
|
|
|
|
|
|
if deductions.get('rent_relief', 0) > 0:
|
|
|
key_points.append(
|
|
|
KeyPoint(text=f"Rent relief applied: N{deductions['rent_relief']:,.0f}")
|
|
|
)
|
|
|
|
|
|
action_items = [
|
|
|
ActionItem(action="Verify your payslip shows correct deductions", priority="high"),
|
|
|
ActionItem(action="Keep records for annual tax filing"),
|
|
|
]
|
|
|
|
|
|
legal_citations = [
|
|
|
LegalCitation(document=citation) for citation in (citations or [regime])
|
|
|
]
|
|
|
|
|
|
data = {
|
|
|
"income": {"monthly": monthly_income, "annual": monthly_income * 12},
|
|
|
"tax": {"monthly": monthly_tax, "annual": monthly_tax * 12},
|
|
|
"net": {"monthly": monthly_net, "annual": monthly_net * 12},
|
|
|
"effective_rate": effective_rate,
|
|
|
"deductions": deductions,
|
|
|
"bands": bands,
|
|
|
}
|
|
|
|
|
|
return StandardResponse(
|
|
|
headline=headline,
|
|
|
summary=summary,
|
|
|
key_points=key_points,
|
|
|
action_items=action_items,
|
|
|
data=data,
|
|
|
citations=legal_citations,
|
|
|
regime=regime,
|
|
|
)
|
|
|
|
