construir_relatorio()Objetivo: gerar um relatório HTML executivo e auditável consolidando:
- Inspeção de esquema (metadados das tabelas)
- Métricas Antes × Depois e Δ (delta)
- Resumo da execução (quantos vínculos inseridos, quantos usuários atualizados, consultas executadas)
- Amostra de registros para verificação manual
Esse relatório é pensado para auditoria, troubleshooting e tomada de decisão na GSERV. Ele documenta o que mudou, por que mudou e como foi feito — com evidências.
Acesse: Relatório das Dags
def construir_relatorio(inspec: dict, before: dict, resumo: dict, after: dict) -> dict:
"""Retorna {"subject": <assunto>, "html": <conteúdo_html>}"""
inspec: saída de inspecionar_tabelas() (schema e observações)
before: métricas antes (saída de _collect_metrics())
resumo: retorno de aplicar_regra() (IDs afetados, totals, queries)
after: métricas depois (saída de _collect_metrics())
O retorno é um dicionário serializável pronto para ser:
1. salvo como arquivo .html em /opt/airflow/output/…, e/ou
2. enviado por e-mail (ex.: EmailOperator/SMTP custom).
m = re.search(r"%OU=([^_%]+)", USER_DN_LIKE, re.IGNORECASE)
org_name = (m.group(1).title() if m else "SEAS")
Extrai a OU do padrão (ex.: «%OU=SEAS_%» → Seas)
Usa como título do relatório e identificação visual.
Dica: padronize USER_DN_LIKE por órgão (SEAS, PC, SEDUC…) para nomes corretos.
render_sample(rows): recebe uma lista de dicionários (ex.: amostra de usuários) e retorna uma tabela HTML simples, pronta para embutir no relatório.
render_schema(schema): percorre as tabelas/colunas inspecionadas em information_schema.COLUMNS e produz tabelas HTML com column, type, null, key, default.
Essas funções encapsulam o HTML e deixam a função principal limpa.
diffs = {
"total_users": after["total_users"] - before["total_users"],
"seas_users_like": after["seas_users_like"] - before["seas_users_like"],
"users_profiles_id_147": after["users_profiles_id_147"] - before["users_profiles_id_147"],
"links_profiles_users_147": after["links_profiles_users_147"] - before["links_profiles_users_147"],
}
O delta mostra o impacto real da execução (ver seção “Diferenças (Δ)”).
Se sua DAG for parametrizada para outro perfil/OU, ajuste as chaves (ou gere dinamicamente pelo PROFILE_ID).
Recomendação GSERV: padronizar nomes das chaves com placeholders (ex.: users_profiles_id_{PROFILE_ID}) para reuso entre órgãos.
to_link_ids = resumo.get("to_link_ids", []) or []
to_update_ids = resumo.get("to_update_ids", []) or []
IDs vinculados em glpi_profiles_users
IDs atualizados em glpi_users.profiles_id
Esses artefatos são ouro para auditoria e debugging (podem virar CSV).
O HTML usa Bootstrap 5 (CDN) e algumas classes padrões (table, table-striped, table-bordered) para:
legibilidade
consistência visual
visual “executivo” que funciona tanto no browser quanto por e-mail (boa compatibilidade)
Observação: relatórios de e-mail têm limitações de CSS. Evite recursos avançados, mantenha o layout simples e tabelas legíveis.
Resumo da Regra
Mostra PROFILE_ID, USER_DN_LIKE e as ações que a DAG realiza (vínculo e update).
Métricas (Antes → Depois) com Δ
Tabela comparativa para leitura rápida do impacto.
Ações Executadas
Totais e “cards” com os primeiros 20 IDs afetados (útil para uma revisão manual rápida).
Amostra de Usuários (antes)
Amostra do segmento filtrado, para conferir reais padrões de user_dn e profiles_id.
Esquema (colunas) — tabelas alvo
Metadados do schema atual, garantindo transparência técnica (o que havia no momento da execução).
Consultas principais
Trecho pre com o array de queries executadas (quando populado no resumo).
a) Gerar o HTML após medir o “depois”
from airflow.decorators import task
from datetime import datetime
import pathlib
@task
def gerar_relatorio_html(inspec: dict, before: dict, resumo: dict, after: dict):
rep = construir_relatorio(inspec, before, resumo, after) # {"subject":..., "html":...}
# Persistir artefatos (padronize pasta por run)
run_ts = datetime.now().strftime("%Y%m%d_%H%M%S")
base = pathlib.Path(f"/opt/airflow/output/glpi_seas/{run_ts}")
base.mkdir(parents=True, exist_ok=True)
# Salvar HTML
html_path = base / "relatorio_glpi_seas.html"
html_path.write_text(rep["html"], encoding="utf-8")
# (Opcional) Também salvar JSON/CSV de deltas e IDs
# ...
return {"subject": rep["subject"], "path": str(html_path)}
b) Enviar por e-mail (opções)
Com EmailOperator (se configurado):
Monte o EmailOperator com html_content lendo o arquivo salvo.
Com smtplib (custom):
Envie usando SMTP da instituição.
Lembre-se de usar Variáveis do Airflow/Connections para segredos.
Boas práticas:
Nunca embuta credenciais no código.
Evite anexos pesados (HTML puro costuma ser suficiente).
Configure reply-to e assunto padronizado (ex.: [Airflow][GLPI][SEAS] Relatório ).
Estrutura de pastas:
/opt/airflow/output/<dag_id>/<run_ts>/
├── relatorio_glpi_<org>.html
├── deltas.json
├── to_link_ids.csv
└── to_update_ids.csv
| Sintoma | Possível causa | Ação |
|---|---|---|
| Δ = 0 com candidatos > 0 | Falta permissão de UPDATE/INSERT |
Verificar grants do usuário do DB |
| Muitos candidatos, pouco Δ | Falhas parciais (chunk) ou transação | Ver logs, procurar erro entre chunks, validar commit |
| Δ alto inesperado | Filtro muito amplo (USER_DN_LIKE) |
Refinar LIKE, validar com amostra |
| Divergência em vínculos | Inserção falhou/foi parcial | Conferir índices/locks; reprocessar apenas vínculos |
Importante: O relatório não é apenas visual — é documento de auditoria.
Armazene o HTML e os CSVs com timestamp por execução; isso constrói um histórico confiável para inspeções futuras.
Mantenha o HTML autocontido (sem dependências externas sensíveis). O uso de CDN do Bootstrap é ok; se necessário, inline CSS crítico.
Abaixo, o snippet-base da sua função (com comentários já explicados nesta seção). Mantenha sincronizado com as variáveis/constantes de perfil e OU da DAG em produção.
def construir_relatorio(inspec: dict, before: dict, resumo: dict, after: dict):
import re
from textwrap import dedent
from datetime import datetime
import json
m = re.search(r"%OU=([^_%]+)", USER_DN_LIKE, re.IGNORECASE)
org_name = (m.group(1).title() if m else "SEAS")
report_date = datetime.now().strftime('%d/%m/%Y')
def render_sample(rows):
if not rows:
return "<em>Nenhum exemplo disponível.</em>"
head = "<tr><th>id</th><th>name</th><th>profiles_id</th><th>user_dn</th></tr>"
body = "\n".join(
f"<tr><td>{r.get('id','')}</td><td>{r.get('name','')}</td>"
f"<td>{r.get('profiles_id','')}</td>"
f"<td style='font-size:12px'>{r.get('user_dn','')}</td></tr>"
for r in rows
)
return f"<table border='1' cellpadding='6' cellspacing='0'>{head}{body}</table>"
def render_schema(schema):
parts = []
for t, cols in schema.items():
cols_rows = "".join(
f"<tr><td>{c['column']}</td><td>{c['type']}</td><td>{c['nullable']}</td>"
f"<td>{c['key']}</td><td>{c['default']}</td></tr>"
for c in cols
)
parts.append(dedent(f"""
<h4 style="margin-bottom:6px">{t}</h4>
<table border="1" cellpadding="6" cellspacing="0">
<tr><th>column</th><th>type</th><th>null</th><th>key</th><th>default</th></tr>
{cols_rows}
</table>
"""))
return "\n".join(parts) if parts else "<em>Sem metadados.</em>"
diffs = {
"total_users": after["total_users"] - before["total_users"],
"seas_users_like": after["seas_users_like"] - before["seas_users_like"],
"users_profiles_id_147": after["users_profiles_id_147"] - before["users_profiles_id_147"],
"links_profiles_users_147": after["links_profiles_users_147"] - before["links_profiles_users_147"],
}
to_link_ids = resumo.get("to_link_ids", []) or []
to_update_ids = resumo.get("to_update_ids", []) or []
subject = f"[Airflow][GLPI] Rotina {org_name} concluída — {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
html = f"""... HTML conforme explicado acima ..."""
return {"subject": subject, "html": html}
salvar_resultados()Esta função finaliza o pipeline consolidando todos os artefatos da execução em disco:
relatório HTML, CSVs de usuários afetados e métricas comparativas.
É uma etapa essencial para auditoria, rastreabilidade e análise de impacto dentro da GSERV.
out_dir = f"/opt/airflow/output/glpi_seduc/{datetime.now().strftime('%Y%m%d_%H%M%S')}"
os.makedirs(out_dir, exist_ok=True)
Cria um diretório exclusivo por execução (baseado em timestamp).
Centraliza todos os artefatos da DAG (HTML, CSVs, logs auxiliares).
⚠️ Importante: o caminho /opt/airflow/output/glpi_seduc/ deve ser ajustado conforme o órgão (ex.: glpi_seas, glpi_pc etc.).
Em produção, recomenda-se padronizar via Airflow Variable (Variable.get(«output_dir_base»)).
html_path = os.path.join(out_dir, "relatorio_glpi_seduc.html")
with open(html_path, "w", encoding="utf-8") as f:
f.write(relatorio["html"])
Grava o HTML gerado pela função construir_relatorio().
O relatório contém:
Metadados de execução (data, perfil alvo, critério de DN)
Métricas “Antes → Depois” com Δ
IDs afetados e queries executadas
Amostras e esquema das tabelas
O arquivo pode ser aberto via navegador ou anexado em e-mail.
Finalidade: manter evidência visual e técnica do que foi processado.
A função realiza uma consulta detalhada no banco para obter as informações completas dos usuários alterados (vinculados ou atualizados).
fetch_details(resumo.get("to_link_ids", []), "linked")
fetch_details(resumo.get("to_update_ids", []), "updated")
Detalhes:
Cada grupo é processado em blocos de 1000 IDs (função chunked()), evitando sobrecarga no banco.
Busca apenas os campos essenciais: id, name, user_dn e o tipo de ação.
Os resultados são compilados em uma lista de dicionários rows.
csv_users_path = os.path.join(out_dir, "usuarios_afetados.csv")
Saída:
Saída:
usuarios_afetados.csv — com os usuários realmente modificados pela DAG.
| id | name | action | user_dn |
|---|---|---|---|
| 101 | Ana Souza | linked | CN=ana,OU=SEDUC,... |
| 103 | Bruno Nunes | updated | CN=bruno,OU=SEDUC,... |
💡 Essa amostra é crucial para auditoria e rollback caso algo precise ser revertido.
Após a coleta, são calculadas as diferenças (Δ) entre as métricas de antes e depois da execução.
diffs = {
"total_users": after["total_users"] - before["total_users"],
"seduc_users_like": after["seduc_users_like"] - before["seduc_users_like"],
"users_profiles_id_115": after["users_profiles_id_115"] - before["users_profiles_id_115"],
"links_profiles_users_115": after["links_profiles_users_115"] - before["links_profiles_users_115"],
}
Essas informações são registradas no CSV metricas.csv:
| metrica | antes | depois | diff |
|---|---|---|---|
| total_users | 52.731 | 52.731 | 0 |
| seduc_users_like | 8.412 | 8.412 | 0 |
| users_profiles_id_115 | 7.890 | 8.410 | +520 |
| links_profiles_users_115 | 7.885 | 8.408 | +523 |
📈 Essas métricas são usadas para análises posteriores, geração de KPIs e monitoramento da efetividade da automação.
Ao final, a função retorna um dicionário com os caminhos completos dos arquivos gerados:
return {
"html_path": html_path,
"csv_users_path": csv_users_path,
"csv_metrics_path": csv_metrics_path,
"output_dir": out_dir,
}
Esse retorno pode ser usado para:
Logar os artefatos no Airflow (XCom ou log da task)
Encaminhar por e-mail (com EmailOperator)
Alimentar pipelines de relatórios (Power BI, Streamlit, etc.)
Ordem típica das tarefas no pipeline:
insp = inspecionar_tabelas()
before = quantificar_antes()
resumo = aplicar_regra(before)
after = quantificar_depois()
rel = construir_relatorio(inspec=insp, before=before, resumo=resumo, after=after)
paths = salvar_resultados(inspec=insp, before=before, resumo=resumo, after=after, relatorio=rel)
Garante que a coleta e o relatório sempre ocorram ao final do fluxo.
A variável paths permite acesso rápido aos arquivos gerados (para logs, e-mail ou upload).
| Prática | Descrição |
|---|---|
| Separar pastas por órgão | Ex.: /opt/airflow/output/glpi_seas/, /glpi_seduc/, /glpi_pc/. |
| Timestamp único por execução | Evita sobreposição e garante rastreabilidade. |
| Nomes consistentes de arquivos | relatorio_glpi_<org>.html, metricas.csv, usuarios_afetados.csv. |
| Manter encoding UTF-8 | Evita erros com acentuação ao abrir CSVs no Excel ou BI. |
| Automatizar rotação de pastas antigas | Para evitar acúmulo de logs antigos no container. |
A função salvar_resultados() é o fechamento do ciclo do pipeline, garantindo que:
As alterações realizadas sejam documentadas.
As métricas fiquem registradas para análise histórica.
O time da GSERV tenha evidência clara do comportamento do fluxo.
✅ Em resumo: nada é perdido — cada execução deixa um rastro completo, técnico e auditável.