Modern ERP Reports Application with microservices architecture Tech Stack: - Backend: FastAPI + python-oracledb (Oracle DB integration) - Frontend: Vue.js 3 + PrimeVue + Vite - Telegram Bot: python-telegram-bot + SQLite - Infrastructure: Shared database pool, JWT authentication, SSH tunnel Features: - FastAPI backend with async Oracle connection pool - Vue.js 3 responsive frontend with PrimeVue components - Telegram bot alternative interface - Microservices architecture with shared components - Complete deployment support (Linux Docker + Windows IIS) - Comprehensive testing (Playwright E2E + pytest) Repository Structure: - reports-app/ - Main application (backend, frontend, telegram-bot) - shared/ - Shared components (database pool, auth, utils) - deployment/ - Deployment scripts (Linux & Windows) - docs/ - Project documentation - security/ - Security scanning and git hooks
516 lines
18 KiB
Python
516 lines
18 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"
|
|
|
|
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[:5]: # Max 5
|
|
name = acc.get('name', 'N/A')
|
|
balance = round(acc.get('balance', 0))
|
|
text += f" - {name}: {balance:,} RON\n"
|
|
|
|
if len(casa_accounts) > 5:
|
|
text += f" ... si inca {len(casa_accounts) - 5} conturi"
|
|
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[:5]: # Max 5
|
|
name = acc.get('name', 'N/A')
|
|
balance = round(acc.get('balance', 0))
|
|
text += f" - {name}: {balance:,} RON\n"
|
|
|
|
if len(bank_accounts) > 5:
|
|
text += f" ... si inca {len(bank_accounts) - 5} conturi"
|
|
else:
|
|
text += "Nu exista conturi bancare configurate."
|
|
|
|
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 clients
|
|
if clients:
|
|
text += f"**Top 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[:5], 1):
|
|
name = client.get('name', 'N/A')
|
|
balance = round(client.get('balance', 0))
|
|
text += f"{idx}. {name}: {balance:,} RON\n"
|
|
|
|
if len(clients) > 5:
|
|
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 suppliers
|
|
if suppliers:
|
|
text += f"**Top 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[:5], 1):
|
|
name = supplier.get('name', 'N/A')
|
|
balance = round(supplier.get('balance', 0))
|
|
text += f"{idx}. {name}: {balance:,} RON\n"
|
|
|
|
if len(suppliers) > 5:
|
|
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 (content only, no header).
|
|
|
|
Args:
|
|
performance_data: Dict with incasari_total, plati_total, net
|
|
monthly_data: Dict with months, incasari, plati arrays
|
|
company_name: Company name (kept for compatibility, not used)
|
|
|
|
Returns:
|
|
Formatted Markdown string for Telegram
|
|
|
|
Example:
|
|
performance = {'incasari_total': 100000, 'plati_total': 80000, 'net': 20000}
|
|
monthly = {'months': ['Ian', 'Feb'], 'incasari': [50000, 50000], 'plati': [40000, 40000]}
|
|
text = format_cashflow_evolution_response(performance, monthly)
|
|
"""
|
|
text = ""
|
|
|
|
# Performance summary - rotunjit la leu (0 zecimale)
|
|
incasari_total = round(performance_data.get('incasari_total', 0))
|
|
plati_total = round(performance_data.get('plati_total', 0))
|
|
net = round(performance_data.get('net', 0))
|
|
|
|
text += "**Rezumat:**\n"
|
|
text += f" - Total Incasari: {incasari_total:,} RON\n"
|
|
text += f" - Total Plati: {plati_total:,} RON\n"
|
|
text += f" - Net Cash Flow: {net:,} RON\n\n"
|
|
|
|
# Monthly breakdown
|
|
months = monthly_data.get('months', [])
|
|
incasari = monthly_data.get('incasari', [])
|
|
plati = monthly_data.get('plati', [])
|
|
|
|
if months and len(months) > 0:
|
|
text += "**Evolutie Lunara** (ultimele luni):\n"
|
|
|
|
# Show last 6 months
|
|
display_count = min(6, len(months))
|
|
for i in range(display_count):
|
|
month = months[-(display_count - i)]
|
|
inc = round(incasari[-(display_count - i)]) if i < len(incasari) else 0
|
|
plt = round(plati[-(display_count - i)]) if i < len(plati) else 0
|
|
net_month = inc - plt
|
|
|
|
# Simple ASCII bar
|
|
net_indicator = "+" if net_month > 0 else "-" if net_month < 0 else "="
|
|
|
|
text += f"\n**{month}:**\n"
|
|
text += f" {net_indicator} Incasari: {inc:,} RON\n"
|
|
text += f" {net_indicator} Plati: {plt:,} RON\n"
|
|
text += f" {net_indicator} Net: {net_month:,} RON"
|
|
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
|