chris/AV-Watch
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
1fee46ef36bcfa81d85f39e2ca9073d94194a887
AV-Watch/avwatch/cli.py
T

866 lines
32 KiB
Python
Raw Normal View History

Initial commit: AVWatch v1.0.0 with .gitignore and CLAUDE.md
2026-09-24 13:07:21 -04:00
"""Command line interface."""
from __future__ import annotations
import argparse
import logging
import sys
from datetime import datetime
from pathlib import Path
from . import config as config_mod
from . import coupling as coupling_mod
from . import inventory as inventory_mod
from . import report as report_mod
from .models import Finding, RunSummary, VersionCheck, utcnow
from .report import emailer
from .scanner import Scanner, sort_findings
from .store import Store, rehydrate_evidence, rehydrate_verdict
from .versions import compare
log = logging.getLogger("avwatch")
def _setup_logging(level: str, quiet: bool) -> None:
logging.basicConfig(
level=logging.ERROR if quiet else getattr(logging, level, logging.INFO),
format="%(asctime)s %(levelname)-7s %(name)s: %(message)s",
datefmt="%H:%M:%S",
stream=sys.stderr,
)
# requests/urllib3 are noisy at INFO.
logging.getLogger("urllib3").setLevel(logging.WARNING)
def _load(args) -> tuple[config_mod.Config, list, Store]:
cfg = config_mod.load(args.config)
_setup_logging(cfg.log_level, getattr(args, "quiet", False))
items = inventory_mod.load(args.inventory or cfg.inventory_path)
store = Store(cfg.db_path)
return cfg, items, store
def _report_path(cfg: config_mod.Config, tag: str) -> Path:
stamp = datetime.now().strftime("%Y-%m-%d_%H%M")
return cfg.report_dir / f"avwatch_{tag}_{stamp}.html"
# --------------------------------------------------------------------------
# commands
# --------------------------------------------------------------------------
def cmd_scan(args) -> int:
cfg, items, store = _load(args)
scanner = Scanner(cfg, store, offline=args.offline)
policy = "none" if args.no_llm else args.llm
findings, run = scanner.scan(
items, mode="scan", llm_policy=policy, only=args.only
)
findings = sort_findings(findings)
out = Path(args.out) if args.out else _report_path(cfg, "scan")
report_mod.render_html(findings, run, out)
print(report_mod.render_text(findings, run))
print(f"\nHTML report: {out}")
latest = cfg.report_dir / "latest.html"
try:
latest.write_text(out.read_text(encoding="utf-8"), encoding="utf-8")
print(f"Also copied to: {latest}")
except OSError as exc:
log.warning("could not write latest.html: %s", exc)
store.close()
return 0
def cmd_digest(args) -> int:
"""Full pass plus the email. This is what the Monday task runs."""
cfg, items, store = _load(args)
problems = emailer.validate(cfg.email, dry_run=args.dry_run)
if problems and not args.no_email:
print("Email is not ready:", file=sys.stderr)
for p in problems:
print(f" - {p}", file=sys.stderr)
if not args.force:
print("\nFix config.toml, or pass --no-email to skip sending, "
"or --force to build the report anyway.", file=sys.stderr)
store.close()
return 2
scanner = Scanner(cfg, store, offline=args.offline)
findings, run = scanner.scan(
items, mode="digest",
llm_policy="none" if args.no_llm else "all",
only=args.only,
)
findings = sort_findings(findings)
run.mode = "digest"
out = Path(args.out) if args.out else _report_path(cfg, "digest")
report_mod.render_html(findings, run, out)
try:
(cfg.report_dir / "latest.html").write_text(
out.read_text(encoding="utf-8"), encoding="utf-8")
except OSError:
pass
groups = report_mod.group(findings)
subject = _subject(groups, run)
if args.no_email:
print(report_mod.render_text(findings, run))
print(f"\nHTML report: {out}\n(email skipped: --no-email)")
store.close()
return 0
# Nothing new and configured not to nag? Still send if asked to.
nothing_new = run.new_since_last == 0 and not groups.actionable
if nothing_new and not cfg.email.send_when_nothing_new:
print(f"Nothing new; email suppressed by "
f"email.send_when_nothing_new = false. Report at {out}")
store.close()
return 0
try:
result = emailer.send(
cfg.email,
subject=subject,
html_body=report_mod.render_email(findings, run, out),
text_body=report_mod.render_text(findings, run),
attachment=out,
dry_run=args.dry_run,
)
except emailer.EmailError as exc:
print(f"Email failed: {exc}", file=sys.stderr)
print(f"The report was still written to {out}", file=sys.stderr)
store.close()
return 1
print(result)
print(f"HTML report: {out}")
for f in findings:
if f.check.latest and f.comparison == "behind":
store.mark_reported(f.item.id, f.check.latest, channel="email")
store.close()
return 0
def _subject(groups: report_mod.Groups, run: RunSummary) -> str:
stamp = datetime.now().strftime("%d %b")
if groups.avoid:
return (f"{stamp}: {len(groups.avoid)} update(s) to AVOID, "
f"{len(groups.go_behind)} cleared")
if groups.go_behind:
return (f"{stamp}: {len(groups.go_behind)} update(s) cleared, "
f"{len(groups.wait)} on hold")
if groups.wait:
return f"{stamp}: {len(groups.wait)} update(s) available, all on hold"
if run.errors:
return f"{stamp}: no updates, {run.errors} source error(s)"
# Never claim the fleet is up to date when we have nothing to compare
# against -- that reads as an all-clear it has not earned.
if groups.no_baseline:
return (f"{stamp}: {len(groups.no_baseline)} item(s) still need an "
f"installed version recorded")
return f"{stamp}: fleet up to date, nothing to do"
def cmd_report(args) -> int:
"""Rebuild the HTML from the last stored scan. No network."""
cfg, items, store = _load(args)
by_id = inventory_mod.by_id(items)
findings: list[Finding] = []
for row in store.all_finding_rows():
item = by_id.get(row["item_id"])
if item is None:
continue # removed from inventory since that scan
check = VersionCheck(
item_id=item.id,
latest=row["latest"],
status=row["check_status"] or "unknown",
detail=row["check_detail"],
)
findings.append(Finding(
item=item, check=check,
comparison=row["comparison"] or compare(item.installed, row["latest"]),
bump=row["bump"] or "unknown",
evidence=rehydrate_evidence(row["evidence_json"]),
verdict=rehydrate_verdict(row["verdict_json"]),
))
if not findings:
print("No stored findings yet. Run 'avwatch scan' first.", file=sys.stderr)
store.close()
return 1
findings = sort_findings(findings)
run = store.last_run() or RunSummary(run_id="cached", started=utcnow())
run.items_checked = len(findings)
run.updates_found = sum(1 for f in findings if f.comparison == "behind")
run.errors = sum(1 for f in findings if f.check.status == "error")
if args.text:
print(report_mod.render_text(findings, run))
else:
out = Path(args.out) if args.out else _report_path(cfg, "cached")
report_mod.render_html(findings, run, out)
print(f"HTML report: {out}")
store.close()
return 0
def cmd_list(args) -> int:
cfg, items, store = _load(args)
kinds: dict[str, int] = {}
print(f"{'ID':<26} {'INSTALLED':<12} {'SOURCE':<11} {'CRIT':<12} NAME")
print("-" * 100)
for i in sorted(items, key=lambda x: (x.category, x.id)):
if args.category and i.category != args.category:
continue
kinds[i.source_kind] = kinds.get(i.source_kind, 0) + 1
print(f"{i.id:<26} {(i.installed or '-'):<12} {i.source_kind:<11} "
f"{i.criticality:<12} {i.name}")
print("-" * 100)
print(f"{len(items)} items. Sources: " +
", ".join(f"{k}={v}" for k, v in sorted(kinds.items())))
store.close()
return 0
def cmd_set_installed(args) -> int:
cfg, items, store = _load(args)
path = args.inventory or cfg.inventory_path
try:
inventory_mod.save_installed(path, args.item_id, args.version)
except inventory_mod.InventoryError as exc:
print(f"error: {exc}", file=sys.stderr)
store.close()
return 1
store.add_decision(args.item_id, "installed", args.version,
note=args.note, by=args.by)
print(f"{args.item_id}: installed version recorded as {args.version}")
store.close()
return 0
def cmd_mark(args) -> int:
cfg, items, store = _load(args)
if args.item_id not in inventory_mod.by_id(items):
print(f"error: no such item id: {args.item_id}", file=sys.stderr)
store.close()
return 1
store.add_decision(args.item_id, args.decision, args.version,
note=args.note, by=args.by)
print(f"{args.item_id}: recorded '{args.decision}'"
+ (f" for {args.version}" if args.version else ""))
store.close()
return 0
def cmd_worksheet(args) -> int:
"""Print a walk-the-rack checklist of versions still to record.
Exists because "record your installed versions" is not an actionable
instruction. This prints, in priority order, what to look at, where the
version is displayed on that thing, and the exact command to type.
"""
cfg, items, store = _load(args)
by_id = inventory_mod.by_id(items)
def needs_version(i):
return (not i.installed and i.is_confirmed
and i.update_class == "normal" and i.source_kind != "manual")
# Default to the user's own inventory.yaml. Mixing their gear with items
# AVWatch proposed makes the whole list feel untrustworthy, and rightly so
# -- they are different kinds of claim.
all_todo = [i for i in items if needs_version(i)]
mine = [i for i in all_todo if i.is_user_declared]
added = [i for i in all_todo if not i.is_user_declared]
todo = all_todo if args.include == "all" else mine
def priority(i):
crit = {"showstopper": 0, "high": 1, "normal": 2, "low": 3}
return (
0 if i.coupled_with or i.update_group else 1, # coupled first
crit.get(i.criticality, 2),
i.name.lower(),
)
todo.sort(key=priority)
total_todo = len(todo)
if args.limit:
todo = todo[: args.limit]
if not todo:
print("Nothing to record -- every confirmed, trackable item already "
"has an installed version. Good.")
store.close()
return 0
coupled = [i for i in todo if i.coupled_with or i.update_group]
print("=" * 74)
print("AVWATCH -- VERSIONS TO WRITE DOWN")
if args.include == "all":
print("(your own list PLUS the items AVWatch proposed)")
else:
print("(your own equipment list only)")
print("=" * 74)
print()
print("Work down this list whenever you are at the rack. For each one:")
print(" 1. find the version where it says to look")
print(" 2. type the command shown, with that version on the end")
print()
print("You do not have to finish it. Every line you do makes the Monday")
print("email more useful, and nothing breaks while lines are missing.")
print()
if coupled:
print(f"The first {len(coupled)} matter most: they are gear that has to")
print("match something else, so AVWatch cannot check the match until")
print("it knows both versions.")
print()
if args.include != "all" and added:
print(f"NOT SHOWN: {len(added)} further item(s) that AVWatch proposed")
print("rather than you listing them -- cameras, intercom, network gear")
print("and so on. Add --include all to see those too, or delete")
print("inventory-expansion.yaml and inventory-optional.yaml to drop")
print("them entirely.")
print()
group_seen: set[str] = set()
for n, i in enumerate(todo, 1):
tag = []
if i.coupled_with or i.update_group:
tag.append("MATCHED SET")
tag.append(i.criticality)
if not i.is_user_declared:
tag.append("NOT FROM YOUR LIST")
print("-" * 74)
print(f"{n}. {i.name} [{', '.join(tag)}]")
# One "must match" line per item, from both sources combined --
# update_group members and explicit must_match edges overlap heavily.
mates: dict[str, None] = {}
if i.update_group:
for m in items:
if m.update_group == i.update_group and m.id != i.id:
mates[m.name] = None
for c in i.coupled_with:
other = by_id.get(c.id)
if other and c.relation == "must_match":
mates[other.name] = None
if mates:
print(f" Must match: {', '.join(mates)}")
if i.version_hint:
print(f" Where to look: {i.version_hint}")
elif i.check_url:
print(f" Where to look: not recorded yet -- vendor page: {i.check_url}")
print(f" Then type: python -m avwatch set-installed {i.id} <version>")
print("-" * 74)
remaining = total_todo - len(todo)
if remaining > 0:
print(f"\nShowing {len(todo)} of {total_todo} worth recording. "
f"{remaining} more after these --")
print(f"see them all with: python -m avwatch worksheet --limit 0")
else:
print(f"\nThat is all {total_todo} of them.")
print("\nRe-run this any time: python -m avwatch worksheet")
if args.out:
out = Path(args.out)
out.parent.mkdir(parents=True, exist_ok=True)
import io
import contextlib
buf = io.StringIO()
with contextlib.redirect_stdout(buf):
args_copy = argparse.Namespace(**vars(args))
args_copy.out = None
cmd_worksheet(args_copy)
out.write_text(buf.getvalue(), encoding="utf-8")
print(f"\nAlso saved to: {out}")
store.close()
return 0
def cmd_enter(args) -> int:
"""Walk through items one at a time and type in the running version.
The alternative was 87 separate `set-installed` commands, or hand-editing
a 1300-line YAML file. Neither is something a person does on a Tuesday.
"""
cfg, items, store = _load(args)
path = args.inventory or cfg.inventory_path
# What the last scan found, so the prompt can show the latest version
# next to the question -- often the answer is "same as latest".
latest_by_id = {}
for row in store.all_finding_rows():
if row.get("latest"):
latest_by_id[row["item_id"]] = row["latest"]
todo = [
i for i in items
if not i.installed and i.is_confirmed
and i.update_class == "normal" and i.source_kind != "manual"
]
if args.only:
wanted = set(args.only)
todo = [i for i in todo if i.id in wanted]
crit = {"showstopper": 0, "high": 1, "normal": 2, "low": 3}
todo.sort(key=lambda i: (
0 if (i.coupled_with or i.update_group) else 1,
crit.get(i.criticality, 2),
i.name.lower(),
))
if not todo:
print("Nothing left to fill in. Every trackable item has a version.")
store.close()
return 0
print()
print("=" * 70)
print(f"ENTER THE VERSIONS YOU ARE RUNNING ({len(todo)} to go)")
print("=" * 70)
print()
print("For each one, type the version and press Enter.")
print()
print(" press Enter on its own skip it, ask again next time")
print(" type same it matches the latest shown")
print(" type q stop and save what you have done")
print()
print("Nothing is lost if you stop early.")
print()
done = skipped = 0
for n, item in enumerate(todo, 1):
latest = latest_by_id.get(item.id)
print("-" * 70)
flag = " [MATCHED SET]" if (item.coupled_with or item.update_group) else ""
print(f"({n}/{len(todo)}) {item.name}{flag}")
print(f" {item.criticality}")
if latest:
print(f" newest available: {latest}")
if item.version_hint:
print(f" where to look: {item.version_hint}")
try:
raw = input(" running version > ")
except (EOFError, KeyboardInterrupt):
print("\n\nStopped.")
break
# Copy-pasting a version out of a web page or a PDF drags along
# zero-width and non-breaking characters that are invisible here but
# break the version comparison later.
raw = raw.replace("", "").replace("​", "")
raw = raw.replace(" ", " ").strip().strip("'\"")
if raw.lower() in ("q", "quit", "exit"):
print("\nStopped.")
break
if not raw:
skipped += 1
continue
if raw.lower() in ("same", "s", "latest") and latest:
raw = latest
elif raw.lower() in ("same", "s", "latest"):
print(" (no latest version known for this one -- skipped)")
skipped += 1
continue
from .versions import parse
if not parse(raw).valid:
print(f" '{raw}' does not look like a version number "
f"-- skipped. Try again next time.")
skipped += 1
continue
try:
inventory_mod.save_installed(path, item.id, raw)
except inventory_mod.InventoryError as exc:
print(f" could not save: {exc}")
continue
store.add_decision(item.id, "installed", raw, note="entered via 'enter'")
print(f" saved: {item.name} = {raw}")
done += 1
print()
print("=" * 70)
print(f"Saved {done}. Skipped {skipped}. "
f"{len(todo) - done - skipped} not reached.")
if done:
print("\nSee what that changed: python -m avwatch scan --no-llm")
remaining = len(todo) - done
if remaining:
print(f"\nRun 'python -m avwatch enter' again any time to carry on "
f"with the other {remaining}.")
print()
store.close()
return 0
def cmd_history(args) -> int:
cfg, items, store = _load(args)
print("Recent runs:")
for r in store.recent_runs(args.limit):
print(f" {r['started'][:16]} {r['mode']:<7} "
f"checked={r['items_checked']:<4} behind={r['updates_found']:<4} "
f"new={r['new_since_last']:<3} errors={r['errors']:<3} "
f"run={r['run_id']}")
if args.item_id:
print(f"\nDecisions for {args.item_id}:")
for d in store.decisions_for(args.item_id):
print(f" {d['decided_at'][:16]} {d['decision']:<10} "
f"{d['version'] or '-':<12} {d['note'] or ''}")
store.close()
return 0
def cmd_doctor(args) -> int:
"""Check the setup end to end without sending anything."""
cfg = config_mod.load(args.config)
_setup_logging(cfg.log_level, quiet=True)
ok = True
print("PATHS")
required = {"config", "inventory"}
for label, p in (("config", args.config or cfg.root / "config.toml"),
("inventory", cfg.inventory_path),
("database", cfg.db_path),
("cache", cfg.cache_dir),
("reports", cfg.report_dir)):
exists = Path(p).exists()
if exists:
mark = "ok "
elif label in required:
mark = "MISSING"
ok = False
else:
# The database and output folders are created on first use.
mark = "new"
print(f" [{mark}] {label:<10} {p}")
print("\nINVENTORY")
try:
items = inventory_mod.load(cfg.inventory_path)
except inventory_mod.InventoryError as exc:
print(f" [FAIL] {exc}")
return 1
kinds: dict[str, int] = {}
no_installed = []
for i in items:
kinds[i.source_kind] = kinds.get(i.source_kind, 0) + 1
if (i.installed is None and i.source_kind != "manual"
and i.is_confirmed and i.update_class == "normal"):
no_installed.append(i.id)
files = inventory_mod.resolve_paths(cfg.inventory_path)
print(f" [ok ] {len(items)} items load cleanly from "
f"{len(files)} file(s):")
for f in files:
n = len(inventory_mod.load_file_ids(f))
print(f" {f.name}: {n} items")
print(" sources: " + ", ".join(f"{k}={v}" for k, v in sorted(kinds.items())))
confirmed = [i for i in items if i.is_confirmed]
unconfirmed = [i for i in items if not i.is_confirmed]
if unconfirmed:
print(f" [warn] {len(unconfirmed)} item(s) are placeholders "
f"(status: unconfirmed). No version lookup and no update advice "
f"runs for these until you name the make/model.")
billable = [i for i in confirmed
if i.source_kind == "websearch" and i.update_class == "normal"]
if billable:
print(f" [ok ] {len(billable)} confirmed item(s) need the paid "
f"web lookup")
if len(billable) > cfg.lookup.max_items_per_run:
print(f" [warn] lookup.max_items_per_run is "
f"{cfg.lookup.max_items_per_run}, below that -- "
f"{len(billable) - cfg.lookup.max_items_per_run} item(s) "
f"will be deferred to a later run each time. Raise it in "
f"config.toml if you want full coverage per run.")
classes: dict[str, int] = {}
for i in items:
if i.update_class != "normal":
classes[i.update_class] = classes.get(i.update_class, 0) + 1
if classes:
print(" special update classes: "
+ ", ".join(f"{k}={v}" for k, v in sorted(classes.items())))
if no_installed:
print(f" [warn] {len(no_installed)} confirmed, auto-checked item(s) "
f"have no 'installed' version, so no comparison is possible:")
for i in no_installed[:8]:
print(f" - {i}")
if len(no_installed) > 8:
print(f" ... and {len(no_installed) - 8} more")
print(" fix with: avwatch set-installed <id> <version>")
print("\nVERSION COUPLING")
by_id = inventory_mod.by_id(items)
problems = coupling_mod.validate(by_id)
n_edges = sum(len(i.coupled_with) for i in items)
grps = coupling_mod.groups(by_id)
if problems:
for p in problems[:10]:
print(f" [FAIL] {p}")
if len(problems) > 10:
print(f" [FAIL] ... and {len(problems) - 10} more")
ok = False
else:
print(f" [ok ] {n_edges} dependency edge(s) across "
f"{len(grps)} update group(s), all resolving")
if grps:
for name, members in sorted(grps.items()):
print(f" {name}: "
+ ", ".join(sorted(m.id for m in members)))
uncheckable = [i for i in items
if i.coupled_with and not i.installed]
if uncheckable:
print(f" [warn] {len(uncheckable)} coupled item(s) have no installed "
f"version, so their dependency can be described but never "
f"checked. These are the highest-value ones to record first.")
print("\nVERDICT ENGINE")
key = cfg.llm.api_key
# A placeholder pasted verbatim from documentation is a real and easy
# mistake, and it fails much later and much more confusingly than it
# should. Catch it here.
looks_fake = bool(key) and (
"..." in key
or key.strip().lower() in ("sk-ant-", "your-api-key", "sk-ant-xxx")
or len(key.strip()) < 25
)
if looks_fake:
print(f" [FAIL] ANTHROPIC_API_KEY is set to {key[:14]!r}..., which "
f"is a placeholder, not a real key.")
print(" Get one at console.anthropic.com -> API Keys, then:")
print(' setx ANTHROPIC_API_KEY "<paste the real key here>"')
print(" Open a NEW terminal afterwards -- setx only affects "
"new windows.")
ok = False
elif key:
try:
import anthropic # noqa: F401
print(f" [ok ] Claude enabled, model {cfg.llm.model} "
f"(escalates to {cfg.llm.escalate_model} for "
f"{', '.join(cfg.llm.escalate_for_criticality)})")
except ImportError:
print(" [warn] ANTHROPIC_API_KEY is set but the 'anthropic' package "
"is not installed -- run: pip install anthropic")
ok = False
else:
print(" [warn] no ANTHROPIC_API_KEY -- heuristics only. Verdicts will "
"be blunter and carry no reasoning.")
print("\nCOMMUNITY SOURCES")
if cfg.risk.reddit_client_id and cfg.risk.reddit_client_secret:
print(" [ok ] Reddit OAuth credentials present")
else:
print(" [warn] no Reddit credentials -- falling back to DuckDuckGo, "
"which is rate-limited and noisier.")
print(" Create a free 'script' app: https://www.reddit.com/prefs/apps")
print(f" [{'ok ' if cfg.risk.use_duckduckgo else 'off'}] DuckDuckGo web search")
print(f" [{'ok ' if cfg.risk.use_github_issues else 'off'}] GitHub issue search")
import os
if not (os.environ.get("AVWATCH_GITHUB_TOKEN") or os.environ.get("GITHUB_TOKEN")):
print(" [warn] no GitHub token -- 60 requests/hour shared between "
"release checks and issue searches. Set AVWATCH_GITHUB_TOKEN.")
print("\nEMAIL")
problems = emailer.validate(cfg.email)
if problems:
for p in problems:
print(f" [warn] {p}")
ok = ok and not cfg.email.enabled
else:
print(f" [ok ] {cfg.email.smtp_host}:{cfg.email.smtp_port} as "
f"{cfg.email.from_addr} -> {', '.join(cfg.email.to)}")
if args.network:
print("\nLIVE SOURCE TEST")
from .http import HttpClient
from .sources import check_item
from .sources.base import Context
http = HttpClient(cfg.cache_dir, cache_ttl=0, timeout=cfg.http_timeout)
ctx = Context(http=http)
sample: dict[str, object] = {}
for i in items:
sample.setdefault(i.source_kind, i)
# The websearch source needs an API key; without one it correctly
# reports "check by hand", which is a skip, not a failure.
ctx.scratch["lookup_cfg"] = cfg.lookup if cfg.lookup.enabled else None
ctx.scratch["store"] = None
for kind, item in sample.items():
if kind == "manual":
continue
chk = check_item(item, ctx) # type: ignore[arg-type]
if chk.status == "ok":
mark, detail = "ok ", chk.latest
elif chk.status == "manual":
mark, detail = "skip", chk.detail
else:
mark, detail = "FAIL", chk.detail
ok = False
print(f" [{mark}] {kind:<11} {item.name[:34]:<34} " # type: ignore[union-attr]
f"{detail}")
if ok:
print("\nSetup looks usable. Warnings above are optional features that "
"are simply not configured yet.")
else:
print("\nSetup has problems -- see the FAIL / MISSING lines above. "
"(Lines marked 'warn' or 'skip' are unconfigured extras, "
"not breakage.)")
return 0 if ok else 1
def cmd_test_email(args) -> int:
cfg = config_mod.load(args.config)
_setup_logging(cfg.log_level, quiet=True)
try:
result = emailer.send(
cfg.email,
subject="test message",
html_body="<p>AVWatch SMTP test. If you can read this, "
"delivery works.</p>",
text_body="AVWatch SMTP test. If you can read this, delivery works.",
dry_run=args.dry_run,
)
except emailer.EmailError as exc:
print(f"failed: {exc}", file=sys.stderr)
return 1
print(result)
return 0
# --------------------------------------------------------------------------
def build_parser() -> argparse.ArgumentParser:
p = argparse.ArgumentParser(
prog="avwatch",
description="Track firmware and software updates across a live "
"production fleet, and judge whether they are safe to install.",
)
p.add_argument("--config", type=Path, help="path to config.toml")
p.add_argument("--inventory", type=Path, help="path to inventory.yaml")
p.add_argument("-q", "--quiet", action="store_true",
help="suppress progress logging")
sub = p.add_subparsers(dest="command", required=True)
def add_scan_args(sp):
sp.add_argument("--only", nargs="+", metavar="ID",
help="check just these item ids")
sp.add_argument("--offline", action="store_true",
help="use only cached HTTP responses")
sp.add_argument("--no-llm", action="store_true",
help="heuristics only, no API calls")
sp.add_argument("--out", type=Path, help="write the HTML report here")
s = sub.add_parser("scan", help="check for updates and write a report")
add_scan_args(s)
s.add_argument("--llm", choices=["new-only", "all", "none"], default="new-only",
help="which items get a Claude verdict (default: new-only)")
s.set_defaults(func=cmd_scan)
d = sub.add_parser("digest", help="full scan plus the email briefing")
add_scan_args(d)
d.add_argument("--dry-run", action="store_true",
help="build everything but do not actually send")
d.add_argument("--no-email", action="store_true", help="skip sending")
d.add_argument("--force", action="store_true",
help="continue even if email config is incomplete")
d.set_defaults(func=cmd_digest)
r = sub.add_parser("report", help="rebuild the report from stored data")
r.add_argument("--out", type=Path)
r.add_argument("--text", action="store_true", help="print plain text instead")
r.set_defaults(func=cmd_report)
l = sub.add_parser("list", help="list the tracked inventory")
l.add_argument("--category")
l.set_defaults(func=cmd_list)
si = sub.add_parser("set-installed",
help="record the version actually running on an item")
si.add_argument("item_id")
si.add_argument("version")
si.add_argument("--note")
si.add_argument("--by")
si.set_defaults(func=cmd_set_installed)
m = sub.add_parser("mark", help="record a decision about an item")
m.add_argument("item_id")
m.add_argument("decision", choices=["installed", "holding", "skipped"])
m.add_argument("--version")
m.add_argument("--note")
m.add_argument("--by")
m.set_defaults(func=cmd_mark)
e = sub.add_parser(
"enter",
help="type in the versions you are running, one item at a time")
e.add_argument("--only", nargs="+", metavar="ID",
help="only ask about these item ids")
e.set_defaults(func=cmd_enter)
w = sub.add_parser(
"worksheet",
help="print a walk-the-rack checklist of versions still to record")
w.add_argument("--limit", type=int, default=15,
help="how many to list (default 15; use 0 for all)")
w.add_argument("--include", choices=["mine", "all"], default="mine",
help="'mine' (default) = only your own inventory.yaml; "
"'all' also includes the items AVWatch proposed")
w.add_argument("--out", type=Path, help="also save the checklist to a file")
w.set_defaults(func=cmd_worksheet)
h = sub.add_parser("history", help="show recent runs and decisions")
h.add_argument("--limit", type=int, default=10)
h.add_argument("--item-id")
h.set_defaults(func=cmd_history)
doc = sub.add_parser("doctor", help="check the setup")
doc.add_argument("--network", action="store_true",
help="also test one live fetch per source kind")
doc.set_defaults(func=cmd_doctor)
te = sub.add_parser("test-email", help="send a one-line test email")
te.add_argument("--dry-run", action="store_true")
te.set_defaults(func=cmd_test_email)
return p
def main(argv: list[str] | None = None) -> int:
args = build_parser().parse_args(argv)
try:
return args.func(args)
except inventory_mod.InventoryError as exc:
print(f"inventory error: {exc}", file=sys.stderr)
return 1
except KeyboardInterrupt:
print("\ninterrupted", file=sys.stderr)
return 130
if __name__ == "__main__":
sys.exit(main())
Reference in New Issue Copy Permalink