File size: 2,203 Bytes
eed1cab | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 | """CLI subcommand: ``dataforge audit <txn_id>``."""
from __future__ import annotations
import json
from pathlib import Path
from typing import Annotated
import typer
from rich.console import Console
from rich.panel import Panel
from dataforge.transactions import TransactionAuditVerdict, verify_transaction_log
_console = Console(stderr=True)
def audit(
txn_id: Annotated[
str,
typer.Argument(help="Transaction identifier to audit."),
],
search_root: Annotated[
Path | None,
typer.Option(
"--search-root",
help="Root directory used to locate the transaction log.",
exists=True,
file_okay=False,
dir_okay=True,
readable=True,
),
] = None,
log_path: Annotated[
Path | None,
typer.Option(
"--log-path",
help="Explicit JSONL transaction log path.",
exists=True,
file_okay=True,
dir_okay=False,
readable=True,
),
] = None,
json_output: Annotated[
bool,
typer.Option("--json", help="Print the audit report as JSON."),
] = False,
) -> None:
"""Verify a transaction log's local hash chain."""
report = verify_transaction_log(txn_id, log_path=log_path, search_root=search_root)
if json_output:
typer.echo(json.dumps(report.model_dump(mode="json"), indent=2, sort_keys=True))
else:
style = "green" if report.verdict == TransactionAuditVerdict.VERIFIED else "red"
body = (
f"Verdict: [bold]{report.verdict.value}[/bold]\n"
f"Transaction: {report.txn_id or txn_id}\n"
f"Events: {report.event_count}\n"
f"Head SHA-256: {report.head_sha256 or 'n/a'}"
)
if report.errors:
body += "\n\n" + "\n".join(f"- {error}" for error in report.errors)
_console.print(Panel(body, title="Transaction Audit", style=style))
if report.verdict == TransactionAuditVerdict.VERIFIED:
raise typer.Exit(code=0)
if report.verdict == TransactionAuditVerdict.LEGACY_UNVERIFIED:
raise typer.Exit(code=1)
raise typer.Exit(code=2)
|