Marius Mutu
e49e653e12
Rezolva problema cand selectoarele BT se schimba - acum incearca multiple strategii pentru a gasi username, password si submit button. Imbunatateste si gestionarea GDPR cookie banner. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
872 lines
36 KiB
Python
872 lines
36 KiB
Python
"""
|
|
BTGO Scraper - Automatizare citire solduri conturi bancare
|
|
Folosește Playwright pentru automatizare browser
|
|
"""
|
|
from playwright.sync_api import sync_playwright, TimeoutError as PlaywrightTimeout
|
|
import logging
|
|
import csv
|
|
import json
|
|
import time
|
|
import os
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from config import Config
|
|
|
|
|
|
class BTGoScraper:
|
|
"""Scraper pentru extragerea soldurilor de pe btgo.ro"""
|
|
|
|
def __init__(self):
|
|
"""Initializare scraper cu logging si configurare"""
|
|
self.config = Config()
|
|
Config.validate()
|
|
self._setup_logging()
|
|
self._ensure_directories()
|
|
self.page = None
|
|
self.login_page = None # Pagina de login (popup)
|
|
|
|
# Setup pentru progress updates prin Telegram (optional)
|
|
self.telegram_chat_id = os.getenv('TELEGRAM_CHAT_ID')
|
|
self.telegram_message_id = os.getenv('TELEGRAM_MESSAGE_ID')
|
|
self.progress_notifier = None
|
|
|
|
logging.info(f"Environment: TELEGRAM_CHAT_ID={self.telegram_chat_id}, TELEGRAM_MESSAGE_ID={self.telegram_message_id}")
|
|
|
|
# Inițializează notifier pentru progress updates dacă avem chat_id
|
|
if self.telegram_chat_id and self.telegram_message_id:
|
|
try:
|
|
from notifications import TelegramNotifier
|
|
self.progress_notifier = TelegramNotifier(self.config)
|
|
logging.info(f"Progress updates activate pentru chat_id={self.telegram_chat_id}, message_id={self.telegram_message_id}")
|
|
except Exception as e:
|
|
logging.warning(f"Nu am putut inițializa progress notifier: {e}")
|
|
else:
|
|
logging.warning("Progress updates dezactivate - lipsesc TELEGRAM_CHAT_ID sau TELEGRAM_MESSAGE_ID")
|
|
|
|
def _setup_logging(self):
|
|
"""Configurare logging in consola si fisier zilnic"""
|
|
# Creaza director logs daca nu exista
|
|
Path(self.config.LOG_DIR).mkdir(exist_ok=True)
|
|
|
|
# Nume fisier cu data curenta
|
|
log_file = Path(self.config.LOG_DIR) / f"scraper_{datetime.now().strftime('%Y-%m-%d')}.log"
|
|
|
|
# Format log
|
|
log_format = '[%(asctime)s] [%(levelname)s] %(message)s'
|
|
date_format = '%Y-%m-%d %H:%M:%S'
|
|
|
|
# Configurare logging
|
|
logging.basicConfig(
|
|
level=getattr(logging, self.config.LOG_LEVEL),
|
|
format=log_format,
|
|
datefmt=date_format,
|
|
handlers=[
|
|
logging.StreamHandler(), # Console
|
|
logging.FileHandler(log_file, encoding='utf-8') # Fisier
|
|
]
|
|
)
|
|
|
|
logging.info(f"Logging initializat: {log_file}")
|
|
|
|
def _ensure_directories(self):
|
|
"""Creaza directoarele data/ si logs/ daca nu exista"""
|
|
Path(self.config.OUTPUT_DIR).mkdir(exist_ok=True)
|
|
Path(self.config.LOG_DIR).mkdir(exist_ok=True)
|
|
logging.info(f"Directoare verificate: {self.config.OUTPUT_DIR}, {self.config.LOG_DIR}")
|
|
|
|
def _update_progress(self, message: str):
|
|
"""
|
|
Trimite update de progres către Telegram (editează mesajul inițial)
|
|
|
|
Args:
|
|
message: Mesajul de progres
|
|
"""
|
|
if self.progress_notifier and self.telegram_chat_id and self.telegram_message_id:
|
|
full_message = f"*{message}*"
|
|
self.progress_notifier._edit_message(
|
|
self.telegram_chat_id,
|
|
self.telegram_message_id,
|
|
full_message
|
|
)
|
|
logging.info(f"Progress update: {message}")
|
|
|
|
def _dismiss_gdpr_cookies(self, page):
|
|
"""
|
|
Inchide GDPR cookie banner daca este vizibil
|
|
|
|
Args:
|
|
page: Pagina Playwright pe care sa verifice
|
|
"""
|
|
try:
|
|
# Verifica daca exista gdprcookie-wrapper
|
|
gdpr_wrapper = page.locator(".gdprcookie-wrapper")
|
|
if gdpr_wrapper.is_visible(timeout=3000):
|
|
logging.info(" GDPR cookie banner detectat")
|
|
|
|
# Incearca diverse butoane de accept (in ordinea probabilitatii)
|
|
accept_selectors = [
|
|
".gdprcookie-wrapper button:has-text('Accept')",
|
|
".gdprcookie-wrapper button:has-text('Accepta')",
|
|
".gdprcookie-wrapper button:has-text('Sunt de acord')",
|
|
".gdprcookie-wrapper button:has-text('OK')",
|
|
".gdprcookie-wrapper .gdprcookie-buttons button:first-child",
|
|
".gdprcookie-wrapper button",
|
|
]
|
|
|
|
for selector in accept_selectors:
|
|
try:
|
|
accept_btn = page.locator(selector).first
|
|
if accept_btn.is_visible(timeout=1000):
|
|
accept_btn.click()
|
|
logging.info(f" [OK] Cookies acceptate (selector: {selector})")
|
|
time.sleep(1) # Asteapta sa dispara banner-ul
|
|
return True
|
|
except:
|
|
continue
|
|
|
|
logging.warning(" Nu am gasit buton de accept in GDPR wrapper")
|
|
return False
|
|
except:
|
|
logging.info(" Nu exista GDPR cookie banner (sau deja inchis)")
|
|
return False
|
|
|
|
def _find_username_field(self, page):
|
|
"""
|
|
Detecteaza inteligent campul de username folosind multiple strategii.
|
|
Ordinea: selectori specifici -> selectori generici -> detectie structurala
|
|
|
|
Returns:
|
|
Locator daca gasit, None altfel
|
|
"""
|
|
strategies = [
|
|
# 1. Selectori specifici BT (pot sa se schimbe)
|
|
("placeholder_exact", "ID logare"),
|
|
("placeholder_exact", "ID de logare"),
|
|
("id", "user"),
|
|
("name", "user"),
|
|
|
|
# 2. Selectori generici (mai stabili)
|
|
("placeholder_contains", "logare"),
|
|
("placeholder_contains", "user"),
|
|
("placeholder_contains", "utilizator"),
|
|
("name_contains", "user"),
|
|
("name_contains", "login"),
|
|
("id_contains", "user"),
|
|
("id_contains", "login"),
|
|
|
|
# 3. Selectori structurali (foarte stabili)
|
|
("css", "form input[type='text']:not([type='hidden'])"),
|
|
("css", "form input:not([type='password']):not([type='hidden']):not([type='submit'])"),
|
|
("css", "input[type='text']"),
|
|
("css", ".form-control[type='text']"),
|
|
|
|
# 4. Fallback - primul input vizibil care nu e password/submit
|
|
("first_text_input", None),
|
|
]
|
|
|
|
for strategy_type, value in strategies:
|
|
try:
|
|
field = self._try_field_strategy(page, strategy_type, value)
|
|
if field and field.is_visible(timeout=1000):
|
|
logging.info(f" [USERNAME] Gasit cu strategia: {strategy_type}='{value}'")
|
|
return field
|
|
except Exception:
|
|
continue
|
|
|
|
logging.error(" [USERNAME] Nu am gasit campul cu nicio strategie!")
|
|
return None
|
|
|
|
def _find_password_field(self, page):
|
|
"""
|
|
Detecteaza inteligent campul de parola.
|
|
Campul password e foarte stabil - type='password' e standard HTML.
|
|
|
|
Returns:
|
|
Locator daca gasit, None altfel
|
|
"""
|
|
strategies = [
|
|
# 1. Cel mai stabil - type='password' (standard HTML)
|
|
("css", "input[type='password']"),
|
|
|
|
# 2. Selectori specifici BT
|
|
("id", "password"),
|
|
("name", "password"),
|
|
("name", "pass"),
|
|
("placeholder_exact", "Parola"),
|
|
("placeholder_exact", "Password"),
|
|
|
|
# 3. Selectori generici
|
|
("placeholder_contains", "parola"),
|
|
("placeholder_contains", "password"),
|
|
("name_contains", "pass"),
|
|
("id_contains", "pass"),
|
|
|
|
# 4. Fallback structural
|
|
("css", "form input[type='password']"),
|
|
("css", ".form-control[type='password']"),
|
|
]
|
|
|
|
for strategy_type, value in strategies:
|
|
try:
|
|
field = self._try_field_strategy(page, strategy_type, value)
|
|
if field and field.is_visible(timeout=1000):
|
|
logging.info(f" [PASSWORD] Gasit cu strategia: {strategy_type}='{value}'")
|
|
return field
|
|
except Exception:
|
|
continue
|
|
|
|
logging.error(" [PASSWORD] Nu am gasit campul cu nicio strategie!")
|
|
return None
|
|
|
|
def _find_submit_button(self, page):
|
|
"""
|
|
Detecteaza inteligent butonul de submit.
|
|
|
|
Returns:
|
|
Locator daca gasit, None altfel
|
|
"""
|
|
strategies = [
|
|
# 1. Selectori specifici BT
|
|
("css", "input[value='Autentifică-te']"),
|
|
("css", "button:has-text('Autentifică-te')"),
|
|
("css", "input[value*='Autentific']"),
|
|
|
|
# 2. Selectori generici pentru login buttons
|
|
("css", "input[type='submit']"),
|
|
("css", "button[type='submit']"),
|
|
("css", "form button.btn-primary"),
|
|
("css", "form input.btn-primary"),
|
|
|
|
# 3. Text-based (mai putin stabil dar functional)
|
|
("text_contains", "Login"),
|
|
("text_contains", "Conectare"),
|
|
("text_contains", "Autentificare"),
|
|
("text_contains", "Intra"),
|
|
("text_contains", "Submit"),
|
|
|
|
# 4. Fallback - orice buton din form
|
|
("css", "form button"),
|
|
("css", "form input[type='button']"),
|
|
("css", ".btn-primary"),
|
|
]
|
|
|
|
for strategy_type, value in strategies:
|
|
try:
|
|
button = self._try_button_strategy(page, strategy_type, value)
|
|
if button and button.is_visible(timeout=1000):
|
|
logging.info(f" [SUBMIT] Gasit cu strategia: {strategy_type}='{value}'")
|
|
return button
|
|
except Exception:
|
|
continue
|
|
|
|
logging.error(" [SUBMIT] Nu am gasit butonul cu nicio strategie!")
|
|
return None
|
|
|
|
def _try_field_strategy(self, page, strategy_type, value):
|
|
"""Helper pentru a incerca o strategie de gasire a unui camp"""
|
|
if strategy_type == "placeholder_exact":
|
|
return page.get_by_placeholder(value, exact=True)
|
|
elif strategy_type == "placeholder_contains":
|
|
return page.locator(f"input[placeholder*='{value}' i]").first
|
|
elif strategy_type == "id":
|
|
return page.locator(f"#{value}")
|
|
elif strategy_type == "id_contains":
|
|
return page.locator(f"input[id*='{value}' i]").first
|
|
elif strategy_type == "name":
|
|
return page.locator(f"input[name='{value}']")
|
|
elif strategy_type == "name_contains":
|
|
return page.locator(f"input[name*='{value}' i]").first
|
|
elif strategy_type == "css":
|
|
return page.locator(value).first
|
|
elif strategy_type == "label":
|
|
return page.get_by_label(value)
|
|
elif strategy_type == "first_text_input":
|
|
# Gaseste primul input care nu e password, hidden sau submit
|
|
inputs = page.locator("input:visible").all()
|
|
for inp in inputs:
|
|
try:
|
|
inp_type = inp.get_attribute("type", timeout=500) or "text"
|
|
if inp_type.lower() not in ["password", "hidden", "submit", "button", "checkbox", "radio"]:
|
|
return inp
|
|
except:
|
|
continue
|
|
return None
|
|
|
|
def _try_button_strategy(self, page, strategy_type, value):
|
|
"""Helper pentru a incerca o strategie de gasire a butonului"""
|
|
if strategy_type == "css":
|
|
return page.locator(value).first
|
|
elif strategy_type == "text_contains":
|
|
return page.locator(f"button:has-text('{value}'), input[value*='{value}' i]").first
|
|
elif strategy_type == "role":
|
|
return page.get_by_role("button", name=value)
|
|
return None
|
|
|
|
def run(self):
|
|
"""Entry point principal - orchestreaza tot flow-ul"""
|
|
try:
|
|
# Check dacă rulăm în mod balances_only
|
|
balances_only = os.getenv('BALANCES_ONLY', 'false').lower() == 'true'
|
|
|
|
logging.info("=" * 60)
|
|
if balances_only:
|
|
logging.info("Start BTGO Scraper (DOAR SOLDURI)")
|
|
else:
|
|
logging.info("Start BTGO Scraper")
|
|
logging.info("=" * 60)
|
|
|
|
with sync_playwright() as p:
|
|
# Lansare browser
|
|
browser = p.chromium.launch(
|
|
headless=self.config.HEADLESS,
|
|
slow_mo=100 if not self.config.HEADLESS else 0 # Slow motion pentru debugging
|
|
)
|
|
|
|
# Creaza pagina cu viewport standard
|
|
self.page = browser.new_page(viewport={'width': 1920, 'height': 1080})
|
|
logging.info(f"Browser lansat (headless={self.config.HEADLESS})")
|
|
|
|
# Flow complet
|
|
self.login()
|
|
self.handle_2fa_wait()
|
|
self._update_progress("2FA aprobat! Incarc conturi...")
|
|
accounts = self.read_accounts()
|
|
csv_path, json_path = self.save_results(accounts)
|
|
|
|
# Descarcă tranzacții pentru toate conturile (doar dacă nu e balances_only)
|
|
downloaded_files = []
|
|
if balances_only:
|
|
logging.info("Mod DOAR SOLDURI - skip download tranzactii")
|
|
elif self.config.DOWNLOAD_TRANSACTIONS:
|
|
downloaded_files = self.download_transactions(accounts)
|
|
else:
|
|
logging.info("Download tranzacții dezactivat (DOWNLOAD_TRANSACTIONS=false)")
|
|
|
|
# Screenshot final de confirmare
|
|
screenshot_path = Path(self.config.OUTPUT_DIR) / f"dashboard_{datetime.now().strftime('%Y-%m-%d_%H-%M-%S')}.png"
|
|
self.page.screenshot(path=str(screenshot_path))
|
|
logging.info(f"Screenshot final salvat: {screenshot_path}")
|
|
|
|
# Trimite notificări (email, Discord)
|
|
if self.config.ENABLE_NOTIFICATIONS:
|
|
self.send_notifications(csv_path, downloaded_files, accounts)
|
|
|
|
# Afișează rezumat final
|
|
logging.info("=" * 60)
|
|
logging.info("REZUMAT FINAL")
|
|
logging.info("=" * 60)
|
|
logging.info(f"Conturi citite: {len(accounts)}")
|
|
if self.config.DOWNLOAD_TRANSACTIONS:
|
|
logging.info(f"Tranzacții descărcate: {len(downloaded_files)}")
|
|
for df in downloaded_files:
|
|
logging.info(f" - {df['cont']}: {df['fisier']}")
|
|
logging.info("=" * 60)
|
|
|
|
browser.close()
|
|
# Nu mai trimitem update progress aici - mesajul final cu soldurile a fost deja editat de notifications.py
|
|
logging.info("=" * 60)
|
|
logging.info("✓ Scraper finalizat cu succes!")
|
|
logging.info("=" * 60)
|
|
return True
|
|
|
|
except Exception as e:
|
|
self._handle_error(e)
|
|
return False
|
|
|
|
def login(self):
|
|
"""Autentificare cu username si password"""
|
|
self._update_progress("Deschid pagina de login...")
|
|
logging.info("Navigare catre https://go.bancatransilvania.ro/")
|
|
self.page.goto('https://go.bancatransilvania.ro/', wait_until='networkidle')
|
|
logging.info("Pagina incarcata")
|
|
|
|
try:
|
|
# Cookie consent - asteapta si accepta (GDPR wrapper)
|
|
logging.info("Verificare GDPR cookie banner...")
|
|
self._dismiss_gdpr_cookies(self.page)
|
|
|
|
# Click pe butonul LOGIN - deschide popup
|
|
logging.info("Click pe butonul LOGIN...")
|
|
with self.page.expect_popup() as popup_info:
|
|
login_link = self.page.get_by_role("link", name="LOGIN")
|
|
login_link.click()
|
|
|
|
# Preia referinta la popup-ul de login
|
|
self.login_page = popup_info.value
|
|
logging.info("✓ Popup login deschis")
|
|
|
|
# Verifica GDPR cookies si pe popup
|
|
self._dismiss_gdpr_cookies(self.login_page)
|
|
|
|
# Asteapta sa se incarce pagina de login
|
|
time.sleep(2)
|
|
|
|
# Screenshot debug pentru a vedea starea paginii
|
|
debug_path = Path(self.config.OUTPUT_DIR) / f"debug_login_popup_{datetime.now().strftime('%Y%m%d_%H%M%S')}.png"
|
|
self.login_page.screenshot(path=str(debug_path))
|
|
logging.info(f"Screenshot debug salvat: {debug_path}")
|
|
|
|
# Completare username - detectie inteligenta cu fallback
|
|
logging.info("Detectie camp username...")
|
|
username_field = self._find_username_field(self.login_page)
|
|
if not username_field:
|
|
raise Exception("Nu am gasit campul de username cu nicio strategie!")
|
|
username_field.fill(self.config.BTGO_USERNAME)
|
|
logging.info("[OK] Username completat")
|
|
|
|
# Completare password - detectie inteligenta cu fallback
|
|
logging.info("Detectie camp password...")
|
|
password_field = self._find_password_field(self.login_page)
|
|
if not password_field:
|
|
raise Exception("Nu am gasit campul de parola cu nicio strategie!")
|
|
password_field.fill(self.config.BTGO_PASSWORD)
|
|
logging.info("[OK] Password completat")
|
|
|
|
# Click pe butonul de submit - detectie inteligenta cu fallback
|
|
logging.info("Detectie buton submit...")
|
|
submit_button = self._find_submit_button(self.login_page)
|
|
if not submit_button:
|
|
raise Exception("Nu am gasit butonul de submit cu nicio strategie!")
|
|
submit_button.click()
|
|
logging.info("[OK] Credentials trimise, astept 2FA...")
|
|
self._update_progress("Astept aprobare 2FA pe telefon...")
|
|
|
|
except PlaywrightTimeout as e:
|
|
logging.error(f"Timeout la login: {e}")
|
|
raise Exception("Nu am gasit elementele de login. Verifica selectors!")
|
|
except Exception as e:
|
|
logging.error(f"Eroare la login: {e}")
|
|
raise
|
|
|
|
def handle_2fa_wait(self):
|
|
"""Asteapta aprobare 2FA cu auto-detect"""
|
|
|
|
logging.info("=" * 60)
|
|
logging.info("🔐 APROBARE 2FA NECESARA")
|
|
logging.info("=" * 60)
|
|
logging.info("Verifica aplicatia George pe telefon si aproba autentificarea.")
|
|
logging.info(f"Timeout: {self.config.TIMEOUT_2FA_SECONDS} secunde")
|
|
logging.info("=" * 60)
|
|
|
|
# Detectam login reusit prin schimbarea URL-ului la goapp.bancatransilvania.ro
|
|
# sau prin aparitia butonului de conturi (#accountsBtn)
|
|
post_login_selectors = [
|
|
"#accountsBtn", # Butonul pentru conturi (apare post-login)
|
|
]
|
|
|
|
start_time = datetime.now()
|
|
timeout = self.config.TIMEOUT_2FA_SECONDS
|
|
|
|
while (datetime.now() - start_time).seconds < timeout:
|
|
# Verifică dacă butonul de conturi este CLICKABLE (nu doar vizibil)
|
|
# Aceasta este singura verificare sigură - butonul apare doar după 2FA reușit
|
|
try:
|
|
# Așteaptă ca #accountsBtn să fie clickable (nu doar vizibil în DOM)
|
|
accounts_btn = self.login_page.locator("#accountsBtn")
|
|
if accounts_btn.is_visible(timeout=1000):
|
|
# Verifică că este și clickable (enabled)
|
|
if accounts_btn.is_enabled():
|
|
logging.info("✓ Autentificare 2FA reusita! (Buton conturi activ)")
|
|
time.sleep(2) # Asteapta ca pagina sa se stabilizeze complet
|
|
# Update page reference la login_page pentru restul operatiilor
|
|
self.page = self.login_page
|
|
return True
|
|
except:
|
|
pass
|
|
|
|
# Afiseaza countdown
|
|
elapsed = (datetime.now() - start_time).seconds
|
|
remaining = timeout - elapsed
|
|
print(f"\rAsteptam aprobare 2FA... {remaining}s ramas", end='', flush=True)
|
|
|
|
time.sleep(2)
|
|
|
|
print() # New line dupa countdown
|
|
raise TimeoutError(f"Timeout 2FA dupa {timeout} secunde. Verifica ca ai aprobat pe telefon!")
|
|
|
|
def download_transactions(self, accounts):
|
|
"""Descarca CSV-uri cu tranzactiile pentru fiecare cont"""
|
|
logging.info("=" * 60)
|
|
logging.info("Descarcare tranzactii pentru toate conturile...")
|
|
logging.info("=" * 60)
|
|
|
|
downloaded_files = []
|
|
|
|
# IMPORTANT: Collapse toate conturile mai intai
|
|
logging.info("Collapse toate conturile...")
|
|
all_expanded = self.page.locator(".mat-icon.rotate-90").all()
|
|
for expanded_icon in all_expanded:
|
|
try:
|
|
expanded_icon.click()
|
|
time.sleep(0.3)
|
|
except:
|
|
pass
|
|
|
|
time.sleep(1)
|
|
logging.info("✓ Toate conturile sunt collapse")
|
|
|
|
# Re-gaseste toate cardurile de conturi
|
|
all_cards = self.page.locator("fba-account-details-card").all()
|
|
logging.info(f"Gasit {len(all_cards)} carduri de conturi")
|
|
|
|
for idx, account in enumerate(accounts, 1):
|
|
try:
|
|
nume_cont = account['nume_cont']
|
|
iban = account['iban']
|
|
|
|
self._update_progress(f"Descarc tranzactii ({idx}/{len(accounts)})...")
|
|
logging.info(f"[{idx}/{len(accounts)}] Descarcare tranzactii pentru: {nume_cont}")
|
|
|
|
# Doar pentru PRIMUL cont trebuie expand + click Tranzacții
|
|
# Pentru restul, suntem deja pe pagina de tranzacții (din selectarea din modal)
|
|
if idx == 1:
|
|
# Primul cont - expand și click Tranzacții
|
|
if idx - 1 >= len(all_cards):
|
|
logging.error(f" ✗ Nu exista card la pozitia {idx-1}")
|
|
continue
|
|
|
|
card = all_cards[idx - 1]
|
|
|
|
# Expand contul (click pe săgeată)
|
|
expand_button = card.locator(".collapse-account-btn").first
|
|
expand_button.click()
|
|
time.sleep(2) # Așteaptă expandare
|
|
logging.info(f" Contul expandat")
|
|
|
|
# Click pe butonul Tranzacții
|
|
try:
|
|
transactions_button = card.locator(".account-transactions-btn").first
|
|
transactions_button.click()
|
|
time.sleep(3) # Așteaptă încărcarea paginii cu tranzacții
|
|
logging.info(f" Click pe buton Tranzactii - pagina se incarca...")
|
|
except Exception as e:
|
|
logging.error(f" ✗ Nu am gasit butonul Tranzactii: {e}")
|
|
try:
|
|
expand_button.click()
|
|
time.sleep(0.5)
|
|
except:
|
|
pass
|
|
continue
|
|
else:
|
|
# Conturile 2-5: suntem deja pe pagina de tranzacții (din modal)
|
|
logging.info(f" Deja pe pagina tranzactii (selectat din modal)")
|
|
time.sleep(2) # Așteaptă stabilizare pagină
|
|
|
|
# Așteaptă să apară butonul CSV (indica că pagina s-a încărcat)
|
|
try:
|
|
self.page.wait_for_selector('button:has-text("CSV")', timeout=5000)
|
|
logging.info(f" Buton CSV detectat")
|
|
except:
|
|
logging.warning(f" Timeout asteptand butonul CSV")
|
|
|
|
# Click pe butonul CSV și așteaptă download
|
|
try:
|
|
with self.page.expect_download(timeout=15000) as download_info:
|
|
csv_button = self.page.get_by_role("button", name="CSV")
|
|
csv_button.click()
|
|
logging.info(f" Click pe butonul CSV - astept download...")
|
|
|
|
download = download_info.value
|
|
|
|
# Salvează fișierul cu un nume descriptiv
|
|
timestamp = datetime.now().strftime('%Y-%m-%d_%H-%M-%S')
|
|
nume_safe = nume_cont.replace(' ', '_').replace('/', '_')
|
|
filename = f"tranzactii_{nume_safe}_{timestamp}.csv"
|
|
save_path = Path(self.config.OUTPUT_DIR) / filename
|
|
|
|
download.save_as(save_path)
|
|
logging.info(f" ✓ Salvat: {save_path}")
|
|
|
|
downloaded_files.append({
|
|
'cont': nume_cont,
|
|
'iban': iban,
|
|
'fisier': str(save_path)
|
|
})
|
|
|
|
except Exception as e:
|
|
logging.error(f" ✗ Eroare la descarcarea CSV: {e}")
|
|
|
|
# Navighează înapoi la lista de conturi
|
|
try:
|
|
# Click pe butonul back/close (săgeată stânga sau X)
|
|
back_button = self.page.locator('button[aria-label="Back"], .back-button, #selectAccountBtn').first
|
|
back_button.click()
|
|
time.sleep(1.5)
|
|
logging.info(f" Navigat inapoi - verific modal...")
|
|
except Exception as e:
|
|
logging.warning(f" Nu am putut naviga inapoi: {e}")
|
|
time.sleep(1)
|
|
|
|
# Verifică dacă a apărut modal de selectare cont
|
|
try:
|
|
modal_visible = self.page.locator('.modal-content').is_visible(timeout=2000)
|
|
if modal_visible and idx < len(accounts):
|
|
logging.info(f" Modal detectat - selectez contul urmator...")
|
|
|
|
# Calculează ID-ul contului următor
|
|
next_account = accounts[idx] # idx este 0-indexed pentru next
|
|
next_iban = next_account['iban']
|
|
next_iban_digits = ''.join(filter(str.isdigit, next_iban))[-10:]
|
|
next_account_id = f"accountC14RONCRT{next_iban_digits}"
|
|
|
|
# Click pe contul următor din modal
|
|
modal_account = self.page.locator(f'#{next_account_id}').first
|
|
modal_account.click()
|
|
time.sleep(2)
|
|
logging.info(f" ✓ Selectat cont din modal: {next_account['nume_cont']}")
|
|
else:
|
|
# Nu e modal - e ultima iteratie sau nu a aparut modal
|
|
logging.info(f" Nu e modal - continuam normal")
|
|
except Exception as e:
|
|
logging.warning(f" Eroare verificare modal: {e}")
|
|
|
|
# Re-găsește cardurile (pentru flow normal fără modal)
|
|
try:
|
|
all_cards = self.page.locator("fba-account-details-card").all()
|
|
except:
|
|
pass
|
|
|
|
except Exception as e:
|
|
logging.error(f" ✗ Eroare la descarcarea tranzactiilor pentru {nume_cont}: {e}")
|
|
# Încearcă să navighezi înapoi
|
|
try:
|
|
self.page.keyboard.press("Escape")
|
|
time.sleep(1)
|
|
except:
|
|
pass
|
|
continue
|
|
|
|
logging.info("=" * 60)
|
|
logging.info(f"✓ Descarcate {len(downloaded_files)}/{len(accounts)} fisiere CSV cu tranzactii")
|
|
logging.info("=" * 60)
|
|
|
|
return downloaded_files
|
|
|
|
def read_accounts(self):
|
|
"""Extrage soldurile tuturor conturilor"""
|
|
logging.info("Citire conturi si solduri...")
|
|
|
|
try:
|
|
# Click pe butonul de conturi pentru a afisa lista
|
|
logging.info("Click pe butonul Conturi...")
|
|
# Simplu selector - doar butonul, nu div-ul interior
|
|
accounts_button = self.page.locator("#accountsBtn")
|
|
accounts_button.click()
|
|
time.sleep(3) # Asteapta ca lista sa se incarce
|
|
logging.info("✓ Sectiunea conturi deschisa")
|
|
|
|
# Update progres DUPĂ ce lista de conturi s-a încărcat
|
|
self._update_progress("Citesc solduri conturi...")
|
|
|
|
# Gaseste toate cardurile de conturi
|
|
account_cards = self.page.locator("fba-account-details-card").all()
|
|
logging.info(f"Gasit {len(account_cards)} conturi")
|
|
|
|
accounts = []
|
|
|
|
for idx, card in enumerate(account_cards, 1):
|
|
try:
|
|
# Extrage nume cont (din tag <h4>)
|
|
nume = card.locator("h4.fw-normal.mb-0").inner_text()
|
|
|
|
# Extrage IBAN (din span cu clasa specific)
|
|
iban = card.locator("span.small.text-grayscale-label").inner_text()
|
|
|
|
# Extrage sold (din <strong> cu clasa 'sold')
|
|
sold_text = card.locator("strong.sold").inner_text()
|
|
|
|
# Parsing sold: "7,223.26 RON" -> sold_numeric=7223.26, moneda="RON"
|
|
# Elimina spatii, inlocuieste virgula cu punct
|
|
sold_parts = sold_text.strip().split()
|
|
|
|
if len(sold_parts) >= 2:
|
|
sold_str = sold_parts[0] # "7,223.26"
|
|
moneda = sold_parts[1] # "RON"
|
|
else:
|
|
sold_str = sold_text
|
|
moneda = "RON" # Default
|
|
|
|
# Converteste la float (elimina virgule care sunt separator de mii)
|
|
# Format: "7,223.26" -> 7223.26
|
|
sold_numeric = float(sold_str.replace(',', ''))
|
|
|
|
# Determina tip cont din nume (optional)
|
|
nume_lower = nume.lower()
|
|
if 'colector' in nume_lower:
|
|
tip_cont = 'colector'
|
|
elif 'profit' in nume_lower:
|
|
tip_cont = 'profit'
|
|
elif 'operationale' in nume_lower or 'operational' in nume_lower:
|
|
tip_cont = 'operational'
|
|
elif 'taxe' in nume_lower:
|
|
tip_cont = 'taxe'
|
|
elif 'antreprenor' in nume_lower:
|
|
tip_cont = 'antreprenor'
|
|
else:
|
|
tip_cont = 'curent'
|
|
|
|
accounts.append({
|
|
'nume_cont': nume.strip(),
|
|
'iban': iban.strip(),
|
|
'sold': sold_numeric,
|
|
'moneda': moneda.strip(),
|
|
'tip_cont': tip_cont
|
|
})
|
|
|
|
logging.info(f" [{idx}] {nume}: {sold_numeric} {moneda}")
|
|
|
|
except Exception as e:
|
|
logging.warning(f"Eroare la citirea contului {idx}: {e}")
|
|
continue
|
|
|
|
if not accounts:
|
|
raise Exception("Nu am putut citi niciun cont din lista!")
|
|
|
|
logging.info(f"✓ Citite {len(accounts)} conturi cu succes")
|
|
self._update_progress(f"Gasite {len(accounts)} conturi")
|
|
return accounts
|
|
|
|
except Exception as e:
|
|
logging.error(f"Eroare la citirea conturilor: {e}")
|
|
raise
|
|
|
|
def save_results(self, accounts):
|
|
"""Salveaza rezultate in CSV si JSON"""
|
|
if not accounts:
|
|
logging.warning("Nu exista conturi de salvat!")
|
|
return
|
|
|
|
timestamp = datetime.now().strftime('%Y-%m-%d_%H-%M-%S')
|
|
|
|
# Salvare CSV
|
|
csv_path = Path(self.config.OUTPUT_DIR) / f'solduri_{timestamp}.csv'
|
|
with open(csv_path, 'w', newline='', encoding='utf-8') as f:
|
|
fieldnames = ['timestamp', 'nume_cont', 'iban', 'sold', 'moneda', 'tip_cont']
|
|
writer = csv.DictWriter(f, fieldnames=fieldnames)
|
|
writer.writeheader()
|
|
|
|
for account in accounts:
|
|
row = account.copy()
|
|
row['timestamp'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
|
writer.writerow(row)
|
|
|
|
logging.info(f"✓ CSV salvat: {csv_path}")
|
|
|
|
# Salvare JSON
|
|
json_path = Path(self.config.OUTPUT_DIR) / f'solduri_{timestamp}.json'
|
|
|
|
# Calculeaza metadata
|
|
currencies = list(set(a.get('moneda', 'N/A') for a in accounts))
|
|
total_ron = sum(a['sold'] for a in accounts if a.get('moneda') == 'RON')
|
|
|
|
output = {
|
|
'metadata': {
|
|
'timestamp': datetime.now().isoformat(),
|
|
'scraper_version': '1.0.0',
|
|
'total_accounts': len(accounts),
|
|
'total_ron': round(total_ron, 2),
|
|
'currencies': currencies
|
|
},
|
|
'accounts': accounts
|
|
}
|
|
|
|
with open(json_path, 'w', encoding='utf-8') as f:
|
|
json.dump(output, f, indent=2, ensure_ascii=False)
|
|
|
|
logging.info(f"✓ JSON salvat: {json_path}")
|
|
logging.info(f"Total conturi: {len(accounts)}, Total RON: {total_ron:.2f}")
|
|
|
|
return csv_path, json_path
|
|
|
|
def send_notifications(self, csv_path, downloaded_files, accounts):
|
|
"""
|
|
Trimite notificări email și Discord cu fișierele descărcate
|
|
|
|
Args:
|
|
csv_path: Calea către fișierul CSV cu solduri
|
|
downloaded_files: Lista de dicționare cu fișierele de tranzacții descărcate
|
|
accounts: Lista cu datele conturilor (solduri, nume, etc.)
|
|
"""
|
|
try:
|
|
from notifications import NotificationService
|
|
|
|
self._update_progress("Trimit rezultate...")
|
|
logging.info("=" * 60)
|
|
logging.info("TRIMITERE NOTIFICĂRI")
|
|
logging.info("=" * 60)
|
|
|
|
# Colectează toate fișierele de trimis
|
|
files_to_send = [str(csv_path)] # CSV cu solduri
|
|
|
|
# Adaugă toate CSV-urile cu tranzacții
|
|
for df in downloaded_files:
|
|
if 'fisier' in df and Path(df['fisier']).exists():
|
|
files_to_send.append(df['fisier'])
|
|
logging.info(f"Adăugat pentru notificare: {Path(df['fisier']).name}")
|
|
|
|
logging.info(f"Total fișiere de trimis: {len(files_to_send)}")
|
|
|
|
# Trimite prin toate canalele configurate
|
|
service = NotificationService(self.config)
|
|
logging.info(f"Calling send_all with telegram_message_id={self.telegram_message_id}, telegram_chat_id={self.telegram_chat_id}")
|
|
results = service.send_all(
|
|
files_to_send,
|
|
accounts,
|
|
telegram_message_id=self.telegram_message_id,
|
|
telegram_chat_id=self.telegram_chat_id
|
|
)
|
|
|
|
# Afișează rezumat
|
|
if self.config.EMAIL_ENABLED:
|
|
status = "✓ Succes" if results['email'] else "✗ Eșuat"
|
|
logging.info(f"Email: {status}")
|
|
|
|
if self.config.TELEGRAM_ENABLED:
|
|
status = "✓ Succes" if results['telegram'] else "✗ Eșuat"
|
|
logging.info(f"Telegram: {status}")
|
|
|
|
logging.info("=" * 60)
|
|
|
|
except Exception as e:
|
|
logging.warning(f"Notificările au eșuat: {e}")
|
|
logging.warning("Scraper-ul a finalizat cu succes, dar notificările nu au fost trimise")
|
|
# Nu aruncă excepția mai departe - notificările sunt opționale
|
|
|
|
def _handle_error(self, error):
|
|
"""Error handling cu screenshot"""
|
|
logging.error("=" * 60)
|
|
logging.error(f"EROARE: {error}")
|
|
logging.error("=" * 60, exc_info=True)
|
|
|
|
# Salvare screenshot la eroare
|
|
if self.page and self.config.SCREENSHOT_ON_ERROR:
|
|
try:
|
|
error_path = Path(self.config.OUTPUT_DIR) / f"error_{datetime.now().strftime('%Y%m%d_%H%M%S')}.png"
|
|
self.page.screenshot(path=str(error_path))
|
|
logging.error(f"Screenshot eroare salvat: {error_path}")
|
|
except Exception as e:
|
|
logging.error(f"Nu am putut salva screenshot: {e}")
|
|
|
|
|
|
def main():
|
|
"""Functie main pentru rulare script"""
|
|
try:
|
|
scraper = BTGoScraper()
|
|
success = scraper.run()
|
|
exit(0 if success else 1)
|
|
except ValueError as e:
|
|
# Eroare de configurare
|
|
logging.error(f"Eroare configurare: {e}")
|
|
exit(4)
|
|
except NotImplementedError as e:
|
|
# Selectors nu sunt completati
|
|
print(str(e))
|
|
exit(5)
|
|
except Exception as e:
|
|
logging.error(f"Eroare neasteptata: {e}", exc_info=True)
|
|
exit(99)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|