266 lines
9.1 KiB
Python
266 lines
9.1 KiB
Python
#!/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())
|