201 lines
7.4 KiB
Python
201 lines
7.4 KiB
Python
"""为定开项目补充『入库时间』字段并回填历史数据。
|
||
|
||
取值优先级（当前口径 = 原始数据的月/周第一天）：
|
||
1. 原始 Excel 里该项目**最早出现**的周期 → 换算成该周期第一天
|
||
- 《周工作情况总结》中「定制开发」块下的子项目行
|
||
- 《项目明细表》按周分组的台账行（该列是稀疏填充，需向下继承）
|
||
2. 该项目在 custom_updates 里最早一条记录的周期
|
||
3. 项目自身 period_id 对应周期
|
||
4. 兜底：记录创建日期
|
||
|
||
实现说明：
|
||
- `Base.metadata.create_all()` 只建新表，**不会给已有表加列**，故此处显式 ALTER TABLE。
|
||
- Excel 里是平移前的年份，这里按 (月, 周) 匹配现有周期并取最早的一年，避免年份口径耦合。
|
||
|
||
用法：
|
||
python scripts/add_entry_date.py --dry-run
|
||
python scripts/add_entry_date.py
|
||
python scripts/add_entry_date.py --force
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import sys
|
||
from collections import defaultdict
|
||
from pathlib import Path
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||
|
||
import openpyxl # noqa: E402
|
||
from sqlalchemy import inspect, text # noqa: E402
|
||
|
||
from app.constants import normalize_custom_name, parse_period_label, period_first_day # noqa: E402
|
||
from app.db import SessionLocal, engine # noqa: E402
|
||
from app.models import ( # noqa: E402
|
||
Base,
|
||
CustomProject,
|
||
CustomUpdate,
|
||
Period,
|
||
)
|
||
|
||
DEFAULT_XLSX = "/Users/jiliu/WorkSpace/定开管理/软件开发部管理工作执行表.xlsx"
|
||
COLUMN = "entry_date"
|
||
|
||
|
||
def _clean(v):
|
||
if v is None:
|
||
return None
|
||
s = str(v).strip()
|
||
return s or None
|
||
|
||
|
||
def ensure_column() -> bool:
|
||
Base.metadata.create_all(bind=engine)
|
||
insp = inspect(engine)
|
||
if "custom_projects" not in insp.get_table_names():
|
||
return False
|
||
cols = [c["name"] for c in insp.get_columns("custom_projects")]
|
||
if COLUMN in cols:
|
||
return False
|
||
with engine.begin() as conn:
|
||
conn.execute(text(f"ALTER TABLE custom_projects ADD COLUMN {COLUMN} VARCHAR(16)"))
|
||
print(f"[ok] 已新增字段 custom_projects.{COLUMN}")
|
||
return True
|
||
|
||
|
||
def load_excel_periods(xlsx: str) -> dict[str, tuple[int, int]]:
|
||
"""从原始 Excel 提取『项目名 → (月, 周)』。
|
||
|
||
优先级：**《项目明细表》优先**（项目正式登记入册的地方），
|
||
其次才是《周工作情况总结》「定制开发」块（该项目在办的周次）。
|
||
因此这里先处理明细表，用 setdefault 占位。
|
||
"""
|
||
wb = openpyxl.load_workbook(xlsx, data_only=True, read_only=True)
|
||
found: dict[str, tuple[int, int]] = {}
|
||
|
||
# 1) 项目明细表：周期列向下继承（优先来源）
|
||
if "项目明细表" in wb.sheetnames:
|
||
ws2 = wb["项目明细表"]
|
||
last_label = None
|
||
for r in ws2.iter_rows(min_row=1, values_only=True):
|
||
cells = list(r) + [None] * 7
|
||
label, name = _clean(cells[1]), _clean(cells[2])
|
||
if label:
|
||
last_label = label
|
||
if not name or name in ("项目", "项目名称"):
|
||
continue
|
||
parsed = parse_period_label(last_label, 0) if last_label else None
|
||
if not parsed:
|
||
continue
|
||
clean_name, _amt = normalize_custom_name(name)
|
||
if clean_name:
|
||
found.setdefault(clean_name, (parsed[1], parsed[2]))
|
||
|
||
# 2) 周工作情况总结：「定制开发」块下的子项目（补充来源）
|
||
if "周工作情况总结" in wb.sheetnames:
|
||
ws = wb["周工作情况总结"]
|
||
cur_period = None
|
||
cur_group = None
|
||
for r in ws.iter_rows(min_row=3, values_only=True):
|
||
cols = list(r) + [None] * 6
|
||
time_v, work_v, desc_v = _clean(cols[1]), _clean(cols[2]), _clean(cols[3])
|
||
if time_v and time_v != "时间":
|
||
parsed = parse_period_label(time_v, 0)
|
||
cur_period = (parsed[1], parsed[2]) if parsed else None
|
||
cur_group = None
|
||
if work_v:
|
||
cur_group = work_v
|
||
if cur_period and cur_group == "定制开发" and desc_v:
|
||
name, _amt = normalize_custom_name(desc_v)
|
||
if name:
|
||
found.setdefault(name, cur_period)
|
||
|
||
return found
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("xlsx", nargs="?", default=DEFAULT_XLSX)
|
||
ap.add_argument("--dry-run", action="store_true")
|
||
ap.add_argument("--force", action="store_true", help="覆盖已有入库时间")
|
||
args = ap.parse_args()
|
||
|
||
ensure_column()
|
||
|
||
db = SessionLocal()
|
||
try:
|
||
periods = db.query(Period).all()
|
||
by_mw: dict[tuple[int, int], list[Period]] = defaultdict(list)
|
||
for p in periods:
|
||
by_mw[(p.month, p.week)].append(p)
|
||
for lst in by_mw.values():
|
||
lst.sort(key=lambda x: x.sort_key) # 取最早的一年
|
||
|
||
excel_map: dict[str, tuple[int, int]] = {}
|
||
try:
|
||
excel_map = load_excel_periods(args.xlsx)
|
||
print(f"[i] 从原始表格读到 {len(excel_map)} 个项目的周期归属")
|
||
except FileNotFoundError:
|
||
print(f"[!] 找不到 Excel：{args.xlsx}，跳过回查")
|
||
|
||
updated = skipped = 0
|
||
src_count = defaultdict(int)
|
||
samples = []
|
||
|
||
for cp in db.query(CustomProject).all():
|
||
if cp.entry_date and not args.force:
|
||
skipped += 1
|
||
continue
|
||
|
||
mw = excel_map.get(cp.name)
|
||
period = by_mw[mw][0] if mw and by_mw.get(mw) else None
|
||
source = "原始表格"
|
||
|
||
if period is None:
|
||
pid = (
|
||
db.query(CustomUpdate.period_id)
|
||
.filter(CustomUpdate.custom_project_id == cp.id)
|
||
.order_by(CustomUpdate.period_id)
|
||
.limit(1)
|
||
.scalar()
|
||
) or cp.period_id
|
||
period = next((p for p in periods if p.id == pid), None)
|
||
source = "周期记录"
|
||
|
||
if period is not None:
|
||
value = period_first_day(period.year, period.month, period.week)
|
||
origin = f"{period.year}年{period.month}月第{period.week}周"
|
||
else:
|
||
value = cp.created_at.date().isoformat() if cp.created_at else None
|
||
origin = "记录创建日期"
|
||
source = "创建日期"
|
||
|
||
cp.entry_date = value
|
||
updated += 1
|
||
src_count[source] += 1
|
||
if len(samples) < 10:
|
||
samples.append((cp.name, origin, value))
|
||
|
||
if args.dry_run:
|
||
db.rollback()
|
||
print(f"[dry-run] 将回填 {updated} 个项目（跳过已有 {skipped} 个）")
|
||
print(" 数据来源：", dict(src_count))
|
||
for name, origin, value in samples:
|
||
print(f" · {name[:32]:34s} {origin:>18s} → {value}")
|
||
return
|
||
|
||
db.commit()
|
||
print(f"[ok] 已回填 {updated} 个项目（跳过已有 {skipped} 个）")
|
||
print(" 数据来源：", dict(src_count))
|
||
for name, origin, value in samples:
|
||
print(f" · {name[:32]:34s} {origin:>18s} → {value}")
|
||
with_value = db.query(CustomProject).filter(CustomProject.entry_date.isnot(None)).count()
|
||
total = db.query(CustomProject).count()
|
||
print(f"[ok] 覆盖率：{with_value}/{total}")
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|