auto: sync OpenClaw config 2026-08-27 11:17
This commit is contained in:
@@ -0,0 +1,265 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
周报数据库同步工具
|
||||
将 Markdown 周报的元数据同步到 MySQL 数据库
|
||||
|
||||
用法：
|
||||
python3 sync_reports_to_db.py --all # 同步所有周报
|
||||
python3 sync_reports_to_db.py --file xxx.md # 同步单个文件
|
||||
python3 sync_reports_to_db.py --week 2026-W35 # 同步指定周
|
||||
"""
|
||||
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
import argparse
|
||||
import json
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
|
||||
# 数据库配置
|
||||
DB_CONFIG = {
|
||||
'host': 'localhost',
|
||||
'user': 'root',
|
||||
'password': '', # 从 .env 读取
|
||||
'database': 'resume'
|
||||
}
|
||||
|
||||
REPORTS_DIR = Path(__file__).parent / 'weekly-reports'
|
||||
|
||||
|
||||
def load_env():
|
||||
"""读取 .env 文件"""
|
||||
env = {}
|
||||
env_path = Path(__file__).parent.parent / '.env'
|
||||
try:
|
||||
with open(env_path, 'r', encoding='utf-8') as f:
|
||||
for line in f:
|
||||
line = line.strip()
|
||||
if not line or line.startswith('#') or '=' not in line:
|
||||
continue
|
||||
key, _, value = line.partition('=')
|
||||
env[key.strip()] = value.strip().strip('"').strip("'")
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
return env
|
||||
|
||||
|
||||
def parse_markdown_report(filepath):
|
||||
"""解析 Markdown 周报文件，提取元数据"""
|
||||
with open(filepath, 'r', encoding='utf-8') as f:
|
||||
content = f.read()
|
||||
|
||||
data = {
|
||||
'file_path': str(filepath),
|
||||
'year': None,
|
||||
'week_number': None,
|
||||
'start_date': None,
|
||||
'end_date': None,
|
||||
'project': None,
|
||||
'main_task': None,
|
||||
'daily_work': {},
|
||||
'problems': [],
|
||||
'word_count': len(content)
|
||||
}
|
||||
|
||||
# 提取年份和周数（从文件名或内容）
|
||||
filename = filepath.name
|
||||
match = re.match(r'(\d{4})-W(\d{2})', filename)
|
||||
if match:
|
||||
data['year'] = int(match.group(1))
|
||||
data['week_number'] = int(match.group(2))
|
||||
|
||||
# 提取日期范围
|
||||
date_match = re.search(r'\*\*日期范围：\*\*\s*(\d{4}-\d{2}-\d{2})\s*~\s*(\d{4}-\d{2}-\d{2})', content)
|
||||
if date_match:
|
||||
data['start_date'] = date_match.group(1)
|
||||
data['end_date'] = date_match.group(2)
|
||||
|
||||
# 提取项目名称
|
||||
project_match = re.search(r'项目名称：\s*(维云智造\s*(G\d+)|G\d+)', content)
|
||||
if project_match:
|
||||
project_text = project_match.group(1)
|
||||
g_match = re.search(r'G(\d+)', project_text)
|
||||
if g_match:
|
||||
data['project'] = f"G{g_match.group(1)}"
|
||||
else:
|
||||
data['project'] = 'G5' # 默认
|
||||
|
||||
# 提取主要任务
|
||||
task_match = re.search(r'主要任务：\s*(.+?)(?:\n|$)', content)
|
||||
if task_match:
|
||||
data['main_task'] = task_match.group(1).strip()
|
||||
|
||||
# 提取每日工作
|
||||
daily_section = re.search(r'## 本周工作内容\s*\n([\s\S]*?)(?:\n##|\Z)', content)
|
||||
if daily_section:
|
||||
daily_text = daily_section.group(1)
|
||||
# 匹配每日内容
|
||||
days = re.findall(r'###?\s*(?:周 [一二三四五六日]|(\d{4}-\d{2}-\d{2}))\s*\n([\s\S]*?)(?=###?|$)', daily_text)
|
||||
for date_str, work_text in days:
|
||||
if date_str:
|
||||
items = re.findall(r'^\s*[-•*]\s*(.+)$', work_text, re.MULTILINE)
|
||||
data['daily_work'][date_str] = items
|
||||
|
||||
# 提取存在问题
|
||||
problem_section = re.search(r'## 存在问题\s*\n([\s\S]*?)(?:\n##|\Z)', content)
|
||||
if problem_section:
|
||||
problem_text = problem_section.group(1).strip()
|
||||
if problem_text and problem_text != '无':
|
||||
data['problems'] = [line.strip() for line in problem_text.split('\n') if line.strip()]
|
||||
|
||||
return data
|
||||
|
||||
|
||||
def sync_to_mysql(report_data, dry_run=False):
|
||||
"""同步到 MySQL 数据库"""
|
||||
try:
|
||||
import mysql.connector
|
||||
except ImportError:
|
||||
print("❌ 需要安装 mysql-connector-python: pip3 install mysql-connector-python")
|
||||
return False
|
||||
|
||||
env = load_env()
|
||||
db_config = {
|
||||
'host': 'localhost',
|
||||
'user': 'root',
|
||||
'password': env.get('MYSQL_ROOT_PASSWORD', ''),
|
||||
'database': 'resume',
|
||||
'charset': 'utf8mb4'
|
||||
}
|
||||
|
||||
try:
|
||||
conn = mysql.connector.connect(**db_config)
|
||||
cursor = conn.cursor()
|
||||
|
||||
# 插入或更新周报元数据
|
||||
sql = """
|
||||
INSERT INTO weekly_reports
|
||||
(year, week_number, start_date, end_date, project, main_task, file_path, has_problems, word_count)
|
||||
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
|
||||
ON DUPLICATE KEY UPDATE
|
||||
main_task = VALUES(main_task),
|
||||
file_path = VALUES(file_path),
|
||||
has_problems = VALUES(has_problems),
|
||||
word_count = VALUES(word_count),
|
||||
updated_at = CURRENT_TIMESTAMP
|
||||
"""
|
||||
|
||||
values = (
|
||||
report_data['year'],
|
||||
report_data['week_number'],
|
||||
report_data['start_date'],
|
||||
report_data['end_date'],
|
||||
report_data['project'],
|
||||
report_data['main_task'],
|
||||
report_data['file_path'],
|
||||
1 if report_data['problems'] else 0,
|
||||
report_data['word_count']
|
||||
)
|
||||
|
||||
if dry_run:
|
||||
print(f"📝 [DRY RUN] 会执行：{sql % values}")
|
||||
else:
|
||||
cursor.execute(sql, values)
|
||||
report_id = cursor.lastrowid
|
||||
|
||||
# 同步每日工作明细
|
||||
if report_id:
|
||||
for work_date, items in report_data['daily_work'].items():
|
||||
day_of_week = datetime.strptime(work_date, '%Y-%m-%d').weekday() + 1
|
||||
items_json = json.dumps(items, ensure_ascii=False)
|
||||
|
||||
daily_sql = """
|
||||
INSERT INTO weekly_report_daily
|
||||
(report_id, day_of_week, work_date, work_items)
|
||||
VALUES (%s, %s, %s, %s)
|
||||
ON DUPLICATE KEY UPDATE work_items = VALUES(work_items)
|
||||
"""
|
||||
|
||||
cursor.execute(daily_sql, (report_id, day_of_week, work_date, items_json))
|
||||
|
||||
# 同步问题记录
|
||||
if report_data['problems']:
|
||||
for problem in report_data['problems']:
|
||||
problem_sql = """
|
||||
INSERT INTO weekly_report_problems
|
||||
(report_id, problem_description)
|
||||
VALUES (%s, %s)
|
||||
"""
|
||||
cursor.execute(problem_sql, (report_id, problem))
|
||||
|
||||
conn.commit()
|
||||
print(f"✅ 同步成功：{report_data['file_path']}")
|
||||
|
||||
cursor.close()
|
||||
conn.close()
|
||||
return True
|
||||
|
||||
except mysql.connector.Error as err:
|
||||
print(f"❌ 数据库错误：{err}")
|
||||
return False
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description='周报数据库同步工具')
|
||||
group = parser.add_mutually_exclusive_group(required=True)
|
||||
group.add_argument('--all', action='store_true', help='同步所有周报')
|
||||
group.add_argument('--file', type=str, help='同步单个文件')
|
||||
group.add_argument('--week', type=str, help='同步指定周，如 2026-W35')
|
||||
group.add_argument('--dry-run', action='store_true', help='预览执行，不实际写入数据库')
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
files_to_sync = []
|
||||
|
||||
if args.file:
|
||||
files_to_sync = [Path(args.file)]
|
||||
elif args.week:
|
||||
match = re.match(r'(\d{4})-W(\d{2})', args.week)
|
||||
if match:
|
||||
year, week = match.groups()
|
||||
pattern = f"{year}-W{week}-*.md"
|
||||
files_to_sync = list(REPORTS_DIR.glob(f"{year}/{pattern}"))
|
||||
elif args.all:
|
||||
# 查找所有 Markdown 周报文件
|
||||
for year_dir in REPORTS_DIR.iterdir():
|
||||
if year_dir.is_dir() and year_dir.name.isdigit():
|
||||
files_to_sync.extend(year_dir.glob("*.md"))
|
||||
|
||||
if not files_to_sync:
|
||||
print("⚠️ 未找到需要同步的文件")
|
||||
return 1
|
||||
|
||||
print(f"📂 找到 {len(files_to_sync)} 个文件待同步")
|
||||
|
||||
success_count = 0
|
||||
for filepath in sorted(files_to_sync):
|
||||
if not filepath.exists():
|
||||
print(f"⚠️ 文件不存在：{filepath}")
|
||||
continue
|
||||
|
||||
print(f"\n📄 处理：{filepath.name}")
|
||||
report_data = parse_markdown_report(filepath)
|
||||
|
||||
if not report_data['year'] or not report_data['week_number']:
|
||||
print(f"⚠️ 无法解析年份/周数，跳过")
|
||||
continue
|
||||
|
||||
if args.dry_run:
|
||||
print(f" 年份：{report_data['year']}, 周数：{report_data['week_number']}")
|
||||
print(f" 日期：{report_data['start_date']} ~ {report_data['end_date']}")
|
||||
print(f" 项目：{report_data['project']}")
|
||||
print(f" 任务：{report_data['main_task']}")
|
||||
print(f" 问题：{len(report_data['problems'])} 个")
|
||||
success_count += 1
|
||||
else:
|
||||
if sync_to_mysql(report_data):
|
||||
success_count += 1
|
||||
|
||||
print(f"\n{'✅' if success_count == len(files_to_sync) else '⚠️'} 完成：{success_count}/{len(files_to_sync)} 个文件")
|
||||
return 0 if success_count == len(files_to_sync) else 1
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
exit(main())
|
||||
Reference in New Issue
Block a user