import os
import io
import requests
import logging
from pathlib import Path
from flask import Flask, request, jsonify, send_file, after_this_request
from pydantic import BaseModel, ValidationError, Field
from typing import Optional, List
from datetime import datetime
import mysql.connector
from mysql.connector import Error
from mysql.connector import pooling
import copy
import uuid
import concurrent.futures
import threading
import time
from flask import render_template_string
import json
from playwright.sync_api import sync_playwright
app = Flask(__name__)
# auxpara compartir multiples workers de Gunicorn
def set_job_status(job_id, status_dict):
try:
with open(f"/tmp/{job_id}.json", "w") as f:
json.dump(status_dict, f)
except Exception as e:
logger.error(f"Error guardando status JSON: {e}")
def get_job_status(job_id):
try:
path = f"/tmp/{job_id}.json"
if os.path.exists(path):
with open(path, "r") as f:
return json.load(f)
except:
pass
return {'status': 'not_found'}
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# Pydantic para validar los campos
class TransactionSchema(BaseModel):
cliente: str
sco_number: str
sucursal: str
pos_transaction_id: int
started_at: datetime
end_transaction_at: Optional[datetime] = None
closed_at: datetime
status: str
item_count: int = 0
total_amount: float = 0.0
paid_amount: float = 0.0
change_amount: float = 0.0
amounts_reconstructed: bool = False
payment_type: Optional[str] = None
split_payment: bool = False
age_verification_requested: bool = False
age_verification_value: Optional[int] = Field(default=None, gt=0)
bag_item_count: int = 0
# config mysql con .env general
DB_HOST = os.environ.get("DB_HOST")
DB_PORT = int(os.environ.get("DB_PORT", "3306"))
DB_USER = os.environ.get("DB_USER")
DB_PASSWORD = os.environ.get("DB_PASSWORD")
DB_NAME = os.environ.get("DB_NAME")
if not all([DB_HOST, DB_USER, DB_PASSWORD, DB_NAME]):
raise ValueError("Faltan credenciales de MySQL.")
# manejo de db pool dentro de la instancia
db_pool = None
dbconfig = {
"host": DB_HOST,
"port": DB_PORT,
"user": DB_USER,
"password": DB_PASSWORD,
"database": DB_NAME
}
max_retries = 5
for attempt in range(max_retries):
try:
db_pool = pooling.MySQLConnectionPool(
pool_name="sco_pool",
pool_size=10,
pool_reset_session=True,
**dbconfig
)
logger.info("Connection pool de MySQL creado con éxito.")
break
except Error as e:
logger.error(f"Error al crear el pool de conexiones (Intento {attempt + 1}/{max_retries}): {e}")
if attempt < max_retries - 1:
time.sleep(3)
else:
logger.error("No se pudo conectar a MySQL después de varios intentos.")
def get_db_pool():
return db_pool
# Pantalla de carga HTML (PRUEBA grafica)
LOADING_HTML = """
Generando Reporte...
Construyendo su reporte PDF...
Estamos recolectando y procesando los datos de Grafana. Esto tomará entre 1 y 4 minutos dependiendo del tamaño. Por favor, no cierre esta ventana.
"""
def background_export_task(job_id, dashboard_uid, dashboard_name, cliente, grafana_vars, time_from, time_to, tz, grafana_session):
logger.info(f"[{job_id}] Iniciando hilo de trabajo en segundo plano (Modo Playwright Headless)...")
set_job_status(job_id, {'status': 'processing'})
try:
# URL hacia la UI real de Grafana
target_url = f"http://grafana:3000/d/{dashboard_uid}?orgId=1"
# Inyectamos dinámicamente TODAS las variables que empiezan con var-
for var_key, var_values in grafana_vars.items():
for val in var_values:
target_url += f"&{var_key}={val}"
# Forzamos timezone
target_url += "&kiosk=1&hideNav=1&timezone=browser"
if time_from: target_url += f"&from={time_from}"
if time_to: target_url += f"&to={time_to}"
with sync_playwright() as p:
# Apagamos GPU y WebGL para evitar limites de hardware
browser = p.chromium.launch(headless=True, args=['--disable-gpu', '--disable-dev-shm-usage', '--disable-webgl'])
context = browser.new_context(
viewport={'width': 1920, 'height': 1080},
ignore_https_errors=True,
timezone_id=tz,
)
if grafana_session:
context.add_cookies([{
'name': 'grafana_session',
'value': grafana_session,
'domain': 'grafana', # host interno del contenedor
'path': '/'
}])
logger.info(f"[{job_id}] Clonación de Cookie exitosa.")
else:
logger.warning(f"[{job_id}] ERROR: No se recibió cookie de sesión desde. Petición rechazada por Grafana.")
page = context.new_page()
# Bloquear Websockets
page.route("**/*api/live/ws*", lambda route: route.abort())
page.on("pageerror", lambda err: logger.error(f"[{job_id}] Playwright Page Error: {err}"))
# escuchamos dinamicamente las querys
active_queries = [0]
def handle_request(request):
if "api/ds/query" in request.url:
active_queries[0] += 1
def handle_request_finished(request):
if "api/ds/query" in request.url:
active_queries[0] = max(0, active_queries[0] - 1)
def handle_request_failed(request):
if "api/ds/query" in request.url:
active_queries[0] = max(0, active_queries[0] - 1)
page.on("request", handle_request)
page.on("requestfinished", handle_request_finished)
page.on("requestfailed", handle_request_failed)
page.goto(target_url, timeout=0, wait_until='load')
logger.info(f"[{job_id}] Validando permisos de acceso al dashboard...")
# Dar tiempo a que React termine sus peticiones
try:
page.wait_for_load_state("networkidle", timeout=10000)
except:
pass
page.wait_for_timeout(2000)
if "/login" in page.url or page.get_by_placeholder("email or username").is_visible():
error_msg = "Acceso Denegado - No tienes permisos para exportar este Dashboard o tu sesion caducó."
logger.error(f"[{job_id}] {error_msg}")
raise Exception(error_msg)
# check "Forbidden" o "Access denied" en el contenido de la página
page_text = page.locator("body").inner_text().lower()
if "failed to load dashboard" in page_text or "forbidden" in page_text or "access denied" in page_text or "not found" in page_text:
error_msg = "Error 404 - El dashboard solicitado no existe o fue eliminado."
logger.error(f"[{job_id}] Intento de acceso a dashboard sin permisos.")
raise Exception(error_msg)
logger.info(f"[{job_id}] Acceso concedido. Esperando 10s para la primera fila de paneles...")
page.wait_for_timeout(10000)
try:
# # Localizamos ROWS colapsadas y las abrimos
page.evaluate("""
const buttons = document.querySelectorAll('[aria-expanded="false"]');
buttons.forEach(btn => {
const testId = btn.getAttribute('data-testid') || '';
const className = btn.className || '';
const parentClass = (btn.parentElement && btn.parentElement.className) || '';
const ariaLabel = btn.getAttribute('aria-label') || '';
// Es una fila si cumple alguna de estas heurísticas de Grafana:
const isRow = testId.includes('dashboard-row') ||
className.includes('row') ||
parentClass.includes('row') ||
ariaLabel.toLowerCase().includes('row') ||
ariaLabel.toLowerCase().includes('fila') ||
btn.offsetWidth > 500; // Las barras de fila ocupan mucho ancho
if (isRow) {
btn.setAttribute('data-pdf-row', 'true');
}
});
// Soporte legacy para Grafana 8
document.querySelectorAll('.dashboard-row--collapsed').forEach(div => {
div.setAttribute('data-pdf-row', 'true');
});
""")
# Recolectamos todas las filas
row_elements = page.locator('[data-pdf-row="true"]').all()
rows_to_click = []
for el in row_elements:
if el.is_visible():
box = el.bounding_box()
if box:
rows_to_click.append({
'el': el,
'y': box['y'],
'text': el.inner_text().strip()[:40]
})
# ORDENAMOS de mayor a menor (important)
rows_to_click.sort(key=lambda item: item['y'], reverse=True)
logger.info(f"[{job_id}] Se encontraron {len(rows_to_click)} filas dinamicas para expandir.")
for item in rows_to_click:
logger.info(f"[{job_id}] Expandiendo fila interactiva: {item['text']}")
item['el'].scroll_into_view_if_needed()
page.wait_for_timeout(500)
item['el'].click(force=True)
page.wait_for_timeout(1000)
except Exception as e:
logger.warning(f"[{job_id}] Error secundario al hacer click en filas: {e}")
# 5 segundos para que Grafana renderize
page.wait_for_timeout(5000)
total_height = page.evaluate("document.body.scrollHeight")
# salto de 1080px para que las querys carguen tranquilas
viewport_height = 1080
current_y = total_height
while current_y > 0:
page.evaluate(f"window.scrollTo(0, {current_y})")
logger.info(f"[{job_id}] Scrolleamos a {current_y}px...")
# 2s iniciales para que grafana/react reaccione
page.wait_for_timeout(2000)
# Espera reactiva ---> vigila hasta que las queries activas terminen.
wait_time = 0
while active_queries[0] > 0 and wait_time < 15000:
page.wait_for_timeout(1000)
wait_time += 1000
if wait_time % 5000 == 0:
logger.info(f"[{job_id}] Waiting DB... Queries actives: {active_queries[0]}")
current_y -= viewport_height
page.evaluate("window.scrollTo(0, 0)")
logger.info(f"[{job_id}] Tope alcanzado. Esperando ultima resolución.")
page.wait_for_timeout(2000)
wait_time = 0
# tiempo de seguridad por precaucion
while active_queries[0] > 0 and wait_time < 45000:
page.wait_for_timeout(1000)
wait_time += 1000
if wait_time % 5000 == 0:
logger.info(f"[{job_id}] Tope - Still waiting DB... Queries actives: {active_queries[0]}")
logger.info(f"[{job_id}] Cambiando a Viewport original de la pagina de: {total_height}px.")
final_height = min(total_height, 25000)
page.set_viewport_size({'width': 1920, 'height': final_height})
page.evaluate("window.scrollTo(0, 0)")
logger.info(f"[{job_id}] 15000ms para que grafana procese...")
page.wait_for_timeout(15000)
# checkeamos si NO ESTAMOS EN LOGIN == tenemos cookies validas?
if "/login" in page.url or page.get_by_placeholder("email or username").is_visible():
error_msg = "Acceso Denegado - Sesión caducada o rechazada en el último momento."
logger.error(f"[{job_id}] {error_msg}")
raise Exception(error_msg)
# sacamos titulo para que no diga: " - Dashboards - Grafana"
page.evaluate("document.title = document.title.replace(' - Dashboards - Grafana', '');")
# Eliminar todos los (href) para que no sean clickeables en el PDF generado
logger.info(f"[{job_id}] Eliminando enlaces interactivos del HTML...")
page.evaluate("""
document.querySelectorAll('a[href]').forEach(el => {
el.removeAttribute('href');
});
""")
logger.info(f"[{job_id}] Generando PDF...")
pdf_filepath = f"/tmp/{job_id}.pdf"
# Forzamos a Chrome a creer que sigue en una pantalla usando CSS @media print
# Esto evita que Grafana oculte el reloj + botones
page.emulate_media(media="screen")
page.pdf(
path=pdf_filepath,
print_background=True,
width="1920px",
height=f"{final_height}px",
margin={"top": "0", "right": "0", "bottom": "0", "left": "0"},
page_ranges="1"
)
browser.close()
logger.info(f"[{job_id}] Work Done! Guardado en: {pdf_filepath}")
import re
# convertimos a lowercase el 'dashboard_name' para que el archivo salga mejor
safe_dashboard_name = re.sub(r'[^a-z0-9]+', '-', dashboard_name.lower()).strip('-')
filename_final = f"{safe_dashboard_name}.pdf"
set_job_status(job_id, {'status': 'done', 'filepath': pdf_filepath, 'filename': filename_final})
except Exception as e:
logger.error(f"[{job_id}] FATAL ERROR en Playwright: {str(e)}")
set_job_status(job_id, {'status': 'error', 'error': str(e)})
@app.route('/api/v1/export/image', methods=['GET'])
def export_image_trigger():
# Capturar TODAS las variables (empiezan con var-)
grafana_vars = {}
for key in request.args.keys():
if key.startswith('var-'):
grafana_vars[key] = request.args.getlist(key)
cliente = request.args.get('cliente')
if not cliente and 'var-cliente' in grafana_vars:
cliente = grafana_vars['var-cliente'][0]
dashboard_uid = request.args.get('dashboard_uid')
dashboard_name = request.args.get('dashboard_name', 'Masivo')
time_from = request.args.get('from')
time_to = request.args.get('to')
tz = request.args.get('tz', 'America/Argentina/Buenos_Aires')
if "${" in tz or not tz:
tz = 'America/Argentina/Buenos_Aires'
grafana_session = request.cookies.get('grafana_session')
if not cliente or not dashboard_uid:
return jsonify({'error': 'Faltan parametros obligatorios'}), 400
# Crear el ticket asincrono
job_id = uuid.uuid4().hex
set_job_status(job_id, {'status': 'starting'})
# Lanzar el thread
thread = threading.Thread(
target=background_export_task,
args=(job_id, dashboard_uid, dashboard_name, cliente, grafana_vars, time_from, time_to, tz, grafana_session)
)
thread.start()
# Responder INMEDIATAMENTE a el usuario asi no falla cloudflare
html_content = LOADING_HTML.replace("{{job_id}}", job_id)
return render_template_string(html_content)
@app.route('/api/v1/export/status/', methods=['GET'])
def export_status(job_id):
status_info = get_job_status(job_id)
return jsonify(status_info)
@app.route('/api/v1/export/download/', methods=['GET'])
def export_download(job_id):
status_info = get_job_status(job_id)
if status_info.get('status') != 'done':
return "Archivo no encontrado o no está listo", 404
filepath = status_info.get('filepath')
download_name = status_info.get('filename', 'reporte.pdf')
try:
# delete del archivo after download
@after_this_request
def remove_file(response):
try:
os.remove(filepath)
os.remove(f"/tmp/{job_id}.json")
logger.info(f"[{job_id}] Limpieza temporal exitosa tras descarga.")
except Exception as e:
logger.error(f"[{job_id}] Error borrando temp files: {e}")
return response
return send_file(
filepath,
mimetype='application/pdf',
as_attachment=True,
download_name=download_name
)
except Exception as e:
return str(e), 500
@app.route('/api/v1/transactions', methods=['POST'])
def receive_transaction():
data = request.get_json()
if not data or not isinstance(data, list):
return jsonify({"error": "Se esperaba un array JSON de transacciones"}), 400
# Pydantic validation
valid_transactions = []
failed_messages = []
for i, item in enumerate(data):
try:
valid_transactions.append(TransactionSchema(**item))
except ValidationError as e:
logger.warning(f"JSON rechazado en índice {i}: {e.errors()}")
tx_id = item.get("pos_transaction_id", f"índice {i}")
failed_messages.append(f"La transaccion con id {tx_id} no cumple el formato esperado: {str(e.errors())}")
if not valid_transactions:
return jsonify({
"status": "error",
"data": "Todas las transacciones fallaron la validación de formato.\n" + "\n".join(failed_messages)
}), 400
pool = get_db_pool()
if not pool:
return jsonify({"error": "Error interno del servidor"}), 500
# Extraemos una conexion disponible del pool
try:
connection = pool.get_connection()
except Error as e:
logger.error(f"Error obteniendo conexión del pool: {e}")
return jsonify({"error": "Error de base de datos"}), 500
cursor = connection.cursor()
insert_query = """
INSERT INTO transactions (
cliente, sco_number, sucursal, pos_transaction_id, started_at, end_transaction_at,
closed_at, status, item_count, total_amount, paid_amount, change_amount,
amounts_reconstructed, payment_type, split_payment, age_verification_requested,
age_verification_value, bag_item_count
) VALUES (
%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s
) ON DUPLICATE KEY UPDATE
started_at = VALUES(started_at),
end_transaction_at = VALUES(end_transaction_at),
closed_at = VALUES(closed_at),
status = VALUES(status),
item_count = VALUES(item_count),
total_amount = VALUES(total_amount),
paid_amount = VALUES(paid_amount),
change_amount = VALUES(change_amount),
amounts_reconstructed = VALUES(amounts_reconstructed),
payment_type = VALUES(payment_type),
split_payment = VALUES(split_payment),
age_verification_requested = VALUES(age_verification_requested),
age_verification_value = VALUES(age_verification_value),
bag_item_count = VALUES(bag_item_count)
"""
values_list = []
for transaction in valid_transactions:
# Forzamos lower case en Python también como doble capa de seguridad
values_list.append((
transaction.cliente.lower(),
transaction.sco_number,
transaction.sucursal,
transaction.pos_transaction_id,
transaction.started_at,
transaction.end_transaction_at,
transaction.closed_at,
transaction.status,
transaction.item_count,
transaction.total_amount,
transaction.paid_amount,
transaction.change_amount,
transaction.amounts_reconstructed,
transaction.payment_type,
transaction.split_payment,
transaction.age_verification_requested,
transaction.age_verification_value,
transaction.bag_item_count
))
try:
cursor.executemany(insert_query, values_list)
connection.commit()
logger.info(f"Bulk Upsert exitoso: {len(valid_transactions)} guardadas. {len(failed_messages)} fallidas.")
response_data = "Batch procesado correctamente."
if failed_messages:
response_data = "\n".join(failed_messages)
return jsonify({
"status": "ok",
"data": response_data
}), 200
except Error as e:
logger.error(f"Error ejecutando query MySQL: {e}")
connection.rollback()
return jsonify({"error": "Error al guardar en la base de datos"}), 500
finally:
if cursor:
cursor.close()
if connection:
# no cerramos la conexión - la devolvemos al pool
connection.close()
ALERTS_LOG_FILE = os.environ.get("ALERTS_LOG_FILE", "/data/alerts/alerts.txt")
@app.route('/api/v1/alerts/webhook', methods=['POST'])
def receive_grafana_alert():
try:
data = request.get_json(force=True)
if not data:
return jsonify({"status": "error", "message": "No JSON payload received"}), 400
now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
status = data.get("status", "unknown").upper()
title = data.get("title", data.get("groupLabels", {}).get("alertname", "Sin Titulo"))
alerts = data.get("alerts", [])
# Asegurar que el directorio exista
log_path = Path(ALERTS_LOG_FILE)
log_path.parent.mkdir(parents=True, exist_ok=True)
with open(str(log_path), "a", encoding="utf-8") as f:
f.write(f"\n{'='*75}\n")
f.write(f"[{now_str}] ALERTA GRAFANA - ESTADO: {status}\n")
f.write(f"Título: {title}\n")
if alerts:
for idx, alert in enumerate(alerts, 1):
labels = alert.get("labels", {})
annotations = alert.get("annotations", {})
starts_at = alert.get("startsAt", "")
ends_at = alert.get("endsAt", "")
alert_status = alert.get("status", status)
f.write(f"\n [Alerta #{idx}] Status: {alert_status}\n")
f.write(f" - Nombre : {labels.get('alertname', 'N/A')}\n")
f.write(f" - Cliente : {labels.get('cliente', 'N/A')}\n")
f.write(f" - Sucursal : {labels.get('sucursal', 'N/A')}\n")
f.write(f" - SCO : {labels.get('sco_number', labels.get('instance', 'N/A'))}\n")
f.write(f" - Severidad : {labels.get('severity', 'N/A')}\n")
f.write(f" - Resumen : {annotations.get('summary', annotations.get('description', 'N/A'))}\n")
if starts_at:
f.write(f" - Inicio : {starts_at}\n")
if ends_at and ends_at != "0001-01-01T00:00:00Z":
f.write(f" - Fin : {ends_at}\n")
else:
f.write(f" Detalle: {data.get('message', 'Sin descripción adicional')}\n")
f.write(f"\nRAW JSON: {json.dumps(data, ensure_ascii=False)}\n")
f.flush()
logger.info(f"Alerta de Grafana guardada en {ALERTS_LOG_FILE} [{status}] {title}")
return jsonify({"status": "ok", "message": "Alerta registrada correctamente"}), 200
except Exception as e:
logger.error(f"Error procesando webhook de alerta: {e}")
return jsonify({"status": "error", "message": str(e)}), 500
if __name__ == '__main__':
app.run(host='0.0.0.0', port=5000, debug=True)