Coverage for merco/observability/audit.py: 0%

1"""审计日志""" 

2 

3import json 

4import time 

5from pathlib import Path 

6 

7 

8class AuditLogger: 

9 """记录安全与操作审计日志""" 

10 

11 def __init__(self, log_path: str = None): 

12 self.log_path = Path(log_path or "~/.merco/audit.log").expanduser() 

13 self.log_path.parent.mkdir(parents=True, exist_ok=True) 

14 

15 def log(self, action: str, user: str, details: dict = None): 

16 """记录审计事件""" 

17 entry = { 

18 "timestamp": time.time(), 

19 "action": action, 

20 "user": user, 

21 "details": details or {}, 

22 } 

23 

24 with open(self.log_path, "a") as f: 

25 f.write(json.dumps(entry) + "\n") 

26 

27 def read(self, limit: int = 100) -> list[dict]: 

28 """读取最近的审计日志""" 

29 if not self.log_path.exists(): 

30 return [] 

31 

32 entries = [] 

33 with open(self.log_path) as f: 

34 for line in f: 

35 line = line.strip() 

36 if line: 

37 entries.append(json.loads(line)) 

38 

39 return entries[-limit:]