"""domain 层:判定「该做什么」,不含 HTTP、不含 SQL。 这里集中三个最容易被写散的规则: 1. 文件 ↔ 项目 的匹配(精确,绝不猜) 2. 工单是否已存在 / 是否可续跑(幂等的判断源) 3. 版本推进(确认工单)的下一步参数 """ from __future__ import annotations import os import zipfile from dataclasses import dataclass from enum import StrEnum from .models import Project, WorkSheet, normalize_name class Action(StrEnum): CREATE = "create" RESUME = "resume" SKIP_DONE = "skip-done" SKIP_EXISTS = "skip-exists" SKIP_PUSHED = "skip-pushed" ERROR = "error" @property def writes(self) -> bool: return self in {Action.CREATE, Action.RESUME} @dataclass class PlanRow: project_name: str file_path: str action: Action reason: str = "" project: Project | None = None worksheet: WorkSheet | None = None existing_count: int = 0 # 版本链(前端 EffortListVersionInfo 的等价物) # version_code -> (next_operator, next_version_code, percent);None 表示流程结束 VERSION_TRANSITIONS: dict[str, tuple[str, str, int] | None] = { "": ("CustomerLeader", "客户责任人确认版", 20), # 新建草稿视同「技术责任人确认版」 "技术责任人确认版": ("CustomerLeader", "客户责任人确认版", 20), "客户责任人确认版": ("CustomerLeader", "甲方确认版", 20), "甲方确认版": None, } def next_transition(version_code: str) -> tuple[str, str, int]: """返回 (NextOperator, NextVersionCode, Percent)。到终点则报错。""" if version_code not in VERSION_TRANSITIONS: raise ValueError(f"未知版本「{version_code}」,无法推导下一步") step = VERSION_TRANSITIONS[version_code] if step is None: raise ValueError(f"版本「{version_code}」已是流程终点,无需确认") return step def decide(existing: list[WorkSheet], *, ledger_stage: str | None, adopt: bool) -> tuple[Action, str, WorkSheet | None]: """根据「服务器现状 + 台账记录」决定动作。""" if not existing: return Action.CREATE, "本项目该 Sprint 无工单", None # 优先看同一 Sprint 的工单(调用方已按 SystemName 过滤) worksheet = existing[0] if worksheet.done: return Action.SKIP_DONE, f"工单 {worksheet.code} 已确认完成(Status={worksheet.status})", worksheet if ledger_stage == "TFS_PUSHED": return Action.SKIP_PUSHED, f"工单 {worksheet.code} 已推送 TFS", worksheet if ledger_stage: return Action.RESUME, f"台账记录停在 {ledger_stage},续跑", worksheet if adopt: return Action.RESUME, "工单存在但无台账记录,--adopt 接管续跑", worksheet return Action.SKIP_EXISTS, f"工单 {worksheet.code} 已存在(无台账记录,未接管)", worksheet def list_xlsx(folder: str) -> list[str]: if not os.path.isdir(folder): raise SystemExit(f"目录不存在:{folder}") result = [] for name in sorted(os.listdir(folder)): path = os.path.join(folder, name) if os.path.isfile(path) and name.lower().endswith(".xlsx") and not name.startswith("~$"): result.append(path) return result def match_project(stem: str, projects: list[Project]) -> tuple[Project | None, str]: """精确匹配项目;命中多个或零个都算失败,不做「包含即命中」的猜测。""" target = normalize_name(stem) exact = [p for p in projects if normalize_name(p.project_name) == target] if len(exact) == 1: return exact[0], "" if len(exact) > 1: names = "、".join(p.project_name for p in exact) return None, f"项目名歧义({names})" # 只用于提示,不给结果 near = [p.project_name for p in projects if target in normalize_name(p.project_name)] hint = f",相近项目:{'、'.join(near[:3])}" if near else "" return None, f"QuickTeam 中没有该项目{hint}" def precheck_xlsx(path: str) -> tuple[bool, str]: """上传前的轻量自检:确认是「工作量评估表」模板,避免白跑一趟服务器解析。""" try: with zipfile.ZipFile(path) as archive: names = archive.namelist() if "xl/workbook.xml" not in names: return False, "不是 xlsx(缺少 xl/workbook.xml)" if "xl/sharedStrings.xml" not in names: return False, "文件为空表或已损坏" text = archive.read("xl/sharedStrings.xml").decode("utf-8", "ignore") except zipfile.BadZipFile: return False, "不是合法的 xlsx(可能仍是 Excel 打开中的临时文件)" except OSError as exc: return False, f"读取失败:{exc}" if "工作量评估表" not in text: return False, "未找到「工作量评估表」标题,可能不是工单模板" if "服务模块" not in text: return False, "未找到明细表头(服务模块),模板可能已变更" return True, ""