From 980233840edcaadfb8826238cfcff5d1d2dcc1e9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Domagoj=20Andri=C4=87?= Date: Sat, 27 Jun 2026 14:00:59 +0200 Subject: [PATCH] Additional bugfixes --- .../analytics/internal/scheduler/scheduler.go | 33 ------------------- backend/app/core/executor.py | 2 +- backend/app/core/export.py | 14 ++++++++ backend/app/routers/wwi.py | 20 ++++++++++- 4 files changed, 34 insertions(+), 35 deletions(-) diff --git a/backend/analytics/internal/scheduler/scheduler.go b/backend/analytics/internal/scheduler/scheduler.go index d0fc9e6..2fda8ee 100644 --- a/backend/analytics/internal/scheduler/scheduler.go +++ b/backend/analytics/internal/scheduler/scheduler.go @@ -68,7 +68,6 @@ func (s *Scheduler) Start() { s.cron.AddFunc("0 30 3 * * *", s.jobAWAnomalyDetection) s.cron.AddFunc("0 0 * * * *", s.jobWWIReorder) s.cron.AddFunc("0 30 3 * * *", s.jobWWISupplierScores) - s.cron.AddFunc("0 30 * * * *", s.jobWWIEvents) s.cron.AddFunc("0 0 4 * * *", s.jobWWIDataQuality) s.cron.Start() slog.Info("scheduler started", "jobs", len(s.cron.Entries())) @@ -101,7 +100,6 @@ func (s *Scheduler) TriggerWWIJob(jobName string) error { fns := map[string]func(){ "reorder": s.jobWWIReorder, "supplier_scores": s.jobWWISupplierScores, - "events": s.jobWWIEvents, "data_quality": s.jobWWIDataQuality, } fn, ok := fns[jobName] @@ -220,12 +218,6 @@ func (s *Scheduler) jobAWDataQuality() { if err != nil { return 0, err } - persistence.AppendAudit(ctx, s.pgPool, persistence.AuditEntry{ - Action: "job.completed", ActorType: "scheduler", ActorID: "aw.daily.data_quality", - Domain: "aw", Service: "otel-bi-analytics", EntityType: "data_quality", - Status: report.Status, - Payload: map[string]any{"status": report.Status, "failed_checks": report.FailedChecks}, - }) return len(report.Checks), nil }) } @@ -270,37 +262,12 @@ func (s *Scheduler) jobWWISupplierScores() { }) } -func (s *Scheduler) jobWWIEvents() { - s.runJob("wwi.hourly.events", "wwi", func(ctx context.Context) (int, error) { - data, err := analytics.WWIGetReorderRecommendations(ctx, s.wwiDB) - if err != nil { - return 0, err - } - var highUrgency []analytics.ReorderRecommendation - for _, item := range data { - if item.Urgency == "HIGH" { - highUrgency = append(highUrgency, item) - } - } - if err := persistence.GenerateStockEvents(ctx, s.pgPool, highUrgency); err != nil { - slog.Warn("generate_stock_events (events job) failed", "err", err) - } - return len(highUrgency), nil - }) -} - func (s *Scheduler) jobWWIDataQuality() { s.runJob("wwi.daily.data_quality", "wwi", func(ctx context.Context) (int, error) { report, err := analytics.WWIRunDataQualityCheck(ctx, s.wwiDB) if err != nil { return 0, err } - persistence.AppendAudit(ctx, s.pgPool, persistence.AuditEntry{ - Action: "job.completed", ActorType: "scheduler", ActorID: "wwi.daily.data_quality", - Domain: "wwi", Service: "otel-bi-analytics", EntityType: "data_quality", - Status: report.Status, - Payload: map[string]any{"status": report.Status, "failed_checks": report.FailedChecks}, - }) return len(report.Checks), nil }) } diff --git a/backend/app/core/executor.py b/backend/app/core/executor.py index 4540298..6727579 100644 --- a/backend/app/core/executor.py +++ b/backend/app/core/executor.py @@ -23,5 +23,5 @@ def get_executor() -> ThreadPoolExecutor: def shutdown_executor() -> None: global _executor if _executor is not None: - _executor.shutdown(wait=False) + _executor.shutdown(wait=True) _executor = None diff --git a/backend/app/core/export.py b/backend/app/core/export.py index af46aba..1ce6ee9 100644 --- a/backend/app/core/export.py +++ b/backend/app/core/export.py @@ -2,6 +2,7 @@ from __future__ import annotations import io +import openpyxl from reportlab.lib import colors from reportlab.lib.pagesizes import A4, landscape from reportlab.lib.styles import getSampleStyleSheet @@ -57,6 +58,19 @@ def _pdf_table(rows: list[dict]) -> Table: return t +def to_xlsx_bytes(rows: list[dict]) -> bytes: + """Serialise *rows* to a single-sheet XLSX workbook and return the raw bytes.""" + wb = openpyxl.Workbook() + ws = wb.active + if rows: + ws.append(list(rows[0].keys())) + for row in rows: + ws.append([str(v) if v is not None else "" for v in row.values()]) + buf = io.BytesIO() + wb.save(buf) + return buf.getvalue() + + def to_pdf_bytes(rows: list[dict], title: str, subtitle: str = "") -> bytes: """Serialise *rows* to a single-sheet PDF and return the raw bytes.""" buf = io.BytesIO() diff --git a/backend/app/routers/wwi.py b/backend/app/routers/wwi.py index 95883ac..4c1ebbe 100644 --- a/backend/app/routers/wwi.py +++ b/backend/app/routers/wwi.py @@ -14,7 +14,7 @@ from sqlalchemy.orm import sessionmaker, Session from app.core.audit import ExportRecord, append_audit, current_span_context from app.core.config import settings from app.core.executor import get_executor -from app.core.export import to_pdf_bytes +from app.core.export import to_pdf_bytes, to_xlsx_bytes from app.core.security import FrontendPrincipal, require_frontend_principal from app.domain.wwi import analytics @@ -379,6 +379,24 @@ async def export_wwi_business_events( data = await asyncio.get_running_loop().run_in_executor( get_executor(), lambda: analytics.get_business_events(pg_factory, limit=limit) ) + + if format == "xlsx": + today = datetime.now(timezone.utc).strftime("%Y%m%d") + filename = f"wwi_business_events_{today}.xlsx" + trace_id, span_id = current_span_context() + + def _build_xlsx(): + content = to_xlsx_bytes(data) + _record_export(pg_factory, "wwi", "business-events", "xlsx", filters, + len(data), len(content), actor_id, trace_id, span_id) + return content + + content = await asyncio.get_running_loop().run_in_executor(get_executor(), _build_xlsx) + return Response( + content=content, media_type=_XLSX_MEDIA, + headers={"Content-Disposition": f'attachment; filename="{filename}"'}, + ) + return await asyncio.get_running_loop().run_in_executor( get_executor(), lambda: _make_pdf(data, "wwi_business_events",