Automacao com Scripts

Scripts para operacoes em massa, monitoramento de SLA e relatorios automaticos.

---

Criar entregas em massa via CSV

Formato do CSV

title,origin_address,origin_lat,origin_lng,destination_address,destination_lat,destination_lng
Entrega 001,Rua das Flores 100 Goiania,-16.6869,-49.2648,Av Anhanguera 500 Goiania,-16.6643,-49.2744
Entrega 002,Rua 1 Quadra 1 Goiania,-16.7012,-49.2789,Rua 2 Quadra 5 Goiania,-16.7100,-49.2900

Script de importacao

import csv
import time
import requests
from pathlib import Path


def bulk_create_deliveries(
    token: str,
    csv_file: str,
    delay_seconds: float = 0.5,
) -> list[dict]:
    """
    Cria entregas a partir de um CSV.
    delay_seconds: pausa entre requisicoes para nao sobrecarregar a API.
    """
    BASE_URL = "https://fast-deliv-backend.vercel.app"
    headers = {"Authorization": f"Bearer {token}"}
    results = []

    with open(csv_file, newline="", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        rows = list(reader)

    print(f"Criando {len(rows)} entregas...")

    for i, row in enumerate(rows, 1):
        try:
            resp = requests.post(
                f"{BASE_URL}/api/v1/deliveries",
                headers=headers,
                json={
                    "title":               row["title"],
                    "origin_address":      row["origin_address"],
                    "origin_lat":          float(row["origin_lat"]),
                    "origin_lng":          float(row["origin_lng"]),
                    "destination_address": row["destination_address"],
                    "destination_lat":     float(row["destination_lat"]),
                    "destination_lng":     float(row["destination_lng"]),
                },
            )
            resp.raise_for_status()
            delivery = resp.json()
            results.append({"row": i, "id": delivery["id"], "status": "ok"})
            print(f"  [{i}/{len(rows)}] Criada: {delivery['id']}")

        except Exception as e:
            results.append({"row": i, "title": row.get("title"), "status": "error", "error": str(e)})
            print(f"  [{i}/{len(rows)}] ERRO: {e}")

        time.sleep(delay_seconds)

    ok    = sum(1 for r in results if r["status"] == "ok")
    error = sum(1 for r in results if r["status"] == "error")
    print(f"\nConcluido: {ok} criadas, {error} erros")
    return results


if __name__ == "__main__":
    import os
    bulk_create_deliveries(
        token=os.environ["FAST_DELIV_TOKEN"],
        csv_file="entregas.csv",
    )

---

Monitorar entregas pendentes com alerta de SLA

import requests
from datetime import datetime, timezone, timedelta


def check_sla_violations(
    token: str,
    max_pending_hours: float = 2.0,
) -> list[dict]:
    """
    Retorna entregas 'pending' ha mais de max_pending_hours.
    """
    BASE_URL = "https://fast-deliv-backend.vercel.app"
    headers = {"Authorization": f"Bearer {token}"}

    resp = requests.get(
        f"{BASE_URL}/api/v1/deliveries",
        headers=headers,
        params={"status": "pending"},
    )
    resp.raise_for_status()
    pending = resp.json()

    now = datetime.now(timezone.utc)
    threshold = timedelta(hours=max_pending_hours)
    violations = []

    for d in pending:
        created = datetime.fromisoformat(d["created_at"].replace("Z", "+00:00"))
        age = now - created
        if age > threshold:
            violations.append({
                "id":      d["id"],
                "title":   d["title"],
                "created": d["created_at"],
                "age_hours": round(age.total_seconds() / 3600, 1),
            })

    return violations


def send_alert(violations: list[dict]) -> None:
    """Substituir por email, Slack, PagerDuty, etc."""
    if not violations:
        return
    print(f"ALERTA: {len(violations)} entregas pendentes ha mais de 2h:")
    for v in violations:
        print(f"  - {v['title']} ({v['id']}) — {v['age_hours']}h sem motorista")


if __name__ == "__main__":
    import os
    violations = check_sla_violations(
        token=os.environ["FAST_DELIV_TOKEN"],
        max_pending_hours=2.0,
    )
    send_alert(violations)

---

Relatorio diario automatico

import json
import requests
from datetime import date, datetime, timezone
from decimal import Decimal


def generate_daily_report(token: str, report_date: date | None = None) -> dict:
    if report_date is None:
        report_date = date.today()

    date_str = report_date.isoformat()
    BASE_URL = "https://fast-deliv-backend.vercel.app"
    headers = {"Authorization": f"Bearer {token}"}

    all_deliveries = requests.get(
        f"{BASE_URL}/api/v1/deliveries",
        headers=headers,
    ).json()

    today = [
        d for d in all_deliveries
        if d["created_at"].startswith(date_str)
    ]

    completed = [d for d in today if d["status"] == "completed"]
    cancelled = [d for d in today if d["status"] == "cancelled"]
    pending   = [d for d in today if d["status"] == "pending"]

    total_earnings = sum(
        Decimal(d["driver_earnings"])
        for d in completed
        if d.get("driver_earnings")
    )

    distances = [d["distance_km"] for d in completed if d.get("distance_km")]
    avg_distance = sum(distances) / len(distances) if distances else 0

    report = {
        "date": date_str,
        "total": len(today),
        "completed": len(completed),
        "cancelled": len(cancelled),
        "pending": len(pending),
        "completion_rate": round(len(completed) / len(today) * 100, 1) if today else 0,
        "total_earnings_brl": str(total_earnings),
        "avg_distance_km": round(avg_distance, 2),
        "generated_at": datetime.now(timezone.utc).isoformat(),
    }
    return report


def save_and_print_report(report: dict) -> None:
    filename = f"report_{report['date']}.json"
    with open(filename, "w", encoding="utf-8") as f:
        json.dump(report, f, ensure_ascii=False, indent=2)

    print(f"=== Relatorio {report['date']} ===")
    print(f"Total de entregas: {report['total']}")
    print(f"Concluidas:        {report['completed']} ({report['completion_rate']}%)")
    print(f"Canceladas:        {report['cancelled']}")
    print(f"Pendentes:         {report['pending']}")
    print(f"Ganhos totais:     R$ {report['total_earnings_brl']}")
    print(f"Distancia media:   {report['avg_distance_km']} km")
    print(f"Salvo em:          {filename}")


if __name__ == "__main__":
    import os
    report = generate_daily_report(token=os.environ["FAST_DELIV_TOKEN"])
    save_and_print_report(report)

---

Configurar cron (Linux/macOS)

# Editar crontab
crontab -e

# Verificar SLA a cada 30 minutos (horario comercial)
*/30 8-18 * * 1-5 /usr/bin/python3 /opt/scripts/check_sla.py >> /var/log/fast_deliv_sla.log 2>&1

# Relatorio diario as 23h
0 23 * * * /usr/bin/python3 /opt/scripts/daily_report.py >> /var/log/fast_deliv_report.log 2>&1

---

Script de conciliacao de pagamentos

#!/bin/bash
# reconcile.sh — verifica saques pendentes ha mais de 1 hora

AUTH_RESP=$(curl -s -X POST \
  "${SUPABASE_URL}/auth/v1/token?grant_type=password" \
  -H "apikey: ${SUPABASE_ANON_KEY}" \
  -H "Content-Type: application/json" \
  -d "{\"email\":\"${FAST_DELIV_EMAIL}\",\"password\":\"${FAST_DELIV_PASSWORD}\"}")

TOKEN=$(echo "$AUTH_RESP" | jq -r '.access_token')

TRANSACTIONS=$(curl -s "${BASE_URL}/api/v1/wallets/me/transactions" \
  -H "Authorization: Bearer ${TOKEN}")

# Saques do dia
TODAY=$(date -u +"%Y-%m-%d")
DEBITS=$(echo "$TRANSACTIONS" | jq --arg d "$TODAY" \
  '[.[] | select(.type == "debit" and (.created_at | startswith($d)))]')

echo "Saques de hoje: $(echo "$DEBITS" | jq 'length')"
echo "Total: R$ $(echo "$DEBITS" | jq '[.[].amount | tonumber] | add // 0')"
echo "$DEBITS" | jq '.[] | {id, amount, description, created_at}'

---

Notificacao por Slack

import requests

def send_slack_alert(webhook_url: str, message: str) -> None:
    """Envia alerta para canal do Slack via Incoming Webhook."""
    requests.post(webhook_url, json={"text": message})


# Usar no monitor de SLA
violations = check_sla_violations(token=TOKEN, max_pending_hours=2.0)
if violations:
    msg = f":warning: *{len(violations)} entregas pendentes ha mais de 2h!*\n"
    for v in violations:
        msg += f"• {v['title']} — {v['age_hours']}h sem motorista\n"
    send_slack_alert(SLACK_WEBHOOK_URL, msg)