324 lines
11 KiB
Python
324 lines
11 KiB
Python
#!/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 '否'}")
|