FEAT: Migração para WiFire

- Substitui uso da Wifeed pela Wifere como fonde de dados de coleta dos usuarios - Correção de cron - Melhoria de logging
2026-06-29 17:35:04 -03:00 · 2026-06-29 17:35:04 -03:00 · 4af74ab458
commit 4af74ab458
parent 555a2e5eff
11 changed files with 250 additions and 95 deletions
--- a/.env.example
+++ b/.env.example
@ -19,10 +19,12 @@ RUIJIE_SECRET=10493c81e8e94f56b8710d78ed2527c7
 RUIJIE_ACCESS_TOKEN=          # opcional — ETL renova automaticamente
 RUIJIE_GROUP_ID=9290679

-# WiFeed (autenticação usuários)
-WIFEED_BASE_URL=https://api.wifeed.com.br
-WIFEED_CLIENT_ID=60e40ee2-f39f-4556-8a22-840a2e3fa686
-WIFEED_CLIENT_SECRET=dRpd6FB2hjbyvcA
+# WiFire (autenticação usuários)
+WIFIRE_BASE_URL=https://api.wifire.me
+WIFIRE_USERNAME=seu_email@empresa.com.br
+WIFIRE_API_KEY=sua_api_key_aqui
+WIFIRE_ESTABLISHMENT_ID=   # ID numérico do estabelecimento (use list_establishments() para descobrir)
+WIFIRE_GROUP_ID=           # Alternativa ao ESTABLISHMENT_ID — preencha um dos dois

 # Logging
 LOG_LEVEL=INFO
@ -41,7 +43,7 @@ No **Gitea**, cadastre em **Settings → Secrets**:
 | `DB_PASSWORD` | Senha do PostgreSQL (pode ser secret) |
 | `RUIJIE_ACCESS_TOKEN` | Token Ruijie (opcional — ETL pode renovar) |
 | `RUIJIE_SECRET` | Secret Ruijie |
-| `WIFEED_CLIENT_SECRET` | Client Secret WiFeed |
+| `WIFIRE_API_KEY` | API Key WiFire |

 ---

@ -59,8 +61,9 @@ Em **Settings → Variables** (não sensíveis):
 | `RUIJIE_APPID` | AppID Ruijie | — |
 | `RUIJIE_SECRET` | Secret Ruijie | — |
 | `RUIJIE_GROUP_ID` | Group ID Ruijie | `9290679` |
-| `WIFEED_BASE_URL` | URL base WiFeed | — |
-| `WIFEED_CLIENT_ID` | Client ID WiFeed | — |
+| `WIFIRE_BASE_URL` | URL base WiFire | `https://api.wifire.me` |
+| `WIFIRE_USERNAME` | Usuário WiFire | — |
+| `WIFIRE_ESTABLISHMENT_ID` | ID do estabelecimento WiFire | — |
 | `LOG_LEVEL` | Nível de log | `INFO` |
 | `REGISTRY` | Docker Registry (ex: `ghcr.io/user`) | vazio |
 | `REGISTRY_USERNAME` | Username do registry | vazio |
--- a/app/core/config.py
+++ b/app/core/config.py
@ -17,9 +17,11 @@ RUIJIE_SECRET       = os.getenv("RUIJIE_SECRET")
 RUIJIE_ACCESS_TOKEN = os.getenv("RUIJIE_ACCESS_TOKEN") 
 RUIJIE_GROUP_ID     = os.getenv("RUIJIE_GROUP_ID")

-# WiFeed (pendente)
-WIFEED_BASE_URL = os.getenv("WIFEED_BASE_URL", "")
-WIFEED_CLIENT_ID = os.getenv("WIFEED_CLIENT_ID", "")
-WIFEED_CLIENT_SECRET = os.getenv("WIFEED_CLIENT_SECRET", "")
+# WiFire
+WIFIRE_BASE_URL          = os.getenv("WIFIRE_BASE_URL", "https://api.wifire.me")
+WIFIRE_USERNAME          = os.getenv("WIFIRE_USERNAME", "")
+WIFIRE_API_KEY           = os.getenv("WIFIRE_API_KEY", "")
+WIFIRE_ESTABLISHMENT_ID  = os.getenv("WIFIRE_ESTABLISHMENT_ID", "")
+WIFIRE_GROUP_ID          = os.getenv("WIFIRE_GROUP_ID", "")

 LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO")
--- a/app/extractor/ruijie.py
+++ b/app/extractor/ruijie.py
@ -1,6 +1,7 @@
 import requests
 import logging
 from typing import Tuple, Optional
+from datetime import datetime, timezone

 from app.core.config import (
    RUIJIE_BASE_URL,
@ -20,9 +21,10 @@ def get_access_token() -> str:
    payload = {"appid": RUIJIE_APPID, "secret": RUIJIE_SECRET}
    resp = requests.post(url, json=payload, timeout=15)
    resp.raise_for_status()
-    token = resp.json().get("data", {}).get("access_token")
+    data = resp.json()
+    token = data.get("accessToken") or data.get("data", {}).get("access_token")
    if not token:
-        raise ValueError(f"Token não retornado: {resp.json()}")
+        raise ValueError(f"Token não retornado: {data}")
    return token


@ -98,7 +100,13 @@ def extract_all_users(
        if page_max > max_online_ms:
            max_online_ms = page_max

-        logger.info(f"Ruijie: página {page} → {len(records)} registros (watermark={watermark_ms})")
+        times = [r.get('onlineTime', 0) for r in records if r.get('onlineTime')]
+        if times:
+            t_min = datetime.fromtimestamp(min(times) / 1000, tz=timezone.utc).strftime('%Y-%m-%d %H:%M')
+            t_max = datetime.fromtimestamp(max(times) / 1000, tz=timezone.utc).strftime('%Y-%m-%d %H:%M')
+            logger.info(f"Ruijie: página {page} → {len(records)} registros | range {t_min} → {t_max} UTC | coletados={len(all_records)}")
+        else:
+            logger.info(f"Ruijie: página {page} → {len(records)} registros (sem onlineTime)")

        if len(records) < page_size or stopped_early:
            break
--- a/app/extractor/wifire.py
+++ b/app/extractor/wifire.py
@ -0,0 +1,98 @@
+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
--- a/app/main.py
+++ b/app/main.py
@ -7,12 +7,11 @@ from psycopg2.extras import DictCursor

 from app.core.config import (
    RUIJIE_ACCESS_TOKEN, LOG_LEVEL,
-    WIFEED_CLIENT_ID, WIFEED_CLIENT_SECRET,
    DB_HOST, DB_PORT, DB_NAME, DB_USER, DB_PASSWORD
 )
-from app.extractor.ruijie import refresh_token, extract_all_users
-from app.extractor.wifeed import get_access_token as wf_token, extract_all_access, WiFeedIPBlockedError
-from app.transform.merge_mac import transform_ruijie, transform_wifeed
+from app.extractor.ruijie import refresh_token, get_access_token as ruijie_login, extract_all_users
+from app.extractor.wifire import extract_customers
+from app.transform.merge_mac import transform_ruijie, transform_wifire
 from app.load.load_database import load

 logging.basicConfig(
@ -74,43 +73,51 @@ def run():
    try:
        # === Ruijie ===
        log.info("--- Ruijie ---")
-        token_r = refresh_token(RUIJIE_ACCESS_TOKEN) if RUIJIE_ACCESS_TOKEN else __import__('app.extractor.ruijie', fromlist=['get_access_token']).get_access_token()
+        if RUIJIE_ACCESS_TOKEN:
+            try:
+                token_r = refresh_token(RUIJIE_ACCESS_TOKEN)
+            except Exception as e:
+                log.warning(f"Ruijie: refresh falhou ({e}), fazendo login completo...")
+                token_r = ruijie_login()
+        else:
+            token_r = ruijie_login()

        # Watermark Ruijie (epoch ms como string)
        wm_val_str, last_run = get_watermark(conn, 'ruijie')
        use_wm = should_use_watermark(last_run, max_age_hours=1)
        wm_ms = int(wm_val_str) if use_wm and wm_val_str else None

-        raw_r, new_wm_ms = extract_all_users(token_r, watermark_ms=wm_ms)
-        log.info(f"Ruijie: raw_r tem {len(raw_r)} registros extraídos")
-        sessions = [transform_ruijie(r) for r in raw_r]  # Sem filtro if
-        log.info(f"Ruijie: {len(sessions)} sessions transformadas")
-        log.info(f"Ruijie: {len(sessions)} sessions (wm={'used' if use_wm else 'ignored'})")
+        if wm_val_str and not use_wm:
+            from datetime import datetime, timezone
+            wm_dt = datetime.fromtimestamp(int(wm_val_str) / 1000, tz=timezone.utc)
+            log.warning(f"Ruijie: watermark IGNORADO — last_run_at={last_run} (>1h atrás). Extraindo tudo desde o início.")
+        elif wm_ms:
+            from datetime import datetime, timezone
+            wm_dt = datetime.fromtimestamp(wm_ms / 1000, tz=timezone.utc)
+            log.info(f"Ruijie: watermark ATIVO — extraindo apenas registros após {wm_dt.strftime('%Y-%m-%d %H:%M UTC')}")
+        else:
+            log.info("Ruijie: sem watermark — primeira execução, extraindo tudo")

-        # === WiFeed ===
-        log.info("--- WiFeed ---")
-        token_w = None
+        raw_r, new_wm_ms = extract_all_users(token_r, watermark_ms=wm_ms)
+        sessions = [transform_ruijie(r) for r in raw_r]
+        log.info(f"Ruijie: {len(raw_r)} registros extraídos → {len(sessions)} sessions transformadas")
+
+        # === WiFire ===
+        log.info("--- WiFire ---")
+        users = []
+        new_wm_date = date.today().strftime("%Y-%m-%d")
        try:
-            token_w = wf_token(WIFEED_CLIENT_ID, WIFEED_CLIENT_SECRET)
-        except WiFeedIPBlockedError as e:
-            log.error(f"⛔ WiFeed desabilitado: {e}")
-            log.error("⏳ Aguarde desbloqueio do IP e tente novamente amanhã")
-            token_w = None
-            users = []
-            new_wm_date = date.today().strftime("%Y-%m-%d")
-        
-        if token_w:
-            wm_date_str, last_run_w = get_watermark(conn, 'wifeed')
+            wm_date_str, last_run_w = get_watermark(conn, 'wifire')
            use_wm_w = should_use_watermark(last_run_w, max_age_hours=1)
            wm_date = date.fromisoformat(wm_date_str) if use_wm_w and wm_date_str else None

-            raw_w, new_wm_date = extract_all_access(token_w, watermark_date=wm_date)
-            log.info(f"WiFeed: raw_w tem {len(raw_w)} registros extraídos")
-            users = [transform_wifeed(r) for r in raw_w]  # Sem filtro if
-            log.info(f"WiFeed: {len(users)} users transformados")
-            log.info(f"WiFeed: {len(users)} users (wm={'used' if use_wm_w else 'ignored'})")
-        else:
-            log.info("WiFeed: pulando (IP bloqueado ou não autenticado)")
+            raw_w, new_wm_date = extract_customers(watermark_date=wm_date)
+            log.info(f"WiFire: raw_w tem {len(raw_w)} registros extraídos")
+            users = [transform_wifire(r) for r in raw_w]
+            log.info(f"WiFire: {len(users)} users transformados (wm={'used' if use_wm_w else 'ignored'})")
+        except Exception as e:
+            log.error(f"WiFire: erro na extração: {e}", exc_info=True)
+            users = []

        # === Load ===
        load(conn, users, sessions)
@ -123,11 +130,7 @@ def run():
        else:
            log.info("Ruijie: sem registros novos, watermark mantido")

-        # WiFeed: atualiza apenas se conseguiu autenticar
-        if token_w:
-            set_watermark(conn, 'wifeed', new_wm_date)
-        else:
-            log.info("WiFeed: watermark não atualizado (autenticação falhou)")
+        set_watermark(conn, 'wifire', new_wm_date)
        
        conn.commit()
        log.info("=== ETL DONE ===")
--- a/app/transform/merge_mac.py
+++ b/app/transform/merge_mac.py
@ -31,33 +31,38 @@ def transform_ruijie(record: dict) -> dict:
    }


-def transform_wifeed(record: dict) -> dict:
+def transform_wifire(record: dict) -> dict:
    """
-    Access WiFeed → dict (usuário + MAC do dispositivo).
-    
-    Dados vêm de /v2/report/access e incluem:
-    - hostMacAddress: MAC do dispositivo
-    - clientName, clientEmail, clientPhoneNumber: Info do cliente
-    - clientGender, clientBirthdate: Dados demográficos
-    - clientExtraFields: CPF e outros extras
-    - clientId: ID único do cliente
-    - hostType: Tipo de dispositivo (Android, iPhone, etc)
-    - localName: Local/network onde conectou
+    Customer WiFire → dict para tabela users.
+
+    Campos esperados de /establishments/customers:
+    - mac / mac_address: MAC do dispositivo (tenta os dois)
+    - name: nome do cliente
+    - email: e-mail
+    - cpf: CPF (campo direto na resposta WiFire)
+    - sex: gênero ('M'/'F')
+    - birthday / birthdate: data de nascimento 'YYYY-MM-DD'
+    - phone / celphone: telefone
+    - id: ID único do cliente
+    - device_type / host_type: tipo de dispositivo
+    - establishment_name / local_name: nome do local
    """
-    cpf = ''
-    extra = record.get('clientExtraFields')
-    if isinstance(extra, dict):
-        cpf = extra.get('CPF', '')
+    mac = (
+        record.get('mac') or
+        record.get('mac_address') or
+        record.get('device_mac') or
+        ''
+    )

    return {
-        'mac_address':  normalize_mac(record.get('hostMacAddress', '')),
-        'name':         record.get('clientName', ''),
-        'email':        record.get('clientEmail', ''),
-        'cpf':          cpf,
-        'gender':       record.get('clientGender', ''),
-        'birthdate':    record.get('clientBirthdate'),   # string 'YYYY-MM-DD'
-        'phone':        record.get('clientPhoneNumber', ''),
-        'client_id':    int(record.get('clientId', 0) or 0),
-        'host_type':    record.get('hostType', ''),
-        'local_name':   record.get('localName', ''),
+        'mac_address':  normalize_mac(mac),
+        'name':         record.get('name', ''),
+        'email':        record.get('email', ''),
+        'cpf':          record.get('cpf', ''),
+        'gender':       record.get('sex', ''),
+        'birthdate':    record.get('birthday') or record.get('birthdate'),
+        'phone':        record.get('phone') or record.get('celphone', ''),
+        'client_id':    int(record.get('id', 0) or 0),
+        'host_type':    record.get('device_type') or record.get('host_type', ''),
+        'local_name':   record.get('establishment_name') or record.get('local_name', ''),
    }
--- a/backfill.py
+++ b/backfill.py
@ -2,31 +2,27 @@ import time
 from datetime import date, timedelta
 import psycopg2

-from app.extractor.wifeed import get_access_token, extract_all_access
-from app.core.config import WIFEED_CLIENT_ID, WIFEED_CLIENT_SECRET
+from app.extractor.wifire import extract_customers
 from app.core.config import DB_HOST, DB_PORT, DB_NAME, DB_USER, DB_PASSWORD
-from app.transform.merge_mac import transform_wifeed
+from app.transform.merge_mac import transform_wifire
 from app.load.load_database import load

 START_DATE = date(2026, 4, 21)
 END_DATE   = date.today() - timedelta(days=1)

-SLEEP_BETWEEN = 20  # 20s entre requisições = 3 req/min, bem abaixo do limite
+SLEEP_BETWEEN = 20  # 20s entre requisições = 3 req/min, abaixo do limite de 10 req/min

 conn = psycopg2.connect(host=DB_HOST, port=DB_PORT, dbname=DB_NAME, user=DB_USER, password=DB_PASSWORD)
 conn.autocommit = False

-token = get_access_token(WIFEED_CLIENT_ID, WIFEED_CLIENT_SECRET)
-time.sleep(SLEEP_BETWEEN)  # pausa já após o login
-
 current = START_DATE
 while current <= END_DATE:
    print(f"Processando {current}...")
    try:
-        raw, _ = extract_all_access(token, watermark_date=current)
-        users = [transform_wifeed(r) for r in raw if transform_wifeed(r)]
+        raw, _ = extract_customers(watermark_date=current)
+        users = [transform_wifire(r) for r in raw]
        if users:
-            load(conn, users, [])  # sessions vazio no backfill
+            load(conn, users, [])
            conn.commit()
            print(f"  {len(users)} usuários inseridos")
        else:
@ -38,9 +34,9 @@ while current <= END_DATE:
            time.sleep(90)
        else:
            print(f"  ERRO em {current}: {e}")
-    
+
    current += timedelta(days=1)
    time.sleep(SLEEP_BETWEEN)

 conn.close()
-print("Backfill concluído!")
+print("Backfill concluído!")
--- a/infra/Dockerfile
+++ b/infra/Dockerfile
@ -22,10 +22,9 @@ COPY infra/init.sql /docker-entrypoint-initdb.d/
 COPY infra/entrypoint.sh /usr/local/bin/
 RUN chmod +x /usr/local/bin/entrypoint.sh

-# Crontab
+# Crontab (system cron.d format: MIN HOUR DOM MON DOW USER CMD)
 COPY infra/crontab /etc/cron.d/wifi-etl
-RUN chmod 0644 /etc/cron.d/wifi-etl && \
-    crontab /etc/cron.d/wifi-etl
+RUN chmod 0644 /etc/cron.d/wifi-etl

 # Log dir
 RUN mkdir -p /var/log && touch /var/log/wifi-etl.log /var/log/wifi-etl-cleanup.log
--- a/infra/crontab
+++ b/infra/crontab
@ -1,10 +1,7 @@
 # WiFi ETL Pipeline — Cron Jobs
-# Formato: MIN HOUR DOM MON DOW COMMAND
-# Fuso horário: UTC (ajustar se necessário)
+# Formato: MIN HOUR DOM MON DOW USER COMMAND
+# Fuso horário: UTC

 # A cada 5 minutos: executa o ETL
-# WiFi ETL Pipeline — Cron Jobs
+*/5 * * * * root cd /app && /usr/local/bin/python -m app.main >> /var/log/wifi-etl.log 2>&1

-*/5 * * * * cd /app && /usr/local/bin/python -m app.main >> /var/log/wifi-etl.log 2>&1
-
-#END
--- a/infra/docker-compose.yml
+++ b/infra/docker-compose.yml
@ -1,9 +1,15 @@
 version: '3.8'

+networks:
+  wifi-etl-net:
+    driver: bridge
+
 services:
  postgres:
    image: postgres:15-alpine
    container_name: wifi_etl_db
+    networks:
+      - wifi-etl-net
    environment:
      POSTGRES_DB: wifi_etl
      POSTGRES_USER: postgres
@ -25,6 +31,8 @@ services:
      context: ..
      dockerfile: infra/Dockerfile
    container_name: wifi_etl_worker
+    networks:
+      - wifi-etl-net
    env_file:
      - ../.env
    depends_on:
--- a/preview.py
+++ b/preview.py
@ -0,0 +1,36 @@
+"""
+Preview dos dados coletados da WiFire.
+Rode da raiz do projeto:  python preview.py
+"""
+import json
+from datetime import date, timedelta
+from app.extractor.wifire import extract_customers
+from app.transform.merge_mac import transform_wifire
+
+# Busca dados de ontem
+ontem = date.today() - timedelta(days=1)
+print(f"\nBuscando clientes com last_access em {ontem}...\n")
+
+raw, date_str = extract_customers(watermark_date=ontem)
+
+print(f"Total encontrado: {len(raw)} clientes\n")
+
+if not raw:
+    print("Nenhum dado retornado. Tente outra data ou verifique o establishment ID.")
+else:
+    print("=" * 60)
+    print("EXEMPLO — 1 registro bruto da API WiFire:")
+    print("=" * 60)
+    print(json.dumps(raw[0], indent=2, ensure_ascii=False))
+
+    print("\n" + "=" * 60)
+    print("EXEMPLO — após transform (como vai pro banco):")
+    print("=" * 60)
+    print(json.dumps(transform_wifire(raw[0]), indent=2, ensure_ascii=False))
+
+    print("\n" + "=" * 60)
+    print(f"RESUMO dos {min(5, len(raw))} primeiros clientes:")
+    print("=" * 60)
+    for r in raw[:5]:
+        t = transform_wifire(r)
+        print(f"  {t['name']:<30} mac={t['mac_address']}  cpf={t['cpf'] or '---'}  email={t['email'] or '---'}")