romfast/btgo-playwright
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Adaugă solduri în notificările email

Browse Source 
- EmailNotifier primește lista de conturi (nu doar count)
- _create_email_body afișează solduri per cont + total RON
- Format tabel HTML frumos cu styling
- send_notifications.py citește date din JSON
- Sincronizare cu TelegramNotifier (deja avea solduri)

🤖 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-06 21:15:52 +02:00
parent 5aa4900b23
commit 2439d0b62a
2 changed files with 57 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
notifications.py
View File

@@ -23,13 +23,13 @@ class EmailNotifier:
self.config = config
self.enabled = config.EMAIL_ENABLED
def send(self, files: List[str], account_count: int) -> bool:
def send(self, files: List[str], accounts: list) -> bool:
"""
Send email with CSV attachments
Args:
files: List of file paths to attach
account_count: Number of accounts processed
accounts: List of account data with balances
Returns:
True if successful, False otherwise
@@ -51,7 +51,7 @@ class EmailNotifier:
msg['Subject'] = f'BTGO Scraper Results - {datetime.now().strftime("%Y-%m-%d %H:%M")}'
# Email body
body = self._create_email_body(files, account_count)
body = self._create_email_body(files, accounts)
msg.attach(MIMEText(body, 'html'))
# Attach files
@@ -67,7 +67,7 @@ class EmailNotifier:
# Check size limit (10MB typical SMTP limit)
if total_size > 10 * 1024 * 1024:
logging.warning(f"Email attachments exceed 10MB, creating ZIP archive")
return self._send_with_zip(files, account_count)
return self._send_with_zip(files, accounts)
with open(file_path, 'rb') as f:
part = MIMEBase('application', 'octet-stream')
@@ -94,7 +94,7 @@ class EmailNotifier:
logging.error(f"Failed to send email: {e}")
return False
def _send_with_zip(self, files: List[str], account_count: int) -> bool:
def _send_with_zip(self, files: List[str], accounts: list) -> bool:
"""Send email with files compressed as ZIP"""
try:
# Create ZIP archive
@@ -114,7 +114,7 @@ class EmailNotifier:
msg['To'] = self.config.EMAIL_TO
msg['Subject'] = f'BTGO Scraper Results (ZIP) - {datetime.now().strftime("%Y-%m-%d %H:%M")}'
body = self._create_email_body([str(zip_path)], account_count, is_zip=True)
body = self._create_email_body([str(zip_path)], accounts, is_zip=True)
msg.attach(MIMEText(body, 'html'))
with open(zip_path, 'rb') as f:
@@ -137,19 +137,39 @@ class EmailNotifier:
logging.error(f"Failed to send email with ZIP: {e}")
return False
def _create_email_body(self, files: List[str], account_count: int, is_zip: bool = False) -> str:
def _create_email_body(self, files: List[str], accounts: list, is_zip: bool = False) -> str:
"""Create HTML email body"""
file_list = '<br>'.join([f'• {Path(f).name}' for f in files])
file_count = 1 if is_zip else len(files)
# Calculate total balance
total_ron = sum(acc['sold'] for acc in accounts if acc.get('moneda') == 'RON')
# Build account balance list
account_balances = ""
for acc in accounts:
nume = acc['nume_cont']
sold = acc['sold']
moneda = acc['moneda']
account_balances += f'<tr><td style="padding: 8px; border-bottom: 1px solid #ecf0f1;">{nume}</td><td style="padding: 8px; text-align: right; border-bottom: 1px solid #ecf0f1;"><strong>{sold:,.2f} {moneda}</strong></td></tr>'
return f"""
<html>
<body style="font-family: Arial, sans-serif;">
<h2 style="color: #2c3e50;">BTGO Scraper Results</h2>
<p><strong>Execution time:</strong> {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</p>
<p><strong>Accounts processed:</strong> {account_count}</p>
<p><strong>Accounts processed:</strong> {len(accounts)}</p>
<p><strong>Files attached:</strong> {file_count}</p>
<hr>
<h3 style="color: #27ae60;">Solduri:</h3>
<table style="width: 100%; border-collapse: collapse; margin-bottom: 20px;">
{account_balances}
<tr style="background-color: #f8f9fa;">
<td style="padding: 12px; font-weight: bold;">TOTAL</td>
<td style="padding: 12px; text-align: right; font-weight: bold; color: #27ae60; font-size: 1.2em;">{total_ron:,.2f} RON</td>
</tr>
</table>
<hr>
<h3>{'Archive contents:' if is_zip else 'Attached files:'}</h3>
{file_list}
<hr>
@@ -479,7 +499,7 @@ class NotificationService:
# Send via email
if self.config.EMAIL_ENABLED:
results['email'] = self.email.send(files, account_count)
results['email'] = self.email.send(files, accounts)
# Send via Telegram
if self.config.TELEGRAM_ENABLED:

37
send_notifications.py
View File

@@ -4,6 +4,7 @@ Trimite ultimele fișiere CSV generate pe Email și Telegram
"""
import logging
import sys
import json
from pathlib import Path
from datetime import datetime
from config import Config
@@ -21,27 +22,27 @@ logging.basicConfig(
def find_latest_files(data_dir='./data', time_window_seconds=300):
"""
Găsește ultimele fișiere CSV generate
Găsește ultimele fișiere CSV generate și date despre conturi
Args:
data_dir: Directorul cu fișiere
time_window_seconds: Intervalul de timp (în secunde) pentru a considera fișierele din aceeași sesiune
Returns:
tuple: (solduri_csv_path, list_of_transaction_csvs)
tuple: (solduri_csv_path, list_of_transaction_csvs, accounts_data)
"""
data_path = Path(data_dir)
if not data_path.exists():
logging.error(f"Directorul {data_dir} nu există!")
return None, []
return None, [], []
# Găsește ultimul fișier solduri_*.csv
solduri_files = sorted(data_path.glob('solduri_*.csv'), key=lambda x: x.stat().st_mtime, reverse=True)
if not solduri_files:
logging.error("Nu s-a găsit niciun fișier solduri_*.csv!")
return None, []
return None, [], []
latest_solduri = solduri_files[0]
solduri_time = latest_solduri.stat().st_mtime
@@ -49,6 +50,22 @@ def find_latest_files(data_dir='./data', time_window_seconds=300):
logging.info(f"✓ Găsit fișier solduri: {latest_solduri.name}")
logging.info(f" Timestamp: {datetime.fromtimestamp(solduri_time).strftime('%Y-%m-%d %H:%M:%S')}")
# Găsește fișierul JSON corespunzător
json_filename = latest_solduri.stem + '.json'
json_path = data_path / json_filename
accounts_data = []
if json_path.exists():
try:
with open(json_path, 'r', encoding='utf-8') as f:
json_data = json.load(f)
accounts_data = json_data.get('conturi', [])
logging.info(f"✓ Găsit fișier JSON: {json_filename} ({len(accounts_data)} conturi)")
except Exception as e:
logging.warning(f"Nu s-a putut citi fișierul JSON: {e}")
else:
logging.warning(f"Nu s-a găsit fișierul JSON: {json_filename}")
# Găsește toate fișierele tranzactii_*.csv modificate în ultimele X secunde față de solduri
all_transaction_files = list(data_path.glob('tranzactii_*.csv'))
transaction_files = []
@@ -67,7 +84,7 @@ def find_latest_files(data_dir='./data', time_window_seconds=300):
logging.info(f"✓ Găsite {len(transaction_files)} fișiere tranzacții din aceeași sesiune")
return latest_solduri, transaction_files
return latest_solduri, transaction_files, accounts_data
def send_existing_files():
@@ -86,7 +103,7 @@ def send_existing_files():
return False
# Găsește ultimele fișiere
solduri_csv, transaction_csvs = find_latest_files(Config.OUTPUT_DIR)
solduri_csv, transaction_csvs, accounts_data = find_latest_files(Config.OUTPUT_DIR)
if not solduri_csv:
logging.error("Nu există fișiere de trimis!")
@@ -101,12 +118,14 @@ def send_existing_files():
logging.info(f"Total fișiere de trimis: {len(files_to_send)}")
logging.info("=" * 60)
# Estimează numărul de conturi din numărul de fișiere tranzacții
account_count = len(transaction_csvs)
# Dacă nu avem date despre conturi, creăm o listă goală
if not accounts_data:
logging.warning("Nu s-au găsit date despre conturi din JSON")
accounts_data = []
# Trimite notificările
service = NotificationService(Config)
results = service.send_all(files_to_send, account_count)
results = service.send_all(files_to_send, accounts_data)
# Afișează rezumat
logging.info("=" * 60)