团队休假与节假日覆盖规划器

--- name: team-leave-and-holiday-coverage-planner description: 规划团队休假，确保在节假日期间满足最低人员配置要求 - 根据角色覆盖规则（一人填补一个岗位）检查每个工作日，按公平的优先级顺序决定哪些请求可以批准，提供无空缺的替代日期，尊重禁休期、兼职日和半天假，并跟踪额度、结转和高峰休假天数。包含一个离线的 stdlib Python 规划器，支持 --self-test。 --- # 团队休假与节假日覆盖规划器 你帮助团队负责人、支持和运营经理、诊所和商店老板回答十二月的问题：谁可以休假，什么时候休，而不会让岗位空缺，并且是否公平？休假电子表格显示谁不在，但不显示剩下的人是否真的能覆盖这些角色（一个同时是唯一二级工程师的负责人不可能在同一天身兼两职）。你将团队、覆盖规则和休假请求转化为逐日检查，按明确的优先级顺序决定请求，提供替代方案而不是直接拒绝，并在问题变成人力资源对话之前标记出额度和公平性问题。 ## 此技能中的文件 - `scripts/leave_cover.py` - 读取团队 JSON（人员、角色、兼职日、额度、覆盖最低要求、节假日、覆盖覆盖、禁休期和高峰日）和休假 CSV；使用一人一岗位匹配检查每个工作日，找出已批准休假造成的空缺，决定未决请求（去年工作过高峰期的优先，然后先到先得），建议最近的无空缺日期或更短的行程，并报告日历网格、额度表和发现项 LV01-LV07（仅使用 Python 3.8+ 标准库） - `references/coverage-rules.md` - 如何设置每个角色的最低要求、安静日的覆盖覆盖、兼职人员、半天假、值班以及匹配的作用 - `references/fair-leave-policy.md` - 优先级规则、禁休期、额度和结转、如何妥善拒绝，以及每个发现代码的含义 - `templates/leave-plan-brief.md` - 发送给团队的计划 - `examples/support-team.json` - 一个 7 人客户支持团队，涵盖圣诞节和新年：负责人、二级、值班、一名兼职人员、节假日、12 月 24 日和 31 日较轻的覆盖、账单运行禁休期和结转限制 - `examples/support-leave.csv` - 3 个已批准和 7 个请求的休假，包括一个半天假 - `examples/support-plan.md` - 规划器在这些文件上的实际输出 ## 工作流程 ### 1. 收集输入 询问：时间段（覆盖整个假期段外加前后一周）、团队及每个人的角色和工作日、额度和本休假年度已休天数、团队关闭的公共假日（如果开放则为减少的覆盖）、每个角色和总体的最低覆盖、较低最低要求的安静日、禁休期（发布、盘点、账单运行、审计）以及人们竞争的峰值日。然后是休假：已批准的、请求的（附提交日期）、培训、已知的育儿假或医疗假。询问团队关于优先级的约定；如果没有，建议“去年工作过高峰期的优先，然后先到先得”。 ### 2. 运行规划器 ``` python3 scripts/leave_cover.py examples/support-team.json examples/support-leave.csv python3 scripts/leave_cover.py team.json leave.csv --json python3 scripts/leave_cover.py --self-test ``` 退出码 0 = 已批准的休假保持覆盖且无人超过额度，1 = 已批准的休假已破坏覆盖或超过额度，2 = 输入错误（消息指明行号）。如果没有 shell，手动构建日历：每行一个人，每列一个工作日，然后每天将工作的人员分配到角色岗位，一人一岗位，并标记短缺。 ### 3. 读取结果 - 已批准休假造成的空缺（LV01）优先：它们现在就需要修复（临时覆盖、交换、中介或值班安排），而不是新规则。 - 额度问题（LV02）：已批准休假超过额度需要人力资源确认结转或无薪天数。 - 请求：批准标记为批准的那些；对于冲突和禁休期，向个人提供替代方案，而不是拒绝（LV04）。 - 公平性（LV05）和高峰休假天数：检查是否有人连续两年获得最佳天数，而工作过高峰期的人却被拒绝。 - 脆弱性：没有备用人员的角色（LV03）和零松弛天数（LV07）需要指定后备人员。 - 用或丢（LV06）：将失去天数的人应在安静周预订。 ### 4. 调整并重新运行 使用相同的文件尝试明显的修复：交叉培训的后备人员（给某人添加一个角色）、安静日的较低最低要求（覆盖）、两人之间的交换，或拆分长行程。重新运行，直到已批准的休假没有空缺，每个未决请求都有决定或具体替代方案。 ### 5. 交付 填写 `templates/leave-plan-brief.md`：使用的规则、每个请求的决定及原因和替代方案、日历、薄弱日的覆盖安排和额度说明。在发布计划前与请求未获批准的人交谈。 ## 示例结果 在支持团队示例中，Omar 已批准的行程已经超过他 25 天额度 3 天（LV02，高：人力资源必须确认结转或无薪天数）。Sam 去年工作过圣诞节，因此他 12 月 24 日至 1 月 4 日的请求首先被决定并批准；Jonas（12 月 21-23 日）和 Priya 的半天假及 12 月 21-23 日也被批准。Ade 的 12 月 29-31 日会使 12 月 29 日和 30 日的二级和总覆盖短缺；规划器建议改为 12 月 16-18 日。Lena 的 12 月 30 日至 1 月 8 日跨越 1 月 5-6 日的账单运行禁休期，只有 12 月 31 日可以在不产生空缺的情况下批准。Jonas 的第二次行程需要 5 天但只剩 4 天。Lena 将因 5 天结转限制损失 10 天，而 12 月 21-23 日和 29 日完全没有松弛。 ## 规则 - 按角色检查覆盖，一人一岗位，而不仅仅是人数。 - 对所有人应用一个明确的优先级规则并显示顺序；绝不由谁喊得最响来决定。 - 始终将“这次不行”与替代日期、更短的行程或交换配对。 - 将病假、育儿假和其他受保护的休假视为固定：绝不要求人们移动它们，并将医疗细节排除在计划之外。 - 额度、结转和无薪休假遵循合同和当地法律；标记给人力资源而不是自行决定。 - 仅与需要知道的团队和经理分享计划。 ## 覆盖规则 ### 最低要求 - 为每个**角色**设置最低要求（必须覆盖的内容：负责人/审批人、二级、值班、药剂师、钥匙持有人、叉车执照）和一个**任意**总数（足够的人手处理工作量）。 - 基于需求而非习惯设定数字：每天工单/电话/订单除以一个人处理的数量，加上法律或合同要求必须在场的人员。 - 安静日（12 月 24 日和 31 日，许多 B2B 团队的假期周）通常证明较低的最低要求是合理的：对这些日期使用 `coverage_overrides`，而不是降低整个时间段。 - 业务关闭的日子放在 `holidays` 中：无需覆盖，不扣额度。 ### 一人一岗位 一个拥有多个角色的人每天仍然只填补一个岗位。规划器使用二分匹配将人员分配到岗位，因此如果存在可行的安排，它会找到： - Maya（负责人、二级）+ Lena（一级、二级）可以覆盖负责人 1 + 二级 1，但不能覆盖负责人 1 + 二级 2。 - 短缺按角色报告；“任意”短缺意味着总体人员太少。 ### 人员 - `days`：兼职人员的工作日；他们在非工作日的休假不扣额度且不影响覆盖。 - 半天假（`half_day=yes`）扣 0.5 天额度；规划器当天不计算该人员的覆盖（保守）。如果仅上午缺席仍能覆盖下午班次，请在简报中说明。 - 培训、病假、育儿假和无薪休假像假期一样减少覆盖，但不计入额度。 ### 值班和远程覆盖 - 如果某人在休假期间在家值班，不要将该角色标记为休假；明确同意并按政策支付。规划器将休假视为完全离岗。 - 外部或中介后备人员可以添加为仅拥有该角色且无额度的人员。 ### 松弛 - 零松弛日（恰好达到最低要求）会因一天病假而崩溃。在时间段开始前为每个这样的日子指定后备人员。 - 只有最低要求 + 1 个持有人的角色无法承受两次缺席；计划在新的一年进行交叉培训。 ## 公平休假决定 ### 优先级规则（在请求到来之前说明） 常见、可辩护的顺序： 1. 已批准的休假保持不变（除非个人同意更改）。 2. 受保护的休假（育儿假、医疗假、法定假）是固定的。 3. 对于高峰日：去年工作过高峰期的人优先。 4. 然后按提交日期先到先得。 5. 平局：轮换（上次输掉平局的人这次赢）。 避免仅基于资历的高峰日规则；它们对新员工感觉不公平，且常常失去优秀人才。 ### 禁休期 - 保持简短并关联到真实事件（发布、账单运行、盘点、审计、辅导业务的学期开始）。尽早宣布。 - 触及禁休期的请求不会自动拒绝：规划器提供最近的无空缺日期或禁休期前的最长部分。例外需要指定的审批人。 ### 额度和结转 - 剩余 = 额度 - 时间段前已休 - 已批准 - 建议。 - 已批准休假超过额度（LV02 高）是人力资源问题：从去年结转、购买额外天数或无薪休假。 - 结转限制意味着未预订的天数在年底丢失（LV06）：推动人们预订安静周；在一些国家，法定最低休假不能简单地过期，因此请检查当地规则。 ### 妥善拒绝 - 解释规则和具体空缺（“12 月 29 日和 30 日二级会少一个人”）。 - 提供规划器找到的替代方案、与指定同事的交换或更短的行程。 - 尽早决定（假期前至少 4-6 周），以便人们仍能预订旅行。 - 记录决定和原因；明年的优先级取决于它。 ### 发现代码（leave_cover.py） | 代码 | 严重性 | 含义 | |---|---|---| | LV01 | 高 | 已批准的休假已使某天某个角色或总数短缺 | | LV02 | 高 / 中 | 已批准休假超过额度 / 请求需要的天数超过剩余 | | LV03 | 中 / 低 | 角色持有人数不超过最低要求 / 只有一个备用人员 | | LV04 | 中 | 请求与覆盖或禁休期冲突，提供替代方案 | | LV05 | 中 | 去年工作过高峰期的人被拒绝，而其他人获得高峰日 | | LV06 | 低 | 超过结转限制的剩余天数将丢失 | | LV07 | 信息 | 计划后零松弛的天数 | ## 节假日休假计划模板 # 节假日休假计划：<团队>，<时间段> **覆盖规则：** <负责人 1，二级 2（12 月 24/31 日为 1），值班 1，至少 4 人> **关闭：** <12 月 25 日，12 月 28 日，1 月 1 日> **使用的优先级：** <去年工作过高峰期的优先，然后先到先得> **禁休期：** <1 月 5-6 日账单运行> ## 决定 | 人员 | 请求 | 决定 | 原因 | 提供的替代方案 | |---|---|---|---|---| | <Sam> | <12 月 24 日 - 1 月 4 日> | <批准> | <去年工作过圣诞节> | - | | <Ade> | <12 月 29-31 日> | <不是这些日期> | <12 月 29-30 日二级短缺> | <12 月 16-18 日，或与...交换> | ## 日历 ``` <粘贴规划器网格> ``` ## 薄弱日和后备人员 | 日 | 薄弱原因 | 后备人员 | |---|---|---| | <12 月 21-23 日> | <恰好 4 人工作> | <在家值班：...> | ## 额度说明（供人力资源参考） - <Omar：已批准行程超过额度 3 天 - 确认结转 / 无薪> - <Lena：剩余 15 天，结转限制 5 - 建议预订二月> ## 后续步骤 - <在发布前与 Ade、Lena、Jonas 交谈> - <在...发布计划；通过交换请求更改...> - <在第一季度交叉培训第二个负责人后备人员> ## 示例团队 JSON { "team": "Northwind 客户支持", "period": "2026-12-14..2027-01-08", "workdays": ["mon", "tue", "wed", "thu", "fri"], "holidays": ["2026-12-25", "2026-12-28", "2027-01-01"], "coverage": {"lead": 1, "tier2": 2, "oncall": 1, "any": 4}, "coverage_overrides": [ {"dates": "2026-12-24", "coverage": {"tier2": 1, "any": 3}}, {"dates": "2026-12-31", "coverage": {"tier2": 1, "any": 3}} ], "blackout": [ {"dates": "2027-01-05..2027-01-06", "reason": "一月账单运行，全员参与"} ], "peak_days": "2026-12-21..2027-01-01", "carryover_max": 5, "people": [ {"name": "Maya", "roles": ["lead", "tier2"], "allowance": 25, "taken": 17, "worked_peak_last_year": true}, {"name": "Ade", "roles": ["lead", "tier2"], "allowance": 25, "taken": 19}, {"name": "Omar", "roles": ["tier2", "oncall"], "allowance": 25, "taken": 20}, {"name": "Lena", "roles": ["tier1", "tier2"], "allowance": 25, "taken": 10}, {"name": "Jonas", "roles": ["tier1", "oncall"], "allowance": 25, "taken": 18}, {"name": "Sam", "roles": ["tier1", "oncall"], "allowance": 25, "taken": 15, "worked_peak_last_year": true}, {"name": "Priya", "roles": ["tier1"], "days": ["mon", "wed", "thu"], "allowance": 15, "taken": 9} ] } ## 示例休假 CSV person,start,end,type,status,submitted,half_day,notes Maya,2026-12-21,2026-12-24,vacation,approved,2026-09-30,,family visit Omar,2026-12-29,2027-01-08,vacation,approved,2026-10-02,,"flights booked, Lagos" Lena,2026-12-14,2026-12-15,training,approved,2026-11-01,,Zendesk admin course Jonas,2026-12-21,2026-12-23,vacation,requested,2026-10-20,, Ade,2026-12-29,2026-12-31,vacation,requested,2026-10-25,, Sam,2026-12-24,2027-01-04,vacation,requested,2026-11-05,,"skiing, worked last Christmas" Lena,2026-12-30,2027-01-08,vacation,requested,2026-11-12,, Priya,2026-12-16,2026-12-16,vacation,requested,2026-11-20,yes,school nativity play Jonas,2027-01-04,2027-01-08,vacation,requested,2026-11-25,, Priya,2026-12-21,2026-12-23,vacation,requested,2026-11-28,, ## 示例规划器输出 # 休假覆盖计划：Northwind 客户支持（2026-12-14 至 2027-01-08） 每个工作日最低覆盖：任意 4，负责人 1，值班 1，二级 2。 节假日（关闭）：12 月 25 日，12 月 28 日，01 月 01 日。 **结论：已批准休假超过额度** - 7 个未决请求中有 4 个可以在无空缺的情况下批准；3 个需要决定。 ## 请求（按决定顺序：去年工作过高峰期的优先，然后先到先得） | # | 人员 | 日期 | 工作日 | 类型 | 提交 | 决定 | 原因 | 替代方案 | |---|---|---|---|---|---|---|---|---| | 1 | Sam | 周四 12 月 24 日 - 周一 01 月 04 日 | 5 | vacation | 2026-11-05 | 批准 | - | - | | 2 | Jonas | 周一 12 月 21 日 - 周三 12 月 23 日 | 3 | vacation | 2026-10-20 | 批准 | - | - | | 3 | Ade | 周二 12 月 29 日 - 周四 12 月 31 日 | 3 | vacation | 2026-10-25 | 冲突 | 12 月 29 日、12 月 30 日任意/二级短缺 | 周三 12 月 16 日至周五 12 月 18 日 | | 4 | Lena | 周三 12 月 30 日 - 周五 01 月 08 日 | 7 | vacation | 2026-11-12 | 禁休期 | 一月账单运行，全员参与（01 月 05 日、01 月 06 日） | 缩短至周四 12 月 31 日 - 周四 12 月 31 日（1 个工作日，在禁休期前返回） | | 5 | Priya | 周三 12 月 16 日 - 周三 12 月 16 日（半天） | 0.5 | vacation | 2026-11-20 | 批准 | - | - | | 6 | Jonas | 周一 01 月 04 日 - 周五 01 月 08 日 | 5 | vacation | 2026-11-25 | 超过额度 | 需要 5 天，剩余 4 天 | - | | 7 | Priya | 周一 12 月 21 日 - 周三 12 月 23 日 | 2 | vacation | 2026-11-28 | 批准 | - | - | ## 日历（已批准 + 建议） ``` D J 11111222222233000000 45678123458901145678 MTWTFMTWTFMTWTFMTWTF Maya .....VVVVHH...H..... Ade .........HHxxxH..... Omar .........HHVVVHVVVVV Lena TT.......HH.xxHxxxxx Jonas .....vvv.HH...Hxxxxx Sam ........vHHvvvHv.... Priya .-h.-v-v.HH-..H.-..- cover any +++++++++ +++ +++++ cover lead +++++++++ +++ +++++ cover oncall +++++++++ +++ +++++ cover tier2 +++++++++ +++ +++++ ``` V/T/S/P/U/O 已批准的假期/培训/病假/育儿假/无薪/其他，小写 = 建议批准的请求，h = 半天，x = 未批准的请求，H = 节假日，- = 非工作日，+ / ! = 角色已覆盖 / 短缺。 ## 额度（天） | 人员 | 额度 | 之前已休 | 此处批准 | 建议 | 剩余 | 高峰休假天数 | |---|---|---|---|---|---|---| | Maya | 25 | 17 | 4 | 0 | 4 | 4 | | Ade | 25 | 19 | 0 | 0 | 6 | 0 | | Omar | 25 | 20 | 8 | 0 | -3 | 3 | | Lena | 25 | 10 | 0 | 0 | 15 | 0 | | Jonas | 25 | 18 | 0 | 3 | 4 | 3 | | Sam | 25 | 15 | 0 | 5 | 5 | 4 | | Priya | 15 | 9 | 0 | 2.5 | 3.5 | 2 | ## 发现项 - **高 LV02** Omar 的已批准休假超过额度 3 天（20 + 8 已休，额度 25）；与人力资源确认结转或无薪休假。 - **中 LV02** Jonas 01 月 04 日-01 月 08 日：需要 5 天，剩余 4 天；批准更短的行程或同意无薪天数 - **中 LV04** Ade 12 月 29 日-12 月 31 日：冲突（12 月 29 日、12 月 30 日任意/二级短缺）；提供周三 12 月 16 日至周五 12 月 18 日 - **中 LV04** Lena 12 月 30 日-01 月 08 日：禁休期（一月账单运行，全员参与（01 月 05 日、01 月 06 日））；提供缩短至周四 12 月 31 日 - 周四 12 月 31 日（1 个工作日，在禁休期前返回） - **低 LV03** 负责人有一个备用人员（Maya、Ade，最低 1）；同时两次缺席会破坏它。 - **低 LV06** Ade 将剩余 6 天（结转限制 5）；除非预订，否则丢失 1 天。 - **低 LV06** Lena 将剩余 15 天（结转限制 5）；除非预订，否则丢失 10 天。 - **信息 LV07** 4 天无松弛（12 月 21 日、12 月 22 日、12 月 23 日、12 月 29 日）：一天病假会破坏覆盖；安排后备人员。 ## Python 规划器脚本 ```python #!/usr/bin/env python3 """团队休假与节假日覆盖规划器（仅使用 Python 3.8+ 标准库）。 逐日检查团队的休假请求是否符合最低人员配置规则，决定哪些请求可以在不留下空缺的情况下批准，为其余请求建议替代日期，并跟踪额度和谁获得高峰假日。 团队 JSON： { "team": "支持台", "period": "2026-12-14..2027-01-08", "workdays": ["mon", "tue", "wed", "thu", "fri"], "holidays": ["2026-12-25", "2026-12-28", "2027-01-01"], （办公室关闭，无需覆盖） "coverage": {"lead": 1, "tier2": 2, "oncall": 1, "any": 4}, （"any" = 工作总人数） "coverage_overrides": [{"dates": "2026-12-24", "coverage": {"tier2": 1, "any": 3}}], "blackout": [{"dates": "2027-01-05..2027-01-06", "reason": "账单运行"}], "peak_days": "2026-12-21..2027-01-01", "people": [ {"name": "Maya", "roles": ["lead", "tier2"], "allowance": 25, "taken": 17, "worked_peak_last_year": true}, {"name": "Priya", "roles": ["tier1"], "days": ["mon", "wed", "thu"], "allowance": 15, "taken": 9} ] } 每个工作日的人员填补一个角色岗位（同时是二级的负责人只计一次），作为二分匹配求解。 休假 CSV（需要表头）：person,start,end,type,status,submitted,half_day,notes type vacation（计入额度） | sick | training | parental | unpaid | other status approved | requested | declined（忽略） half_day yes = 半天额度；该人员当天不计入覆盖 用法： python3 leave_cover.py team.json leave.csv [--json] python3 leave_cover.py --self-test 退出码：0 = 已批准的休假保持覆盖且无人超过额度，1 = 已批准的休假已破坏覆盖或超过额度，2 = 输入错误。 """ import argparse import csv import datetime as dt import json import sys from collections import OrderedDict, defaultdict DAYS = ["mon", "tue", "wed", "thu", "fri", "sat", "sun"] TYPE_CODE = {"vacation": "V", "sick": "S", "training": "T", "parental": "P", "unpaid": "U", "other": "O"} class InputError(ValueError): pass def d(s, what="date"): try: return dt.date.fromisoformat(str(s).strip()) except ValueError: raise InputError("%s %r 不是 YYYY-MM-DD" % (what, s)) def date_range(spec): spec = str(spec) a, _, b = spec.partition("..") a, b = d(a), d(b or a) if b < a: raise InputError("范围 %r 在开始前结束" % spec) return [a + dt.timedelta(days=i) for i in range((b - a).days + 1)] # ----------------------------------------------------------------------------- 输入 def load_team(cfg): if "period" not in cfg or not cfg.get("people"): raise InputError("团队 JSON 需要 'period' 和 'people'") t = dict(cfg) t["days"] = date_range(cfg["period"]) t["workdays"] = [w.lower()[:3] for w in cfg.get("workdays", DAYS[:5])] t["holidays"] = {d(h, "holiday") for h in cfg.get("holidays", [])} t["coverage"] = {k: int(v) for k, v in (cfg.get("coverage") or {}).items()} over = {} for o in cfg.get("coverage_overrides", []): for day in date_range(o["dates"]): over[day] = {k: int(v) for k, v in o["coverage"].items()} t["overrides"] = over t["blackout"] = [(set(date_range(b["dates"])), b.get("reason", "blackout")) for b in cfg.get("blackout", [])] t["peak"] = set(date_range(cfg["peak_days"])) if cfg.get("peak_days") else set() people = OrderedDict() for p in cfg["people"]: if "name" not in p: raise InputError("每个人都需要一个名字") q = dict(p) q["roles"] = list(p.get("roles", [])) q["days"] = [w.lower()[:3] for w in p.get("days", t["workdays"])] q["allowance"] = float(p.get("allowance", 0)) q["taken"] = float(p.get("taken", 0)) q["carryover_max"] = p.get("carryover_max", cfg.get("carryover_max")) people[p["name"]] = q t["people"] = people roles = {r for p in people.values() for r in p["roles"]} for r in list(t["coverage"]) + [r for o in over.values() for r in o]: if r != "any" and r not in roles: raise InputError("覆盖角色 %r 无人持有" % r) return t def load_leave(path, team): out = [] with open(path, newline="", encoding="utf-8") as f: rows = [r for r in f if r.strip() and not r.lstrip().startswith("#")] for n, row in enumerate(csv.DictReader(rows), 2): name = (row.get("person") or "").strip() if name not in team["people"]: raise InputError("休假行 %d：未知人员 %r" % (n, name)) typ = (row.get("type") or "vacation").strip().lower() if typ not in TYPE_CODE: raise InputError("休假行 %d：类型必须是 %s 之一" % (n, ", ".join(TYPE_CODE))) status = (row.get("status") or "requested").strip().lower() if status not in ("approved", "requested", "declined"): raise InputError("休假行 %d：状态必须是 approved、requested 或 declined" % n) start, end = d(row.get("start"), "start"), d(row.get("end") or row.get("start"), "end") if end < start: raise InputError("休假行 %d：结束在开始之前" % n) out.append({"id": n, "person": name, "start": start, "end": end, "type": typ, "status": status, "submitted": (row.get("submitted") or "").strip() or "9999-12-31", "half": (row.get("half_day") or "").strip().lower() in ("yes", "y", "true", "1"), "notes": (row.get("notes") or "").strip()}) return out # ----------------------------------------------------------------------------- 核心 def is_workday(team, day): return DAYS[day.weekday()] in team["workdays"] and day not in team["holidays"] def person_works(team, p, day): return is_workday(team, day) and DAYS[day.weekday()] in p["days"] def leave_days(team, lv): p = team["people"][lv["person"]] days = [] cur = lv["start"] while cur <= lv["end"]: if person_works(team, p, cur): days.append(cur) cur += dt.timedelta(days=1) return days def charge(team, lv): n = len(leave_days(team, lv)) return 0.5 * n if lv["half"] else float(n) def need_for(team, day): need = dict(team["coverage"]) need.update(team["overrides"].get(day, {})) return need def shortfall(team, day, off): """返回 {role: missing} 给定一天和休假人员集合。""" if not is_workday(team, day): return {} avail = [p for n, p in team["people"].items() if n not in off and person_works(team, p, day)] need = need_for(team, day) slots = [r for r, k in sorted(need.items()) if r != "any" for _ in range(k)] match = {} # 岗位索引 -> 人员姓名 def try_assign(pi, seen): for si, role in enumerate(slots): if role in avail[pi]["roles"] and si not in seen: seen.add(si) if si not in match or try_assign(match[si], seen): match[si] = pi return True return False for pi in range(len(avail)): try_assign(pi, set()) short = defaultdict(int) for si, role in enumerate(slots): if si not in match: short[role] += 1 if need.get("any", 0) > len(avail): short["any"] = need["any"] - len(avail) return dict(short) def off_map(team, leaves): off = defaultdict(set) for lv in leaves: for day in leave_days(team, lv): off[day].add(lv["person"]) return off def gaps(team, leaves): off = off_map(team, leaves) out = OrderedDict() for day in team["days"]: s = shortfall(team, day, off.get(day, set())) if s: out[day] = s return out def total(g): return sum(sum(v.values()) for v in g.values()) def new_gaps(before, after): out = OrderedDict() for day, s in after.items(): b = before.get(day, {}) worse = {r: k - b.get(r, 0) for r, k in s.items() if k > b.get(r, 0)} if worse: out[day] = worse return out def blackout_hit(team, lv): days = set(leave_days(team, lv)) for bdays, reason in team["blackout"]: hit = sorted(days & bdays) if hit: return hit, reason return None, None def suggest(team, current, lv, max_shift=15): """同一请求（相同日历长度）的最近移动，保持覆盖。""" length = (lv["end"] - lv["start"]).days want = len(leave_days(team, lv)) base = gaps(team, current) first, last = team["days"][0], team["days"][-1] for shift in sorted(range(-max_shift, max_shift + 1), key=lambda x: (abs(x), x)): if shift == 0: continue cand = dict(lv, start=lv["start"] + dt.timedelta(days=shift), end=lv["start"] + dt.timedelta(days=shift + length)) if cand["start"] < first or cand["end"] > last: continue if len(leave_days(team, cand)) < want or blackout_hit(team, cand)[0]: continue if not new_gaps(base, gaps(team, current + [cand])): return cand return None def trimmed(team, current, lv): """触及禁休期的请求的最长部分，在第一个禁休日之前，保持覆盖。""" hit, _ = blackout_hit(team, lv) days = [x for x in leave_days(team, lv) if x < hit[0]] base = gaps(team, current) best = None for i in range(len(days)): for j in range(len(days) - 1, i - 1, -1): if best and j - i + 1 <= len(leave_days(team, best)): break cand = dict(lv, start=days[i], end=days[j]) if not new_gaps(base, gaps(team, current + [cand])): best = cand break return best def plan(team, leaves): findings = [] def add(sev, code, msg): findings.append({"severity": sev, "code": code, "message": msg}) approved = [l for l in leaves if l["status"] == "approved"] requests = [l for l in leaves if l["status"] == "requested"] used = defaultdict(float) for l in approved: if l["type"] == "vacation": used[l["person"]] += charge(team, l) g_approved = gaps(team, approved) for day, s in g_approved.items(): add("high", "LV01", "%s %s：已批准的休假已使 %s 短缺" % ( day.isoformat(), DAYS[day.weekday()].title(), ", ".join("%s 缺 %d" % (r, k) for r, k in sorted(s.items())))) def priority(l): p = team["people"][l["person"]] touches_peak = bool(set(leave_days(team, l)) & team["peak"]) return (0 if touches_peak and p.get("worked_peak_last_year") else 1, l["submitted"], l["id"]) current = list(approved) decisions = [] for l in sorted(requests, key=priority): p = team["people"][l["person"]] days = leave_days(team, l) c = charge(team, l) if l["type"] == "vacation" else 0.0 dec = {"leave": l, "workdays": len(days), "charge": c, "decision": "approve", "reason": "", "suggestion": None} if not days: dec.update(decision="no workdays", reason="仅在节假日/非工作日") elif l["type"] == "vacation" and p["allowance"] and p["taken"] + used[l["person"]] + c > p["allowance"] + 1e-9: left = p["allowance"] - p["taken"] - used[l["person"]] dec.update(decision="over allowance", reason="需要 %g 天，剩余 %g 天" % (c, max(left, 0))) else: hit, reason = blackout_hit(team, l) if hit: dec.update(decision="blackout", reason="%s（%s）" % (reason, ", ".join(x.strftime("%d %b") for x in hit))) else: ng = new_gaps(gaps(team, current), gaps(team, current + [l])) if ng: roles = sorted({r for s in ng.values() for r in s}) dec.update(decision="conflict", reason="%s 在 %s 短缺" % ( "/".join(roles), ", ".join(x.strftime("%d %b") for x in ng))) if dec["decision"] == "approve": current.append(l) used[l["person"]] += c elif dec["decision"] in ("conflict", "blackout"): s = suggest(team, current, l) if s: dec["suggestion"] = "%s 至 %s" % (s["start"].strftime("%a %d %b"), s["end"].strftime("%a %d %b")) elif dec["decision"] == "blackout": t = trimmed(team, current, l) if t: dec["suggestion"] = "缩短至 %s - %s（%d 个工作日，在禁休期前返回）" % ( t["start"].strftime("%a %d %b"), t["end"].strftime("%a %d %b"), len(leave_days(team, t))) decisions.append(dec) approved_peak = defaultdict(int) for l in current: approved_peak[l["person"]] += len(set(leave_days(team, l)) & team["peak"]) for dec in decisions: l = dec["leave"] p = team["people"][l["person"]] if dec["decision"] != "approve" and p.get("worked_peak_last_year") and set(leave_days(team, l)) & team["peak"]: lucky = [n for n, k in approved_peak.items() if k and not team["people"][n].get("worked_peak_last_year")] add("medium", "LV05", "%s 去年工作过高峰期，但此请求为 %s，而 %s 获得高峰休假；考虑交换。" % (l["person"], dec["decision"], ", ".join(sorted(lucky)) or "无人")) g_final = gaps(team, current) allowance = [] for n, p in team["people"].items(): planned_req = sum(dec["charge"] for dec in decisions if dec["leave"]["person"] == n and dec["decision"] == "approve") appr = used[n] - planned_req left = p["allowance"] - p["taken"] - used[n] if p["allowance"] else None allowance.append({"person": n, "allowance": p["allowance"], "taken_before": p["taken"], "approved": appr, "recommended": planned_req, "left": left, "peak_days_off": approved_peak.get(n, 0)}) if p["allowance"] and p["taken"] + appr > p["allowance"] + 1e-9: add("high", "LV02", "%s 的已批准休假超过额度 %g 天（%g + %g 已休，额度 %g）；与人力资源确认结转或无薪休假。" % (n, p["taken"] + appr - p["allowance"], p["taken"], appr, p["allowance"])) cm = p.get("carryover_max") if left is not None and cm is not None and left > float(cm): add("low", "LV06", "%s 将剩余 %g 天（结转限制 %g）；除非预订，否则丢失 %g 天。" % (n, left, float(cm), left - float(cm))) for role, k in sorted(team["coverage"].items()): if role == "any": continue holders = [n for n, p in team["people"].items() if role in p["roles"]] if len(holders) <= k: add("medium", "LV03", "只有 %d 人可以覆盖 %s（最低 %d）：%s 的任何缺席都会破坏覆盖；交叉培训后备人员。" % (len(holders), role, k, ", ".join(holders))) elif len(holders) == k + 1: add("low", "LV03", "%s 有一个备用人员（%s，最低 %d）；同时两次缺席会破坏它。" % (role, ", ".join(holders), k)) for day, s in g_final.items(): if day not in g_approved: add("high", "LV01", "%s 决定后仍短缺：%s" % (day, s)) for dec in decisions: if dec["decision"] in ("conflict", "blackout"): l = dec["leave"] add("medium", "LV04", "%s %s-%s：%s（%s）%s" % ( l["person"], l["start"].strftime("%d %b"), l["end"].strftime("%d %b"), dec["decision"], dec["reason"], ("；提供 " + dec["suggestion"]) if dec["suggestion"] else "；15 天内无无空缺替代方案")) elif dec["decision"] == "over allowance": l = dec["leave"] add("medium", "LV02", "%s %s-%s：%s；批准更短的行程或同意无薪天数" % ( l["person"], l["start"].strftime("%d %b"), l["end"].strftime("%d %b"), dec["reason"])) thin = [] off = off_map(team, current) for day in team["days"]: if not is_workday(team, day): continue working = [n for n, p in team["people"].items() if n not in off.get(day, set()) and person_works(team, p, day)] if len(working) == need_for(team, day).get("any", 0): thin.append(day) if thin: add("info", "LV07", "%d 天无松弛（%s）：一天病假会破坏覆盖；安排后备人员。" % ( len(thin), ", ".join(x.strftime("%d %b") for x in thin))) order = {"high": 0, "medium": 1, "low": 2, "info": 3} findings.sort(key=lambda f: (order[f["severity"]], f["code"])) return {"team": team.get("team", "Team"), "period": team["period"], "gaps_approved": g_approved, "gaps_final": g_final, "decisions": decisions, "allowance": allowance, "findings": findings, "current": current, "thin": thin} # ----------------------------------------------------------------------------- 报告 def grid(team, leaves_final, all_leaves): days = [x for x in team["days"] if x.weekday() < 5 or DAYS[x.weekday()] in team["workdays"]] final_ids = {l["id"] for l in leaves_final} cell = defaultdict(dict) for l in all_leaves: if l["status"] == "declined": continue for day in leave_days(team, l): if l["id"] in final_ids: c = TYPE_CODE[l["type"]] if l["status"] == "approved" else TYPE_CODE[l["type"]].lower() else: c = "x" if l["half"]: c = "h" if c != "x" else "x" cell[l["person"]][day] = c width = max([len(n) for n in team["people"]] + [len("cover " + r) for r in team["coverage"]]) + 2 months = " " * width + "".join(x.strftime("%b")[0] if i == 0 or x.month != days[i - 1].month else " " for i, x in enumerate(days)) L = [months, " " * width + "".join(str(x.day // 10) for x in days), " " * width + "".join(str(x.day % 10) for x in days), " " * width + "".join(DAYS[x.weekday()][0].upper() for x in days)] for n, p in team["people"].items(): row = [] for x in days: if x in team["holidays"]: row.append("H") elif not person_works(team, p, x): row.append("-") else: row.append(cell[n].get(x, ".")) L.append(n.ljust(width) + "".join(row)) off = off_map(team, leaves_final) for role in sorted(team["coverage"]): row = [] for x in days: if not is_workday(team, x): row.append(" ") continue s = shortfall(team, x, off.get(x, set())) row.append("!" if role in s else "+") L.append(("cover " + role).ljust(width) + "".join(row)) return L def report(team, leaves, r): L = ["# 休假覆盖计划：%s（%s）" % (r["team"], r["period"].replace("..", " 至 ")), ""] need = ", ".join("%s %d" % (k, v) for k, v in sorted(team["coverage"].items())) L.append("每个工作日最低覆盖：%s。 节假日（关闭）：%s。" % ( need, ", ".join(x.strftime("%d %b") for x in sorted(team["holidays"])) or "无")) n_appr = sum(1 for dec in r["decisions"] if dec["decision"] == "approve") verdict = ("已批准的休假已破坏覆盖" if r["gaps_approved"] else "已批准休假超过额度" if any(f["code"] == "LV02" and f["severity"] == "high" for f in r["findings"]) else "已覆盖") L += ["", "**结论：%s** - %d 个未决请求中有 %d 个可以在无空缺的情况下批准；%d 个需要决定。" % ( verdict, n_appr, len(r["decisions"]), len(r["decisions"]) - n_appr), ""] L += ["## 请求（按决定顺序：去年工作过高峰期的优先，然后先到先得）", "", "| # | 人员 | 日期 | 工作日 | 类型 | 提交 | 决定 | 原因 | 替代方案 |", "|---|---|---|---|---|---|---|---|---|"] for i, dec in enumerate(r["decisions"], 1): l = dec["leave"] L.append("| %d | %s | %s - %s%s | %s | %s | %s | %s | %s | %s |" % ( i, l["person"], l["start"].strftime("%a %d %b"), l["end"].strftime("%a %d %b"), "（半天）" if l["half"] else "", ("%g" % (0.5 * dec["workdays"])) if l["half"] else dec["workdays"], l["type"], l["submitted"][:10], dec["decision"], dec["reason"] or "-", dec["suggestion"] or "-")) L += ["", "## 日历（已批准 + 建议）", "", "```"] + grid(team, r["current"], leaves) + [ "```", "V/T/S/P/U/O 已批准的假期/培训/病假/育儿假/无薪/其他，小写 = 建议批准的请求，h = 半天，x = 未批准的请求，H = 节假日，- = 非工作日，+ / ! = 角色已覆盖 / 短缺。", ""] L += ["## 额度（天）", "", "| 人员 | 额度 | 之前已休 | 此处批准 | 建议 | 剩余 | 高峰休假天数 |", "|---|---|---|---|---|---|---|"] for a in r["allowance"]: L.append("| %s | %g | %g | %g | %g | %s | %d |" % (a["person"], a["allowance"], a["taken_before"], a["approved"], a["recommended"], "-" if a["left"] is None else "%g" % a["left"], a["peak_days_off"])) L += ["", "## 发现项", ""] if not r["findings"]: L.append("- 无") for f in r["findings"]: L.append("- **%s %s** %s" % (f["severity"].upper(), f["code"], f["message"])) return "\n".join(L) + "\n" # ----------------------------------------------------------------------------- 自测 def self_test(): import os import tempfile cases = [] def ok(name, cond, detail=""): cases.append((name, bool(cond), detail)) base = {"team": "T", "period": "2026-12-14..2026-12-25", "holidays": ["2026-12-25"], "peak_days": "2026-12-21..2026-12-24", "coverage": {"lead": 1, "dev": 1, "any": 2}, "people": [{"name": "A", "roles": ["lead", "dev"], "allowance": 10, "taken": 0}, {"name": "B", "roles": ["dev"], "allowance": 10, "taken": 9}, {"name": "C", "roles": ["lead"], "allowance": 10, "taken": 0, "worked_peak_last_year": True}, {"name": "D", "roles": ["dev"], "days": ["mon", "tue"], "allowance": 5, "taken": 0}]} t = load_team(base) ok("周末和节假日不是工作日", not is_workday(t, d("2026-12-19")) and not is_workday(t, d("2026-12-25"))) sf = shortfall(t, d("2026-12-16"), {"B", "C", "D"}) ok("一人填补一个岗位（A 不能同时是负责人和开发）", sf.get("any") == 1 and sf.get("lead", 0) + sf.get("dev", 0) == 1, sf) ok("匹配找到 A=开发，C=负责人", shortfall(t, d("2026-12-16"), {"B", "D"}) == {}) ok("兼职人员仅在自己的日子计数", shortfall(t, d("2026-12-17"), {"A", "B"}) == {"dev": 1, "any": 1}, shortfall(t, d("2026-12-17"), {"A", "B"})) lv = {"id": 1, "person": "D", "start": d("2026-12-14"), "end": d("2026-12-20"), "type": "vacation", "status": "requested", "submitted": "2026-11-01", "half": False, "notes": ""} ok("休假天数跳过非工作日", len(leave_days(t, lv)) == 2) ok("半天扣 0.5", charge(t, dict(lv, half=True)) == 1.0) d0 = tempfile.mkdtemp() csvp = os.path.join(d0, "l.csv") with open(csvp, "w") as f: f.write("person,start,end,type,status,submitted,half_day,notes\n" "A,2026-12-21,2026-12-22,vacation,approved,2026-10-01,,\n" "C,2026-12-21,2026-12-21,vacation,requested,2026-11-10,,\n" "B,2026-12-14,2026-12-16,vacation,requested,2026-10-05,,\n" "D,2026-12-15,2026-12-15,training,requested,2026-11-01,,\n") leaves = load_leave(csvp, t) r = plan(t, leaves) by = {(dec["leave"]["person"]): dec for dec in r["decisions"]} ok("C 与 A 的已批准休假冲突（无负责人）", by["C"]["decision"] == "conflict", by["C"]) ok("冲突获得替代日期", by["C"]["suggestion"] is not None, by["C"]) ok("B 超过额度（3 天，剩余 1 天）", by["B"]["decision"] == "over allowance", by["B"]) ok("D 培训批准", by["D"]["decision"] == "approve", by["D"]) ok("工作过高峰期的人首先决定", r["decisions"][0]["leave"]["person"] == "C", [x["leave"]["person"] for x in r["decisions"]]) codes = {f["code"] for f in r["findings"]} ok("LV03 负责人的单点故障", "LV03" in codes, codes) ok("LV05 C 的公平性标记", "LV05" in codes, codes) ok("已批准休假无空缺", not r["gaps_approved"]) with open(csvp, "a") as f: f.write("C,2026-12-21,2026-12-21,sick,approved,2026-12-01,,\n") r2 = plan(t, load_leave(csvp, t)) ok("LV01 已批准休假中的空缺", any(f["code"] == "LV01" and f["severity"] == "high" for f in r2["findings"])) t2 = load_team(dict(base, blackout=[{"dates": "2026-12-16", "reason": "发布"}])) r3 = plan(t2, load_leave(csvp, t2)) ok("禁休期先于覆盖", any(x["decision"] == "blackout" for x in r3["decisions"]) or all(x["leave"]["person"] != "B" or x["decision"] == "over allowance" for x in r3["decisions"])) lvb = dict(lv, person="A", start=d("2026-12-16"), end=d("2026-12-16")) ok("blackout_hit 检测重叠", blackout_hit(t2, lvb)[0] == [d("2026-12-16")]) try: load_team(dict(base, coverage={"qa": 1})) ok("未知覆盖角色被拒绝", False) except InputError: ok("未知覆盖角色被拒绝", True) rep = report(t, leaves, r) ok("报告有网格和结论", "Verdict" in rep and "cover lead" in rep) for name, good, detail in cases: print("%s %s%s" % ("PASS" if good else "FAIL", name, "" if good else " -> %r" % (detail,))) passed = sum(1 for _, g, _ in cases if g) print("%d/%d 通过" % (passed, len(cases))) return 0 if passed == len(cases) else 1 def to_json(r): def conv(o): if isinstance(o, dt.date): return o.isoformat() if isinstance(o, dict): return {(k.isoformat() if isinstance(k, dt.date) else k): conv(v) for k, v in o.items()} if isinstance(o, (list, tuple)): return [conv(x) for x in o] return o return json.dumps(conv(r), indent=2) def main(argv=None): ap = argparse.ArgumentParser(description="根据最低覆盖规划团队休假。") ap.add_argument("team", nargs="?") ap.add_argument("leave", nargs="?") ap.add_argument("--json", action="store_true") ap.add_argument("--self-test", action="store_true") a = ap.parse_args(argv) if a.self_test: return self_test() if not (a.team and a.leave): ap.print_usage(sys.stderr) return 2 try: with open(a.team, encoding="utf-8") as f: team = load_team(json.load(f)) leaves = load_leave(a.leave, team) r = plan(team, leaves) except (InputError, OSError, json.JSONDecodeError, KeyError) as e: print("错误：%s" % e, file=sys.stderr) return 2 print(to_json(r) if a.json else report(team, leaves, r), end="\n" if a.json else "") bad = r["gaps_approved"] or any(f["code"] == "LV02" and f["severity"] == "high" for f in r["findings"]) return 1 if bad else 0 if __name__ == "__main__": sys.exit(main()) ```