#!/usr/bin/env python3 """ Script para importar transações do arquivo Excel para o WebMoney """ import openpyxl import requests import json from datetime import datetime import hashlib # Configurações API_BASE = "https://webmoney.cnxifly.com/api" EXCEL_FILE = "moneypro_backup_2025-12-06_093049.xlsx" # Mapeamento de contas Excel -> IDs do banco # BBVA (Conta Corrente) -> Conta "Viviane" (bank_name: BBVA) -> id: 3 # Santander (Conta Corrente) -> Conta "Marco" (bank_name: Santander) -> id: 2 ACCOUNT_MAP = { "BBVA": 3, # Viviane - BBVA "Santander": 2, # Marco - Santander } def get_token(): """Obter token de autenticação""" response = requests.post( f"{API_BASE}/login", headers={"Content-Type": "application/json", "Accept": "application/json"}, json={"email": "marco@cnxifly.com", "password": "Master9354"} ) data = response.json() return data["data"]["token"] def get_existing_transactions(token): """Obter transações existentes para verificar duplicidades""" response = requests.get( f"{API_BASE}/transactions?per_page=10000", headers={"Authorization": f"Bearer {token}", "Accept": "application/json"} ) transactions = response.json().get("data", []) # Criar conjunto de hashes únicos existing = set() for t in transactions: # Criar hash baseado em data planejada, descrição, valor e conta key = f"{t['planned_date'][:10]}|{t['description']}|{float(t['planned_amount']):.2f}|{t['account_id']}" existing.add(key) return existing def parse_date(date_str): """Converter data do formato DD/MM/YYYY para YYYY-MM-DD""" if not date_str: return None try: dt = datetime.strptime(date_str, "%d/%m/%Y") return dt.strftime("%Y-%m-%d") except: return None def read_excel_transactions(): """Ler transações do arquivo Excel""" wb = openpyxl.load_workbook(EXCEL_FILE) ws = wb["Transações"] transactions = [] for row_idx in range(2, ws.max_row + 1): planned_date = ws.cell(row=row_idx, column=1).value effective_date = ws.cell(row=row_idx, column=2).value description = ws.cell(row=row_idx, column=3).value notes = ws.cell(row=row_idx, column=4).value planned_amount = ws.cell(row=row_idx, column=5).value effective_amount = ws.cell(row=row_idx, column=6).value extra_amount = ws.cell(row=row_idx, column=7).value status = ws.cell(row=row_idx, column=8).value account_name = ws.cell(row=row_idx, column=9).value account_type = ws.cell(row=row_idx, column=10).value if not description or not account_name: continue transactions.append({ "planned_date": planned_date, "effective_date": effective_date, "description": description.replace('\n', ' ').strip() if description else "", "notes": notes.strip() if notes else None, "planned_amount": float(planned_amount) if planned_amount else 0, "effective_amount": float(effective_amount) if effective_amount else None, "status": status, "account_name": account_name, }) return transactions def create_transaction(token, trans, account_id): """Criar uma transação via API""" planned_date = parse_date(trans["planned_date"]) effective_date = parse_date(trans["effective_date"]) # Determinar tipo e valor (positivo = credit, negativo = debit) amount = trans["planned_amount"] if amount < 0: trans_type = "debit" amount = abs(amount) else: trans_type = "credit" # Determinar status if trans["status"] == "Efetivada": status = "completed" elif trans["status"] == "Pendente": status = "pending" else: status = "pending" # Determinar valor efetivo eff_amount = trans["effective_amount"] if eff_amount is not None: eff_amount = abs(eff_amount) else: eff_amount = amount payload = { "account_id": account_id, "type": trans_type, "description": trans["description"], "planned_amount": amount, "amount": eff_amount, "planned_date": planned_date, "effective_date": effective_date if status == "completed" else None, "status": status, "notes": trans["notes"], } response = requests.post( f"{API_BASE}/transactions", headers={ "Authorization": f"Bearer {token}", "Accept": "application/json", "Content-Type": "application/json" }, json=payload ) return response.status_code == 201, response.text def main(): print("=" * 60) print("IMPORTAÇÃO DE TRANSAÇÕES DO MONEYPRO") print("=" * 60) # 1. Obter token print("\n[1/4] Obtendo token de autenticação...") token = get_token() print(f"✓ Token obtido") # 2. Obter transações existentes print("\n[2/4] Verificando transações existentes...") existing = get_existing_transactions(token) print(f"✓ {len(existing)} transações existentes encontradas") # 3. Ler Excel print("\n[3/4] Lendo arquivo Excel...") transactions = read_excel_transactions() print(f"✓ {len(transactions)} transações lidas do Excel") # 4. Importar transações print("\n[4/4] Importando transações...") imported = 0 duplicates = 0 errors = 0 error_details = [] for i, trans in enumerate(transactions): account_id = ACCOUNT_MAP.get(trans["account_name"]) if not account_id: errors += 1 error_details.append(f"Conta não mapeada: {trans['account_name']}") continue # Verificar duplicidade planned_date = parse_date(trans["planned_date"]) amount = abs(trans["planned_amount"]) key = f"{planned_date}|{trans['description']}|{amount:.2f}|{account_id}" if key in existing: duplicates += 1 continue # Criar transação success, response = create_transaction(token, trans, account_id) if success: imported += 1 existing.add(key) # Evitar duplicatas no mesmo lote else: errors += 1 if len(error_details) < 10: error_details.append(f"Erro ao criar: {trans['description'][:30]}... - {response[:100]}") # Progresso if (i + 1) % 100 == 0: print(f" Processadas: {i + 1}/{len(transactions)} | Importadas: {imported} | Duplicatas: {duplicates} | Erros: {errors}") # Resultado final print("\n" + "=" * 60) print("RESULTADO DA IMPORTAÇÃO") print("=" * 60) print(f"Total no Excel: {len(transactions)}") print(f"Importadas: {imported}") print(f"Duplicatas ignoradas: {duplicates}") print(f"Erros: {errors}") print(f"Total processadas: {imported + duplicates + errors}") if error_details: print(f"\nPrimeiros erros:") for err in error_details[:5]: print(f" - {err}") print("\n✓ Importação concluída!") if __name__ == "__main__": main()