Add YAML-driven report generation

scripts/generate_reports.py

@@ -3,77 +3,91 @@ import sqlite3
 from pathlib import Path
 from typing import List, Dict
 
+import yaml
+
 import typer
 from jinja2 import Environment, FileSystemLoader
 
 DB_PATH = Path("database/ngxstat.db")
 OUTPUT_DIR = Path("output")
 TEMPLATE_DIR = Path("templates")
+REPORT_CONFIG = Path("reports.yml")
 
 app = typer.Typer(help="Generate aggregated log reports")
 
-def _load_existing(path: Path) -> List[Dict]:
-    if path.exists():
-        try:
-            return json.loads(path.read_text())
-        except Exception:
-            return []
-    return []
+def _load_config() -> List[Dict]:
+    if not REPORT_CONFIG.exists():
+        typer.echo(f"Config file not found: {REPORT_CONFIG}")
+        raise typer.Exit(1)
+    with REPORT_CONFIG.open("r") as fh:
+        data = yaml.safe_load(fh) or []
+    if not isinstance(data, list):
+        typer.echo("reports.yml must contain a list of report definitions")
+        raise typer.Exit(1)
+    return data
 
 def _save_json(path: Path, data: List[Dict]) -> None:
     path.parent.mkdir(parents=True, exist_ok=True)
     path.write_text(json.dumps(data, indent=2))
 
-def _render_html(interval: str, json_name: str, out_path: Path) -> None:
+def _render_html(interval: str, reports: List[Dict], out_path: Path) -> None:
     env = Environment(loader=FileSystemLoader(TEMPLATE_DIR))
     template = env.get_template("report.html")
-    out_path.write_text(template.render(interval=interval, json_path=json_name))
+    out_path.write_text(template.render(interval=interval, reports=reports))
 
-def _aggregate(interval: str, fmt: str) -> None:
-    json_path = OUTPUT_DIR / f"{interval}.json"
-    html_path = OUTPUT_DIR / f"{interval}.html"
-
-    existing = _load_existing(json_path)
-    last_bucket = existing[-1]["bucket"] if existing else None
+def _generate_interval(interval: str) -> None:
+    cfg = _load_config()
+    defs = [d for d in cfg if d.get("interval") == interval]
+    if not defs:
+        typer.echo(f"No reports defined for {interval}")
+        return
 
     conn = sqlite3.connect(DB_PATH)
     cur = conn.cursor()
 
-    query = f"SELECT strftime('{fmt}', datetime(time)) as bucket, COUNT(*) as hits FROM logs"
-    params = []
-    if last_bucket:
-        query += " WHERE datetime(time) > datetime(?)"
-        params.append(last_bucket)
-    query += " GROUP BY bucket ORDER BY bucket"
+    out_dir = OUTPUT_DIR / interval
+    out_dir.mkdir(parents=True, exist_ok=True)
 
-    rows = cur.execute(query, params).fetchall()
-    for bucket, hits in rows:
-        existing.append({"bucket": bucket, "hits": hits})
+    report_list = []
+    for definition in defs:
+        name = definition["name"]
+        query = definition["query"]
+        cur.execute(query)
+        rows = cur.fetchall()
+        headers = [c[0] for c in cur.description]
+        data = [dict(zip(headers, row)) for row in rows]
+        json_path = out_dir / f"{name}.json"
+        _save_json(json_path, data)
+        report_list.append({
+            "name": name,
+            "label": definition.get("label", name.title()),
+            "chart": definition.get("chart", "line"),
+            "json": f"{name}.json",
+        })
 
-    existing.sort(key=lambda x: x["bucket"])
-    _save_json(json_path, existing)
-    _render_html(interval, json_path.name, html_path)
-    typer.echo(f"Generated {json_path} and {html_path}")
+    _save_json(out_dir / "reports.json", report_list)
+    _render_html(interval, report_list, out_dir / "index.html")
+    typer.echo(f"Generated {interval} reports")
 
 @app.command()
 def hourly() -> None:
-    """Aggregate logs into hourly buckets."""
-    _aggregate("hourly", "%Y-%m-%d %H:00:00")
+    """Generate hourly reports."""
+    _generate_interval("hourly")
 
 @app.command()
 def daily() -> None:
-    """Aggregate logs into daily buckets."""
-    _aggregate("daily", "%Y-%m-%d")
+    """Generate daily reports."""
+    _generate_interval("daily")
 
 @app.command()
 def weekly() -> None:
-    """Aggregate logs into weekly buckets."""
-    _aggregate("weekly", "%Y-%W")
+    """Generate weekly reports."""
+    _generate_interval("weekly")
 
 @app.command()
 def monthly() -> None:
-    """Aggregate logs into monthly buckets."""
-    _aggregate("monthly", "%Y-%m")
+    """Generate monthly reports."""
+    _generate_interval("monthly")
 
 if __name__ == "__main__":
     app()