You've already forked DataMate
- 添加检查脚本 scripts/check_yesterdays_changes.py - 配置 cron 定时任务,每天 UTC 2:00(北京时间上午10:00)执行 - 更新 SOUL.md 和 HEARTBEAT.md,配置系统事件处理逻辑 - 报告发送到当前 Telegram 会话(-1003879848304)
138 lines
4.0 KiB
Python
Executable File
138 lines
4.0 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
定时检查昨天修改的代码和待测试项目
|
|
使用方法:python3 check_yesterdays_changes.py
|
|
"""
|
|
|
|
import subprocess
|
|
import json
|
|
import os
|
|
from datetime import datetime, timedelta, timezone
|
|
import sys
|
|
|
|
# 项目路径
|
|
PROJECT_PATH = "/root/.openclaw/workspace/Code/DataMate"
|
|
|
|
# 跟踪文件路径
|
|
TRACKING_FILE = "/root/.openclaw/workspace/memory/yesterday_changes.json"
|
|
|
|
def get_yesterdays_commits():
|
|
"""获取昨天修改的提交"""
|
|
# 昨天 UTC 00:00:00 到 今天 UTC 00:00:00
|
|
now_utc = datetime.now(timezone.utc)
|
|
yesterday = (now_utc - timedelta(days=1)).strftime("%Y-%m-%d")
|
|
today = now_utc.strftime("%Y-%m-%d")
|
|
|
|
# 获取昨天的提交
|
|
cmd = f"git log --since='{yesterday} 00:00:00 +0000' --until='{today} 00:00:00 +0000' --pretty=format:'%H|%s|%an|%ad' --date=iso"
|
|
result = subprocess.run(cmd, shell=True, cwd=PROJECT_PATH, capture_output=True, text=True)
|
|
|
|
if result.returncode != 0:
|
|
print(f"获取 git log 失败: {result.stderr}")
|
|
return []
|
|
|
|
commits = []
|
|
for line in result.stdout.strip().split('\n'):
|
|
if line:
|
|
parts = line.split('|')
|
|
if len(parts) >= 3:
|
|
commits.append({
|
|
'hash': parts[0],
|
|
'message': parts[1],
|
|
'author': parts[2],
|
|
'date': parts[3] if len(parts) > 3 else ''
|
|
})
|
|
|
|
return commits
|
|
|
|
def extract_test_items_from_commits(commits):
|
|
"""从提交信息中提取测试项"""
|
|
test_items = []
|
|
|
|
# 测试相关的关键词
|
|
test_keywords = [
|
|
'测试', 'test', 'verify', '验证', 'check',
|
|
'功能', 'feature', '优化', 'optimize', '修复', 'fix'
|
|
]
|
|
|
|
for commit in commits:
|
|
message = commit['message'].lower()
|
|
# 检查是否包含测试相关关键词
|
|
if any(keyword in message for keyword in test_keywords):
|
|
# 提提交信息的第一部分作为测试项
|
|
test_item = commit['message'].split(':')[0] if ':' in commit['message'] else commit['message'][:50]
|
|
test_items.append({
|
|
'message': commit['message'],
|
|
'hash': commit['hash'],
|
|
'date': commit['date']
|
|
})
|
|
|
|
return test_items
|
|
|
|
def save_tracking_data(commits, test_items):
|
|
"""保存跟踪数据到文件"""
|
|
now_utc = datetime.now(timezone.utc)
|
|
data = {
|
|
'date': now_utc.strftime("%Y-%m-%d"),
|
|
'commits': commits,
|
|
'test_items': test_items,
|
|
'generated_at': now_utc.isoformat()
|
|
}
|
|
|
|
# 确保目录存在
|
|
os.makedirs(os.path.dirname(TRACKING_FILE), exist_ok=True)
|
|
|
|
with open(TRACKING_FILE, 'w', encoding='utf-8') as f:
|
|
json.dump(data, f, ensure_ascii=False, indent=2)
|
|
|
|
def load_tracking_data():
|
|
"""加载跟踪数据"""
|
|
if not os.path.exists(TRACKING_FILE):
|
|
return None
|
|
|
|
with open(TRACKING_FILE, 'r', encoding='utf-8') as f:
|
|
return json.load(f)
|
|
|
|
def generate_report():
|
|
"""生成检查报告"""
|
|
commits = get_yesterdays_commits()
|
|
|
|
if not commits:
|
|
return None # 昨天没有修改代码
|
|
|
|
test_items = extract_test_items_from_commits(commits)
|
|
|
|
# 保存跟踪数据
|
|
save_tracking_data(commits, test_items)
|
|
|
|
# 生成报告
|
|
report = []
|
|
|
|
if commits:
|
|
report.append(f"📅 昨天(UTC 时间)共 {len(commits)} 次代码提交:\n")
|
|
|
|
if test_items:
|
|
report.append("📋 待测试项目:\n")
|
|
for i, item in enumerate(test_items, 1):
|
|
report.append(f"{i}. {item['message']}")
|
|
report.append(f" 提交: {item['hash'][:8]} | 时间: {item['date'][:19]}\n")
|
|
else:
|
|
report.append("✅ 没有明显的测试项\n")
|
|
|
|
report.append("\n请确认以上测试项是否已测试完毕或验证通过。")
|
|
|
|
return '\n'.join(report)
|
|
|
|
def main():
|
|
"""主函数"""
|
|
report = generate_report()
|
|
|
|
if report is None:
|
|
print("昨天没有修改代码,无需发送通知。")
|
|
return
|
|
|
|
print(report)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|