Rafael Lopes
331a021d9a
Some checks failed
Deploy WiFi-ETL Prod / deploy (push) Failing after 0s
- Adicionado suporte para extração de dados do Ruijie e WiFeed, incluindo autenticação e tratamento de erros. - Adicionado suporte para watermarking em ambas as fontes para extração incremental. - Criado script de transformação para mesclagem de MAC addresses. - Implementado Backfill para WiFeed, permitindo extração histórica com controle de taxa. - Adicionado script de depuração para testes de transformação do WiFeed. - Desenvolvido scripts de implantação e configurações do Docker para setup de produção. - Criado script de inicialização do schema do PostgreSQL em infra/init.sql. - Adicionado teste automatizado para lógica de transformação e carregamento em test_transform_load.py. - Atualizado documentation para implantação e setup de produção.
84 lines
2.7 KiB
Python
84 lines
2.7 KiB
Python
import psycopg2
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
from typing import Optional
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
INSERT_USER = """
|
|
INSERT INTO users (mac_address, name, cpf, gender, email, birthdate,
|
|
phone, client_id, host_type, local_name)
|
|
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
|
|
ON CONFLICT (mac_address) DO UPDATE SET
|
|
name = EXCLUDED.name,
|
|
cpf = EXCLUDED.cpf,
|
|
gender = EXCLUDED.gender,
|
|
email = EXCLUDED.email,
|
|
birthdate = EXCLUDED.birthdate,
|
|
phone = EXCLUDED.phone,
|
|
client_id = EXCLUDED.client_id,
|
|
host_type = EXCLUDED.host_type,
|
|
local_name = EXCLUDED.local_name,
|
|
updated_at = NOW();
|
|
"""
|
|
|
|
INSERT_SESSION = """
|
|
INSERT INTO sessions (
|
|
mac_address, access_point_name, building_name,
|
|
band, channel, rssi, user_ip,
|
|
bytes_up, bytes_down, bytes_total, serial_number,
|
|
online_time, offline_time, active_time_ms
|
|
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
|
|
ON CONFLICT (mac_address, online_time) DO NOTHING;
|
|
"""
|
|
|
|
|
|
def epoch_to_utc(ms: Optional[int]) -> Optional[datetime]:
|
|
return datetime.fromtimestamp(ms / 1000, tz=timezone.utc) if ms else None
|
|
|
|
|
|
def load(conn, users: list[dict], sessions: list[dict]):
|
|
cur = conn.cursor()
|
|
|
|
# Users upsert
|
|
if users:
|
|
user_params = [
|
|
(
|
|
u['mac_address'], u.get('name'), u.get('cpf'), u.get('gender'),
|
|
u.get('email'), u.get('birthdate'), u.get('phone'),
|
|
u.get('client_id'), u.get('host_type'), u.get('local_name')
|
|
) for u in users
|
|
]
|
|
cur.executemany(INSERT_USER, user_params)
|
|
logger.info(f"Users upsert: {cur.rowcount} rows")
|
|
else:
|
|
logger.info(f"Users upsert: 0 rows (nenhum usuário para processar)")
|
|
|
|
# Sessions insert (converte epoch → datetime)
|
|
if sessions:
|
|
session_params = [
|
|
(
|
|
s['mac_address'],
|
|
s['access_point_name'],
|
|
s['building_name'],
|
|
s['band'],
|
|
s['channel'],
|
|
s['rssi'],
|
|
s['user_ip'],
|
|
s['bytes_up'],
|
|
s['bytes_down'],
|
|
s['bytes_total'],
|
|
s['serial_number'],
|
|
epoch_to_utc(s['online_time_ms']),
|
|
epoch_to_utc(s['offline_time_ms']),
|
|
s['active_time_ms']
|
|
) for s in sessions
|
|
]
|
|
cur.executemany(INSERT_SESSION, session_params)
|
|
logger.info(f"Sessions insert: {cur.rowcount} rows")
|
|
else:
|
|
logger.info(f"Sessions insert: 0 rows (nenhuma sessão para processar)")
|
|
|
|
conn.commit()
|
|
cur.close()
|