Refactor reports to share definitions across intervals

This commit is contained in:
Jordan Wages 2025-07-18 01:50:19 -05:00
commit 657c30b260
4 changed files with 38 additions and 18 deletions

View file

@ -13,6 +13,17 @@ OUTPUT_DIR = Path("output")
TEMPLATE_DIR = Path("templates")
REPORT_CONFIG = Path("reports.yml")
# Mapping of interval names to SQLite strftime formats. These strings are
# substituted into report queries whenever the special ``{bucket}`` token is
# present so that a single report definition can be reused for multiple
# intervals.
INTERVAL_FORMATS = {
"hourly": "%Y-%m-%d %H:00:00",
"daily": "%Y-%m-%d",
"weekly": "%Y-%W",
"monthly": "%Y-%m",
}
app = typer.Typer(help="Generate aggregated log reports")
def _load_config() -> List[Dict]:
@ -35,13 +46,23 @@ def _render_html(interval: str, reports: List[Dict], out_path: Path) -> None:
template = env.get_template("report.html")
out_path.write_text(template.render(interval=interval, reports=reports))
def _bucket_expr(interval: str) -> str:
"""Return the SQLite strftime expression for the given interval."""
fmt = INTERVAL_FORMATS.get(interval)
if not fmt:
typer.echo(f"Unsupported interval: {interval}")
raise typer.Exit(1)
return f"strftime('{fmt}', datetime(time))"
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}")
if not cfg:
typer.echo("No report definitions found")
return
bucket = _bucket_expr(interval)
conn = sqlite3.connect(DB_PATH)
cur = conn.cursor()
@ -49,9 +70,9 @@ def _generate_interval(interval: str) -> None:
out_dir.mkdir(parents=True, exist_ok=True)
report_list = []
for definition in defs:
for definition in cfg:
name = definition["name"]
query = definition["query"]
query = definition["query"].replace("{bucket}", bucket)
cur.execute(query)
rows = cur.fetchall()
headers = [c[0] for c in cur.description]