79 lines
2.3 KiB
Python
79 lines
2.3 KiB
Python
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()
|