AiDOPWarehouse/ai-dop-platform/tools/fill_weekly_plan_from_person.py

# -*- coding: utf-8 -*-
"""
将「每周详细计划_人员」同步到「每周详细计划_时间」（保持时间页版式），
并更新日报「20260413」中「二、项目周进度」为 W2（2026-04-13 所在周），
补充「三、项目日进度」与合并区风险/预案文案。
"""
from __future__ import annotations

import re
import shutil
from datetime import datetime
from pathlib import Path

import openpyxl
from openpyxl.utils import range_boundaries

SRC = Path(r"d:\Projects\Ai-DOP\AI-DOP项目总计划_20260413.xlsx")
BACKUP = SRC.with_name(SRC.stem + "_before_fill_" + datetime.now().strftime("%Y%m%d_%H%M%S") + ".xlsx")


def cell_tl(ws, row: int, col: int) -> tuple[int, int]:
    for m in ws.merged_cells.ranges:
        min_col, min_row, max_col, max_row = range_boundaries(str(m))
        if min_row <= row <= max_row and min_col <= col <= max_col:
            return min_row, min_col
    return row, col


def get_cell(ws, row: int, col: int):
    rr, cc = cell_tl(ws, row, col)
    return ws.cell(rr, cc).value


def set_cell(ws, row: int, col: int, value):
    rr, cc = cell_tl(ws, row, col)
    ws.cell(rr, cc).value = value


def norm_week(val) -> str | None:
    if val is None:
        return None
    s = str(val).strip()
    m = re.match(r"^(W\d+)\b", s, re.I)
    return m.group(1).upper() if m else None


def merge_cell_text(a, b, tag: str) -> str:
    ta = str(a).strip() if a is not None else ""
    tb = str(b).strip() if b is not None else ""
    if not tb or tb == ta:
        return ta
    if not ta:
        return f"【{tag}】{tb}"
    return f"{ta}\n【{tag}】{tb}"


def set_merged_top_left(ws, row: int, col: int, value):
    for m in ws.merged_cells.ranges:
        min_col, min_row, max_col, max_row = range_boundaries(str(m))
        if min_row <= row <= max_row and min_col <= col <= max_col:
            ws.cell(min_row, min_col).value = value
            return
    ws.cell(row, col).value = value


def fill_time_sheet(time_ws, person_ws):
    for c in range(1, 9):
        set_cell(time_ws, 1, c, get_cell(person_ws, 1, c))
    for r in range(2, 9):
        for c in range(1, 9):
            set_cell(time_ws, r, c, get_cell(person_ws, r, c))

    # W1：合并彭熙玉轨道（人员第 14 行）
    tag = "彭熙玉"
    for c in range(3, 9):
        set_cell(
            time_ws,
            5,
            c,
            merge_cell_text(get_cell(time_ws, 5, c), get_cell(person_ws, 14, c), tag),
        )

    # W2：合并彭熙玉轨道（人员第 15 行）
    for c in range(3, 9):
        set_cell(
            time_ws,
            6,
            c,
            merge_cell_text(get_cell(time_ws, 6, c), get_cell(person_ws, 15, c), tag),
        )

    # W8–W14：人员表第 16–22 行 → 时间页行号 = 4 + 周序号（与原有 W5→第9行一致）
    for pr in range(16, 23):
        wk = norm_week(get_cell(person_ws, pr, 1))
        if not wk:
            continue
        try:
            n = int(wk[1:])
        except ValueError:
            continue
        tr = 4 + n
        if tr < 9:
            continue
        for c in range(1, 9):
            set_cell(time_ws, tr, c, get_cell(person_ws, pr, c))


def build_w2_daily_block(person_ws) -> dict[str, object]:
    """日报「周进度」区：日期/状态以刘文博 W2（第 6 行）为准，目标与事项合并彭熙玉（第 15 行）。"""
    tag = "彭熙玉"
    out: dict[str, object] = {
        "G": norm_week(get_cell(person_ws, 6, 1)) or "W2",
        "H": get_cell(person_ws, 6, 8) or "进行中",
        "I": get_cell(person_ws, 6, 2) or "2026/04/13 - 2026/04/18",
        "J": merge_cell_text(get_cell(person_ws, 6, 3), get_cell(person_ws, 15, 3), tag),
        "K": merge_cell_text(get_cell(person_ws, 6, 4), get_cell(person_ws, 15, 4), tag),
        "L": merge_cell_text(get_cell(person_ws, 6, 5), get_cell(person_ws, 15, 5), tag),
        "M": merge_cell_text(get_cell(person_ws, 6, 6), get_cell(person_ws, 15, 6), tag),
        "N": merge_cell_text(get_cell(person_ws, 6, 7), get_cell(person_ws, 15, 7), tag),
    }
    return out


def fill_daily_20260413(ws, g_block: dict[str, object]):
    # 周进度 G–N 只写首行；多人一行由「时间」页拆分后运行 split_weekly_time_by_person 同步
    for r in (4,):
        for key, ci in zip("GHIJKLMN", range(7, 15)):
            val = g_block.get(key)
            if val is not None:
                ws.cell(r, ci).value = val

    ws.cell(16, 2).value = (
        "1.按W2计划推进S0对外接口与多变量推演联调；\n"
        "2.对照周验收物推进S8研发节点（与S0联动相关接口自检）。"
    )
    set_merged_top_left(
        ws,
        16,
        3,
        "1.继续S0与S8联动联调收尾；\n2.整理当日联调记录与待确认问题清单。",
    )
    ws.cell(16, 5).value = "与周计划一致：关注模块间契约变更需拉通确认。"
    risk = str(g_block.get("N") or "").strip() or "模块间契约、接口口径与联调窗口需当日对齐。"
    set_merged_top_left(ws, 16, 7, risk)
    set_merged_top_left(
        ws,
        16,
        12,
        "1.接口/口径变更走会议纪要确认；\n"
        "2.关键路径延误则当日升级并调整次日优先级（对齐W2验收物）。",
    )

    ws.cell(17, 2).value = (
        "1.按W2「S1-S4开发推进」拆解：梳理产销/计划/供应/采购相关表结构与页面清单；\n"
        "2.对齐与S8/S9数据依赖字段，标记待确认项。"
    )
    set_merged_top_left(
        ws,
        17,
        3,
        "1.继续完善模块功能清单；\n2.准备与刘文博侧S8联动所需的数据字段对齐材料。",
    )

    ws.cell(18, 2).value = (
        "1.组织W2联调节奏：S0对外接口、S8节点对照检查；\n"
        "2.同步S9指标建模与S4试点相关风险；\n"
        "3.团队工具/框架答疑与Demo路线澄清。"
    )
    set_merged_top_left(
        ws,
        18,
        3,
        "1.继续推进S9指标建模与S4试点演示准备；\n"
        "2.跟进S0/S8联调问题闭环与跨模块口径确认。",
    )


def main():
    if not SRC.exists():
        raise SystemExit(f"missing: {SRC}")
    shutil.copy2(SRC, BACKUP)
    wb = openpyxl.load_workbook(SRC)
    person = wb["每周详细计划_人员"]
    time_ws = wb["每周详细计划_时间"]
    day = wb["20260413"]

    fill_time_sheet(time_ws, person)
    w2_daily = build_w2_daily_block(person)
    fill_daily_20260413(day, w2_daily)

    wb.save(SRC)
    print("saved:", SRC)
    print("backup:", BACKUP)


if __name__ == "__main__":
    main()