Eidan T.
152 lines
3.2 KiB
Python
152 lines
3.2 KiB
Python
from App import *
|
|
from requests import *
|
|
|
|
|
|
pg_config = PGConfig()
|
|
database = DatabaseManager(config.database, config.databaseID)
|
|
run_etl(config, database, data_report)
|
|
|
|
|
|
|
|
pg_config = PGConfig()
|
|
|
|
|
|
|
|
projects = {'PG':pg_config,
|
|
#'Feduro':feduroConfig,
|
|
#'Cimberton':cimbertonConfig}
|
|
}
|
|
|
|
|
|
print(projects)
|
|
for project, config in projects.items():
|
|
print(f"Iniciando servicio ETL para el proyecto {project} y {config}")
|
|
report_list = config.reportsID
|
|
run_etl(project, config, report_list=report_list)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
import time
|
|
import importlib
|
|
import pyodbc
|
|
from datetime import datetime
|
|
from apscheduler.schedulers.background import BackgroundScheduler
|
|
|
|
# 🔌 conexión a SQL Server
|
|
CONN_STR = """
|
|
DRIVER={ODBC Driver 18 for SQL Server};
|
|
SERVER=sql.glmanalytics.com;
|
|
DATABASE=glm_pg;
|
|
UID=vmglm001;
|
|
PWD=tu_password;
|
|
Encrypt=yes;
|
|
TrustServerCertificate=yes;
|
|
"""
|
|
|
|
# 📥 cargar configuración desde BD
|
|
def get_clients_config():
|
|
conn = pyodbc.connect(CONN_STR)
|
|
cursor = conn.cursor()
|
|
|
|
cursor.execute("""
|
|
SELECT DISTINCT client_name, frequency_minutes
|
|
FROM etl_report_config
|
|
WHERE is_active = 1
|
|
""")
|
|
|
|
rows = cursor.fetchall()
|
|
conn.close()
|
|
|
|
configs = []
|
|
for row in rows:
|
|
configs.append({
|
|
"client": row[0],
|
|
"frequency": row[1]
|
|
})
|
|
|
|
return configs
|
|
|
|
|
|
# 📦 cargar config desde carpeta /config
|
|
def load_client_config(client_name):
|
|
try:
|
|
module = importlib.import_module(f"config.{client_name}")
|
|
return module.CONFIG
|
|
except Exception as e:
|
|
raise Exception(f"No se pudo cargar config de {client_name}: {e}")
|
|
|
|
|
|
# ⚙️ lógica ETL por cliente
|
|
def run_etl(client_name):
|
|
try:
|
|
print(f"\n[{datetime.now()}] 🚀 Iniciando ETL para {client_name}")
|
|
|
|
config = load_client_config(client_name)
|
|
|
|
# 🔥 aquí usas tu config
|
|
connection = config.get("connection")
|
|
tables = config.get("tables", [])
|
|
|
|
print(f"Tablas a procesar: {tables}")
|
|
|
|
# 👉 ejemplo de flujo
|
|
for table in tables:
|
|
print(f"Procesando {client_name} - {table}")
|
|
# extract()
|
|
# transform()
|
|
# load()
|
|
|
|
print(f"[{datetime.now()}] ✅ ETL finalizado {client_name}")
|
|
|
|
except Exception as e:
|
|
print(f"[ERROR] ❌ {client_name}: {e}")
|
|
|
|
|
|
# 🧠 job que ejecuta el scheduler
|
|
def job(client_name):
|
|
run_etl(client_name)
|
|
|
|
|
|
# 🚀 iniciar scheduler
|
|
def start_scheduler():
|
|
scheduler = BackgroundScheduler()
|
|
|
|
configs = get_clients_config()
|
|
|
|
for config in configs:
|
|
client = config["client"]
|
|
frequency = config["frequency"]
|
|
|
|
scheduler.add_job(
|
|
job,
|
|
'interval',
|
|
minutes=frequency,
|
|
args=[client],
|
|
id=f"job_{client}",
|
|
replace_existing=True
|
|
)
|
|
|
|
print(f"✅ Job registrado: {client} cada {frequency} minutos")
|
|
|
|
scheduler.start()
|
|
print("\n🚀 Scheduler iniciado correctamente\n")
|
|
|
|
# mantener vivo (IMPORTANTE en Docker)
|
|
try:
|
|
while True:
|
|
time.sleep(60)
|
|
except (KeyboardInterrupt, SystemExit):
|
|
scheduler.shutdown()
|
|
|
|
|
|
# ▶️ entrypoint
|
|
if __name__ == "__main__":
|
|
start_scheduler() |