Rafael Lopes
4af74ab458
All checks were successful
Deploy WiFi-ETL Prod / deploy (push) Successful in 47s
- Substitui uso da Wifeed pela Wifere como fonde de dados de coleta dos usuarios - Correção de cron - Melhoria de logging
99 lines
3.1 KiB
Python
99 lines
3.1 KiB
Python
import requests
|
|
import logging
|
|
from typing import Dict, List, Optional, Tuple
|
|
from datetime import date
|
|
|
|
from app.core.config import (
|
|
WIFIRE_BASE_URL, WIFIRE_USERNAME, WIFIRE_API_KEY,
|
|
WIFIRE_ESTABLISHMENT_ID, WIFIRE_GROUP_ID,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
BASE_URL = (WIFIRE_BASE_URL or "https://api.wifire.me").rstrip('/')
|
|
|
|
|
|
def _auth() -> tuple:
|
|
if not WIFIRE_USERNAME or not WIFIRE_API_KEY:
|
|
raise ValueError("WiFire: WIFIRE_USERNAME e WIFIRE_API_KEY são obrigatórios")
|
|
return (WIFIRE_USERNAME, WIFIRE_API_KEY)
|
|
|
|
|
|
def list_establishments() -> list:
|
|
"""Lista estabelecimentos disponíveis — útil para descobrir o WIFIRE_ESTABLISHMENT_ID."""
|
|
url = f"{BASE_URL}/establishments"
|
|
resp = requests.get(url, auth=_auth(), timeout=15)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
return data.get("data", data)
|
|
|
|
|
|
def extract_customers(
|
|
watermark_date: Optional[date] = None,
|
|
) -> Tuple[List[Dict], str]:
|
|
"""
|
|
Extrai clientes do WiFire via GET /establishments/customers.
|
|
|
|
Filtra pelo dia exato passado em watermark_date (initialPeriod = finalPeriod = data).
|
|
Percorre todas as páginas via campo 'nextpage' na resposta.
|
|
|
|
Args:
|
|
watermark_date: Data a extrair (padrão: hoje).
|
|
|
|
Returns:
|
|
(lista_de_clientes, date_str "YYYY-MM-DD")
|
|
"""
|
|
target_date = watermark_date or date.today()
|
|
date_str = target_date.strftime("%Y-%m-%d")
|
|
|
|
if not WIFIRE_ESTABLISHMENT_ID and not WIFIRE_GROUP_ID:
|
|
raise ValueError(
|
|
"WiFire: configure WIFIRE_ESTABLISHMENT_ID ou WIFIRE_GROUP_ID. "
|
|
"Use list_establishments() para descobrir o valor correto."
|
|
)
|
|
|
|
# nextpage retorna apenas ?page=N sem os outros params, então paginamos manualmente
|
|
params: Dict = {}
|
|
if WIFIRE_ESTABLISHMENT_ID:
|
|
params["establishment"] = WIFIRE_ESTABLISHMENT_ID
|
|
else:
|
|
params["group"] = WIFIRE_GROUP_ID
|
|
params["periodType"] = "last_access"
|
|
params["initialPeriod"] = date_str
|
|
params["finalPeriod"] = date_str
|
|
|
|
logger.info(f"WiFire: extraindo clientes com last_access em {date_str}")
|
|
|
|
url = f"{BASE_URL}/establishments/customers"
|
|
all_records: List[Dict] = []
|
|
page = 1
|
|
|
|
while True:
|
|
try:
|
|
params["page"] = page
|
|
resp = requests.get(url, auth=_auth(), params=params, timeout=30)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
|
|
records = data.get("data", [])
|
|
if not isinstance(records, list):
|
|
logger.warning(f"WiFire: campo 'data' não é lista: {type(records)}")
|
|
break
|
|
|
|
all_records.extend(records)
|
|
logger.debug(f"WiFire: página {page} — {len(records)} registros")
|
|
|
|
if not records or not data.get("nextpage"):
|
|
break
|
|
page += 1
|
|
|
|
except Exception as e:
|
|
logger.error(f"WiFire: erro durante extração: {e}", exc_info=True)
|
|
raise
|
|
|
|
if all_records:
|
|
logger.debug(f"WiFire: campos disponíveis no registro: {list(all_records[0].keys())}")
|
|
|
|
logger.info(f"WiFire: {len(all_records)} clientes extraídos para {date_str}")
|
|
return all_records, date_str
|