from __future__ import annotations import argparse import html import json import re from datetime import datetime, timezone from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parents[1] DEFAULT_REPORT_ROOT = ROOT / "reports" / "1c-access" def slugify(value: str, *, max_length: int = 80) -> str: slug = re.sub(r"[^0-9A-Za-zА-Яа-яЁё._-]+", "-", value.strip()) slug = re.sub(r"-+", "-", slug).strip("-._") return (slug or "role-audit")[:max_length] def esc(value: Any) -> str: return html.escape("" if value is None else str(value), quote=True) def read_json(path: Path) -> dict[str, Any]: data = json.loads(path.read_text(encoding="utf-8-sig")) if not isinstance(data, dict): raise ValueError(f"JSON root is not an object: {path}") return data def load_export(path: Path) -> dict[str, Any]: data = read_json(path) artifacts = data.get("artifacts") if isinstance(data.get("artifacts"), dict) else {} export_path = artifacts.get("json") if export_path: return read_json(Path(export_path)) return data def summary_role(summary: dict[str, Any]) -> str | None: query = summary.get("query") if isinstance(summary.get("query"), dict) else {} role = query.get("role") if isinstance(query, dict) else None return str(role) if role is not None else None def find_latest_summaries(report_root: Path, base_id: str, *, role: str | None = None, count: int = 2) -> list[Path]: folder = report_root / slugify(base_id, max_length=60) summaries: list[tuple[str, Path]] = [] role_filter = role.casefold() if role else None for path in folder.glob("*.summary.json"): try: summary = read_json(path) except (OSError, json.JSONDecodeError, ValueError): continue summary_role_value = summary_role(summary) if role_filter is not None and (summary_role_value or "").casefold() != role_filter: continue generated_at = str(summary.get("generated_at") or "") summaries.append((generated_at, path)) summaries.sort(key=lambda item: item[0], reverse=True) return [path for _, path in summaries[:count]] def user_key(row: dict[str, Any]) -> str: value = row.get("user_id") or row.get("user_name") return str(value or "").strip() def row_user(row: dict[str, Any]) -> dict[str, Any]: return { "user_id": row.get("user_id"), "user_name": row.get("user_name"), "user_type": row.get("user_type"), "user_active": row.get("user_active"), "user_marked": row.get("user_marked"), } def group_rows_by_user(export: dict[str, Any]) -> dict[str, dict[str, Any]]: users: dict[str, dict[str, Any]] = {} rows = export.get("rows") if isinstance(export.get("rows"), list) else [] for row in rows: if not isinstance(row, dict): continue key = user_key(row) if not key: continue item = users.setdefault(key, {"user": row_user(row), "access_paths": set(), "rows": []}) if row.get("access_path"): item["access_paths"].add(str(row.get("access_path"))) item["rows"].append(row) for item in users.values(): item["access_paths"] = sorted(item["access_paths"]) return users def compare_exports(old_export: dict[str, Any], new_export: dict[str, Any]) -> dict[str, Any]: old_users = group_rows_by_user(old_export) new_users = group_rows_by_user(new_export) old_keys = set(old_users) new_keys = set(new_users) added_keys = sorted(new_keys - old_keys) removed_keys = sorted(old_keys - new_keys) common_keys = sorted(old_keys & new_keys) changed_paths = [ { "user": new_users[key]["user"], "old_access_paths": old_users[key]["access_paths"], "new_access_paths": new_users[key]["access_paths"], } for key in common_keys if old_users[key]["access_paths"] != new_users[key]["access_paths"] ] return { "schema": "onec_access_role_audit_compare.v1", "status": "ok", "generated_at": datetime.now(timezone.utc).isoformat(), "counts": { "old_users": len(old_keys), "new_users": len(new_keys), "added_users": len(added_keys), "removed_users": len(removed_keys), "unchanged_users": len(common_keys), "changed_access_paths": len(changed_paths), }, "added_users": [new_users[key]["user"] for key in added_keys], "removed_users": [old_users[key]["user"] for key in removed_keys], "changed_access_paths": changed_paths, } def render_html_report(compare: dict[str, Any]) -> str: counts = compare.get("counts") if isinstance(compare.get("counts"), dict) else {} def user_rows(name: str) -> str: users = compare.get(name) if isinstance(compare.get(name), list) else [] return "\n".join( "" f"{esc(item.get('user_name'))}" f"{esc(item.get('user_id'))}" f"{esc(item.get('user_type'))}" f"{esc(item.get('user_active'))}" f"{esc(item.get('user_marked'))}" "" for item in users if isinstance(item, dict) ) or "Absent." changed = compare.get("changed_access_paths") if isinstance(compare.get("changed_access_paths"), list) else [] changed_rows = "\n".join( "" f"{esc((item.get('user') or {}).get('user_name') if isinstance(item.get('user'), dict) else None)}" f"
{esc('\\n'.join(item.get('old_access_paths') or []))}
" f"
{esc('\\n'.join(item.get('new_access_paths') or []))}
" "" for item in changed if isinstance(item, dict) ) or "Absent." return f""" 1C Access Audit Compare

1C Access Audit Compare

Generated: {esc(compare.get('generated_at'))}
{esc(counts.get('old_users'))}old users
{esc(counts.get('new_users'))}new users
{esc(counts.get('added_users'))}added
{esc(counts.get('removed_users'))}removed
{esc(counts.get('changed_access_paths'))}path changes

Added users

{user_rows('added_users')}
UserIDTypeActiveMarked

Removed users

{user_rows('removed_users')}
UserIDTypeActiveMarked

Changed access paths

{changed_rows}
UserOld pathsNew paths
""" def compare_files(old_path: Path, new_path: Path, *, output: Path | None = None, html_output: Path | None = None) -> dict[str, Any]: result = compare_exports(load_export(old_path), load_export(new_path)) result["sources"] = {"old": str(old_path), "new": str(new_path)} result["artifacts"] = { **({"json": str(output)} if output is not None else {}), **({"html": str(html_output)} if html_output is not None else {}), } if output is not None: output.parent.mkdir(parents=True, exist_ok=True) output.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8") if html_output is not None: html_output.parent.mkdir(parents=True, exist_ok=True) html_output.write_text(render_html_report(result), encoding="utf-8") return result def main() -> int: parser = argparse.ArgumentParser(description="Compare two 1C access role audit exports or summaries.") parser.add_argument("old", type=Path, nargs="?") parser.add_argument("new", type=Path, nargs="?") parser.add_argument("--latest", action="store_true", help="Compare the latest two summaries for a base/role.") parser.add_argument("--base-id", default="upo_test") parser.add_argument("--role") parser.add_argument("--report-root", type=Path, default=DEFAULT_REPORT_ROOT) parser.add_argument("--output", type=Path) parser.add_argument("--html", type=Path) args = parser.parse_args() old_path = args.old new_path = args.new output = args.output html_output = args.html if args.latest: latest = find_latest_summaries(args.report_root, args.base_id, role=args.role, count=2) if len(latest) < 2: parser.error("Not enough matching summaries for --latest; need at least two.") new_path, old_path = latest[0], latest[1] folder = args.report_root / slugify(args.base_id, max_length=60) stem = f"compare-{old_path.stem.replace('.summary', '')}-{new_path.stem.replace('.summary', '')}" output = output or folder / f"{stem}.json" html_output = html_output or folder / f"{stem}.html" if old_path is None or new_path is None: parser.error("Either provide OLD and NEW paths or use --latest.") result = compare_files(old_path, new_path, output=output, html_output=html_output) print(json.dumps(result, ensure_ascii=False, indent=2)) return 0 if result.get("status") == "ok" else 1 if __name__ == "__main__": raise SystemExit(main())