"""
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, wait_for_banner=True):
"""
Inchide GDPR cookie banner daca este vizibil
Args:
page: Pagina Playwright pe care sa verifice
wait_for_banner: Daca True, asteapta aparitia banner-ului inainte de a incerca
"""
try:
# Asteapta ca banner-ul GDPR sa apara (poate intarzia dupa page load)
if wait_for_banner:
try:
page.wait_for_selector(".gdprcookie-wrapper", timeout=5000, state="visible")
logging.info(" [INFO] Banner GDPR detectat, incerc sa-l inchid...")
except:
# Banner-ul nu a aparut in 5 secunde, continuam
pass
# Strategii pentru cookie consent (in ordinea probabilitatii)
cookie_strategies = [
# 1. Noul buton BT (2024+)
("role", "button", "Accept toate"),
("role", "button", "Accepta toate"),
("role", "button", "Acceptă toate"),
# 2. Vechiul GDPR wrapper - text specific
("css", ".gdprcookie-wrapper button:has-text('Accept')"),
("css", ".gdprcookie-wrapper button:has-text('Sunt de acord')"),
("css", ".gdprcookie-wrapper button:has-text('Accepta')"),
("css", ".gdprcookie-wrapper button:has-text('Acceptă')"),
("css", ".gdprcookie-wrapper button:has-text('OK')"),
# 3. GDPR wrapper - orice buton (fallback agresiv)
("css", ".gdprcookie-wrapper button"),
("css", ".gdprcookie-wrapper .btn"),
("css", ".gdprcookie-wrapper [role='button']"),
# 4. Fallback generic
("role", "button", "Accept"),
("role", "button", "Accepta"),
("role", "button", "OK"),
]
for strategy in cookie_strategies:
try:
if strategy[0] == "role":
btn = page.get_by_role(strategy[1], name=strategy[2])
else:
btn = page.locator(strategy[2]).first
if btn.is_visible(timeout=3000):
btn.click(force=True)
logging.info(f" [OK] Cookies acceptate ({strategy})")
time.sleep(1)
# Verifica ca banner-ul a disparut
try:
page.wait_for_selector(".gdprcookie-wrapper", timeout=2000, state="hidden")
logging.info(" [OK] Banner GDPR inchis cu succes")
except:
# Incearca inca o data cu force click
logging.info(" [WARN] Banner inca vizibil, reincercare...")
try:
btn.click(force=True)
time.sleep(1)
except:
pass
return True
except:
continue
logging.info(" Nu exista cookie banner (sau deja acceptat)")
return False
except:
logging.info(" Nu exista cookie banner (sau deja acceptat)")
return False
def _dismiss_one_time_consent(self, page):
"""
Inchide dialoguri one-time (ex: 'Am inteles') daca apar
Args:
page: Pagina Playwright pe care sa verifice
"""
consent_buttons = [
("text", "Am înțeles"),
("text", "Am inteles"),
("role", "button", "Am înțeles"),
("role", "button", "OK"),
("role", "button", "Continua"),
]
for strategy in consent_buttons:
try:
if strategy[0] == "text":
btn = page.get_by_text(strategy[1], exact=True)
elif strategy[0] == "role":
btn = page.get_by_role(strategy[1], name=strategy[2])
if btn.is_visible(timeout=2000):
btn.click()
logging.info(f" [OK] Consent inchis ({strategy})")
time.sleep(1)
return True
except:
continue
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
# Texte care apar in mesajele BT cand autentificarea e respinsa
LOGIN_ERROR_PATTERNS = (
"invalid", "incorect", "gresit", "greșit", "eronat",
"blocat", "suspendat", "date de autentificare",
)
def _debug_shot(self, page, tag):
"""Salveaza un screenshot de diagnostic; nu arunca daca esueaza."""
try:
path = Path(self.config.OUTPUT_DIR) / f"debug_{tag}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.png"
page.screenshot(path=str(path))
logging.info(f"Screenshot diagnostic salvat: {path}")
return path
except Exception as e:
logging.warning(f"Nu am putut salva screenshot ({tag}): {e}")
return None
def _detect_login_error(self, page):
"""
Returneaza textul erorii afisate de pagina de login (parola gresita,
cont blocat etc.) sau None. Fara asta, credentialele respinse arata
identic cu un 2FA neaprobat: asteptam degeaba pana la timeout.
"""
try:
candidates = page.locator(
"mat-error, .mat-error, [role='alert'], .alert, "
".error, .error-message, .invalid-feedback"
)
count = min(candidates.count(), 10)
except Exception:
return None
for i in range(count):
try:
el = candidates.nth(i)
if not el.is_visible(timeout=500):
continue
text = (el.inner_text() or "").strip()
except Exception:
continue
if text and any(pat in text.lower() for pat in self.LOGIN_ERROR_PATTERNS):
return " ".join(text.split())
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)
# Nu `with`: iesirea din el opreste Playwright inainte de except,
# iar screenshot-ul din _handle_error pica ("Event loop is closed").
p = sync_playwright().start()
try:
# Viewport 1366x768: BT afiseaza un splash "redirect to store"
# sub ~1100px latime; 1366 e sigur peste prag. Fata de 1920x1080
# randeaza ~50% mai putini pixeli -> CPU mult mai mic la randarea
# software din Session 0 (serviciu Windows, fara GPU).
# --window-size forteaza dimensiunea la nivel de SO (viewport-ul
# CSS de context nu se propaga la fereastra fizica in session 0).
launch_args = [
'--window-size=1366,768',
# Flag-uri care taie overhead inutil in Session 0 (fara GPU)
'--disable-gpu',
'--disable-dev-shm-usage',
'--disable-extensions',
'--disable-background-networking',
'--disable-renderer-backgrounding',
'--disable-features=Translate',
]
context_kwargs = {'viewport': {'width': 1366, 'height': 768}}
if self.config.HEADLESS:
# Playwright 1.48: headless=True foloseste VECHIUL mod headless,
# blocat de WAF-ul Akamai al BT. Fortam NOUL mod headless
# (--headless=new) cu headless=False ca Playwright sa nu adauge
# --headless=old. Randeaza ca un Chrome real, fara fereastra ->
# CPU mult mai mic decat headed software-rendering in session 0.
launch_args.append('--headless=new')
# UA-ul noului mod headless contine 'HeadlessChrome' - semnatura
# pe care o detecteaza Akamai. Il inlocuim cu UA de Chrome normal.
context_kwargs['user_agent'] = (
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) '
'AppleWebKit/537.36 (KHTML, like Gecko) '
'Chrome/130.0.0.0 Safari/537.36'
)
browser = p.chromium.launch(
headless=False,
slow_mo=0 if self.config.HEADLESS else 100,
args=launch_args,
)
context = browser.new_context(**context_kwargs)
self.page = context.new_page()
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
finally:
p.stop()
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...")
# Din 09.2026 go.bancatransilvania.ro redirecteaza (302) spre pagina de
# marketing, fara link Login. Formularul e servit direct de goapp
# (/ -> /app/auth/login), in acelasi tab, fara popup.
# 'networkidle' nu se atinge niciodata (pixeli de tracking) -> 'domcontentloaded'
# + asteptarea pe input#user de mai jos.
logging.info("Navigare catre https://goapp.bancatransilvania.ro/")
self.page.goto('https://goapp.bancatransilvania.ro/', wait_until='domcontentloaded', timeout=30000)
try:
# Cookie consent - asteapta si accepta (GDPR wrapper)
logging.info("Verificare GDPR cookie banner...")
self._dismiss_gdpr_cookies(self.page)
# Verificare finala ca banner-ul nu mai blocheaza
try:
gdpr_wrapper = self.page.locator(".gdprcookie-wrapper")
if gdpr_wrapper.is_visible(timeout=1000):
logging.warning(" [WARN] Banner GDPR inca vizibil, fortez inchiderea...")
# Incearca sa inchida prin JavaScript
self.page.evaluate("document.querySelector('.gdprcookie-wrapper')?.remove()")
time.sleep(0.5)
except:
pass
self.login_page = self.page
# Asteapta ca Angular SPA sa randeze formularul de login.
# goapp.bancatransilvania.ro afiseaza intai un splash ("Aplicatia ta
# de banking business") cat se bootstrap-eaza Angular-ul. Pe masina
# prod (Session 0, randare software) bootstrap-ul e lent si uneori
# depaseste timeout-ul -> asteptam generos si reincarcam la nevoie.
logging.info("Astept incarcarea formularului de login...")
form_loaded = False
for attempt in range(1, 3): # 2 incercari
try:
self.login_page.wait_for_load_state("load", timeout=30000)
except PlaywrightTimeout:
pass
try:
self.login_page.wait_for_selector(
"input#user", timeout=60000, state="visible"
)
form_loaded = True
break
except PlaywrightTimeout:
if attempt == 1:
logging.warning(
" [WARN] Formularul nu a aparut in 60s "
"(SPA blocat pe splash) - reincarc popup-ul..."
)
self.login_page.reload(wait_until="domcontentloaded")
# la a 2-a incercare nu ridicam aici: iesim din bucla si
# raportam explicit mai jos (pagina de login s-a schimbat).
if not form_loaded:
# BT a mutat formularul de login: salvam dovezi si spunem exact
# ce URL/titlu am vazut, in loc de un timeout anonim de selector.
self._debug_shot(self.login_page, "login_page_changed")
try:
ts = datetime.now().strftime('%Y%m%d_%H%M%S')
html_path = Path(self.config.OUTPUT_DIR) / f"login_page_changed_{ts}.html"
html_path.write_text(self.login_page.content(), encoding='utf-8')
logging.error(f" [DIAG] HTML salvat: {html_path}")
except Exception as dump_err:
logging.warning(f" [DIAG] Nu am putut salva HTML-ul: {dump_err}")
raise Exception(
"Pagina de login BT s-a schimbat: "
f"URL final {self.login_page.url}, "
f"titlu {self.login_page.title()!r}"
)
logging.info("[OK] Formular login incarcat")
# Verifica GDPR cookies si pe popup
self._dismiss_gdpr_cookies(self.login_page)
# 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}")
# Capteaza starea paginii CAT browser-ul e inca viu, ca sa stim
# ce a vazut de fapt: Access Denied, pagina goala, alt formular etc.
diag_page = getattr(self, 'login_page', None) or self.page
try:
logging.error(f" [DIAG] URL: {diag_page.url}")
logging.error(f" [DIAG] Titlu: {diag_page.title()!r}")
body = diag_page.evaluate(
"document.body ? document.body.innerText.slice(0, 400) : '(fara body)'"
)
logging.error(f" [DIAG] Text vizibil: {body!r}")
inputs = diag_page.evaluate(
"Array.from(document.querySelectorAll('input'))"
".map(i => i.id + '|' + i.type + '|' + i.name)"
)
logging.error(f" [DIAG] Input-uri in pagina: {inputs}")
shot = Path(self.config.OUTPUT_DIR) / f"diag_login_timeout_{datetime.now().strftime('%Y%m%d_%H%M%S')}.png"
diag_page.screenshot(path=str(shot), full_page=True)
logging.error(f" [DIAG] Screenshot salvat: {shot}")
except Exception as diag_err:
logging.error(f" [DIAG] Nu am putut capta starea paginii: {diag_err}")
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
iteration = 0
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("[OK] 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
# Inchide dialoguri one-time (ex: "Am inteles") daca apar
self._dismiss_one_time_consent(self.page)
return True
except:
pass
# Daca BT a respins credentialele, push-ul 2FA nu pleaca niciodata -
# iesim imediat cu motivul real, nu dupa timeout.
iteration += 1
if iteration % 3 == 0:
login_error = self._detect_login_error(self.login_page)
if login_error:
print()
self._debug_shot(self.login_page, 'login_respins')
raise Exception(
f"Autentificare respinsa de BT: {login_error} "
"(2FA nu a fost trimis - verifica user/parola din .env)"
)
# 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
# Screenshot cat browser-ul e inca viu (dupa raise, event loop-ul e inchis)
self._debug_shot(self.login_page, 'timeout_2fa')
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.
Flux nou (2024+):
1. Primul cont: expand card -> click tranzactii -> select CSV -> Genereaza -> download
2. Conturile urmatoare: #selectAccountBtn -> select cont by heading -> Genereaza -> download
"""
logging.info("=" * 60)
logging.info("Descarcare tranzactii pentru toate conturile...")
logging.info("=" * 60)
downloaded_files = []
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}")
if idx == 1:
# PRIMUL CONT: expand -> click tranzactii -> select CSV
downloaded = self._download_first_account(account)
else:
# CONTURILE URMATOARE: selecteaza din dropdown -> Genereaza -> download
downloaded = self._download_subsequent_account(account)
if downloaded:
downloaded_files.append(downloaded)
except Exception as e:
logging.error(f" [EROARE] Eroare la descarcarea tranzactiilor pentru {nume_cont}: {e}")
# Incearca sa revii la o stare stabila
try:
self.page.keyboard.press("Escape")
time.sleep(1)
except:
pass
continue
logging.info("=" * 60)
logging.info(f"[OK] Descarcate {len(downloaded_files)}/{len(accounts)} fisiere CSV cu tranzactii")
logging.info("=" * 60)
return downloaded_files
def _download_first_account(self, account):
"""
Descarca tranzactii pentru primul cont.
Flow: expand card (daca nu e deja) -> click buton tranzactii -> select CSV -> Genereaza -> download
"""
nume_cont = account['nume_cont']
iban = account['iban']
try:
# Gaseste primul card
first_card = self.page.locator("fba-account-details-card").first
time.sleep(1)
# Verifica daca butonul de tranzactii e DEJA vizibil (cont expandat)
transactions_btn = first_card.locator("fba-account-buttons svg, fba-account-buttons, .account-transactions-btn").first
is_already_expanded = False
try:
is_already_expanded = transactions_btn.is_visible(timeout=2000)
except:
pass
if is_already_expanded:
logging.info(" Contul deja expandat - skip expand click")
else:
# Click pe expand icon (sageata din card)
logging.info(" Contul collapsed - expandez...")
expand_icon = first_card.locator(".mx-auto .mat-icon svg, .collapse-account-btn").first
expand_icon.click()
time.sleep(2)
logging.info(" Contul expandat")
# Click pe butonul de tranzactii (NU pe delete/inchide cont!)
# Butonul corect are: clasa .account-transactions-btn, SVG cu documentChartList, cu "Tranzactii"
transactions_btn = None
# Strategie 1 (PRINCIPALA): container cu clasa account-transactions-btn
try:
btn = first_card.locator(".account-transactions-btn").first
if btn.is_visible(timeout=2000):
transactions_btn = btn
logging.info(" Buton tranzactii gasit prin .account-transactions-btn")
except:
pass
# Strategie 2: SVG cu data-mat-icon-name="documentChartList"
if not transactions_btn:
try:
btn = first_card.locator("mat-icon[data-mat-icon-name='documentChartList']").first
if btn.is_visible(timeout=1000):
transactions_btn = btn
logging.info(" Buton tranzactii gasit prin documentChartList icon")
except:
pass
# Strategie 3: element care contine
cu text "Tranzactii"
if not transactions_btn:
try:
btn = first_card.locator("div:has(p:text('Tranzacții')), div:has(p:text('Tranzactii'))").first
if btn.is_visible(timeout=1000):
transactions_btn = btn
logging.info(" Buton tranzactii gasit prin
text")
except:
pass
# Strategie 4: click direct pe textul "Tranzactii" din card
if not transactions_btn:
try:
btn = first_card.get_by_text("Tranzacții", exact=True)
if btn.is_visible(timeout=1000):
transactions_btn = btn
logging.info(" Buton tranzactii gasit prin get_by_text")
except:
pass
if not transactions_btn:
raise Exception("Nu am gasit butonul de tranzactii!")
transactions_btn.click()
time.sleep(3)
logging.info(" Pagina tranzactii se incarca...")
# Selecteaza format CSV (click pe text "CSV")
csv_option = self.page.get_by_text("CSV", exact=True)
csv_option.click()
time.sleep(1)
logging.info(" Format CSV selectat")
# Marcheaza documentele existente in "Recente" (lista e cumulativa)
self._mark_existing_documents()
# Click pe butonul Genereaza
generate_btn = self.page.get_by_role("button", name="Generează")
generate_btn.click()
time.sleep(2)
logging.info(" Click Genereaza - astept generare...")
# Asteapta si descarca fisierul
return self._wait_and_download(nume_cont, iban)
except Exception as e:
logging.error(f" [EROARE] Download primul cont: {e}")
return None
def _download_subsequent_account(self, account):
"""
Descarca tranzactii pentru conturile 2+.
Flow: click #selectAccountBtn -> select cont by heading -> Genereaza -> download
"""
nume_cont = account['nume_cont']
iban = account['iban']
try:
# Click pe butonul de selectare cont (#selectAccountBtn)
select_btn = self.page.locator("#selectAccountBtn svg, #selectAccountBtn").first
select_btn.click()
time.sleep(2)
logging.info(" Dropdown conturi deschis")
# Debug: listeaza toate heading-urile vizibile din dropdown
visible_names = []
try:
headings = self.page.locator("fba-account-details h4, .account-name, h4").all()
for h in headings[:10]: # Max 10
try:
if h.is_visible(timeout=500):
visible_names.append(h.inner_text().strip())
except:
pass
if visible_names:
logging.info(f" Conturi in dropdown: {visible_names}")
except:
pass
if visible_names and nume_cont not in visible_names:
logging.warning(f" [WARN] Contul '{nume_cont}' lipseste din lista dropdown: {visible_names}")
# Selecteaza contul dupa nume - strategii multiple
account_selected = False
# Strategie 1: heading cu numele exact
try:
heading = self.page.get_by_role("heading", name=nume_cont, exact=True)
if heading.is_visible(timeout=2000):
heading.click()
account_selected = True
logging.info(f" Cont selectat prin heading exact: {nume_cont}")
except Exception as e:
logging.debug(f" Heading exact failed: {e}")
# Strategie 2: heading cu numele partial (fara exact match)
if not account_selected:
try:
heading = self.page.get_by_role("heading", name=nume_cont)
if heading.is_visible(timeout=2000):
heading.click()
account_selected = True
logging.info(f" Cont selectat prin heading partial: {nume_cont}")
except Exception as e:
logging.debug(f" Heading partial failed: {e}")
# Strategie 3: fba-account-details cu has_text
if not account_selected:
try:
account_item = self.page.locator("fba-account-details").filter(has_text=nume_cont).first
if account_item.is_visible(timeout=2000):
account_item.click()
account_selected = True
logging.info(f" Cont selectat prin fba-account-details: {nume_cont}")
except Exception as e:
logging.debug(f" fba-account-details failed: {e}")
# Strategie 4: locator h4 care contine textul
if not account_selected:
try:
h4_elem = self.page.locator(f"h4:has-text('{nume_cont}')").first
if h4_elem.is_visible(timeout=2000):
h4_elem.click()
account_selected = True
logging.info(f" Cont selectat prin h4:has-text: {nume_cont}")
except Exception as e:
logging.debug(f" h4:has-text failed: {e}")
# Strategie 5: orice element cu textul contului
if not account_selected:
try:
text_elem = self.page.get_by_text(nume_cont, exact=True)
if text_elem.is_visible(timeout=2000):
text_elem.click()
account_selected = True
logging.info(f" Cont selectat prin text exact: {nume_cont}")
except Exception as e:
logging.debug(f" text exact failed: {e}")
# Strategie 6: text partial match
if not account_selected:
try:
text_elem = self.page.get_by_text(nume_cont)
if text_elem.is_visible(timeout=2000):
text_elem.click()
account_selected = True
logging.info(f" Cont selectat prin text partial: {nume_cont}")
except Exception as e:
logging.debug(f" text partial failed: {e}")
if not account_selected:
logging.error(f" [EROARE] Nu am putut selecta contul: {nume_cont}")
# Screenshot pentru debug
try:
debug_path = Path(self.config.OUTPUT_DIR) / f"debug_dropdown_{nume_cont.replace(' ', '_')}_{datetime.now().strftime('%H%M%S')}.png"
self.page.screenshot(path=str(debug_path))
logging.error(f" Screenshot debug salvat: {debug_path}")
except:
pass
return None
time.sleep(2)
# Marcheaza documentele existente in "Recente"
self._mark_existing_documents()
# Click pe butonul Genereaza (CSV deja selectat de la primul cont)
generate_btn = self.page.get_by_role("button", name="Generează")
generate_btn.click()
time.sleep(2)
logging.info(" Click Genereaza - astept generare...")
# Asteapta si descarca fisierul
return self._wait_and_download(nume_cont, iban)
except Exception as e:
logging.error(f" [EROARE] Download cont ulterior {nume_cont}: {e}")
return None
# Texte afisate de BT cand exportul nu are continut (nu exista fisier de descarcat)
EMPTY_EXPORT_PATTERNS = (
"nu exist", "nu s-au gasit", "nu s-au găsit",
"fara tranzactii", "fără tranzacții",
)
def _mark_existing_documents(self):
"""Marcheaza documentele deja prezente in lista 'Recente'.
Lista este cumulativa, deci `wait_for_selector("fba-document-item")`
se satisfacea instant pe un document vechi. Marcand cele existente,
putem astepta exact documentul NOU generat.
"""
try:
n = self.page.eval_on_selector_all(
"fba-document-item",
"els => { els.forEach(e => e.setAttribute('data-seen', '1')); return els.length; }",
)
logging.info(f" Documente deja in 'Recente': {n}")
except Exception as e:
logging.debug(f" Nu am putut marca documentele existente: {e}")
def _dump_download_debug(self, nume_cont, tag):
"""La esec: screenshot + outerHTML al listei de documente si al dropdown-ului."""
safe = nume_cont.replace(' ', '_').replace('/', '_').replace('\\', '_')
stamp = datetime.now().strftime('%Y%m%d_%H%M%S')
self._debug_shot(self.page, f"download_{tag}_{safe}")
blocks = {
"documente": (
"() => { const it=[...document.querySelectorAll('fba-document-item')];"
" if(!it.length) return null; let p=it[0].parentElement;"
" while(p && !it.every(i=>p.contains(i))) p=p.parentElement;"
" return p ? p.outerHTML : null; }"
),
"dropdown": (
"() => { const o=document.querySelector('.cdk-overlay-container');"
" return o ? o.outerHTML : null; }"
),
}
for name, js in blocks.items():
try:
html = self.page.evaluate(js)
if not html:
logging.error(f" [DIAG] {name}: negasit in DOM")
continue
path = Path(self.config.OUTPUT_DIR) / f"debug_download_{tag}_{safe}_{name}_{stamp}.html"
path.write_text(html, encoding='utf-8')
logging.error(f" [DIAG] HTML {name} salvat: {path}")
except Exception as e:
logging.error(f" [DIAG] Nu am putut salva HTML {name}: {e}")
try:
texts = self.page.locator("fba-document-item").all_inner_texts()
logging.error(f" [DIAG] fba-document-item ({len(texts)}): {texts}")
except Exception:
pass
def _wait_and_download(self, nume_cont, iban, timeout=20000):
"""
Asteapta generarea fisierului NOU si il descarca.
Args:
nume_cont: Numele contului (pentru filename)
iban: IBAN-ul contului
timeout: Timeout pentru download (ms)
Returns:
Dict cu informatii despre fisierul descarcat sau None
"""
try:
# Asteapta documentul NOU (nu unul vechi ramas in "Recente")
new_selector = "fba-document-item:not([data-seen])"
try:
self.page.wait_for_selector(new_selector, state="visible", timeout=timeout)
except PlaywrightTimeout:
logging.error(f" [EROARE] Niciun document nou generat pentru {nume_cont} ({timeout}ms)")
self._dump_download_debug(nume_cont, "no_new_document")
return None
new_item = self.page.locator(new_selector).first
item_text = (new_item.inner_text() or "").strip()
first_line = item_text.splitlines()[0] if item_text else "(fara text)"
logging.info(f" Document nou: {first_line}")
# BT afiseaza un item de EROARE (fara buton de descarcare) cand
# exportul nu are tranzactii in perioada selectata.
if any(p in item_text.lower() for p in self.EMPTY_EXPORT_PATTERNS):
logging.warning(f" [SKIP] Export fara tranzactii pentru {nume_cont}: {item_text[:160]}")
return None
# Click pe documentul NOU pentru a descarca
with self.page.expect_download(timeout=timeout) as download_info:
download_btn = new_item.locator("svg, path").first
download_btn.click()
download = download_info.value
# Salveaza fisierul cu nume descriptiv
timestamp = datetime.now().strftime('%Y-%m-%d_%H-%M-%S')
nume_safe = nume_cont.replace(' ', '_').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" [OK] Salvat: {save_path}")
return {
'cont': nume_cont,
'iban': iban,
'fisier': str(save_path)
}
except Exception as e:
logging.error(f" [EROARE] Download fisier: {e}")
self._dump_download_debug(nume_cont, "download_exceptie")
return None
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()
# Asteapta explicit aparitia cardurilor de cont. Pe masina prod
# (Session 0, randare software) lista se incarca mult mai lent de
# 3s - un time.sleep fix gasea 0 conturi.
try:
self.page.wait_for_selector(
"fba-account-details-card", state="visible", timeout=45000
)
except PlaywrightTimeout:
logging.warning(" [WARN] Timeout asteptand cardurile de cont")
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")
# Daca tot 0, capteaza starea paginii pentru diagnostic
if not account_cards:
try:
logging.error(f" [DIAG] URL: {self.page.url}")
body = self.page.evaluate(
"document.body ? document.body.innerText.slice(0, 400) : '(fara body)'"
)
logging.error(f" [DIAG] Text vizibil: {body!r}")
fba_tags = self.page.evaluate(
"Array.from(new Set(Array.from(document.querySelectorAll('*'))"
".map(e => e.tagName.toLowerCase()).filter(t => t.startsWith('fba-'))))"
)
logging.error(f" [DIAG] Tag-uri fba-* in pagina: {fba_tags}")
shot = Path(self.config.OUTPUT_DIR) / f"diag_conturi_{datetime.now().strftime('%Y%m%d_%H%M%S')}.png"
self.page.screenshot(path=str(shot), full_page=True)
logging.error(f" [DIAG] Screenshot salvat: {shot}")
except Exception as diag_err:
logging.error(f" [DIAG] Nu am putut capta starea: {diag_err}")
accounts = []
for idx, card in enumerate(account_cards, 1):
try:
# Extrage nume cont (din tag
)
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 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()