romfast/roa2web-service-auto
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Implement unified Telegram bot interface with Login/Logout and fix callback_data limits

Browse Source 
**Interface improvements:**
- Add persistent Login/Logout buttons to main menu
- Help button now updates text above menu (not separate message)
- Expired token automatically transforms menu to Login state
- Consolidate linking success messages (single welcome + menu)

**Fix callback_data length limits (Telegram 64-byte limit):**
- Truncate client/supplier names to 40 chars in callback_data
- Use full names for API calls but truncated for buttons
- Fix pagination buttons for long partner names (30 chars limit)
- Search entities by prefix match to handle truncated names

**Treasury improvements:**
- Show ALL bank/cash accounts (removed 5-item limit)
- Remove unnecessary Refresh/Export buttons from treasury views
- Handle "Message is not modified" error gracefully

**Bug fixes:**
- Fix Markdown parsing errors (replace <cod> with `CODUL_TAU`)
- Fix silent errors when token expires (show user-friendly message)
- Fix Button_data_invalid errors on pagination and details

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Marius Mutu
2025-11-07 01:02:03 +02:00
parent 20f6c52785
commit a4ee394091
3 changed files with 405 additions and 131 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
reports-app/telegram-bot/app/bot/formatters.py
View File

@@ -124,13 +124,10 @@ def format_treasury_casa_response(data: Dict[str, Any], company_name: str = None
casa_accounts = data.get('accounts', [])
if casa_accounts:
text += "**Conturi de Casa:**\n"
for acc in casa_accounts[:5]: # Max 5
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"
if len(casa_accounts) > 5:
text += f" ... si inca {len(casa_accounts) - 5} conturi"
else:
text += "Nu exista conturi de casa configurate."
@@ -162,13 +159,10 @@ def format_treasury_banca_response(data: Dict[str, Any], company_name: str = Non
bank_accounts = data.get('accounts', [])
if bank_accounts:
text += "**Conturi Bancare:**\n"
for acc in bank_accounts[:5]: # Max 5
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"
if len(bank_accounts) > 5:
text += f" ... si inca {len(bank_accounts) - 5} conturi"
else:
text += "Nu exista conturi bancare configurate."

453
reports-app/telegram-bot/app/bot/handlers.py
View File

@@ -69,24 +69,6 @@ async def start_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
if result:
# Success!
username = result['username']
companies = result.get('companies', [])
companies_text = ""
if companies:
companies_text = "\n\n**Companiile tale:**\n"
for comp_id in companies[:3]: # Show first 3 (companies are IDs as strings)
companies_text += f"- Companie ID: {comp_id}\n"
if len(companies) > 3:
companies_text += f"- ... si inca {len(companies) - 3} companii\n"
await update.message.reply_text(
f"**Cont conectat cu succes**\n\n"
f"Bun venit, **{username}**\n"
f"{companies_text}\n"
f"Foloseste meniul sau /help pentru comenzi.",
parse_mode=ParseMode.MARKDOWN
)
# Show main menu with buttons for newly linked user
session_manager = get_session_manager()
@@ -95,12 +77,27 @@ async def start_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
company_name = company['name'] if company else None
company_cui = company.get('cui') if company else None
from app.bot.menus import create_main_menu, get_menu_message
keyboard = create_main_menu(company_name, company_cui)
menu_text = get_menu_message(company_name, company_cui)
from app.bot.menus import create_main_menu, pad_message_for_wide_buttons
keyboard = create_main_menu(company_name, company_cui, is_authenticated=True)
# Single welcome message with menu
if company_name:
welcome_text = (
f"**Cont conectat cu succes**\n\n"
f"Bun venit, **{username}**!\n\n"
f"{company_name}"
)
else:
welcome_text = (
f"**Cont conectat cu succes**\n\n"
f"Bun venit, **{username}**!\n\n"
f"Selectează o companie pentru a continua"
)
welcome_message = pad_message_for_wide_buttons(welcome_text)
await update.message.reply_text(
menu_text,
welcome_message,
reply_markup=keyboard,
parse_mode=ParseMode.MARKDOWN
)
@@ -111,9 +108,9 @@ async def start_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
# Failed linking
await update.message.reply_text(
"**Cod invalid sau expirat**\n\n"
"Genereaza un cod nou din aplicatia web si trimite:\n"
"/start <cod>\n\n"
"Codul expira in 15 minute.",
"Generează un cod nou din aplicația web și trimite:\n"
"`/start CODUL_TAU`\n\n"
"Codul expiră în 15 minute.",
parse_mode=ParseMode.MARKDOWN
)
@@ -135,11 +132,12 @@ async def start_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
company = session.get_active_company()
company_name = company['name'] if company else None
company_cui = company.get('cui') if company else None
# Create main menu
from app.bot.menus import create_main_menu, get_menu_message
keyboard = create_main_menu(company_name)
menu_text = get_menu_message(company_name)
keyboard = create_main_menu(company_name, company_cui, is_authenticated=True)
menu_text = get_menu_message(company_name, company_cui)
await update.message.reply_text(
f"Bun venit, **{username}**\n\n{menu_text}",
@@ -212,10 +210,10 @@ Dupa conectarea contului, foloseste **butoanele interactive** pentru:
/unlink - Deconecteaza contul (securitate)
**Conectare cont:**
1. Logheaza-te in aplicatia web ROA2WEB
2. Acceseaza Setari > Telegram Linking
3. Genereaza cod (valabil 15 minute)
4. Trimite codul in Telegram: /start <cod>
1. Loghează-te în aplicația web ROA2WEB
2. Accesează Setări → Telegram Linking
3. Generează cod (valabil 15 minute)
4. Trimite codul în Telegram: `/start CODUL_TAU`
**Securitate:**
Datele sunt protejate prin autentificare JWT.
@@ -647,9 +645,9 @@ async def menu_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
company_name = company['name'] if company else None
company_cui = company.get('cui') if company else None
# Create main menu
# Create main menu (user is authenticated if they passed the is_linked check)
from app.bot.menus import create_main_menu, get_menu_message
keyboard = create_main_menu(company_name, company_cui)
keyboard = create_main_menu(company_name, company_cui, is_authenticated=True)
menu_text = get_menu_message(company_name, company_cui)
await update.message.reply_text(
@@ -1049,25 +1047,6 @@ async def handle_text_message(update: Update, context: ContextTypes.DEFAULT_TYPE
if result:
# Success!
username = result['username']
companies = result.get('companies', [])
companies_text = ""
if companies:
companies_text = "\n\n**Companiile tale:**\n"
for comp_id in companies[:3]: # Show first 3
companies_text += f"- Companie ID: {comp_id}\n"
if len(companies) > 3:
companies_text += f"- ... si inca {len(companies) - 3} companii\n"
await update.message.reply_text(
f"**Cont conectat cu succes**\n\n"
f"Bun venit, **{username}**\n"
f"{companies_text}\n"
f"Foloseste meniul sau /help pentru comenzi.",
parse_mode=ParseMode.MARKDOWN
)
# Show main menu with buttons for newly linked user
session_manager = get_session_manager()
session = await session_manager.get_or_create_session(telegram_user_id)
@@ -1075,12 +1054,27 @@ async def handle_text_message(update: Update, context: ContextTypes.DEFAULT_TYPE
company_name = company['name'] if company else None
company_cui = company.get('cui') if company else None
from app.bot.menus import create_main_menu, get_menu_message
keyboard = create_main_menu(company_name, company_cui)
menu_text = get_menu_message(company_name, company_cui)
from app.bot.menus import create_main_menu, pad_message_for_wide_buttons
keyboard = create_main_menu(company_name, company_cui, is_authenticated=True)
# Single welcome message with menu
if company_name:
welcome_text = (
f"**Cont conectat cu succes**\n\n"
f"Bun venit, **{username}**!\n\n"
f"{company_name}"
)
else:
welcome_text = (
f"**Cont conectat cu succes**\n\n"
f"Bun venit, **{username}**!\n\n"
f"Selectează o companie pentru a continua"
)
welcome_message = pad_message_for_wide_buttons(welcome_text)
await update.message.reply_text(
menu_text,
welcome_message,
reply_markup=keyboard,
parse_mode=ParseMode.MARKDOWN
)
@@ -1140,6 +1134,29 @@ async def handle_menu_callback(query, telegram_user_id: int, callback_data: str)
# Get auth data
auth_data = await get_user_auth_data(telegram_user_id)
# If user is not authenticated and trying to access financial data, show auth required message
if auth_data is None and action != "select_company":
from app.bot.menus import create_main_menu, pad_message_for_wide_buttons
keyboard = create_main_menu(company_name=None, company_cui=None, is_authenticated=False)
menu_text = pad_message_for_wide_buttons(
"⚠️ **Autentificare necesară**\n\n"
"Pentru a accesa date financiare,\n"
"apasă **Login** și urmează instrucțiunile."
)
await query.edit_message_text(
menu_text,
reply_markup=keyboard,
parse_mode=ParseMode.MARKDOWN
)
return
# If action is select_company and user is not authenticated, allow it (will show empty list or error)
if action == "select_company" and auth_data is None:
await query.answer("Pentru a vedea companiile, trebuie să te autentifici mai întâi.", show_alert=True)
return
jwt_token = auth_data['jwt_token']
# Get active company
@@ -1193,13 +1210,18 @@ async def handle_menu_callback(query, telegram_user_id: int, callback_data: str)
content = format_treasury_casa_response(treasury_data['casa'])
response = format_response_with_company(content, company['name'])
keyboard = create_action_buttons("casa", show_export=True)
keyboard = create_action_buttons("casa", show_export=False, show_refresh=False)
await query.edit_message_text(
response,
reply_markup=keyboard,
parse_mode=ParseMode.MARKDOWN
)
try:
await query.edit_message_text(
response,
reply_markup=keyboard,
parse_mode=ParseMode.MARKDOWN
)
except Exception as e:
# Ignore "Message is not modified" error
if "Message is not modified" not in str(e):
raise
elif action == "banca":
# Trezorerie banca
@@ -1211,13 +1233,18 @@ async def handle_menu_callback(query, telegram_user_id: int, callback_data: str)
content = format_treasury_banca_response(treasury_data['banca'])
response = format_response_with_company(content, company['name'])
keyboard = create_action_buttons("banca", show_export=True)
keyboard = create_action_buttons("banca", show_export=False, show_refresh=False)
await query.edit_message_text(
response,
reply_markup=keyboard,
parse_mode=ParseMode.MARKDOWN
)
try:
await query.edit_message_text(
response,
reply_markup=keyboard,
parse_mode=ParseMode.MARKDOWN
)
except Exception as e:
# Ignore "Message is not modified" error
if "Message is not modified" not in str(e):
raise
elif action == "clienti":
# Sold clienți + listă cu paginare
@@ -1296,10 +1323,10 @@ async def handle_menu_callback(query, telegram_user_id: int, callback_data: str)
async def handle_action_callback(query, telegram_user_id: int, callback_data: str):
"""
Handle action button clicks (Refresh, Export, Menu).
Handle action button clicks (Refresh, Export, Menu, Help, Login, Logout).
Callback format: action:{type}:{view}
Types: refresh, export, menu
Types: refresh, export, menu, help, login, logout
Args:
query: CallbackQuery object
@@ -1315,10 +1342,14 @@ async def handle_action_callback(query, telegram_user_id: int, callback_data: st
session = await session_manager.get_or_create_session(telegram_user_id)
company = session.get_active_company()
# Check if user is authenticated
auth_data = await get_user_auth_data(telegram_user_id)
is_authenticated = auth_data is not None
from app.bot.menus import create_main_menu, get_menu_message
company_name = company['name'] if company else None
company_cui = company.get('cui') if company else None
keyboard = create_main_menu(company_name, company_cui)
keyboard = create_main_menu(company_name, company_cui, is_authenticated)
menu_text = get_menu_message(company_name, company_cui)
await query.edit_message_text(
@@ -1334,10 +1365,14 @@ async def handle_action_callback(query, telegram_user_id: int, callback_data: st
# Check if it's a detail view (client_detail:name or supplier_detail:name)
if view.startswith("client_detail:"):
entity_name = view.split(":", 1)[1] # Extract entity name
await handle_details_callback(query, telegram_user_id, f"details:client:{entity_name}:0")
# Limit name to 40 chars for Telegram callback_data limit (64 bytes)
safe_name = entity_name[:40] if len(entity_name) > 40 else entity_name
await handle_details_callback(query, telegram_user_id, f"details:client:{safe_name}:0")
elif view.startswith("supplier_detail:"):
entity_name = view.split(":", 1)[1] # Extract entity name
await handle_details_callback(query, telegram_user_id, f"details:supplier:{entity_name}:0")
# Limit name to 40 chars for Telegram callback_data limit (64 bytes)
safe_name = entity_name[:40] if len(entity_name) > 40 else entity_name
await handle_details_callback(query, telegram_user_id, f"details:supplier:{safe_name}:0")
else:
# Regular menu view refresh
await handle_menu_callback(query, telegram_user_id, f"menu:{view}")
@@ -1346,6 +1381,101 @@ async def handle_action_callback(query, telegram_user_id: int, callback_data: st
# Export functionality (placeholder for now)
await query.answer("Functia de export va fi disponibila in curand", show_alert=True)
elif action_type == "help":
# Show help message above menu (edit current message)
from app.bot.menus import pad_message_for_wide_buttons, create_main_menu
# Get auth status and company info
session_manager = get_session_manager()
session = await session_manager.get_or_create_session(telegram_user_id)
company = session.get_active_company()
auth_data = await get_user_auth_data(telegram_user_id)
is_authenticated = auth_data is not None
company_name = company['name'] if company else None
company_cui = company.get('cui') if company else None
keyboard = create_main_menu(company_name, company_cui, is_authenticated)
help_text = pad_message_for_wide_buttons(
"**Ghid Rapid**\n\n"
"**Selectare Companie** - Alege compania activă\n\n"
"**Sold Companie** - Dashboard financiar complet\n"
"**Trezorerie Casa** - Situație conturi cash\n"
"**Trezorerie Banca** - Situație conturi bancare\n"
"**Sold Clienti** - Clienți + facturi neplătite\n"
"**Sold Furnizori** - Furnizori + facturi\n"
"**Evolutie Incasari** - Trend lunar încasări\n\n"
"**Logout** - Deconectează contul\n\n"
"_Toate datele sunt în timp real din Oracle._"
)
await query.edit_message_text(
help_text,
reply_markup=keyboard,
parse_mode=ParseMode.MARKDOWN
)
elif action_type == "logout":
# Show logout confirmation
from app.bot.menus import pad_message_for_wide_buttons
confirmation_text = pad_message_for_wide_buttons(
"**Confirmare Deconectare**\n\n"
"Ești sigur că vrei să deconectezi contul?\n\n"
"Accesul la date va fi oprit.\n"
"Poți reconecta oricând cu un cod nou."
)
keyboard = InlineKeyboardMarkup([
[
InlineKeyboardButton("Da, deconectează", callback_data="logout_confirm"),
InlineKeyboardButton("Anulează", callback_data="logout_cancel")
]
])
await query.edit_message_text(
confirmation_text,
reply_markup=keyboard,
parse_mode=ParseMode.MARKDOWN
)
elif action_type == "login":
# Prompt user to enter link code directly (same as login_prompt functionality)
from telegram import ForceReply
from app.bot.menus import pad_message_for_wide_buttons
# Edit the current message with instructions
login_text = pad_message_for_wide_buttons(
"**Conectare Cont ROA2WEB**\n\n"
"Trimite-mi codul primit din aplicația web.\n\n"
"Poți trimite:\n"
"• Doar codul: ABC12XYZ\n"
"• Sau comanda: /start ABC12XYZ\n\n"
"Codul expiră în 15 minute."
)
# Buttons for help or cancel
keyboard = InlineKeyboardMarkup([
[InlineKeyboardButton("Cum obțin codul?", callback_data="login_help")],
[InlineKeyboardButton("« Anulează", callback_data="action:menu")]
])
await query.edit_message_text(
login_text,
reply_markup=keyboard,
parse_mode=ParseMode.MARKDOWN
)
# Send a follow-up message with ForceReply to prompt input
await query.message.reply_text(
"Scrie sau lipește codul aici:",
reply_markup=ForceReply(
selective=True,
input_field_placeholder="ABC12XYZ"
)
)
async def handle_details_callback(query, telegram_user_id: int, callback_data: str):
"""
@@ -1373,26 +1503,35 @@ async def handle_details_callback(query, telegram_user_id: int, callback_data: s
company = session.get_active_company()
if detail_type == "client":
# Get client invoices by name
from app.bot.helpers import get_client_invoices
invoices = await get_client_invoices(company['id'], entity_name, jwt_token)
# Get client details (from clients list)
# entity_name might be truncated to 40 chars, so search by startswith
from app.bot.helpers import get_clients_with_maturity
clients_data = await get_clients_with_maturity(company['id'], jwt_token)
client = next((c for c in clients_data['clients'] if c['name'] == entity_name), None)
# Find client by full or partial name match
client = next((c for c in clients_data['clients'] if c['name'].startswith(entity_name)), None)
if not client:
await query.answer("Client negasit", show_alert=True)
await query.answer("Client negăsit", show_alert=True)
return
# Use FULL client name for invoice search (not truncated)
full_client_name = client['name']
# Get client invoices by FULL name
from app.bot.helpers import get_client_invoices
invoices = await get_client_invoices(company['id'], full_client_name, jwt_token)
# Format response
from app.bot.formatters import format_client_detail_response
from app.bot.menus import create_action_buttons, format_response_with_company
content = format_client_detail_response(client, invoices)
response = format_response_with_company(content, company['name'])
keyboard = create_action_buttons(f"client_detail:{entity_name}", show_export=False, show_back=True)
# Use truncated name for callback_data (to stay within 64 byte limit)
safe_name = entity_name[:40] if len(entity_name) > 40 else entity_name
keyboard = create_action_buttons(f"client_detail:{safe_name}", show_export=False, show_back=True, show_refresh=False)
await query.edit_message_text(
response,
@@ -1401,26 +1540,35 @@ async def handle_details_callback(query, telegram_user_id: int, callback_data: s
)
elif detail_type == "supplier":
# Get supplier invoices by name
from app.bot.helpers import get_supplier_invoices
invoices = await get_supplier_invoices(company['id'], entity_name, jwt_token)
# Get supplier details (from suppliers list)
# entity_name might be truncated to 40 chars, so search by startswith
from app.bot.helpers import get_suppliers_with_maturity
suppliers_data = await get_suppliers_with_maturity(company['id'], jwt_token)
supplier = next((s for s in suppliers_data['suppliers'] if s['name'] == entity_name), None)
# Find supplier by full or partial name match
supplier = next((s for s in suppliers_data['suppliers'] if s['name'].startswith(entity_name)), None)
if not supplier:
await query.answer("Furnizor negasit", show_alert=True)
await query.answer("Furnizor negăsit", show_alert=True)
return
# Use FULL supplier name for invoice search (not truncated)
full_supplier_name = supplier['name']
# Get supplier invoices by FULL name
from app.bot.helpers import get_supplier_invoices
invoices = await get_supplier_invoices(company['id'], full_supplier_name, jwt_token)
# Format response
from app.bot.formatters import format_supplier_detail_response
from app.bot.menus import create_action_buttons, format_response_with_company
content = format_supplier_detail_response(supplier, invoices)
response = format_response_with_company(content, company['name'])
keyboard = create_action_buttons(f"supplier_detail:{entity_name}", show_export=False, show_back=True)
# Use truncated name for callback_data (to stay within 64 byte limit)
safe_name = entity_name[:40] if len(entity_name) > 40 else entity_name
keyboard = create_action_buttons(f"supplier_detail:{safe_name}", show_export=False, show_back=True, show_refresh=False)
await query.edit_message_text(
response,
@@ -1582,6 +1730,63 @@ async def button_callback(update: Update, context: ContextTypes.DEFAULT_TYPE):
"Companie negasita sau nu ai acces la ea."
)
# ========== LOGOUT CALLBACKS ==========
elif callback_data == "logout_confirm":
# Logout user (same as unlink but shows menu after)
from app.auth.linking import unlink_user
success = await unlink_user(telegram_user_id)
if success:
# Clear session too
session_manager = get_session_manager()
await session_manager.delete_session(telegram_user_id)
# Show login menu (non-authenticated)
from app.bot.menus import create_main_menu, get_menu_message, pad_message_for_wide_buttons
keyboard = create_main_menu(company_name=None, company_cui=None, is_authenticated=False)
menu_text = pad_message_for_wide_buttons(
"**Deconectat cu succes**\n\n"
"Contul tău a fost deconectat.\n\n"
"Pentru a te reconecta, apasă **Login**."
)
await query.edit_message_text(
menu_text,
reply_markup=keyboard,
parse_mode=ParseMode.MARKDOWN
)
else:
await query.edit_message_text(
"A apărut o eroare la deconectare.\n"
"Te rog încearcă din nou."
)
elif callback_data == "logout_cancel":
# Cancel logout - return to main menu
session_manager = get_session_manager()
session = await session_manager.get_or_create_session(telegram_user_id)
company = session.get_active_company()
# Check auth
auth_data = await get_user_auth_data(telegram_user_id)
is_authenticated = auth_data is not None
from app.bot.menus import create_main_menu, get_menu_message
company_name = company['name'] if company else None
company_cui = company.get('cui') if company else None
keyboard = create_main_menu(company_name, company_cui, is_authenticated)
menu_text = get_menu_message(company_name, company_cui)
await query.edit_message_text(
menu_text,
reply_markup=keyboard,
parse_mode=ParseMode.MARKDOWN
)
# ========== UNLINK CALLBACKS (LEGACY) ==========
elif callback_data == "unlink_confirm":
# Unlink user
from app.auth.linking import unlink_user
@@ -1595,8 +1800,8 @@ async def button_callback(update: Update, context: ContextTypes.DEFAULT_TYPE):
await query.edit_message_text(
"**Cont deconectat cu succes**\n\n"
"Datele tale au fost sterse din sistem.\n\n"
"Pentru a te reconecta, foloseste /start <cod>",
"Datele tale au fost șterse din sistem.\n\n"
"Pentru a te reconecta, folosește `/start CODUL_TAU`",
parse_mode=ParseMode.MARKDOWN
)
else:
@@ -1616,16 +1821,16 @@ async def button_callback(update: Update, context: ContextTypes.DEFAULT_TYPE):
elif callback_data == "login_help":
# Show detailed help on how to get link code
await query.edit_message_text(
"**Cum obtin codul de link?**\n\n"
"1. Logheaza-te in aplicatia web ROA2WEB\n"
"2. Mergi la: Setari -> Telegram Linking\n"
"3. Apasa \"Genereaza Cod\"\n"
"**Cum obțin codul de link?**\n\n"
"1. Loghează-te în aplicația web ROA2WEB\n"
"2. Mergi la: Setări → Telegram Linking\n"
"3. Apasă **Generează Cod**\n"
"4. Vei primi un cod din 8 caractere (ex: ABC12XYZ)\n"
"5. Trimite-mi comanda: /start <codul_tau>\n\n"
"Important: Codul expira in 15 minute.",
"5. Trimite-mi comanda: `/start CODUL_TAU`\n\n"
"**Important:** Codul expiră în 15 minute.",
reply_markup=InlineKeyboardMarkup([
[InlineKeyboardButton("Am deja cod - Linkez acum", callback_data="login_prompt")],
[InlineKeyboardButton("« Inapoi la Bun Venit", callback_data="login_back")]
[InlineKeyboardButton("« Înapoi la Meniu", callback_data="action:menu")]
]),
parse_mode=ParseMode.MARKDOWN
)
@@ -1845,6 +2050,62 @@ async def error_handler(update: Update, context: ContextTypes.DEFAULT_TYPE):
# COMMON HANDLER FUNCTIONS (pentru consistență comenzi/butoane)
# ============================================================================
async def _handle_expired_auth(query_or_update, telegram_user_id: int, auth_data: Optional[Dict[str, Any]]) -> bool:
"""
Check if auth_data is None (token expired/refresh failed) and send user-friendly message.
Args:
query_or_update: Update (command) or CallbackQuery (button)
telegram_user_id: Telegram user ID
auth_data: Authentication data from get_user_auth_data()
Returns:
True if auth is expired (stop execution), False if auth is valid (continue)
"""
if auth_data is None:
logger.warning(f"Auth expired for user {telegram_user_id}, sending re-authentication message")
# Create re-authentication message
message = (
"⚠️ **Sesiunea ta a expirat**\n\n"
"Token-ul de autentificare a expirat și nu a putut fi reînnoit automat.\n\n"
"**Pentru a continua:**\n"
"1. Accesează aplicația web ROA2WEB\n"
"2. Loginează-te cu contul tău Oracle\n"
"3. Generează un nou cod de link pentru Telegram\n"
"4. Trimite comanda `/start CODUL_TAU`\n\n"
"_Sau folosește `/unlink` pentru a deconecta contul actual._"
)
# Send message based on source type
from telegram import CallbackQuery
if isinstance(query_or_update, CallbackQuery):
# It's a button callback - transform menu to Login menu
await query_or_update.answer("Sesiunea a expirat. Te rog să te reconectezi.", show_alert=True)
# Transform the current message (menu) to Login menu
from app.bot.menus import create_main_menu, pad_message_for_wide_buttons
keyboard = create_main_menu(company_name=None, company_cui=None, is_authenticated=False)
menu_text = pad_message_for_wide_buttons(
"⚠️ **Sesiunea a expirat**\n\n"
"Pentru a continua, apasă **Login**\n"
"și urmează instrucțiunile."
)
await query_or_update.edit_message_text(
menu_text,
reply_markup=keyboard,
parse_mode=ParseMode.MARKDOWN
)
else:
# It's a command (Update) - just send message
await query_or_update.message.reply_text(message, parse_mode=ParseMode.MARKDOWN)
return True # Stop execution
return False # Continue execution
async def _handle_sold_view(
query_or_update,
telegram_user_id: int,

73
reports-app/telegram-bot/app/bot/menus.py
View File

@@ -187,7 +187,8 @@ def get_menu_message(
def create_main_menu(
company_name: Optional[str] = None,
company_cui: Optional[str] = None
company_cui: Optional[str] = None,
is_authenticated: bool = True
) -> InlineKeyboardMarkup:
"""
Create main menu keyboard (Level 1) with financial options.
@@ -197,6 +198,7 @@ def create_main_menu(
Args:
company_name: Active company name, or None if no company selected
company_cui: Company fiscal code (CUI), or None
is_authenticated: Whether user is authenticated (affects Login/Logout button)
Returns:
InlineKeyboardMarkup with main menu buttons
@@ -240,48 +242,52 @@ def create_main_menu(
]
])
# Row 5: Help button (full width)
keyboard.append([
InlineKeyboardButton("Help", callback_data="action:help")
])
# Row 5: Help/Logout buttons (authenticated) or Login button (non-authenticated)
if is_authenticated:
keyboard.append([
InlineKeyboardButton("Help", callback_data="action:help"),
InlineKeyboardButton("Logout", callback_data="action:logout")
])
else:
keyboard.append([
InlineKeyboardButton("Login", callback_data="action:login")
])
return InlineKeyboardMarkup(keyboard)
def create_action_buttons(current_view: str, show_export: bool = True, show_back: bool = False) -> InlineKeyboardMarkup:
def create_action_buttons(current_view: str, show_export: bool = True, show_back: bool = False, show_refresh: bool = True) -> InlineKeyboardMarkup:
"""
Create action buttons for responses (Refresh, Export, Back, Menu).
Layout (buttons made wide by message text padding):
[Refresh] [Export] (if show_export=True)
[Refresh] [Export] (if show_refresh=True and show_export=True)
[Refresh] (if show_refresh=True and show_export=False)
[Înapoi] (if show_back=True, full width)
[Menu] (full width)
Or:
[Refresh] (if show_export=False)
[Înapoi] (if show_back=True, full width)
[Menu] (full width)
[Menu] (full width, always shown)
Args:
current_view: View identifier for refresh callback (e.g., "sold", "clienti")
show_export: Whether to show Export button
show_back: Whether to show Back button to list
show_refresh: Whether to show Refresh button
Returns:
InlineKeyboardMarkup with action buttons
"""
keyboard = []
# Row 1: Refresh and optionally Export
if show_export:
keyboard.append([
InlineKeyboardButton("Refresh", callback_data=f"action:refresh:{current_view}"),
InlineKeyboardButton("Export", callback_data=f"action:export:{current_view}")
])
else:
keyboard.append([
InlineKeyboardButton("Refresh", callback_data=f"action:refresh:{current_view}")
])
# Row 1: Refresh and optionally Export (only if show_refresh is True)
if show_refresh:
if show_export:
keyboard.append([
InlineKeyboardButton("Refresh", callback_data=f"action:refresh:{current_view}"),
InlineKeyboardButton("Export", callback_data=f"action:export:{current_view}")
])
else:
keyboard.append([
InlineKeyboardButton("Refresh", callback_data=f"action:refresh:{current_view}")
])
# Row 2: Back to List (if show_back is True)
if show_back:
@@ -344,10 +350,15 @@ def create_client_list_keyboard(clients: List[Dict], max_items: int = 10, page:
balance_str = f"{balance:,.0f}" if balance else "0"
button_text = f"{client_name} - {balance_str} RON"
# Limit callback_data to 64 bytes (Telegram limit)
# Use only first 40 chars of name to stay within limit
safe_name = client_name[:40] if len(client_name) > 40 else client_name
keyboard.append([
InlineKeyboardButton(
button_text,
callback_data=f"details:client:{client_name}:0" # name:page
callback_data=f"details:client:{safe_name}:0" # name:page
)
])
@@ -417,10 +428,15 @@ def create_supplier_list_keyboard(suppliers: List[Dict], max_items: int = 10, pa
balance_str = f"{balance:,.0f}" if balance else "0"
button_text = f"{supplier_name} - {balance_str} RON"
# Limit callback_data to 64 bytes (Telegram limit)
# Use only first 40 chars of name to stay within limit
safe_name = supplier_name[:40] if len(supplier_name) > 40 else supplier_name
keyboard.append([
InlineKeyboardButton(
button_text,
callback_data=f"details:supplier:{supplier_name}:0" # name:page
callback_data=f"details:supplier:{safe_name}:0" # name:page
)
])
@@ -480,6 +496,9 @@ def create_invoice_list_keyboard(
"""
keyboard = []
# Limit partner_name to 30 chars for Telegram callback_data limit (64 bytes)
safe_partner_name = partner_name[:30] if len(partner_name) > 30 else partner_name
# Calculate pagination
total_invoices = len(invoices)
total_pages = (total_invoices + max_items - 1) // max_items # Ceiling division
@@ -517,7 +536,7 @@ def create_invoice_list_keyboard(
# Previous button
if page > 0:
nav_buttons.append(
InlineKeyboardButton("< Anterior", callback_data=f"invoices_page:{partner_type}:{partner_name}:{page-1}")
InlineKeyboardButton("< Anterior", callback_data=f"invoices_page:{partner_type}:{safe_partner_name}:{page-1}")
)
# Page indicator (non-clickable)
@@ -528,7 +547,7 @@ def create_invoice_list_keyboard(
# Next button
if page < total_pages - 1:
nav_buttons.append(
InlineKeyboardButton("Următor >", callback_data=f"invoices_page:{partner_type}:{partner_name}:{page+1}")
InlineKeyboardButton("Următor >", callback_data=f"invoices_page:{partner_type}:{safe_partner_name}:{page+1}")
)
keyboard.append(nav_buttons)