250 lines
10 KiB
Python
250 lines
10 KiB
Python
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(
|
||
"<tr>"
|
||
f"<td>{esc(item.get('user_name'))}</td>"
|
||
f"<td>{esc(item.get('user_id'))}</td>"
|
||
f"<td>{esc(item.get('user_type'))}</td>"
|
||
f"<td>{esc(item.get('user_active'))}</td>"
|
||
f"<td>{esc(item.get('user_marked'))}</td>"
|
||
"</tr>"
|
||
for item in users
|
||
if isinstance(item, dict)
|
||
) or "<tr><td colspan=\"5\">Absent.</td></tr>"
|
||
|
||
changed = compare.get("changed_access_paths") if isinstance(compare.get("changed_access_paths"), list) else []
|
||
changed_rows = "\n".join(
|
||
"<tr>"
|
||
f"<td>{esc((item.get('user') or {}).get('user_name') if isinstance(item.get('user'), dict) else None)}</td>"
|
||
f"<td><pre>{esc('\\n'.join(item.get('old_access_paths') or []))}</pre></td>"
|
||
f"<td><pre>{esc('\\n'.join(item.get('new_access_paths') or []))}</pre></td>"
|
||
"</tr>"
|
||
for item in changed
|
||
if isinstance(item, dict)
|
||
) or "<tr><td colspan=\"3\">Absent.</td></tr>"
|
||
|
||
return f"""<!doctype html>
|
||
<html lang="ru">
|
||
<head>
|
||
<meta charset="utf-8">
|
||
<title>1C Access Audit Compare</title>
|
||
<style>
|
||
body {{ font-family: Segoe UI, Arial, sans-serif; margin: 24px; color: #1f2933; }}
|
||
.cards {{ display: flex; flex-wrap: wrap; gap: 12px; margin: 18px 0; }}
|
||
.card {{ border: 1px solid #d9e2ec; border-radius: 8px; padding: 12px 14px; min-width: 130px; }}
|
||
.card b {{ display: block; font-size: 20px; }}
|
||
table {{ border-collapse: collapse; width: 100%; margin: 12px 0 24px; }}
|
||
th, td {{ border: 1px solid #d9e2ec; padding: 8px; vertical-align: top; }}
|
||
th {{ background: #f0f4f8; text-align: left; }}
|
||
pre {{ white-space: pre-wrap; margin: 0; font-family: Consolas, monospace; }}
|
||
</style>
|
||
</head>
|
||
<body>
|
||
<h1>1C Access Audit Compare</h1>
|
||
<div>Generated: {esc(compare.get('generated_at'))}</div>
|
||
<div class="cards">
|
||
<div class="card"><b>{esc(counts.get('old_users'))}</b>old users</div>
|
||
<div class="card"><b>{esc(counts.get('new_users'))}</b>new users</div>
|
||
<div class="card"><b>{esc(counts.get('added_users'))}</b>added</div>
|
||
<div class="card"><b>{esc(counts.get('removed_users'))}</b>removed</div>
|
||
<div class="card"><b>{esc(counts.get('changed_access_paths'))}</b>path changes</div>
|
||
</div>
|
||
<h2>Added users</h2>
|
||
<table><thead><tr><th>User</th><th>ID</th><th>Type</th><th>Active</th><th>Marked</th></tr></thead><tbody>{user_rows('added_users')}</tbody></table>
|
||
<h2>Removed users</h2>
|
||
<table><thead><tr><th>User</th><th>ID</th><th>Type</th><th>Active</th><th>Marked</th></tr></thead><tbody>{user_rows('removed_users')}</tbody></table>
|
||
<h2>Changed access paths</h2>
|
||
<table><thead><tr><th>User</th><th>Old paths</th><th>New paths</th></tr></thead><tbody>{changed_rows}</tbody></table>
|
||
</body>
|
||
</html>
|
||
"""
|
||
|
||
|
||
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())
|