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()