diff --git a/.env.example b/.env.example index f9e6017..3d2f3d6 100644 --- 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 | diff --git a/app/core/config.py b/app/core/config.py index ff503e2..2fc2637 100644 --- 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") \ No newline at end of file diff --git a/app/extractor/ruijie.py b/app/extractor/ruijie.py index 32e8d58..9d1d2f8 100644 --- 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 diff --git a/app/extractor/wifire.py b/app/extractor/wifire.py new file mode 100644 index 0000000..f2dffd6 --- /dev/null +++ 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 diff --git a/app/main.py b/app/main.py index e5bf447..b17bcf4 100644 --- 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 ===") diff --git a/app/transform/merge_mac.py b/app/transform/merge_mac.py index ad70a8c..104bddc 100644 --- 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', ''), } diff --git a/backfill.py b/backfill.py index 72a4bf9..87503f4 100644 --- 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!") \ No newline at end of file +print("Backfill concluído!") diff --git a/infra/Dockerfile b/infra/Dockerfile index c661b0c..c388e12 100644 --- 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 diff --git a/infra/crontab b/infra/crontab index 38ebe78..4657ec8 100644 --- 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 diff --git a/infra/docker-compose.yml b/infra/docker-compose.yml index b09db3a..5df1c6d 100644 --- 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: diff --git a/preview.py b/preview.py new file mode 100644 index 0000000..795508f --- /dev/null +++ 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 '---'}")