Integracao ERP / Financeiro

Exporte transacoes, reconcilie pagamentos e integre com sistemas de contabilidade e emissao de notas fiscais.

---

Exportar transacoes em JSON

import requests
import json
from datetime import date, datetime

BASE_URL = "https://fast-deliv-backend.vercel.app"

def export_transactions(token: str, output_file: str = "transactions.json") -> int:
    """Exporta todas as transacoes para JSON."""
    headers = {"Authorization": f"Bearer {token}"}

    # Buscar transacoes (endpoint driver — necessario token de driver ou iterar por drivers)
    resp = requests.get(
        f"{BASE_URL}/api/v1/wallets/me/transactions",
        headers=headers,
    )
    resp.raise_for_status()
    transactions = resp.json()

    with open(output_file, "w", encoding="utf-8") as f:
        json.dump(transactions, f, ensure_ascii=False, indent=2)

    print(f"Exportadas {len(transactions)} transacoes para {output_file}")
    return len(transactions)

---

Exportar transacoes em CSV

import csv
from decimal import Decimal

def export_transactions_csv(transactions: list[dict], output_file: str = "transactions.csv") -> None:
    """Exporta transacoes em CSV para importar em Excel/ERP."""
    fieldnames = [
        "id",
        "data",
        "tipo",
        "valor_brl",
        "descricao",
        "entrega_id",
    ]

    with open(output_file, "w", newline="", encoding="utf-8-sig") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames, delimiter=";")
        writer.writeheader()

        for t in transactions:
            writer.writerow({
                "id": t["id"],
                "data": t["created_at"][:10],  # YYYY-MM-DD
                "tipo": "Credito" if t["type"] == "credit" else "Debito",
                "valor_brl": t["amount"].replace(".", ","),  # formato brasileiro
                "descricao": t["description"],
                "entrega_id": t.get("delivery_id", ""),
            })

    print(f"CSV gerado: {output_file}")

---

Extrato diario

from datetime import date, timedelta

def daily_statement(
    token: str,
    target_date: date | None = None,
) -> dict:
    """Gera extrato do dia com totais de creditos e debitos."""
    if target_date is None:
        target_date = date.today()

    date_str = target_date.isoformat()
    headers = {"Authorization": f"Bearer {token}"}

    transactions = requests.get(
        f"{BASE_URL}/api/v1/wallets/me/transactions",
        headers=headers,
    ).json()

    # Filtrar por data
    day_txns = [
        t for t in transactions
        if t["created_at"].startswith(date_str)
    ]

    credits = [t for t in day_txns if t["type"] == "credit"]
    debits  = [t for t in day_txns if t["type"] == "debit"]

    total_credits = sum(Decimal(t["amount"]) for t in credits)
    total_debits  = sum(Decimal(t["amount"]) for t in debits)

    wallet = requests.get(
        f"{BASE_URL}/api/v1/wallets/me",
        headers=headers,
    ).json()

    return {
        "date": date_str,
        "opening_balance": str(Decimal(wallet["balance"]) + total_debits - total_credits),
        "closing_balance": wallet["balance"],
        "total_credits": str(total_credits),
        "total_debits": str(total_debits),
        "credit_count": len(credits),
        "debit_count": len(debits),
        "transactions": day_txns,
    }

# Uso
statement = daily_statement(TOKEN, target_date=date(2026, 5, 19))
print(f"Creditos: R$ {statement['total_credits']}")
print(f"Debitos:  R$ {statement['total_debits']}")
print(f"Saldo:    R$ {statement['closing_balance']}")

---

Reconciliacao de pagamentos

Correlacione transactions do fast_deliv com registros do seu ERP:

def reconcile_payments(
    transactions: list[dict],
    erp_records: list[dict],
) -> dict:
    """
    Reconcilia transacoes do fast_deliv com registros do ERP.
    erp_records deve ter: external_id, amount, date
    """
    # Indexar por delivery_id
    fd_by_delivery = {
        t["delivery_id"]: t
        for t in transactions
        if t["type"] == "credit" and t["delivery_id"]
    }

    matched = []
    unmatched_fd = []
    unmatched_erp = []

    for erp in erp_records:
        fd = fd_by_delivery.get(erp["external_id"])
        if fd:
            if Decimal(fd["amount"]) == Decimal(erp["amount"]):
                matched.append({"erp": erp, "fast_deliv": fd, "ok": True})
            else:
                matched.append({
                    "erp": erp,
                    "fast_deliv": fd,
                    "ok": False,
                    "divergence": str(Decimal(fd["amount"]) - Decimal(erp["amount"])),
                })
        else:
            unmatched_erp.append(erp)

    # Transacoes no fast_deliv sem correspondencia no ERP
    erp_ids = {e["external_id"] for e in erp_records}
    unmatched_fd = [
        t for t in transactions
        if t["type"] == "credit"
        and t["delivery_id"]
        and t["delivery_id"] not in erp_ids
    ]

    return {
        "matched": matched,
        "unmatched_fast_deliv": unmatched_fd,
        "unmatched_erp": unmatched_erp,
        "summary": {
            "total_matched": len(matched),
            "ok": sum(1 for m in matched if m["ok"]),
            "divergent": sum(1 for m in matched if not m["ok"]),
            "unmatched_fd": len(unmatched_fd),
            "unmatched_erp": len(unmatched_erp),
        },
    }

---

Campos relevantes para notas fiscais

Campo	Fonte	Obs
Valor do servico	transaction.amount	Credito da entrega
Data de competencia	delivery.completed_at	Data da prestacao do servico
Descricao	delivery.title	Nome da entrega
Tomador	delivery.destination_address	Endereco do tomador
Prestador	Perfil do admin	Empresa contratante
Codigo de referencia	delivery.id	Para conciliacao futura

Nota fiscal

O fast_deliv nao emite NFS-e automaticamente. Use os campos acima para preencher sua NF em sistemas como NFe.io, NF Paulistana, etc.

---

Script de exportacao automatica via cron

#!/bin/bash
# export_daily.sh — agendar com cron: 0 23 * * * /path/to/export_daily.sh

DATE=$(date +"%Y-%m-%d")
OUTPUT_DIR="/var/reports/fast_deliv"
mkdir -p "$OUTPUT_DIR"

python3 - <<EOF
import os, sys
sys.path.insert(0, '/opt/fast_deliv_integration')
from client import get_token, export_transactions_csv, daily_statement
from datetime import date

TOKEN = get_token(os.environ['FAST_DELIV_EMAIL'], os.environ['FAST_DELIV_PASSWORD'])
statement = daily_statement(TOKEN)

import json
with open('${OUTPUT_DIR}/statement_${DATE}.json', 'w') as f:
    json.dump(statement, f, indent=2)

export_transactions_csv(
    statement['transactions'],
    '${OUTPUT_DIR}/transactions_${DATE}.csv'
)
print("Exportacao concluida para ${DATE}")
EOF