65 lines
2.0 KiB
Python
65 lines
2.0 KiB
Python
|
"""
|
||
|
|
JSON-Log für Capability-Checks (M3 Phase 2 — analog club_feature_logger).
|
||
|
|
"""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import json
|
||
|
|
import logging
|
||
|
|
import os
|
||
|
|
from datetime import datetime, timezone
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import Any, Dict, Optional
|
||
|
|
|
||
|
|
|
||
|
|
def _log_dir() -> Path:
|
||
|
|
custom = (os.getenv("CAPABILITY_LOG_DIR") or "").strip()
|
||
|
|
if custom:
|
||
|
|
return Path(custom)
|
||
|
|
return Path("/app/logs")
|
||
|
|
|
||
|
|
|
||
|
|
capability_logger = logging.getLogger("shinkan.capability_usage")
|
||
|
|
capability_logger.setLevel(logging.INFO)
|
||
|
|
capability_logger.propagate = False
|
||
|
|
|
||
|
|
if not capability_logger.handlers:
|
||
|
|
log_dir = _log_dir()
|
||
|
|
try:
|
||
|
|
log_dir.mkdir(parents=True, exist_ok=True)
|
||
|
|
log_file = log_dir / "capability-usage.log"
|
||
|
|
file_handler = logging.FileHandler(log_file, encoding="utf-8")
|
||
|
|
file_handler.setLevel(logging.INFO)
|
||
|
|
file_handler.setFormatter(logging.Formatter("%(message)s"))
|
||
|
|
capability_logger.addHandler(file_handler)
|
||
|
|
except OSError:
|
||
|
|
stream_handler = logging.StreamHandler()
|
||
|
|
stream_handler.setFormatter(logging.Formatter("[capability-usage] %(message)s"))
|
||
|
|
capability_logger.addHandler(stream_handler)
|
||
|
|
|
||
|
|
|
||
|
|
def log_capability_check(
|
||
|
|
*,
|
||
|
|
club_id: Optional[int],
|
||
|
|
profile_id: Optional[int],
|
||
|
|
capability_id: str,
|
||
|
|
action: str,
|
||
|
|
result: Dict[str, Any],
|
||
|
|
endpoint: Optional[str] = None,
|
||
|
|
phase: str = "probe",
|
||
|
|
) -> None:
|
||
|
|
entry = {
|
||
|
|
"timestamp": datetime.now(timezone.utc).isoformat(),
|
||
|
|
"club_id": club_id,
|
||
|
|
"profile_id": profile_id,
|
||
|
|
"capability": capability_id,
|
||
|
|
"action": action,
|
||
|
|
"endpoint": endpoint,
|
||
|
|
"phase": phase,
|
||
|
|
"allowed": result.get("allowed", True),
|
||
|
|
"reason": result.get("reason", "unknown"),
|
||
|
|
"account_state": result.get("account_state"),
|
||
|
|
"club_roles": result.get("club_roles"),
|
||
|
|
"enforcement": os.getenv("CAPABILITY_ENFORCE", "0") == "1",
|
||
|
|
}
|
||
|
|
capability_logger.info(json.dumps(entry, ensure_ascii=False))
|