281 lines
9.2 KiB
Python
281 lines
9.2 KiB
Python
"""把《软件开发部管理工作执行表.xlsx》导入数据库。
|
||
|
||
用法：
|
||
python scripts/import_excel.py [excel路径] [--reset]
|
||
|
||
解析规则：
|
||
- 「周工作情况总结」：时间(周期) → 工作内容(项目) → 内容说明(子任务) → 完成情况(执行记录)
|
||
* 三列均为稀疏填充，逐级向下继承（forward fill）
|
||
* 仅填「完成情况」的行视为上一条记录的续行，追加到同一条执行记录
|
||
- 「项目明细表」：客户项目台账（区域 / 金额 / 类别 / 状态说明）
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||
|
||
import argparse # noqa: E402
|
||
|
||
import openpyxl # noqa: E402
|
||
|
||
from app.ai_parser import _guess_progress, _guess_status # noqa: E402
|
||
from app.constants import LEDGER_CATEGORIES, parse_period_label # noqa: E402
|
||
from app.db import SessionLocal, engine # noqa: E402
|
||
from app.models import ( # noqa: E402
|
||
Base,
|
||
Category,
|
||
LedgerItem,
|
||
Period,
|
||
Project,
|
||
Task,
|
||
Update,
|
||
)
|
||
|
||
DEFAULT_XLSX = "/Users/jiliu/WorkSpace/定开管理/软件开发部管理工作执行表.xlsx"
|
||
|
||
# 重点工作流（自动打上重点标记）
|
||
KEY_PROJECTS = {"NEX云桌面", "AI Agent", "智飞无人机平台", "定制开发", "云屏"}
|
||
|
||
# 分类归属启发式
|
||
CATEGORY_RULES = [
|
||
(("无人机", "智飞"), "智能装备"),
|
||
(("AI", "智能体", "Agent", "大模型"), "AI 能力"),
|
||
(("NEX", "云桌面", "云屏", "平台开发", "UniFlow"), "平台产品"),
|
||
(("定制", "交付", "项目支撑"), "定制项目"),
|
||
]
|
||
|
||
|
||
def guess_category(name: str) -> str | None:
|
||
for keys, cat in CATEGORY_RULES:
|
||
if any(k.lower() in name.lower() for k in keys):
|
||
return cat
|
||
return "内部运营"
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("xlsx", nargs="?", default=DEFAULT_XLSX)
|
||
ap.add_argument("--reset", action="store_true", help="清空业务表后重新导入")
|
||
args = ap.parse_args()
|
||
|
||
Base.metadata.create_all(bind=engine)
|
||
db = SessionLocal()
|
||
|
||
if args.reset:
|
||
for model in (Update, Task, LedgerItem, Project, Period):
|
||
db.query(model).delete()
|
||
db.commit()
|
||
print("[i] 已清空历史业务数据")
|
||
|
||
cats = {c.name: c for c in db.query(Category).all()}
|
||
|
||
wb = openpyxl.load_workbook(args.xlsx, data_only=True, read_only=True)
|
||
|
||
# ---------------------------------------------------------- 周工作情况总结
|
||
ws = wb["周工作情况总结"]
|
||
rows = list(ws.iter_rows(min_row=3, values_only=True))
|
||
|
||
year_hint = 2024
|
||
prev_month = None
|
||
period: Period | None = None
|
||
project: Project | None = None
|
||
task: Task | None = None
|
||
update: Update | None = None
|
||
|
||
def get_period(label: str) -> Period | None:
|
||
nonlocal year_hint, prev_month
|
||
parsed = parse_period_label(label, year_hint)
|
||
if not parsed:
|
||
return None
|
||
year, month, week = parsed
|
||
if prev_month is not None and month < prev_month:
|
||
year_hint += 1
|
||
year = year_hint
|
||
prev_month = month
|
||
sort_key = year * 10000 + month * 100 + week
|
||
p = db.query(Period).filter(Period.sort_key == sort_key).one_or_none()
|
||
if not p:
|
||
p = Period(
|
||
label=str(label).strip(),
|
||
kind="week",
|
||
year=year,
|
||
month=month,
|
||
week=week,
|
||
sort_key=sort_key,
|
||
)
|
||
db.add(p)
|
||
db.flush()
|
||
return p
|
||
|
||
def get_project(name: str) -> Project:
|
||
p = db.query(Project).filter(Project.name == name).one_or_none()
|
||
if p:
|
||
return p
|
||
cat_name = guess_category(name)
|
||
p = Project(
|
||
name=name,
|
||
category_id=cats[cat_name].id if cat_name in cats else None,
|
||
is_key=name in KEY_PROJECTS,
|
||
status="in_progress",
|
||
progress=0,
|
||
)
|
||
db.add(p)
|
||
db.flush()
|
||
return p
|
||
|
||
n_period = n_project = n_task = n_update = 0
|
||
|
||
for r in rows:
|
||
cells = list(r[:6]) + [None] * (6 - min(6, len(r)))
|
||
_, time_v, work_v, desc_v, done_v, _remark = cells
|
||
|
||
def clean(v):
|
||
if v is None:
|
||
return None
|
||
s = str(v).strip()
|
||
return s or None
|
||
|
||
time_v, work_v, desc_v, done_v = clean(time_v), clean(work_v), clean(desc_v), clean(done_v)
|
||
|
||
if time_v and time_v != "时间":
|
||
period = get_period(time_v)
|
||
if period:
|
||
n_period += 1
|
||
project = task = update = None
|
||
|
||
if work_v:
|
||
project = get_project(work_v)
|
||
n_project += 1
|
||
task = update = None
|
||
|
||
if not period or not project:
|
||
continue
|
||
|
||
if desc_v:
|
||
task = (
|
||
db.query(Task)
|
||
.filter(Task.project_id == project.id, Task.name == desc_v[:255])
|
||
.one_or_none()
|
||
)
|
||
if not task:
|
||
task = Task(project_id=project.id, name=desc_v[:255], sort_order=n_task)
|
||
db.add(task)
|
||
db.flush()
|
||
n_task += 1
|
||
update = None
|
||
|
||
if done_v:
|
||
if update is None and task is not None:
|
||
update = (
|
||
db.query(Update)
|
||
.filter(
|
||
Update.project_id == project.id,
|
||
Update.task_id == task.id,
|
||
Update.period_id == period.id,
|
||
)
|
||
.one_or_none()
|
||
)
|
||
if not update:
|
||
update = Update(
|
||
project_id=project.id,
|
||
task_id=task.id,
|
||
period_id=period.id,
|
||
content=done_v,
|
||
progress=_guess_progress(done_v),
|
||
status=_guess_status(done_v),
|
||
source="import",
|
||
)
|
||
db.add(update)
|
||
db.flush()
|
||
n_update += 1
|
||
else:
|
||
update.content = (update.content or "") + "\n" + done_v
|
||
elif update is not None:
|
||
update.content = (update.content or "") + "\n" + done_v
|
||
|
||
db.commit()
|
||
|
||
# 回填项目最新进度 / 状态；长期无更新的历史一次性事项自动归档
|
||
recent_ids = [
|
||
p.id for p in db.query(Period).order_by(Period.sort_key.desc()).limit(10).all()
|
||
]
|
||
for p in db.query(Project).all():
|
||
ups = (
|
||
db.query(Update)
|
||
.filter(Update.project_id == p.id)
|
||
.order_by(Update.period_id.desc(), Update.id.desc())
|
||
.limit(20)
|
||
.all()
|
||
)
|
||
prs = [u.progress for u in ups if u.progress is not None]
|
||
if prs:
|
||
p.progress = max(prs)
|
||
if ups:
|
||
p.status = ups[0].status
|
||
# 进度未满 100% 时不标记为已完成，避免与进度条矛盾
|
||
if p.status == "done" and (p.progress or 0) < 100:
|
||
p.status = "in_progress"
|
||
if not p.is_key and not any(u.period_id in recent_ids for u in ups):
|
||
p.archived = True
|
||
db.commit()
|
||
|
||
# ---------------------------------------------------------- 项目明细表
|
||
ws2 = wb["项目明细表"]
|
||
n_ledger = 0
|
||
all_periods = db.query(Period).all()
|
||
period_index = {(p.month, p.week): p for p in all_periods}
|
||
|
||
for r in ws2.iter_rows(min_row=1, values_only=True):
|
||
cells = list(r) + [None] * (7 - min(7, len(r)))
|
||
_, p_label, name, region, amount, category, note = cells[:7]
|
||
if not name or not str(name).strip():
|
||
continue
|
||
name = str(name).strip()
|
||
if name in ("项目", "项目名称"):
|
||
continue
|
||
try:
|
||
amt = float(amount) if isinstance(amount, (int, float)) else None
|
||
except (TypeError, ValueError):
|
||
amt = None
|
||
cat = str(category).strip() if category else None
|
||
if cat and cat not in LEDGER_CATEGORIES:
|
||
cat = None
|
||
label = str(p_label).strip() if p_label else None
|
||
parsed = parse_period_label(label, 2025) if label else None
|
||
pid = None
|
||
if parsed:
|
||
pid = period_index.get((parsed[1], parsed[2]))
|
||
pid = pid.id if pid else None
|
||
if db.query(LedgerItem).filter(
|
||
LedgerItem.name == name,
|
||
LedgerItem.period_label == label,
|
||
LedgerItem.category == cat,
|
||
).one_or_none():
|
||
continue
|
||
db.add(
|
||
LedgerItem(
|
||
name=name,
|
||
region=str(region).strip() if region else None,
|
||
amount=amt,
|
||
category=cat,
|
||
status_note=str(note).strip() if note else None,
|
||
period_label=label,
|
||
period_id=pid,
|
||
)
|
||
)
|
||
n_ledger += 1
|
||
db.commit()
|
||
|
||
print(
|
||
f"[ok] 导入完成：周期 {len(db.query(Period).all())} 个 / 项目 {len(db.query(Project).all())} 个 / "
|
||
f"子任务 {n_task} 个 / 执行记录 {len(db.query(Update).all())} 条 / 台账 {n_ledger} 条"
|
||
)
|
||
db.close()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|