Update requirements and AGENTS instructions

scripts/generate_reports.py

@@ -0,0 +1,79 @@
+import json
+import sqlite3
+from pathlib import Path
+from typing import List, Dict
+
+import typer
+from jinja2 import Environment, FileSystemLoader
+
+DB_PATH = Path("database/ngxstat.db")
+OUTPUT_DIR = Path("output")
+TEMPLATE_DIR = Path("templates")
+
+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 _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:
+    env = Environment(loader=FileSystemLoader(TEMPLATE_DIR))
+    template = env.get_template("report.html")
+    out_path.write_text(template.render(interval=interval, json_path=json_name))
+
+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
+
+    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"
+
+    rows = cur.execute(query, params).fetchall()
+    for bucket, hits in rows:
+        existing.append({"bucket": bucket, "hits": hits})
+
+    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}")
+
+@app.command()
+def hourly() -> None:
+    """Aggregate logs into hourly buckets."""
+    _aggregate("hourly", "%Y-%m-%d %H:00:00")
+
+@app.command()
+def daily() -> None:
+    """Aggregate logs into daily buckets."""
+    _aggregate("daily", "%Y-%m-%d")
+
+@app.command()
+def weekly() -> None:
+    """Aggregate logs into weekly buckets."""
+    _aggregate("weekly", "%Y-%W")
+
+@app.command()
+def monthly() -> None:
+    """Aggregate logs into monthly buckets."""
+    _aggregate("monthly", "%Y-%m")
+
+if __name__ == "__main__":
+    app()