Files
roa2web-service-auto/backend/modules/telegram/bot/formatters.py
Claude Agent aac49542e4 fix(telegram): edit messages in-place during server switch flow
Server switch flow now edits the original message at each step
(progress, error, company list) instead of sending separate messages.
"Server schimbat" notice is folded into the company selection header.
Also adds budget debt breakdown to dashboard formatter.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-02-23 09:18:59 +00:00

700 lines
25 KiB
Python

"""
Response formatters for bot commands.
Formats API responses into user-friendly Telegram messages.
"""
from typing import Dict, List, Any
def format_dashboard_response(data: Dict[str, Any], company_name: str = None) -> str:
"""
Format dashboard data for Telegram (content only, no header).
Note: company_name parameter kept for backwards compatibility but not used.
Use format_response_with_company() in handlers to add company header.
"""
text = ""
# Sold total trezorerie (casa + banca) - rotunjit la leu
treasury_totals = data.get('treasury_totals_by_currency', {})
sold_trezorerie = round(float(treasury_totals.get('RON', 0)))
text += f"**Sold Trezorerie:** {sold_trezorerie:,} RON\n\n"
# Sold Clienți - rotunjit la leu
clienti_sold = round(float(data.get('clienti_sold_total', 0)))
clienti_in_termen = round(float(data.get('clienti_sold_in_termen', 0)))
clienti_restant = round(float(data.get('clienti_sold_restant', 0)))
text += f"**Sold Clienți:** {clienti_sold:,} RON\n"
text += f" - În termen: {clienti_in_termen:,} RON\n"
text += f" - Restanță: {clienti_restant:,} RON\n\n"
# Sold Furnizori BRUT (pentru consistență cu detaliile) - rotunjit la leu
furnizori_in_termen = round(float(data.get('furnizori_sold_in_termen', 0)))
furnizori_restant = round(float(data.get('furnizori_sold_restant', 0)))
furnizori_sold_brut = furnizori_in_termen + furnizori_restant
furnizori_avansuri = round(float(data.get('furnizori_avansuri', 0)))
furnizori_sold_net = round(float(data.get('furnizori_sold_total', 0)))
text += f"**Sold Furnizori:** {furnizori_sold_brut:,} RON\n"
text += f" - În termen: {furnizori_in_termen:,} RON\n"
text += f" - Restanță: {furnizori_restant:,} RON\n"
if furnizori_avansuri != 0:
text += f" - Avansuri: {furnizori_avansuri:,} RON\n"
text += f" - Net (după avansuri): {furnizori_sold_net:,} RON"
else:
text += f" - Net: {furnizori_sold_net:,} RON"
# Solduri TVA - rotunjit la leu
tva_plata_prec = round(float(data.get('tva_plata_precedent', 0)))
tva_recup_prec = round(float(data.get('tva_recuperat_precedent', 0)))
tva_plata_cur = round(float(data.get('tva_plata_curent', 0)))
tva_recup_cur = round(float(data.get('tva_recuperat_curent', 0)))
# Afișează secțiunea doar dacă există cel puțin o valoare > 0
if tva_plata_prec > 0 or tva_recup_prec > 0 or tva_plata_cur > 0 or tva_recup_cur > 0:
text += "\n\n**Solduri TVA:**\n"
if tva_plata_prec > 0:
text += f" - TVA de plată precedent: {tva_plata_prec:,} RON\n"
if tva_recup_prec > 0:
text += f" - TVA de recuperat precedent: {tva_recup_prec:,} RON\n"
if tva_plata_cur > 0:
text += f" - TVA de plată curent: {tva_plata_cur:,} RON\n"
if tva_recup_cur > 0:
text += f" - TVA de recuperat curent: {tva_recup_cur:,} RON\n"
# Datorii la Buget - breakdown pe grupe (TVA / BASS / CAM), valori luna precedentă
budget_breakdown = data.get('budget_debt_breakdown', [])
if budget_breakdown:
grupe_cu_datorie = [
g for g in budget_breakdown
if round(float(g.get('precedent', 0))) > 0
]
if grupe_cu_datorie:
total_buget = sum(round(float(g.get('precedent', 0))) for g in grupe_cu_datorie)
text += "\n\n**Datorii la Buget:**\n"
for grupa in grupe_cu_datorie:
val = round(float(grupa.get('precedent', 0)))
text += f" - {grupa.get('label', '')}: {val:,} RON\n"
text += f" Total: {total_buget:,} RON\n"
return text
def format_invoices_response(
invoices: List[Dict[str, Any]],
company_name: str = None,
limit: int = 10
) -> str:
"""
Format invoices list for Telegram - COMPACT TABLE FORMAT.
Args:
invoices: List of invoice dicts
company_name: Company name (kept for compatibility, not used)
limit: Maximum number of invoices to display
Returns:
Formatted Markdown string for Telegram (compact, no emojis)
"""
if not invoices:
return "Nu s-au gasit facturi cu aceste criterii."
# Header (o singură dată)
text = f"**Facturi** ({len(invoices)} total)\n\n"
text += "Nr | Client | Suma | Status\n"
text += "---|--------|------|-------\n"
# Lista facturi - compact, o linie per factură
for idx, inv in enumerate(invoices[:limit], 1):
seria = inv.get('seria', '')
numar = inv.get('numar', '')
client = inv.get('client', 'N/A')
suma = inv.get('suma_totala', 0)
status = inv.get('status', 'N/A')
# Truncate long client names for compact display
client_short = client[:20] + "..." if len(client) > 20 else client
# Status marker (no emoji)
status_marker = "PLATIT" if status == "platit" else "NEPLATIT"
text += f"{seria}{numar} | {client_short} | {suma:,.0f} | {status_marker}\n"
if len(invoices) > limit:
text += f"\n+{len(invoices) - limit} facturi"
return text
# =========================================================================
# FAZA 2: New Formatter Functions for Button Interface
# =========================================================================
def format_treasury_casa_response(data: Dict[str, Any], company_name: str = None) -> str:
"""
Format treasury CASH data for Telegram (content only, no header).
Args:
data: Dict with casa accounts and total from treasury breakdown
company_name: Company name (kept for compatibility, not used)
Returns:
Formatted Markdown string for Telegram
Example:
data = {'accounts': [...], 'total': 5000}
text = format_treasury_casa_response(data)
"""
text = ""
# Total cash balance - rotunjit la leu (0 zecimale)
total_cash = round(data.get('total', 0))
text += f"**Sold Total Cash:** {total_cash:,} RON\n\n"
# Cash accounts
casa_accounts = data.get('accounts', [])
if casa_accounts:
text += "**Conturi de Casa:**\n"
for acc in casa_accounts: # Show all accounts
name = acc.get('name', 'N/A')
balance = round(acc.get('balance', 0))
text += f" - {name}: {balance:,} RON\n"
else:
text += "Nu exista conturi de casa configurate."
return text
def format_treasury_banca_response(data: Dict[str, Any], company_name: str = None) -> str:
"""
Format treasury BANK data for Telegram (content only, no header).
Args:
data: Dict with banca accounts and total from treasury breakdown
company_name: Company name (kept for compatibility, not used)
Returns:
Formatted Markdown string for Telegram
Example:
data = {'accounts': [...], 'total': 15000}
text = format_treasury_banca_response(data)
"""
text = ""
# Total bank balance - rotunjit la leu (0 zecimale)
total_bank = round(data.get('total', 0))
text += f"**Sold Total Banca:** {total_bank:,} RON\n\n"
# Bank accounts
bank_accounts = data.get('accounts', [])
if bank_accounts:
text += "**Conturi Bancare:**\n"
for acc in bank_accounts: # Show all accounts
name = acc.get('name', 'N/A')
balance = round(acc.get('balance', 0))
text += f" - {name}: {balance:,} RON\n"
else:
text += "Nu exista conturi bancare configurate."
return text
def format_treasury_combined_response(data: Dict[str, Any], company_name: str = None) -> str:
"""
Format combined treasury data (Casa + Banca) for Telegram.
Shows grand total, Casa section with accounts, and Banca section with accounts
in a single unified message. Compact format without section titles.
Args:
data: Dict with 'casa' and 'banca' keys from get_treasury_breakdown_split()
company_name: Company name (kept for compatibility, not used)
Returns:
Formatted Markdown string with grand total and both sections
Example:
data = {'casa': {...}, 'banca': {...}}
text = format_treasury_combined_response(data)
"""
def format_amount(amount: int) -> str:
"""Format amount with period as thousands separator (Romanian style)."""
return f"{amount:,}".replace(",", ".")
text = ""
# Extract totals - rounded to whole RON
casa_total = round(data.get('casa', {}).get('total', 0))
banca_total = round(data.get('banca', {}).get('total', 0))
grand_total = casa_total + banca_total
# Grand total header
text += f"**Sold Trezorerie:** {format_amount(grand_total)} RON\n\n"
# Casa section - compact
text += f"**Casa:** {format_amount(casa_total)} RON\n"
casa_accounts = data.get('casa', {}).get('accounts', [])
if casa_accounts:
for acc in casa_accounts:
name = acc.get('name', 'N/A')
balance = round(acc.get('balance', 0))
text += f" - {name}: {format_amount(balance)} RON\n"
text += "\n"
# Banca section - compact
text += f"**Banca:** {format_amount(banca_total)} RON\n"
banca_accounts = data.get('banca', {}).get('accounts', [])
if banca_accounts:
for acc in banca_accounts:
name = acc.get('name', 'N/A')
balance = round(acc.get('balance', 0))
text += f" - {name}: {format_amount(balance)} RON\n"
return text
def format_clients_balance_response(
clients: List[Dict[str, Any]],
maturity_data: Dict[str, Any],
company_name: str = None
) -> str:
"""
Format clients balance with maturity breakdown (content only, no header).
Args:
clients: List of client dicts with id, name, balance
maturity_data: Dict with in_term, overdue, total
company_name: Company name (kept for compatibility, not used)
Returns:
Formatted Markdown string for Telegram
Example:
clients = [{'id': 1, 'name': 'Client A', 'balance': 15000}]
maturity = {'in_term': 10000, 'overdue': 5000, 'total': 15000}
text = format_clients_balance_response(clients, maturity)
"""
text = ""
# Maturity breakdown - rotunjit la leu (0 zecimale)
total = round(maturity_data.get('total', 0))
in_term = round(maturity_data.get('in_term', 0))
overdue = round(maturity_data.get('overdue', 0))
text += f"**Sold Total:** {total:,} RON\n\n"
text += "**Defalcare:**\n"
text += f" - In termen: {in_term:,} RON\n"
text += f" - Restanta: {overdue:,} RON\n\n"
# Top 10 clients
if clients:
text += f"**Top 10 Clienti** ({len(clients)} total):\n"
# Sort by balance descending
sorted_clients = sorted(
clients,
key=lambda x: x.get('balance', 0),
reverse=True
)
for idx, client in enumerate(sorted_clients[:10], 1):
name = client.get('name', 'N/A')
balance = round(client.get('balance', 0))
text += f"{idx}. {name}: {balance:,} RON\n"
if len(clients) > 10:
text += f"\nApasa butonul pentru lista completa"
else:
text += "Nu exista clienti cu solduri."
return text
def format_suppliers_balance_response(
suppliers: List[Dict[str, Any]],
maturity_data: Dict[str, Any],
company_name: str = None
) -> str:
"""
Format suppliers balance with maturity breakdown (content only, no header).
Args:
suppliers: List of supplier dicts with id, name, balance
maturity_data: Dict with in_term, overdue, total
company_name: Company name (kept for compatibility, not used)
Returns:
Formatted Markdown string for Telegram
Example:
suppliers = [{'id': 1, 'name': 'Supplier A', 'balance': 5000}]
maturity = {'in_term': 4000, 'overdue': 1000, 'total': 5000}
text = format_suppliers_balance_response(suppliers, maturity)
"""
text = ""
# Maturity breakdown - rotunjit la leu (0 zecimale)
total = round(maturity_data.get('total', 0))
in_term = round(maturity_data.get('in_term', 0))
overdue = round(maturity_data.get('overdue', 0))
text += f"**Sold Total:** {total:,} RON\n\n"
text += "**Defalcare:**\n"
text += f" - In termen: {in_term:,} RON\n"
text += f" - Restanta: {overdue:,} RON\n\n"
# Top 10 suppliers
if suppliers:
text += f"**Top 10 Furnizori** ({len(suppliers)} total):\n"
# Sort by balance descending
sorted_suppliers = sorted(
suppliers,
key=lambda x: x.get('balance', 0),
reverse=True
)
for idx, supplier in enumerate(sorted_suppliers[:10], 1):
name = supplier.get('name', 'N/A')
balance = round(supplier.get('balance', 0))
text += f"{idx}. {name}: {balance:,} RON\n"
if len(suppliers) > 10:
text += f"\nApasa butonul pentru lista completa"
else:
text += "Nu exista furnizori cu solduri."
return text
def format_cashflow_evolution_response(
performance_data: Dict[str, Any],
monthly_data: Dict[str, Any],
company_name: str = None
) -> str:
"""
Format cash flow evolution data - Table format with mini-charts.
Args:
performance_data: Dict with current_year and previous_year YTD data
monthly_data: Dict with months, incasari, plati arrays + prev year data
company_name: Company name (kept for compatibility, not used)
Returns:
Formatted Markdown string for Telegram (monospace table)
Example:
YTD 2024 vs 2023:
2024 2023 Δ Trend
Inc: 500,000 480,000 +4.2% ████░
Plt: 450,000 440,000 +2.3% ███░
Net: 50,000 40,000 +25.0% █████
"""
text = ""
# Helper functions
def calc_percent_change(current: float, previous: float) -> str:
"""Calculate percentage change: +4.2% or -3.5%"""
if previous == 0:
return "+100%" if current > 0 else "0.0%"
change = ((current - previous) / previous) * 100
sign = "+" if change >= 0 else ""
return f"{sign}{change:.1f}%"
def create_mini_chart(current: float, previous: float, width: int = 5) -> str:
"""Create mini bar chart: ████░ (proportional bars)"""
if current == 0 and previous == 0:
return "" * width
max_val = max(current, previous)
if max_val == 0:
return "" * width
curr_bars = int((current / max_val) * width)
prev_bars = int((previous / max_val) * width)
# Use filled and light blocks
filled = "" * curr_bars
light = "" * (width - curr_bars)
return filled + light
def get_trend_arrow(current: float, previous: float) -> str:
"""Get trend arrow: ↑ or ↓ or →"""
if current > previous * 1.02: # More than 2% increase
return ""
elif current < previous * 0.98: # More than 2% decrease
return ""
else:
return ""
# Extract YTD data
current = performance_data.get('current_year', {})
previous = performance_data.get('previous_year', {})
current_year = current.get('year', '2024')
previous_year = previous.get('year', '2023')
inc_cur = round(current.get('incasari', 0))
plt_cur = round(current.get('plati', 0))
net_cur = round(current.get('net', 0))
inc_prev = round(previous.get('incasari', 0))
plt_prev = round(previous.get('plati', 0))
net_prev = round(previous.get('net', 0))
# YTD Table Header
text += f"**YTD {current_year} vs {previous_year}:**\n"
text += f"` {current_year:>10} {previous_year:>10} Δ `\n"
# YTD Rows
inc_pct = calc_percent_change(inc_cur, inc_prev)
text += f"`Inc: {inc_cur:>10,} {inc_prev:>10,} {inc_pct:>6}`\n"
plt_pct = calc_percent_change(plt_cur, plt_prev)
text += f"`Plt: {plt_cur:>10,} {plt_prev:>10,} {plt_pct:>6}`\n"
net_pct = calc_percent_change(net_cur, net_prev)
text += f"`Net: {net_cur:>10,} {net_prev:>10,} {net_pct:>6}`\n\n"
# Monthly Evolution Table - Simplified (Net only)
months = monthly_data.get('months', [])
incasari = monthly_data.get('incasari', [])
plati = monthly_data.get('plati', [])
incasari_prev = monthly_data.get('incasari_prev', [])
plati_prev = monthly_data.get('plati_prev', [])
if months and len(months) > 0:
text += "**Evolutie Net (12 luni):**\n"
text += f"` {current_year:>10} {previous_year:>10} Δ `\n"
for i, month in enumerate(months):
inc = incasari[i] if i < len(incasari) else 0
plt = plati[i] if i < len(plati) else 0
inc_p = incasari_prev[i] if i < len(incasari_prev) else 0
plt_p = plati_prev[i] if i < len(plati_prev) else 0
net = inc - plt
net_p = inc_p - plt_p
# Extract short month name (first 3 chars before apostrophe)
month_short = month.split("'")[0][:3] if "'" in month else month[:3]
# Calculate percentage change
net_pct = calc_percent_change(net, net_p)
# Format row: Luna Net'current Net'prev Δ (aligned with YTD)
text += f"`{month_short:<4} {int(net):>10,} {int(net_p):>10,} {net_pct:>6}`\n"
else:
text += "Nu exista date lunare disponibile."
return text
def format_client_detail_response(
client: Dict[str, Any],
invoices: List[Dict[str, Any]],
company_name: str = None
) -> str:
"""
Format client details with invoices - COMPACT TABLE FORMAT.
Args:
client: Dict with client info (id, name, balance)
invoices: List of invoice dicts for this client
company_name: Company name (kept for compatibility, not used)
Returns:
Formatted Markdown string for Telegram (compact, no emojis)
Example:
client = {'id': 1, 'name': 'Client A', 'balance': 15000}
invoices = [{'id': 1, 'number': 'FV001', 'amount': 5000, 'status': 'unpaid'}]
text = format_client_detail_response(client, invoices)
"""
client_name = client.get('name', 'N/A')
balance = client.get('balance', 0)
# Header with client info
text = f"**{client_name}**\n"
text += f"**Sold total: {balance:,.2f} RON**"
if invoices and len(invoices) > 1:
text += f"{len(invoices)} facturi"
text += "\n\n"
# Invoices - compact table format (no emojis)
if invoices:
from datetime import datetime
# Sort invoices by date (most recent first)
sorted_invoices = sorted(invoices, key=lambda x: x.get('dataact') or datetime.min, reverse=True)
# Invoice list - simple format without table
text += "Facturi cu sold:\n"
text += "━━━━━━━━━━━━━━━━━━━━\n"
# Invoice rows - one line each, simple format
for inv in sorted_invoices[:10]:
# Backend returns: nract, totctva, soldfinal, datascad, dataact, achitat
number = str(inv.get('nract', 'N/A'))
dataact = inv.get('dataact')
# Parse date - handle various formats to ensure dd.mm.yyyy
if dataact:
if isinstance(dataact, str):
try:
# Try ISO format first: "2024-10-25" or "2024-10-25 00:00:00"
if '-' in dataact and len(dataact) >= 10:
parsed_date = datetime.strptime(dataact[:10], '%Y-%m-%d')
date_str = parsed_date.strftime('%d.%m.%Y')
# Already in dd.mm.yyyy format
elif '.' in dataact:
date_str = dataact.split()[0][:10] # Take just date part
else:
date_str = dataact[:10] if len(dataact) >= 10 else dataact
except:
date_str = dataact[:10] if len(dataact) >= 10 else dataact
else:
# Datetime object - format as dd.mm.yyyy
date_str = dataact.strftime('%d.%m.%Y')
else:
date_str = 'N/A'
sold = float(inv.get('soldfinal', 0) or 0)
# Simple format: Nr • Data • Sold
text += f"Nr {number}{date_str}{sold:,.2f} RON\n"
if len(invoices) > 10:
text += f"\n\n+{len(invoices) - 10} facturi"
else:
text += "Nu exista facturi neachitate"
return text
def format_supplier_detail_response(
supplier: Dict[str, Any],
invoices: List[Dict[str, Any]],
company_name: str = None
) -> str:
"""
Format supplier details with invoices - COMPACT TABLE FORMAT.
Args:
supplier: Dict with supplier info (id, name, balance)
invoices: List of invoice dicts for this supplier
company_name: Company name (kept for compatibility, not used)
Returns:
Formatted Markdown string for Telegram (compact, no emojis)
Example:
supplier = {'id': 1, 'name': 'Supplier A', 'balance': 5000}
invoices = [{'id': 1, 'number': 'FC001', 'amount': 2000, 'status': 'unpaid'}]
text = format_supplier_detail_response(supplier, invoices)
"""
supplier_name = supplier.get('name', 'N/A')
balance = supplier.get('balance', 0)
# Header with supplier info
text = f"**{supplier_name}**\n"
text += f"**Sold total: {balance:,.2f} RON**"
if invoices and len(invoices) > 1:
text += f"{len(invoices)} facturi"
text += "\n\n"
# Invoices - compact table format (no emojis)
if invoices:
from datetime import datetime
# Sort invoices by date (most recent first)
sorted_invoices = sorted(invoices, key=lambda x: x.get('dataact') or datetime.min, reverse=True)
# Invoice list - simple format without table
text += "Facturi cu sold:\n"
text += "━━━━━━━━━━━━━━━━━━━━\n"
# Invoice rows - one line each, simple format
for inv in sorted_invoices[:10]:
# Backend returns: nract, totctva, soldfinal, datascad, dataact, achitat
number = str(inv.get('nract', 'N/A'))
dataact = inv.get('dataact')
# Parse date - handle various formats to ensure dd.mm.yyyy
if dataact:
if isinstance(dataact, str):
try:
# Try ISO format first: "2024-10-25" or "2024-10-25 00:00:00"
if '-' in dataact and len(dataact) >= 10:
parsed_date = datetime.strptime(dataact[:10], '%Y-%m-%d')
date_str = parsed_date.strftime('%d.%m.%Y')
# Already in dd.mm.yyyy format
elif '.' in dataact:
date_str = dataact.split()[0][:10] # Take just date part
else:
date_str = dataact[:10] if len(dataact) >= 10 else dataact
except:
date_str = dataact[:10] if len(dataact) >= 10 else dataact
else:
# Datetime object - format as dd.mm.yyyy
date_str = dataact.strftime('%d.%m.%Y')
else:
date_str = 'N/A'
sold = float(inv.get('soldfinal', 0) or 0)
# Simple format: Nr • Data • Sold
text += f"Nr {number}{date_str}{sold:,.2f} RON\n"
if len(invoices) > 10:
text += f"\n\n+{len(invoices) - 10} facturi"
else:
text += "Nu exista facturi neachitate"
return text
# =========================================================================
# FAZA 6: Performance Footer for Cache Monitoring
# =========================================================================
def add_performance_footer(message: str, cache_hit: bool, time_ms: float, cache_source: str = None) -> str:
"""
Add compact performance footer to bot responses.
Shows data source (cached L1/L2 or database) and response time.
Format: "cached L1 | 15ms", "cached L2 | 25ms" or "db | 285ms"
Args:
message: Existing message text
cache_hit: True if data came from cache
time_ms: Response time in milliseconds
cache_source: Cache source ("L1" for memory, "L2" for SQLite) if cache_hit is True
Returns:
Message with performance footer appended
Example:
>>> add_performance_footer("Dashboard data...", True, 52.3, "L1")
"Dashboard data...\n\ncached L1 | 52ms"
>>> add_performance_footer("Dashboard data...", True, 25.8, "L2")
"Dashboard data...\n\ncached L2 | 26ms"
>>> add_performance_footer("Dashboard data...", False, 285.7)
"Dashboard data...\n\ndb | 286ms"
"""
if cache_hit and cache_source:
source = f"cached {cache_source}"
elif cache_hit:
source = "cached" # Fallback if source not provided
else:
source = "db"
footer = f"\n\n`{source} | {time_ms:.0f}ms`"
return message + footer