🚧 Add export feature and refined exclusion info

This commit is contained in:
Jonas Bengtsson
2026-09-13 12:04:42 +02:00
parent 1d67330f27
commit 3bbfd6c161
10 changed files with 637 additions and 16 deletions
+22 -3
View File
@@ -204,8 +204,17 @@ class ApplicationController:
change_set = self._records_from_status(entries)
kept, excluded = filter_change_set(change_set, self.config.filters)
if excluded:
log.info("Excluded by filters: %d file(s)", len(excluded))
# Two-tier reporting (Section 7): the INFO summary shows whether
# filters bit; per-file paths and matching rules go to DEBUG so
# investigations get the details without bloating normal runs.
log.info(
"Detected %d change(s): %d in scope, %d excluded by filters",
len(change_set),
len(kept),
len(excluded),
)
for exclusion in excluded:
log.debug("Excluded: %s (%s)", exclusion.path, exclusion.reason)
# Self-exclusion safety net (FR-028): operational artefacts are
# already outside the repository; belt-and-braces path check here.
@@ -223,7 +232,7 @@ class ApplicationController:
started_at=dt.datetime.now().astimezone(),
host=socket.gethostname(),
)
log.info("Detected %d changes (run %s)", len(kept), run.run_id)
log.info("Collection run %s", run.run_id)
for record in [*kept.added, *kept.modified]:
result = self.resolvers.resolve(self.config.repository_path, record.path, log=log)
@@ -231,6 +240,16 @@ class ApplicationController:
log.info(
"Author resolved: %s -> %s", record.path, result.identifier
)
# Resolution details (source, confidence) go to DEBUG: they are
# operational diagnostics, not part of the audit record.
log.debug(
"Author resolution detail: %s -> source=%s confidence=%s",
record.path,
result.source,
result.confidence.value
if hasattr(result.confidence, "value")
else result.confidence,
)
message = build_commit_message(
kept,
+46 -12
View File
@@ -1,13 +1,16 @@
"""File filter and security validation (Sections 7 and 33).
Pure-path matching: patterns are matched against the repository-relative
POSIX path. Excluded files are neither staged nor committed nor reported.
Forbidden files abort the commit with a security exit code by default.
POSIX path. Excluded files are neither staged nor committed nor reported;
their paths and the rule that matched them are logged at DEBUG level
(Section 7, Section 22). Forbidden files abort the commit with a security
exit code by default.
"""
from __future__ import annotations
import re
from dataclasses import dataclass
from git_auto_tracker.changes.models import ChangeSet
from git_auto_tracker.config import FileFilters, SecurityConfig
@@ -16,6 +19,14 @@ from git_auto_tracker.errors import SecurityError
_PATTERN_CACHE: dict[str, re.Pattern[str]] = {}
@dataclass(frozen=True)
class Exclusion:
"""One file kept out of the ChangeSet, with the reason (Section 7)."""
path: str
reason: str # e.g. "matched exclude 'runtime/**'" or "no include pattern matched"
def matches(path: str, pattern: str) -> bool:
"""Match a repository-relative POSIX path against a glob pattern.
@@ -93,26 +104,41 @@ def _translate_segment(segment: str) -> str:
def is_included(path: str, filters: FileFilters) -> bool:
"""True when the path matches any include pattern (empty include = all)."""
return not filters.include or any(matches(path, p) for p in filters.include)
return inclusion_reason(path, filters) is None
def inclusion_reason(path: str, filters: FileFilters) -> str | None:
"""None when included; otherwise the reason this path is out of scope."""
if not filters.include or any(matches(path, p) for p in filters.include):
return None
return "no include pattern matched"
def is_excluded(path: str, filters: FileFilters) -> bool:
return any(matches(path, p) for p in filters.exclude)
return exclusion_reason(path, filters) is not None
def exclusion_reason(path: str, filters: FileFilters) -> str | None:
"""None when not excluded; otherwise the exclude pattern that matched."""
for pattern in filters.exclude:
if matches(path, pattern):
return f"matched exclude '{pattern}'"
return None
def filter_change_set(
change_set: ChangeSet,
filters: FileFilters,
) -> tuple[ChangeSet, list[str]]:
) -> tuple[ChangeSet, list[Exclusion]]:
"""Apply include/exclude rules to a ChangeSet.
Returns the filtered ChangeSet plus the list of excluded paths
(for logging).
Returns the filtered ChangeSet plus one Exclusion per file kept out,
carrying the path and the rule that matched (for DEBUG logging).
"""
from git_auto_tracker.changes.models import ChangeType
kept = ChangeSet()
excluded: list[str] = []
excluded: list[Exclusion] = []
buckets = {
ChangeType.ADDED: (change_set.added, kept.added),
@@ -124,10 +150,18 @@ def filter_change_set(
for records, target in buckets.values():
for record in records:
path = record.previous_path or record.path
if is_included(path, filters) and not is_excluded(path, filters):
target.append(record)
else:
excluded.append(record.path)
# Exclude matches are checked first: when a path is both
# out of include scope and exclude-matched, the explicit
# exclude rule is the more actionable reason to report.
blocked = exclusion_reason(path, filters)
if blocked is not None:
excluded.append(Exclusion(record.path, blocked))
continue
out_of_scope = inclusion_reason(path, filters)
if out_of_scope is not None:
excluded.append(Exclusion(record.path, out_of_scope))
continue
target.append(record)
return kept, excluded
+28
View File
@@ -34,6 +34,17 @@ def build_parser() -> argparse.ArgumentParser:
subparsers.add_parser("status", help="show detected changes without committing")
subparsers.add_parser("validate", help="validate configuration, repository and resolvers")
subparsers.add_parser("config", help="print the effective configuration (secrets redacted)")
export_parser = subparsers.add_parser(
"export", help="export the tracker commit history for analysis"
)
export_parser.add_argument(
"-o", "--output", type=Path, required=True,
help="output file (.csv or .jsonl); format follows the suffix",
)
export_parser.add_argument(
"--since", help="only commits more recent than this date (e.g. 2026-09-01)",
)
return parser
@@ -74,6 +85,20 @@ def _config_command(config_path: Path) -> int:
return ExitCode.SUCCESS
def _export_command(controller: ApplicationController, args: argparse.Namespace) -> int:
"""export command body: emit the audit trail as CSV/JSONL."""
from git_auto_tracker.exporting.history import export_history
count = export_history(
controller.config,
args.output,
since=args.since,
log=logging.getLogger("git_auto_tracker"),
)
print(f"{count} file-change record(s) exported to {args.output}")
return ExitCode.SUCCESS
def main(argv: list[str] | None = None) -> int:
args = build_parser().parse_args(argv)
@@ -100,6 +125,9 @@ def main(argv: list[str] | None = None) -> int:
if args.command == "validate":
return _validate_command(controller)
if args.command == "export":
return _export_command(controller, args)
if args.command == "status":
lock.acquire()
try:
@@ -0,0 +1 @@
"""History export for analysis (README: analyzing the audit trail)."""
+191
View File
@@ -0,0 +1,191 @@
"""History export for analysis (README: analyzing the audit trail).
Fetches tracker commits with `git log -z` (NUL-separated records, UTF-8;
Section 6.2 rules applied in reverse), parses the structured commit
messages built by message_builder (Sections 15/16/36), and emits one row
per file change ("long format") as CSV or JSON Lines.
The one-line entries use the SAME separator as collection did
(commit.file_separator). Commits created with a different separator than
the currently configured one are skipped with a warning, because their
fields cannot be split reliably.
"""
from __future__ import annotations
import csv
import json
import subprocess
from dataclasses import dataclass, field
from pathlib import Path
from git_auto_tracker.config import AppConfig
from git_auto_tracker.errors import GitError
#: Field delimiter for the git log --format string; cannot occur in
#: commit messages produced by the tracker.
_UNIT_SEPARATOR = "\x1f"
_SUBJECT_PREFIX = "SCADA automatic change capture - "
_FIELDS = ("commit", "collected_at", "run_id", "host", "letter",
"path", "previous_path", "author", "mtime")
@dataclass
class HistoryRecord:
"""One commit, parsed (collection-level fields + per-file rows)."""
commit: str
collected_at: str = ""
run_id: str = ""
host: str = ""
rows: list[dict[str, str]] = field(default_factory=list)
def fetch_commits(repo: Path, since: str | None = None) -> list[tuple[str, str]]:
"""Return (commit hash, full message) tuples, newest first.
Uses -z (NUL-separated records): multi-line bodies cannot corrupt the
record boundaries. Raises GitError when git fails.
"""
cmd = ["git", "log", "-z", f"--format=%H{_UNIT_SEPARATOR}%B"]
if since:
cmd.append(f"--since={since}")
try:
raw = subprocess.run(
cmd, cwd=str(repo), capture_output=True, check=True
).stdout
except subprocess.CalledProcessError as exc:
stderr = (exc.stderr or b"").decode("utf-8", errors="replace").strip()
raise GitError(f"git log failed: {stderr or exc.returncode}") from exc
except FileNotFoundError as exc:
raise GitError("git executable not found on PATH") from exc
pairs: list[tuple[str, str]] = []
for entry in raw.split(b"\x00"):
if not entry:
continue
commit_hash, _, body = entry.decode("utf-8").partition(_UNIT_SEPARATOR)
pairs.append((commit_hash, body))
return pairs
def parse_message(message: str, separator: str) -> HistoryRecord | None:
"""Parse one tracker commit message; None when it is not one of ours.
Non-tracker commits (e.g. the repository's initial import) return None.
Messages written with a DIFFERENT separator than *separator* are
recognized as tracker commits but yield no rows, since their fields
cannot be split reliably; the caller reports them as skipped.
"""
record = HistoryRecord(commit="")
saw_tracker_format = False
lines = message.splitlines()
for line in lines:
stripped = line.strip()
if stripped.startswith(_SUBJECT_PREFIX):
record.collected_at = stripped[len(_SUBJECT_PREFIX):].strip()
saw_tracker_format = True
elif stripped.startswith("Tracker-Run: "):
record.run_id = stripped.split(": ", 1)[1].strip()
saw_tracker_format = True
elif stripped.startswith("Tracker-Host: "):
record.host = stripped.split(": ", 1)[1].strip()
saw_tracker_format = True
elif line.startswith(" ") and f" {separator} " in line:
saw_tracker_format = True
record.rows.append(_parse_change_line(line.strip(), separator))
if not saw_tracker_format:
return None
return record
def _parse_change_line(text: str, separator: str) -> dict[str, str]:
"""Parse one ` <letter> <sep> <path> <sep> ...` entry (Section 15)."""
parts = [p.strip() for p in text.split(f" {separator} ")]
row = {
"letter": parts[0],
"path": parts[1] if len(parts) > 1 else "",
"previous_path": "",
"author": "",
"mtime": "",
}
letter = row["letter"]
if letter in ("A", "M") and len(parts) == 4:
row["author"], row["mtime"] = parts[2], parts[3]
elif letter == "R" and "->" in row["path"]:
old, new = row["path"].split(" -> ", 1)
row["previous_path"], row["path"] = old, new
# D lines stay bare (2 parts); malformed lines keep empty fields
return row
def to_rows(records: list[HistoryRecord]) -> list[dict[str, str]]:
"""Flatten parsed commits to long format: one row per file change."""
rows: list[dict[str, str]] = []
for record in records:
for row in record.rows:
rows.append({
"commit": record.commit,
"collected_at": record.collected_at,
"run_id": record.run_id,
"host": record.host,
**row,
})
return rows
def write_csv(rows: list[dict[str, str]], output: Path) -> None:
with output.open("w", encoding="utf-8", newline="") as handle:
writer = csv.DictWriter(handle, fieldnames=list(_FIELDS))
writer.writeheader()
writer.writerows(rows)
def write_jsonl(rows: list[dict[str, str]], output: Path) -> None:
with output.open("w", encoding="utf-8", newline="\n") as handle:
for row in rows:
handle.write(json.dumps(row, ensure_ascii=False) + "\n")
def export_history(
config: AppConfig,
output: Path,
since: str | None = None,
log: object | None = None,
) -> int:
"""Export all tracker commits of the configured repository to a file.
Returns the number of exported file-change rows. The output format is
chosen from the file suffix (.jsonl -> JSON Lines, otherwise CSV).
"""
separator = config.commit.file_separator
repo = config.repository_path
exported: list[HistoryRecord] = []
skipped_other_separator: list[str] = []
for commit_hash, message in fetch_commits(repo, since):
record = parse_message(message, separator)
if record is None:
continue # not a tracker commit (e.g. initial import)
record.commit = commit_hash
if not record.rows and record.collected_at:
# Tracker commit whose entries used a different separator
skipped_other_separator.append(commit_hash[:8])
continue
exported.append(record)
if skipped_other_separator and log is not None:
log.warning(
"export: %d commit(s) used a different file separator and were "
"skipped: %s",
len(skipped_other_separator),
", ".join(skipped_other_separator),
)
rows = to_rows(exported)
if output.suffix == ".jsonl":
write_jsonl(rows, output)
else:
write_csv(rows, output)
return len(rows)