269 lines
11 KiB
Python
269 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""解析研究院-维云智造 OKR 绩效 Excel（杨轩），归档到 MySQL resume 库。
|
||
支持新旧两种模板：
|
||
- 旧模板（202601 等，多 sheet）：表头行4，目标行5起
|
||
- 新模板（8月起，单 sheet OKR考核）：表头行5-6，目标行8起（跳过示例行）
|
||
|
||
用法：
|
||
python3 parse_okr_to_mysql.py # 解析 inbound 最新 OKR xlsx
|
||
python3 parse_okr_to_mysql.py --month 202601 # 指定考核月
|
||
python3 parse_okr_to_mysql.py --file <xlsx> # 指定文件
|
||
python3 parse_okr_to_mysql.py --dry-run # 预览不写入
|
||
"""
|
||
import glob
|
||
import os
|
||
import re
|
||
import sys
|
||
import openpyxl
|
||
|
||
MYSQL_DB = "resume"
|
||
MYSQL_HOST = "127.0.0.1"
|
||
MYSQL_PORT = 3306
|
||
MYSQL_USER = "root"
|
||
|
||
|
||
def get_mysql_pwd():
|
||
env_path = "/home/yangxuan/.openclaw/.env"
|
||
for line in open(env_path):
|
||
line = line.strip()
|
||
if line.startswith("MYSQL_PWD="):
|
||
return line.split("=", 1)[1].strip()
|
||
return ""
|
||
|
||
|
||
def parse_info(text):
|
||
"""从人员信息行解析 被考核人/岗位/上级/周期/时间（新旧模板通用）"""
|
||
info = {}
|
||
m = re.search(r"(?:被考核人|姓名)[：:]\s*(\S+)", text)
|
||
info["full_name"] = m.group(1) if m else ""
|
||
m = re.search(r"岗位[：:]\s*(\S+)", text)
|
||
info["position"] = m.group(1) if m else ""
|
||
m = re.search(r"(?:直属上级|直接上级)[：:]\s*(\S+)", text)
|
||
info["supervisor"] = m.group(1) if m else ""
|
||
m = re.search(r"考核周期[：:]\s*([^\s]+)", text)
|
||
info["period"] = m.group(1).strip() if m else ""
|
||
m = re.search(r"考核时间[：:]\s*(\S+)", text)
|
||
info["assess_date"] = m.group(1).strip() if m else ""
|
||
return info
|
||
|
||
|
||
def detect_template(ws):
|
||
"""识别模板。'new'=8月新模板，'old'=202601旧模板。"""
|
||
# 新模板：行5-6 含"员工本人填报"或"关键结果"
|
||
for r in (5, 6):
|
||
for c in range(3, 12):
|
||
v = ws.cell(row=r, column=c).value
|
||
if v and ("员工本人填报" in str(v) or "关键结果" in str(v)):
|
||
return "new"
|
||
# 旧模板：行4 含"目标O"或"关键绩效指标"
|
||
for c in range(3, 12):
|
||
v = ws.cell(row=4, column=c).value
|
||
if v and ("目标O" in str(v) or "关键绩效指标" in str(v)):
|
||
return "old"
|
||
return "new"
|
||
|
||
|
||
def parse_sheet_new(ws):
|
||
"""解析 8 月新模板（OKR考核 sheet）"""
|
||
b3 = ws["B3"].value or ""
|
||
info = parse_info(b3)
|
||
# 月度评价：F15 是“月度评价”标签，分数在其右侧（G15 等），上级可能未填
|
||
monthly_rating = None
|
||
for coord in ("G15", "H15", "I15"):
|
||
v = ws[coord].value
|
||
if isinstance(v, (int, float)):
|
||
monthly_rating = float(v)
|
||
break
|
||
objectives = []
|
||
for row in range(8, 40): # 目标行8起
|
||
seq = ws.cell(row=row, column=2).value # B 序号
|
||
obj = ws.cell(row=row, column=3).value # C 任务工作
|
||
krs = ws.cell(row=row, column=4).value # D 关键结果
|
||
compl = ws.cell(row=row, column=5).value # E 完成情况
|
||
score = ws.cell(row=row, column=7).value # G 评分
|
||
weight = ws.cell(row=row, column=9).value # I 权重
|
||
wscore = ws.cell(row=row, column=10).value # J 得分
|
||
if isinstance(obj, str) and "月度工作补充" in obj:
|
||
break
|
||
if not isinstance(seq, (int, float)):
|
||
continue
|
||
objectives.append({
|
||
"seq": int(seq),
|
||
"objective": (obj or "").strip(),
|
||
"krs": (krs or "").strip(),
|
||
"completion": (compl or "").strip(),
|
||
"score": float(score) if isinstance(score, (int, float)) else None,
|
||
"weight": float(weight) if isinstance(weight, (int, float)) else None,
|
||
"weighted_score": float(wscore) if isinstance(wscore, (int, float)) else None,
|
||
})
|
||
return {**info, "monthly_rating": monthly_rating, "objectives": objectives}
|
||
|
||
|
||
def parse_sheet_old(ws):
|
||
"""解析 202601 旧模板"""
|
||
b2 = ws["B2"].value or ""
|
||
info = parse_info(b2)
|
||
monthly_rating = None
|
||
for coord in ("J12", "B12"):
|
||
v = ws[coord].value
|
||
if isinstance(v, (int, float)):
|
||
monthly_rating = float(v)
|
||
break
|
||
objectives = []
|
||
for row in range(5, 11): # 行5-10 对应序号1-6
|
||
seq = ws.cell(row=row, column=2).value # B 序号
|
||
obj = ws.cell(row=row, column=3).value # C 目标O
|
||
krs = ws.cell(row=row, column=5).value # E KR
|
||
compl = ws.cell(row=row, column=6).value # F 完成情况
|
||
score = ws.cell(row=row, column=8).value # H 评分
|
||
weight = ws.cell(row=row, column=9).value # I 权重
|
||
wscore = ws.cell(row=row, column=10).value # J 加权
|
||
if seq is None and obj is None:
|
||
continue
|
||
objectives.append({
|
||
"seq": seq if isinstance(seq, (int, float)) else row - 4,
|
||
"objective": (obj or "").strip(),
|
||
"krs": (krs or "").strip(),
|
||
"completion": (compl or "").strip(),
|
||
"score": float(score) if isinstance(score, (int, float)) else None,
|
||
"weight": float(weight) if isinstance(weight, (int, float)) else None,
|
||
"weighted_score": float(wscore) if isinstance(wscore, (int, float)) else None,
|
||
})
|
||
return {**info, "monthly_rating": monthly_rating, "objectives": objectives}
|
||
|
||
|
||
def month_from_period(period, sheet_name, fallback):
|
||
"""从考核周期/sheet名 确定 assess_month（YYYYMM）
|
||
跨月周期（如“7月27日-8月27日”）取后段月份（考核期末）
|
||
"""
|
||
# 末段月份优先（跨月周期取考核期末，如 7月27日-8月27日 → 8月）
|
||
m = re.search(r"-(\d{1,2})月(\d{1,2})日", period or "")
|
||
if m:
|
||
return f"2026{int(m.group(1)):02d}"
|
||
m = re.search(r"(\d{4})年(\d{1,2})月", period or "")
|
||
if m:
|
||
return f"{m.group(1)}{int(m.group(2)):02d}"
|
||
m = re.search(r"(\d{4})\s*(\d{2})", sheet_name or "")
|
||
if m:
|
||
return f"{m.group(1)}{int(m.group(2)):02d}"
|
||
m = re.search(r"(\d{1,2})月", period or "")
|
||
if m:
|
||
return f"2026{int(m.group(1)):02d}"
|
||
return fallback or ""
|
||
|
||
|
||
def process_sheet(wb, sheet_name, xls, dry_run=False):
|
||
"""解析并归档单个 sheet，返回 (assess_month, 目标数) 或 None"""
|
||
ws = wb[sheet_name]
|
||
tpl = detect_template(ws)
|
||
print(f"📑 Sheet：{sheet_name} | 识别模板：{'新模板(8月起)' if tpl == 'new' else '旧模板(202601)'}")
|
||
data = parse_sheet_new(ws) if tpl == "new" else parse_sheet_old(ws)
|
||
assess_month = month_from_period(data.get("period"), sheet_name, None)
|
||
if not assess_month:
|
||
print(f" ⚠️ 无法确定考核月，跳过")
|
||
return None
|
||
if not data["objectives"]:
|
||
print(f" ⚠️ {sheet_name} 无目标数据，跳过")
|
||
return None
|
||
|
||
print(f" 被考核人：{data['full_name']} | 岗位：{data['position']} | 上级：{data['supervisor']}")
|
||
print(f" 考核周期：{data.get('period')} | assess_month={assess_month} | 月度评价：{data.get('monthly_rating')}")
|
||
print(f" 目标数：{len(data['objectives'])}")
|
||
for o in data["objectives"]:
|
||
print(f" {o['seq']}. {o['objective'][:28]} | 权重{o['weight']} 评分{o.get('score')}")
|
||
|
||
if dry_run:
|
||
print(" (dry-run 未写入)")
|
||
return assess_month
|
||
|
||
pwd = get_mysql_pwd()
|
||
if not pwd:
|
||
print("❌ 未找到 MYSQL_PWD")
|
||
return None
|
||
import pymysql
|
||
conn = pymysql.connect(host=MYSQL_HOST, port=MYSQL_PORT, user=MYSQL_USER,
|
||
password=pwd, database=MYSQL_DB, charset="utf8mb4")
|
||
cur = conn.cursor()
|
||
cur.execute("SELECT id FROM okr_monthly_records WHERE assess_month=%s AND full_name=%s",
|
||
(assess_month, data["full_name"]))
|
||
for (rid,) in cur.fetchall():
|
||
cur.execute("DELETE FROM okr_objectives WHERE record_id=%s", (rid,))
|
||
cur.execute("DELETE FROM okr_monthly_records WHERE id=%s", (rid,))
|
||
cur.execute(
|
||
"INSERT INTO okr_monthly_records (assess_month, full_name, position, supervisor, assess_date, monthly_rating, source_file) "
|
||
"VALUES (%s,%s,%s,%s,%s,%s,%s)",
|
||
(assess_month, data["full_name"], data["position"], data["supervisor"],
|
||
data.get("assess_date", ""), data.get("monthly_rating"), os.path.basename(xls)))
|
||
rid = cur.lastrowid
|
||
for o in data["objectives"]:
|
||
cur.execute(
|
||
"INSERT INTO okr_objectives (record_id, seq, objective, krs, completion, score, weight, weighted_score) "
|
||
"VALUES (%s,%s,%s,%s,%s,%s,%s,%s)",
|
||
(rid, o["seq"], o["objective"], o["krs"], o["completion"], o.get("score"),
|
||
o["weight"], o["weighted_score"]))
|
||
conn.commit()
|
||
cur.close()
|
||
conn.close()
|
||
print(f" ✅ 已入库（{len(data['objectives'])} 项目标）")
|
||
return assess_month
|
||
|
||
|
||
def main():
|
||
dry_run = "--dry-run" in sys.argv
|
||
month_filter = None
|
||
if "--month" in sys.argv:
|
||
month_filter = sys.argv[sys.argv.index("--month") + 1]
|
||
# 定位 Excel
|
||
xls = None
|
||
for i, a in enumerate(sys.argv[1:]):
|
||
if a == "--file" and i + 2 < len(sys.argv):
|
||
xls = sys.argv[i + 2]
|
||
elif a.endswith(".xlsx"):
|
||
xls = a
|
||
if xls is None or not os.path.exists(xls):
|
||
cand = sorted(glob.glob("/home/yangxuan/.openclaw/media/inbound/*OKR*.xlsx"))
|
||
xls = cand[-1] if cand else None
|
||
if not xls or not os.path.exists(xls):
|
||
print("❌ 未找到 OKR Excel 文件")
|
||
return
|
||
print(f"📄 解析文件：{os.path.basename(xls)}")
|
||
|
||
wb = openpyxl.load_workbook(xls, data_only=True)
|
||
|
||
# 决定要处理的 sheet 列表
|
||
if month_filter:
|
||
# 支持范围：如 202602-202607，或单个 202601，或逗号分隔
|
||
if "-" in month_filter and len(month_filter) == 13:
|
||
start_m, end_m = month_filter.split("-")
|
||
start_i, end_i = int(start_m), int(end_m)
|
||
months = [str(m) for m in range(start_i, end_i + 1)]
|
||
sheets = [n for n in wb.sheetnames if n in months]
|
||
else:
|
||
parts = [p.strip() for p in month_filter.replace("，", ",").split(",")]
|
||
sheets = [n for n in wb.sheetnames if n in parts]
|
||
if not sheets:
|
||
print(f"❌ 文件里未找到指定月份 sheet：{month_filter}")
|
||
return
|
||
else:
|
||
# 无 --month：优先 OKR考核（新模板单 sheet）
|
||
sheets = [n for n in wb.sheetnames if "OKR考核" in n]
|
||
if not sheets:
|
||
sheets = wb.sheetnames[:1]
|
||
|
||
print(f"\n📦 待处理 sheet：{sheets}")
|
||
done = []
|
||
for sn in sheets:
|
||
print()
|
||
r = process_sheet(wb, sn, xls, dry_run)
|
||
if r:
|
||
done.append(r)
|
||
if dry_run:
|
||
print(f"\n🔍 dry-run 完成，共 {len(sheets)} 个 sheet（未写入）")
|
||
else:
|
||
print(f"\n🎉 已完成 {len(done)} 个月 OKR 归档：{done}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|