Additional bugfixes
This commit is contained in:
@@ -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
|
||||
})
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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",
|
||||
|
||||
Reference in New Issue
Block a user