# -*- coding: utf-8 -*-
import os
import json
import re
import ipaddress
from datetime import datetime, timedelta, date
import pandas as pd
import plotly.express as px
import plotly.graph_objects as go
from pyzabbix import ZabbixAPI
import time
import locale
import argparse
# --- FUNÇÕES AUXILIARES ---
def conectar_zabbix(server_url, api_token):
"""Conecta-se à API do Zabbix."""
try:
zapi = ZabbixAPI(server_url)
zapi.login(api_token=api_token)
print(f"Conectado com sucesso à API do Zabbix na versão: {zapi.api_version()}")
return zapi
except Exception as e:
print(f"Erro ao conectar ao Zabbix: {e}")
return None
def buscar_dados_zabbix(zapi, host_name, item_name, target_date):
"""Busca o histórico do item para uma data específica."""
try:
host = zapi.host.get(filter={"host": host_name}, output=["hostid"])
if not host:
print(f"Erro: Host '{host_name}' não encontrado.")
return None
hostid = host[0]['hostid']
item = zapi.item.get(filter={"name": item_name, "hostid": hostid}, output=["itemid", "value_type"])
if not item:
print(f"Erro: Item '{item_name}' não encontrado no host '{host_name}'.")
return None
itemid = item[0]['itemid']
value_type = int(item[0]['value_type'])
all_history = []
print(f"Iniciando busca de histórico para o dia {target_date.strftime('%Y-%m-%d')}...")
for hora in range(24):
inicio_hora = datetime.combine(target_date, datetime.min.time()) + timedelta(hours=hora)
fim_hora = inicio_hora + timedelta(hours=1) - timedelta(seconds=1)
time_from, time_till = int(inicio_hora.timestamp()), int(fim_hora.timestamp())
history_chunk = zapi.history.get(
itemids=[itemid], time_from=time_from, time_till=time_till,
history=value_type, output='extend', sortfield='clock', sortorder='ASC'
)
if history_chunk:
all_history.extend(history_chunk)
time.sleep(0.1)
print(f"Busca finalizada. {len(all_history)} registros de log encontrados.")
return all_history
except Exception as e:
print(f"Erro ao buscar dados do Zabbix: {e}")
return None
def parsear_logs(history_data):
"""Analisa os dados brutos de log e extrai informações detalhadas."""
parsed_data = []
if not history_data: return pd.DataFrame()
print("Analisando logs para extrair insights...")
for record in history_data:
log_content = record['value'].strip()
last_brace = log_content.rfind('}')
if last_brace != -1:
log_content = log_content[:last_brace+1]
match = re.search(r'\{.*\}', log_content)
if not match: continue
try:
log_json = json.loads(match.group(0))
user = log_json.get('remote_user') or (re.search(r'nc_username=([^;]+)', log_json.get('http_cookie', '')).group(1) if 'nc_username=' in log_json.get('http_cookie', '') else None)
if user and re.match(r'^[a-zA-Z0-9._-]+$', user):
parsed_data.append({
'remote_user': re.sub(r'@.*$', '', user.lower()),
'request_time': float(log_json.get('request_time', 0)),
'status': int(log_json.get('status', 0)),
'request_uri': log_json.get('request_uri', '/')
})
except (json.JSONDecodeError, AttributeError, ValueError, TypeError): continue
if not parsed_data: return pd.DataFrame()
return pd.DataFrame(parsed_data)
def gerar_relatorio_html(df, dia_relatorio_ymd):
"""Gera o relatório HTML final com os insights de performance e acesso."""
if df.empty:
return f"
' for _, row in top_404_uris.iterrows()])
top_403_html = "".join([f'
{row["request_uri"]}
{row["count"]}
' for _, row in top_403_uris.iterrows()])
top_outros_4xx_html = "".join([f'
{row["request_uri"]}
{row["count"]}
' for _, row in top_outros_4xx_uris.iterrows()])
return f"""
Relatório de Performance - {dia_relatorio_display}
Relatório de Performance e Acesso
{dia_relatorio_display}
Métricas de Tempo de Resposta
Média Total
{avg_total:.3f}s
Média Cloud (Genérico)
{avg_cloud:.3f}s
Média Office Online
{avg_office:.3f}s
{speed_chart_html}
{status_chart_html}
Análise Detalhada de Erros 4xx (Erros do Cliente)
Não Encontrado (404)
{erros_404}
Acesso Bloqueado (403)
{erros_403}
Outros Erros 4xx
{outros_4xx}
Top 5 URIs com Erro 404
URI
Count
{top_404_html}
Top 5 URIs com Erro 403
URI
Count
{top_403_html}
Top 5 URIs com Outros Erros 4xx
URI
Count
{top_outros_4xx_html}
"""
def main():
parser = argparse.ArgumentParser(description="Gera relatórios de performance a partir de logs do Zabbix.")
parser.add_argument("--server", required=True, help="URL do servidor Zabbix")
parser.add_argument("--token", required=True, help="Token da API do Zabbix.")
parser.add_argument("--host", required=True, help="Nome do host no Zabbix.")
parser.add_argument("--item", required=True, help="Nome do item que contém os logs.")
parser.add_argument("--dias", type=int, default=1, help="Número de dias anteriores para gerar relatórios.")
args = parser.parse_args()
try: locale.setlocale(locale.LC_TIME, 'pt_BR.UTF-8')
except locale.Error: print("Locale 'pt_BR.UTF-8' não encontrado.")
zapi = conectar_zabbix(args.server, args.token)
if not zapi: return
today = date.today()
for i in range(args.dias, 0, -1):
target_date = today - timedelta(days=i)
date_str = target_date.strftime('%Y-%m-%d')
print(f"\n--- Processando relatório para {date_str} ---")
history = buscar_dados_zabbix(zapi, args.host, args.item, target_date)
if not history:
print(f"Nenhum dado encontrado para {date_str}.")
continue
logs_df = parsear_logs(history)
if logs_df.empty:
print(f"Nenhum dado válido para processar em {date_str}.")
continue
report_html = gerar_relatorio_html(logs_df, date_str)
file_name = f"relatorio_insights_{date_str}.html"
try:
with open(file_name, 'w', encoding='utf-8') as f: f.write(report_html)
print(f"Relatório gerado com sucesso: {os.path.abspath(file_name)}")
except Exception as e:
print(f"Erro ao salvar o relatório: {e}")
if __name__ == "__main__":
main()