Passerelle Zero-knowledge : Import automatique des e-mails bancaires

pip install budgero rich pandas python-dotenv imap-tools

# --- EMAIL SETTINGS ---EMAIL_ADDRESS="[email protected]"EMAIL_APP_PASSWORD="paste_16_char_app_password_here"# --- BANK FILTER SETTINGS ---BANK_SENDER_EMAIL="[email protected]"BANK_SUBJECT_KEYWORD="Transaction Alert"# --- BUDGERO SECURITY ---BUDGERO_API_KEY="paste_your_api_token_here"BUDGERO_ENCRYPTION_KEY="paste_your_space_key_here"# --- DESTINATION SETTINGS ---TARGET_BUDGET_ID="1"TARGET_ACCOUNT_ID="1"TARGET_CATEGORY_ID="5"

import osimport imaplibimport emailimport refrom datetime import datetime, timedeltafrom dotenv import load_dotenvfrom rich.console import Consolefrom rich.table import Tablefrom rich.panel import Panelfrom budgero import BudgeroClientfrom budgero.exceptions import APIError, EncryptionError# --- CONFIGURATION & SAFETY ---load_dotenv()console = Console()# 1. SETUP_MODE: Set to True to generate the AI Prompt. Set False to run the bridge.SETUP_MODE = False # 2. DRY_RUN: If True, prints a table but DOES NOT upload to Budgero.DRY_RUN = True  # 3. NUM_SAMPLES: How many emails to send to ChatGPT for training (5 is usually fine, if you get incosistent results you can use more sample so LLM can detect more edge cases).NUM_SAMPLES = 5# 4. Lookback in days: How far in the past do you want to search for transaction emails.LOOKBACK_DAYS =  7# --- PASTE YOUR AI-GENERATED FUNCTION BELOW THIS LINE ---def extract_transaction_info(body):    """    DEFAULT PLACEHOLDER.    Run this script in SETUP_MODE = True to generate the code for this function.    """    return None, None, None, None, None# --- END OF PARSING LOGIC ---def imap_since_days(days: int) -> str:    d = datetime.today() - timedelta(days=days)    return d.strftime('%d-%b-%Y')  # IMAP-friendly formatdef connect_imap():    """Helper to connect to Gmail"""    mail = imaplib.IMAP4_SSL('imap.gmail.com')    mail.login(os.getenv('EMAIL_ADDRESS'), os.getenv('EMAIL_APP_PASSWORD'))    mail.select('inbox')    return maildef generate_llm_prompt():    """Fetches real emails and creates a ChatGPT prompt"""    sender = os.getenv('BANK_SENDER_EMAIL')    keyword = os.getenv('BANK_SUBJECT_KEYWORD')        console.print(f"[yellow]📡 Connecting to IMAP... Fetching last {NUM_SAMPLES} emails from: {sender}[/yellow]")    try:        mail = connect_imap()                # Search for ANY email from sender (Read or Unread) to get samples        status, messages = mail.search(None, f'(FROM "{sender}" SUBJECT "{keyword}")')        email_ids = messages[0].split()[-NUM_SAMPLES:] # Get last N                if not email_ids:            console.print("[red]❌ No emails found! Check your .env sender/keyword.[/red]")            return        samples = []        for num in email_ids:            _, data = mail.fetch(num, '(RFC822)')            msg = email.message_from_bytes(data[0][1])            body = msg.get_payload(decode=True).decode(errors='ignore')            # Clean up: remove newlines/tabs to save tokens, keep it readable            clean_body = body[:2500].replace("\r", "").replace("\n", " ")             samples.append(f"--- EMAIL SAMPLE ---\n{clean_body}\n")        mail.logout()        # Construct the Prompt        prompt = f"""I am writing a Python script to parse transaction emails from my bank.I need you to write a Python function called `extract_transaction_info(body)` that takes the raw HTML/text body of an email and returns the transaction details.**Important:** You must import `datetime` inside the function or assume `from datetime import datetime` is available.The Function Signature must be exactly:`return tx_date, amount, note, tx_type, currency`Requirements:1. `tx_date`: **Python datetime object**. Parse the date string found in the email into a real object.2. `amount`: Float. If format is "1.200,50" (European), convert to standard float 1200.50.3. `note`: String. The Merchant Name/Payee. Remove HTML tags like &nbsp; or <br>. Clean extra whitespace.4. `tx_type`: String. "Outflow" or "Inflow".5. `currency`: String (USD, EUR, RSD, etc).6. Return `None, None, None, None, None` if the email is not a transaction receipt.Here are {len(samples)} real samples of my bank emails:{ "".join(samples) }        """        console.print(Panel.fit("✅ SAMPLES FETCHED! COPY THE TEXT BELOW INTO CHATGPT:", border_style="green"))        print(prompt)         console.print(Panel.fit("AFTER CHATGPT REPLIES: Copy the code it gives you and replace the `extract_transaction_info` function in this script.", border_style="blue"))    except Exception as e:        console.print(f"[red]Connection Error: {e}[/red]")def run_bridge():    """The Main Loop"""    sender = os.getenv('BANK_SENDER_EMAIL')    keyword = os.getenv('BANK_SUBJECT_KEYWORD')        console.print(f"[green]🚀 Starting Budgero Bridge for {sender}...[/green]")    if DRY_RUN:        console.print("[bold yellow]⚠️ DRY RUN MODE ACTIVE: No data will be uploaded.[/bold yellow]")    mail = connect_imap()    since = imap_since_days(LOOKBACK_DAYS)  # last 7 days        # Fetch UNREAD only for the actual run    status, messages = mail.search(        None,        'FROM', f'"{sender}"',        'SUBJECT', f'"{keyword}"',        'SINCE', since,    )    email_ids = messages[0].split()    found_transactions = []    # 1. PARSE STAGE    for num in email_ids:        _, data = mail.fetch(num, '(RFC822)')        msg = email.message_from_bytes(data[0][1])        body = msg.get_payload(decode=True).decode(errors='ignore')        # --- CALL THE AI GENERATED FUNCTION ---        try:            tx_date, amount, note, tx_type, currency = extract_transaction_info(body)                        if amount and tx_date:                found_transactions.append({                    "date": tx_date,                    "amount": amount,                    "note": note,                    "type": tx_type,                    "currency": currency,                    "id": num # Keep email ID to mark as read later if needed                })        except Exception as e:             console.print(f"[red]Parsing Error on email {num.decode()}: {e}[/red]")    mail.logout()    if not found_transactions:        console.print("[blue]No new transactions found.[/blue]")        return    # 2. DISPLAY STAGE (The "Nice Table")    table = Table(title=f"📥 Found {len(found_transactions)} New Transactions")    table.add_column("Date", style="cyan")    table.add_column("Amount", style="magenta", justify="right")    table.add_column("Currency", style="blue")    table.add_column("Payee / Note", style="green")    table.add_column("Type", style="yellow")    for tx in found_transactions:        table.add_row(            tx["date"].strftime("%Y-%m-%d"),            f"{tx['amount']:.2f}",            tx["currency"],            tx["note"],            tx["type"]        )        console.print(table)    # 3. UPLOAD STAGE    if DRY_RUN:        console.print("\n[bold yellow]✋ Dry Run Complete. Set DRY_RUN = False to upload.[/bold yellow]")        return    console.print("\n[bold green]🚀 Uploading to Budgero...[/bold green]")        client = BudgeroClient(        api_key=os.getenv('BUDGERO_API_KEY'),         encryption_key=os.getenv('BUDGERO_ENCRYPTION_KEY')    )    success_count = 0    for tx in found_transactions:        try:            client.add_transaction(                account_id=int(os.getenv('TARGET_ACCOUNT_ID')),                category_id=int(os.getenv('TARGET_CATEGORY_ID')),                budget_id=int(os.getenv('TARGET_BUDGET_ID')),                date=tx["date"],                outflow=float(tx["amount"]) if tx["type"] == "Outflow" else 0,                inflow=float(tx["amount"]) if tx["type"] == "Inflow" else 0,                memo=tx["note"],                payee=tx["note"]            )            success_count += 1            # Optional: Mark email as read on server after success            # mail = connect_imap()            # mail.store(tx['id'], '+FLAGS', '\\Seen')            # mail.logout()        except Exception as e:            console.print(f"[red]❌ Upload Error for {tx['note']}: {e}[/red]")        console.print(f"[blue]✨ Sync Complete. Successfully imported {success_count} transactions.[/blue]")if __name__ == '__main__':    if SETUP_MODE:        generate_llm_prompt()    else:        run_bridge()