#!/usr/bin/env python3 """ NAC Activity Monitor 监控编译活动，检测空闲时间 """ import os import time import json import subprocess from datetime import datetime from pathlib import Path class ActivityMonitor: def __init__(self, workspace_path, idle_timeout=2700): """ 初始化活动监控器 Args: workspace_path: NAC工作区路径 idle_timeout: 空闲超时时间（秒），默认2700秒（45分钟） """ self.workspace_path = Path(workspace_path) self.idle_timeout = idle_timeout self.last_activity_time = time.time() self.activities = [] self.activity_log_path = self.workspace_path / 'memory' / 'logs' self.activity_log_path.mkdir(parents=True, exist_ok=True) # 监控的文件扩展名 self.watched_extensions = ['.rs', '.go', '.charter', '.toml', '.json', '.md'] # 监控的项目目录 self.watched_projects = [ 'nac-blockchain', 'nac-asset-onchain', 'nac-quantum-explorer' ] def record_activity(self, activity_type, details): """ 记录活动 Args: activity_type: 活动类型（file_modified, git_commit, lint_check等） details: 活动详情 """ activity = { 'type': activity_type, 'timestamp': datetime.now().isoformat(), 'details': details } self.activities.append(activity) self.last_activity_time = time.time() # 保存到日志文件 self._save_activity_log(activity) def _save_activity_log(self, activity): """保存活动日志""" today = datetime.now().strftime('%Y%m%d') log_file = self.activity_log_path / f'activity_{today}.log' with open(log_file, 'a') as f: f.write(json.dumps(activity, ensure_ascii=False) + '\n') def check_file_modifications(self): """ 检查文件修改 使用git status检测修改的文件 """ for project in self.watched_projects: project_path = self.workspace_path / project if not project_path.exists(): continue try: # 检查未提交的修改 result = subprocess.run( ['git', 'status', '--porcelain'], cwd=project_path, capture_output=True, text=True, timeout=5 ) if result.stdout.strip(): modified_files = [] for line in result.stdout.strip().split('\n'): if line: status = line[:2].strip() filepath = line[3:].strip() # 只关注监控的文件类型 if any(filepath.endswith(ext) for ext in self.watched_extensions): modified_files.append({ 'status': status, 'path': filepath }) if modified_files: self.record_activity('file_modified', { 'project': project, 'files': modified_files, 'count': len(modified_files) }) except Exception as e: print(f"检查文件修改失败 ({project}): {e}") def check_git_commits(self): """ 检查Git提交 检查今日的所有提交 """ for project in self.watched_projects: project_path = self.workspace_path / project if not project_path.exists(): continue try: # 获取今日提交 result = subprocess.run( ['git', 'log', '--since=midnight', '--pretty=format:%h|%s|%an|%ad', '--date=iso'], cwd=project_path, capture_output=True, text=True, timeout=5 ) if result.stdout.strip(): commits = [] for line in result.stdout.strip().split('\n'): if line: parts = line.split('|') if len(parts) == 4: commits.append({ 'hash': parts[0], 'message': parts[1], 'author': parts[2], 'date': parts[3] }) if commits: # 只记录最新的提交（避免重复） last_recorded_hash = self._get_last_recorded_commit_hash(project) new_commits = [] for commit in commits: if commit['hash'] == last_recorded_hash: break new_commits.append(commit) if new_commits: self.record_activity('git_commit', { 'project': project, 'commits': new_commits, 'count': len(new_commits) }) except Exception as e: print(f"检查Git提交失败 ({project}): {e}") def _get_last_recorded_commit_hash(self, project): """获取最后记录的提交哈希""" today = datetime.now().strftime('%Y%m%d') log_file = self.activity_log_path / f'activity_{today}.log' if not log_file.exists(): return None last_hash = None with open(log_file, 'r') as f: for line in f: try: activity = json.loads(line) if activity['type'] == 'git_commit' and activity['details']['project'] == project: commits = activity['details']['commits'] if commits: last_hash = commits[0]['hash'] except: continue return last_hash def check_lint_activity(self): """ 检查Lint活动 检查今日的Lint检查记录 """ today = datetime.now().strftime('%Y%m%d') lint_log_file = self.activity_log_path / f'lint_{today}.log' if lint_log_file.exists(): # 读取Lint日志 with open(lint_log_file, 'r') as f: content = f.read() if content: # 记录Lint活动 self.record_activity('lint_check', { 'log_file': str(lint_log_file), 'size': len(content) }) def is_idle(self): """ 检查是否空闲 Returns: bool: 如果空闲超过idle_timeout秒，返回True """ idle_time = time.time() - self.last_activity_time return idle_time > self.idle_timeout def get_idle_time(self): """ 获取当前空闲时间（秒） Returns: float: 空闲时间（秒） """ return time.time() - self.last_activity_time def get_today_activities(self): """ 获取今日所有活动 Returns: list: 活动列表 """ today = datetime.now().strftime('%Y%m%d') log_file = self.activity_log_path / f'activity_{today}.log' if not log_file.exists(): return [] activities = [] with open(log_file, 'r') as f: for line in f: try: activity = json.loads(line) activities.append(activity) except: continue return activities def get_activity_statistics(self): """ 获取今日活动统计 Returns: dict: 统计信息 """ activities = self.get_today_activities() stats = { 'total_activities': len(activities), 'file_modifications': 0, 'git_commits': 0, 'lint_checks': 0, 'files_modified_count': 0, 'commits_count': 0 } for activity in activities: activity_type = activity['type'] if activity_type == 'file_modified': stats['file_modifications'] += 1 stats['files_modified_count'] += activity['details'].get('count', 0) elif activity_type == 'git_commit': stats['git_commits'] += 1 stats['commits_count'] += activity['details'].get('count', 0) elif activity_type == 'lint_check': stats['lint_checks'] += 1 return stats def reset_idle_timer(self): """重置空闲计时器""" self.last_activity_time = time.time() def scan_all_activities(self): """ 扫描所有活动 一次性检查所有类型的活动 """ self.check_file_modifications() self.check_git_commits() self.check_lint_activity() if __name__ == '__main__': # 测试代码 import sys workspace = '/home/ubuntu/nac-workspace' if len(sys.argv) > 1: workspace = sys.argv[1] monitor = ActivityMonitor(workspace, idle_timeout=60) # 测试用1分钟超时 print(f"NAC Activity Monitor") print(f"工作区: {workspace}") print(f"空闲超时: {monitor.idle_timeout}秒") print() # 扫描活动 print("扫描活动...") monitor.scan_all_activities() # 显示统计 stats = monitor.get_activity_statistics() print(f"\n今日活动统计:") print(f" 总活动数: {stats['total_activities']}") print(f" 文件修改: {stats['file_modifications']}次 ({stats['files_modified_count']}个文件)") print(f" Git提交: {stats['git_commits']}次 ({stats['commits_count']}个提交)") print(f" Lint检查: {stats['lint_checks']}次") # 显示空闲状态 idle_time = monitor.get_idle_time() print(f"\n当前空闲时间: {idle_time:.0f}秒") print(f"是否空闲: {'是' if monitor.is_idle() else '否'}")