148 lines
4.6 KiB
Python
Executable File
148 lines
4.6 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""odysseus-logs — unified view of log files across the app.
|
|
|
|
Right now logs live in two places:
|
|
- logs/ app-level (compound.log, future runtime)
|
|
- /tmp/odysseus-tmux/*.log per-tmux-session model download/serve logs
|
|
|
|
This CLI surfaces them so you don't have to remember which directory
|
|
holds what.
|
|
|
|
odysseus-logs list # every log we know about
|
|
odysseus-logs tail NAME # tail -f a specific log
|
|
odysseus-logs tail NAME --lines 200 # last N lines only (no follow)
|
|
odysseus-logs cat NAME # print full content
|
|
odysseus-logs clean # delete tmux logs older than 7 days
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "_lib"))
|
|
from cli import quiet_logs, emit, fail, common_parser, run, REPO_ROOT as _REPO_ROOT
|
|
|
|
quiet_logs()
|
|
|
|
import argparse, json, subprocess, time
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
|
|
_APP_LOGS = _REPO_ROOT / "logs"
|
|
_TMUX_LOGS = Path("/tmp/odysseus-tmux")
|
|
|
|
|
|
def _enumerate() -> list[dict]:
|
|
"""Return every known log file as {name, path, bytes, modified}."""
|
|
out = []
|
|
for base in (_APP_LOGS, _TMUX_LOGS):
|
|
if not base.is_dir():
|
|
continue
|
|
for p in sorted(base.glob("*.log")):
|
|
try:
|
|
st = p.stat()
|
|
except OSError:
|
|
continue
|
|
out.append({
|
|
"name": p.name,
|
|
"path": str(p),
|
|
"bytes": st.st_size,
|
|
"modified": datetime.fromtimestamp(st.st_mtime).isoformat(),
|
|
})
|
|
out.sort(key=lambda r: r["modified"], reverse=True)
|
|
return out
|
|
|
|
|
|
def _resolve(name: str) -> Path | None:
|
|
"""Match a log by exact filename, basename-without-extension, or
|
|
substring. Returns the most-recently-modified match if there are
|
|
ties."""
|
|
if not isinstance(name, str):
|
|
return None
|
|
candidates = []
|
|
for base in (_APP_LOGS, _TMUX_LOGS):
|
|
if not base.is_dir():
|
|
continue
|
|
for p in base.glob("*.log"):
|
|
if p.name == name or p.stem == name or name in p.name:
|
|
candidates.append(p)
|
|
if not candidates:
|
|
return None
|
|
candidates.sort(key=lambda p: p.stat().st_mtime, reverse=True)
|
|
return candidates[0]
|
|
|
|
|
|
def cmd_list(args):
|
|
emit(_enumerate(), args)
|
|
|
|
|
|
def cmd_tail(args):
|
|
p = _resolve(args.name)
|
|
if p is None:
|
|
fail(f"no log matching {args.name!r}; run `odysseus-logs list`")
|
|
if args.follow:
|
|
# Exec tail -f so signals + buffering are native.
|
|
os.execvp("tail", ["tail", "-n", str(args.lines), "-f", str(p)])
|
|
# Non-follow: print last N lines + exit.
|
|
proc = subprocess.run(["tail", "-n", str(args.lines), str(p)],
|
|
capture_output=True, text=True)
|
|
sys.stdout.write(proc.stdout)
|
|
sys.exit(proc.returncode)
|
|
|
|
|
|
def cmd_cat(args):
|
|
p = _resolve(args.name)
|
|
if p is None:
|
|
fail(f"no log matching {args.name!r}")
|
|
sys.stdout.write(p.read_text(errors="replace"))
|
|
|
|
|
|
def cmd_clean(args):
|
|
"""Delete tmux logs older than --days. Doesn't touch app logs."""
|
|
if not _TMUX_LOGS.is_dir():
|
|
emit({"deleted": 0, "kept": 0}, args)
|
|
return
|
|
cutoff = time.time() - args.days * 86400
|
|
deleted, kept = [], 0
|
|
for p in _TMUX_LOGS.glob("*.log"):
|
|
try:
|
|
if p.stat().st_mtime < cutoff:
|
|
p.unlink()
|
|
deleted.append(p.name)
|
|
else:
|
|
kept += 1
|
|
except OSError:
|
|
continue
|
|
emit({"deleted": deleted, "kept": kept, "cutoff_days": args.days}, args)
|
|
|
|
|
|
def _build_parser():
|
|
p = common_parser("odysseus-logs", "Tail / list logs across the app.")
|
|
sub = p.add_subparsers(dest="cmd", required=True)
|
|
|
|
pl = sub.add_parser("list", parents=p._common_parents)
|
|
pl.set_defaults(func=cmd_list)
|
|
|
|
pt = sub.add_parser("tail", parents=p._common_parents)
|
|
pt.add_argument("name")
|
|
pt.add_argument("-n", "--lines", type=int, default=80)
|
|
pt.add_argument("-f", "--follow", action="store_true",
|
|
help="follow new lines (tail -f); omit for one-shot")
|
|
pt.set_defaults(func=cmd_tail)
|
|
|
|
pc = sub.add_parser("cat", parents=p._common_parents)
|
|
pc.add_argument("name")
|
|
pc.set_defaults(func=cmd_cat)
|
|
|
|
pcl = sub.add_parser("clean", parents=p._common_parents)
|
|
pcl.add_argument("--days", type=int, default=7,
|
|
help="delete tmux logs older than N days (default 7)")
|
|
pcl.set_defaults(func=cmd_clean)
|
|
|
|
return p
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(run(_build_parser()))
|