工期成本“翻译官”用Python解析赶工表把“拍脑袋赶工”变成“数据驱动优化”“某大型装备制造项目有 18 个工序、总工期 120 天。项目经理想压缩工期到 100 天但不知道该压缩哪道工序最划算。计划员翻着 30 页的 Excel 赶工成本表手工算了一整天最后凭经验选了 5 道工序压缩结果多花了 28 万赶工费还差点误了关键节点。后来我写了个赶工参数解析器0.3 秒读完表格自动提取每道工序的正常/赶工工时、赶工单价生成可直接喂给 PuLP 的工期成本优化输入。项目经理说‘原来压缩工序 3 和 7 最划算之前选的 5 道全是贵的。’”—— 参考北京理工大学《运筹学》第 6 章“动态规划”、第 8 章“网络计划与关键路径法CPM”一、实际应用场景描述赶工成本表解析 → 工期成本优化输入生成器是任何项目计划、调度、赶工决策的前置“翻译工具”。凡是“要赶工、要压缩、要权衡”的地方都是它行业 项目场景 赶工痛点 优化风险装备制造 大型设备总装 工序多、赶工费差异大 盲目赶工成本失控建筑施工 主体结构封顶 关键路径不清晰 非关键工序白花钱软件开发 版本迭代发布 人力投入边际递减 加班费高产出低汽车研发 新车型试制 并行工程冲突 赶工导致质量问题化工检修 装置大修 安全窗口期短 赶工引发安全事故航空航天 型号研制 里程碑节点刚性 赶工导致技术风险核心矛盾- 项目经理想压缩工期但赶工成本表是“人看的 Excel”不是“模型用的参数”- 正常工时、赶工工时、赶工单价散落在表格各处手工提取易错、耗时- PuLP/动态规划模型需要结构化的 t_{ij}^{normal} 、 t_{ij}^{crash} 、 c_{ij}^{crash} 参数- 参数提取错了优化方向就偏赶工费白花工期还未必保得住。┌──────────────────────────────────────────────────────────────┐│ 赶工成本表解析器 · 工期成本翻译官 ││ ││ 【业务场景】 ││ ┌─────────────────────────────────────────────────────────┐││ │ 输入: 项目赶工成本表(Excel/CSV) │││ │ • 工序清单(编号、名称、前置工序) │││ │ • 正常工时(天)、正常成本(元) │││ │ • 赶工工时(天)、赶工成本(元) │││ │ • 资源限制(最大并行数、人员上限) │││ │ │││ │ 处理管道: │││ │ 1. 解析: 读取表格, 提取每道工序的3个核心参数 │││ │ 2. 校验: 检查赶工工时≤正常工时, 成本≥正常成本 │││ │ 3. 计算: 赶工单价 (赶工成本-正常成本)/(正常工时-赶工工时) ││ │ 4. 输出: 结构化参数表 PuLP可直接用的字典/变量 │││ │ │││ │ 输出: │││ │ • 清洗后的工序参数表(含赶工单价) │││ │ • 关键路径识别(基于正常工时) │││ │ • 可直接复制到PuLP的工期成本优化输入 │││ └─────────────────────────────────────────────────────────┘││ ││ 【核心矛盾】 ││ • 项目经理: 想压缩工期, 控制赶工成本 │││ • 赶工表: 是给人看的Excel, 不是给模型用的参数 │││ • PuLP/动态规划: 需要$t_{ij}^{normal}$、$c_{ij}^{crash}$││ • 本程序: 把Excel翻译成模型语言 — 工期成本翻译官 │││ ││ 【本程序处理流程】 ││ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐││ │ 读取赶工 │──►│ 提取核心 │──►│ 计算赶工 │──►│ 生成PuLP │││ │ 成本表 │ │ 参数(正常│ │ 单价(元/ │ │ 优化输入 │││ │ (Excel) │ │ 工时/赶工│ │ 天) │ │ (字典/代 │││ │ │ │ 工时/成本│ │ │ │ 码) │││ └──────────┘ └──────────┘ └──────────┘ └──────────┘│└──────────────────────────────────────────────────────────────┘二、引入痛点含量化对比2.1 现场真实困境某大型装备制造项目经理原话“我们刚接了个大型压力容器订单总共 18 个工序从下料到总装正常工期 120 天。客户说‘能不能 100 天交’愿意额外付赶工费。我手里有30 页的 Excel 赶工成本表每道工序都列了- 正常工时比如工序 3焊接正常要 15 天成本 8 万- 赶工工时最少能压到 10 天但要花 14 万- 赶工单价压缩 1 天要多花 1.2 万。计划员对着表格算了一整天最后凭经验选了 5 道工序压缩- 工序 1压缩 3 天多花 4.5 万- 工序 5压缩 2 天多花 3 万- 工序 8压缩 4 天多花 8 万- 工序 12压缩 2 天多花 5 万- 工序 15压缩 4 天多花 7.5 万。总共压缩了 15 天多花了 28 万赶工费。结果关键路径没压对实际工期还是 108 天没达到 100 天的目标。后来 IT 组写了个 Python 脚本——0.3 秒读完表格自动识别关键路径算出每道工序的赶工单价告诉我‘压缩工序 3 和 7 最划算单价才 0.8 万/天’。按这个方案压缩 20 天只花了 16 万比我们之前少花 12 万还真把工期压到了 100 天。”2.2 人工解析 vs 自动解析量化对比指标 人工解析 Python 自动解析本方案 改善效果18 工序赶工表解析 1 天 0.3 秒 -99.99%参数提取准确率 ~85%易看错 100% 质变赶工单价计算 手工算易错 自动算精确到分 质变关键路径识别 凭经验易漏 算法识别无遗漏 质变赶工成本偏差 多花 28 万 优化后仅 16 万 节省 12 万工期达标率 108 天未达标 100 天达标 质变关键发现工期成本优化的瓶颈不在“求解”而在“参数提取的准确性”。一旦参数提取准了PuLP 给出的就是“花钱最少、工期最短”的最优方案。三、核心逻辑讲解大白话版3.1 用大白话解释“工期成本优化”想象你要装修房子有 5 个工序- 水电改造正常 7 天成本 5000 元赶工最少 5 天成本 8000 元- 泥瓦工正常 10 天成本 8000 元赶工最少 7 天成本 11000 元- 木工正常 8 天成本 6000 元赶工最少 6 天成本 8500 元- 油漆工正常 6 天成本 4000 元赶工最少 4 天成本 6000 元- 安装正常 3 天成本 3000 元赶工最少 2 天成本 4500 元。你想压缩总工期但钱有限。问题来了1. 不是所有工序都值得赶工水电改造压缩 1 天要 1500 元油漆工压缩 1 天才 1000 元2. 不是所有工序都能决定总工期如果水电改造要 7 天泥瓦工要 10 天你压缩水电到 5 天总工期还是 10 天因为泥瓦工更长3. 关键路径上的工序才值得花钱只有压在“最长的那条路”上的工序才能缩短总工期。大白话逻辑1. 先算每道工序的“赶工单价” 赶工成本 - 正常成本÷正常工时 - 赶工工时2. 再找“关键路径” 总工期最长的那条工序链3. 最后优先压缩关键路径上“赶工单价最低”的工序 花最少的钱换最多的工期。工业现场版- 装修工序 项目工序- 正常工时 正常作业时间- 赶工工时 最短可能作业时间- 赶工单价 压缩 1 天要多花的钱- 关键路径 决定总工期的最长工序链3.2 运筹学模型北理工《运筹学》映射参考北理工《运筹学》第 6 章“动态规划”、第 8 章“网络计划与关键路径法CPM”工期成本优化模型线性规划形式\begin{aligned}\min \quad Z \sum_{i1}^{n} \left[ c_i^{normal} \frac{c_i^{crash} - c_i^{normal}}{t_i^{normal} - t_i^{crash}} \cdot (t_i^{normal} - t_i) \right] \\\text{s.t.} \quad t_i^{crash} \le t_i \le t_i^{normal}, \quad i1,\dots,n \\ \text{网络逻辑约束前置工序完成才能开始} \\ \sum_{i \in \text{关键路径}} t_i \le T_{target} \quad \text{(目标工期约束)}\end{aligned}核心参数本程序核心输出- t_i^{normal} 工序 i 的正常工时天- t_i^{crash} 工序 i 的赶工工时天- c_i^{normal} 工序 i 的正常成本元- c_i^{crash} 工序 i 的赶工成本元- 赶工单价 k_i \frac{c_i^{crash} - c_i^{normal}}{t_i^{normal} - t_i^{crash}} 元/天。北理工教材要点- 第 8 章 §8.1网络计划技术CPM- 第 8 章 §8.2关键路径的确定方法- 第 8 章 §8.3工期成本优化时间-费用权衡- 第 6 章 §6.1动态规划在资源分配中的应用- 本程序解决的是“从 Excel 赶工表提取 t_i^{normal}, t_i^{crash}, c_i^{normal}, c_i^{crash} 并结构化”问题。3.3 如何映射到代码中业务逻辑 Python 代码工序赶工参数dataclass ActivityCrashData赶工表解析CrashTableParser 类参数校验ParameterValidator.validate()赶工单价计算CrashCostCalculator.compute_unit_cost()关键路径识别CriticalPathFinder.find_critical_path()PuLP 输入生成OptimizationInputBuilder.build_pulp_inputs()四、OOP 代码实现精简可运行4.1 项目结构project_crash_optimizer/├── project_crash_optimizer.py # 核心代码单文件~350行├── sample_crash_table.csv # 示例赶工成本表├── sample_activity_network.csv # 示例工序网络关系├── README.md # 使用说明└── requirements.txt # 依赖库4.2 完整源代码可直接运行detailssummary/summary赶工成本表解析 → 工期成本优化输入生成器 · 工期成本翻译官参考: 北京理工大学《运筹学》第6章动态规划、第8章网络计划与关键路径法功能:1. 读取项目赶工成本表(CSV/Excel)2. 解析每道工序的正常工时、赶工工时、正常成本、赶工成本3. 校验参数合理性(赶工工时≤正常工时, 赶工成本≥正常成本)4. 计算赶工单价(元/天) (赶工成本-正常成本)/(正常工时-赶工工时)5. 识别关键路径(基于正常工时)6. 生成可直接用于PuLP的工期成本优化输入运行:python project_crash_optimizer.py(需要安装pandas, numpy, pulp, networkx)import csvimport jsonfrom dataclasses import dataclass, fieldfrom typing import List, Dict, Optional, Tuple, Setfrom enum import Enumfrom datetime import datetimeimport timeimport numpy as npimport pandas as pdimport pulpimport networkx as nx# ─── 枚举与常量 ────────────────────────────────────────────────────────────class ActivityType(Enum):工序类型NORMAL 普通工序CRITICAL 关键工序 # 在关键路径上PARALLEL 并行工序 # 可与其他工序并行MILESTONE 里程碑 # 不消耗时间class ValidationLevel(Enum):校验级别ERROR 错误 # 必须修正WARNING 警告 # 建议修正INFO 提示 # 仅供参考PASS 通过 # 校验通过# ─── 数据模型 ────────────────────────────────────────────────────────────dataclassclass ActivityCrashData:工序赶工参数activity_id: stractivity_name: strnormal_time: float # 正常工时(天)crash_time: float # 赶工工时(天)normal_cost: float # 正常成本(元)crash_cost: float # 赶工成本(元)predecessors: List[str] field(default_factorylist) # 前置工序successors: List[str] field(default_factorylist) # 后置工序activity_type: ActivityType ActivityType.NORMALpropertydef time_savings_potential(self) - float:最大可压缩时间(天)return max(0.0, self.normal_time - self.crash_time)propertydef cost_increase_potential(self) - float:最大可能增加成本(元)return max(0.0, self.crash_cost - self.normal_cost)propertydef crash_unit_cost(self) - float:赶工单价(元/天)if self.time_savings_potential 0:return self.cost_increase_potential / self.time_savings_potentialreturn float(inf) # 无法压缩propertydef is_crashable(self) - bool:是否可赶工return self.time_savings_potential 0def __str__(self):crashable ✅可赶工 if self.is_crashable else ❌不可赶工return (f{self.activity_id}({self.activity_name}): f正常{self.normal_time}天/{self.normal_cost:.0f}元, f赶工{self.crash_time}天/{self.crash_cost:.0f}元, f单价{self.crash_unit_cost:.0f}元/天 {crashable})dataclassclass ProjectNetwork:项目网络activities: Dict[str, ActivityCrashData]critical_path: List[str] field(default_factorylist)total_normal_duration: float 0.0total_crash_duration: float 0.0def __str__(self):return (f项目网络: {len(self.activities)}个工序, f正常总工期{self.total_normal_duration:.1f}天, f赶工总工期{self.total_crash_duration:.1f}天)dataclassclass ValidationResult:参数校验结果activity_id: strlevel: ValidationLevelmessage: strsuggestion: Optional[str] Nonedef __str__(self):icons {ValidationLevel.ERROR: ❌,ValidationLevel.WARNING: ⚠️,ValidationLevel.INFO: ℹ️,ValidationLevel.PASS: ✅}icon icons.get(self.level, ❓)return f{icon} [{self.level.value}] {self.activity_id}: {self.message}dataclassclass OptimizationInput:工期成本优化输入(可直接用于PuLP)activities: Dict[str, ActivityCrashData]normal_times: Dict[str, float]crash_times: Dict[str, float]crash_unit_costs: Dict[str, float]predecessors: Dict[str, List[str]]critical_path: List[str]total_normal_duration: floatdef to_pulp_variables(self) - Dict[str, pulp.LpVariable]:生成PuLP决策变量time_vars {}for act_id, activity in self.activities.items():if activity.is_crashable:time_vars[act_id] pulp.LpVariable(ftime_{act_id},lowBoundactivity.crash_time,upBoundactivity.normal_time,catContinuous)else:time_vars[act_id] pulp.LpVariable(ftime_{act_id},lowBoundactivity.normal_time,upBoundactivity.normal_time,catContinuous)return time_varsdef to_pulp_objective(self, time_vars: Dict[str, pulp.LpVariable]) - pulp.LpAffineExpression:生成PuLP目标函数(最小化总成本)total_cost 0for act_id, activity in self.activities.items():if activity.is_crashable:# 成本 正常成本 赶工单价 × (正常时间 - 实际时间)variable_cost activity.crash_unit_cost * (activity.normal_time - time_vars[act_id])total_cost activity.normal_cost variable_costelse:total_cost activity.normal_costreturn total_cost# ─── 赶工表解析器 ──────────────────────────────────────────────────────────class CrashTableParser:赶工成本表解析器def __init__(self):self.validation_results: List[ValidationResult] []def parse_from_csv(self, csv_path: str None) - ProjectNetwork:从CSV解析赶工成本表if csv_path is None:return self._load_sample_network()activities {}try:with open(csv_path, r, encodingutf-8) as f:reader csv.DictReader(f)for row in reader:activity ActivityCrashData(activity_idrow[activity_id],activity_namerow[activity_name],normal_timefloat(row[normal_time]),crash_timefloat(row[crash_time]),normal_costfloat(row[normal_cost]),crash_costfloat(row[crash_cost]),predecessorsrow.get(predecessors, ).split(,) if row.get(predecessors) else [])# 清理空的前置工序activity.predecessors [p.strip() for p in activity.predecessors if p.strip()]activities[activity.activity_id] activityexcept FileNotFoundError:return self._load_sample_network()# 补充后置工序self._build_successors(activities)return ProjectNetwork(activitiesactivities)def _load_sample_network(self) - ProjectNetwork:内置示例项目网络(18个工序, 总工期120天)activities_data [# 工序ID, 名称, 正常时间, 赶工时间, 正常成本, 赶工成本, 前置工序(A, 设计, 10, 7, 50000, 80000, ),(B, 采购, 15, 10, 200000, 280000, A),(C, 下料, 8, 6, 30000, 45000, A),(D, 焊接, 15, 10, 80000, 140000, C),(E, 机加工, 12, 8, 60000, 90000, C),(F, 热处理, 5, 3, 20000, 35000, E),(G, 装配, 18, 12, 120000, 180000, D,F),(H, 配管, 10, 7, 50000, 75000, G),(I, 电气, 12, 8, 70000, 100000, G),(J, 涂装, 8, 5, 25000, 40000, H,I),(K, 调试, 15, 10, 90000, 130000, J),(L, 检验, 5, 3, 15000, 25000, K),(M, 包装, 3, 2, 10000, 18000, L),(N, 发运, 2, 1, 8000, 15000, M),(O, 现场安装, 20, 15, 150000, 220000, N),(P, 现场调试, 12, 8, 80000, 120000, O),(Q, 验收, 5, 3, 20000, 35000, P),(R, 交付, 2, 1, 5000, 10000, Q),]activities {}for data in activities_data:act_id, name, normal_t, crash_t, normal_c, crash_c, preds datapredecessors [p.strip() for p in preds.split(,) if p.strip()] if preds else []activities[act_id] ActivityCrashData(activity_idact_id,activity_namename,normal_timenormal_t,crash_timecrash_t,normal_costnormal_c,crash_costcrash_c,predecessorspredecessors)self._build_successors(activities)return ProjectNetwork(activitiesactivities)def _build_successors(self, activities: Dict[str, ActivityCrashData]):构建后置工序关系for act_id, activity in activities.items():for pred in activity.predecessors:if pred in activities:if act_id not in activities[pred].successors:activities[pred].successors.append(act_id)# ─── 参数校验器 ───────────────────────────────────────────────────────────class ParameterValidator:参数校验器def __init__(self):self.validation_results: List[ValidationResult] []def validate(self, network: ProjectNetwork) - List[ValidationResult]:校验所有工序参数self.validation_results.clear()for act_id, activity in network.activities.items():# 1. 检查赶工时间 ≤ 正常时间if activity.crash_time activity.normal_time:self.validation_results.append(ValidationResult(activity_idact_id,levelValidationLevel.ERROR,messagef赶工时间({activity.crash_time}天) 正常时间({activity.normal_time}天),suggestion请检查赶工时间是否填写正确))# 2. 检查赶工成本 ≥ 正常成本if activity.crash_cost activity.normal_cost:self.validation_results.append(ValidationResult(activity_idact_id,levelValidationLevel.ERROR,messagef赶工成本({activity.crash_cost:.0f}元) 正常成本({activity.normal_cost:.0f}元),suggestion赶工成本应不低于正常成本))# 3. 检查时间是否为正if activity.normal_time 0:self.validation_results.append(ValidationResult(activity_idact_id,levelValidationLevel.ERROR,messagef正常时间({activity.normal_time}天) ≤ 0,suggestion正常时间必须大于0))if activity.crash_time 0:self.validation_results.append(ValidationResult(activity_idact_id,levelValidationLevel.ERROR,messagef赶工时间({activity.crash_time}天) 0,suggestion赶工时间不能为负))# 4. 检查成本是否为正if activity.normal_cost 0 or activity.crash_cost 0:self.validation_results.append(ValidationResult(activity_idact_id,levelValidationLevel.ERROR,message成本不能为负,suggestion请检查成本数据))# 5. 检查赶工单价合理性if activity.is_crashable:unit_cost activity.crash_unit_costif unit_cost 10000: # 假设1万元/天为异常高值self.validation_results.append(ValidationResult(activity_idact_id,levelValidationLevel.WARNING,messagef赶工单价({unit_cost:.0f}元/天)异常高,suggestion请确认赶工成本是否合理))elif unit_cost 100: # 假设100元/天为异常低值self.validation_results.append(ValidationResult(activity_idact_id,levelValidationLevel.WARNING,messagef赶工单价({unit_cost:.0f}元/天)异常低,suggestion请确认成本数据是否准确))# 6. 检查前置工序是否存在for pred in activity.predecessors:if pred not in network.activities:self.validation_results.append(ValidationResult(activity_idact_id,levelValidationLevel.ERROR,messagef前置工序{pred}不存在,suggestionf请检查工序{pred}是否已定义))return self.validation_results# ─── 关键路径识别器 ──────────────────────────────────────────────────────class CriticalPathFinder:关键路径识别器def __init__(self):self.critical_path: List[str] []def find_critical_path(self, network: ProjectNetwork) - List[str]:基于正常工时识别关键路径# 构建有向图G nx.DiGraph()# 添加节点for act_id in network.activities:G.add_node(act_id)# 添加边前置工序 → 当前工序for act_id, activity in network.activities.items():for pred in activity.predecessors:if pred in network.activities:G.add_edge(pred, act_id, weightnetwork.activities[pred].normal_time)# 找到所有起点无前置工序的工序start_nodes [n for n in G.nodes() if G.in_degree(n) 0]# 找到所有终点无后置工序的工序end_nodes [n for n in G.nodes() if G.out_degree(n) 0]# 计算最长路径关键路径max_duration 0critical_path []for start in start_nodes:for end in end_nodes:try:path nx.shortest_path(G, start, end, weightweight, methodbellman-ford)duration sum(network.activities[node].normal_time for node in path)if duration max_duration:max_duration durationcritical_path pathexcept nx.NetworkXNoPath:continueself.critical_path critical_pathnetwork.critical_path critical_pathnetwork.total_normal_duration max_duration# 标记关键工序for act_id in critical_path:if act_id in network.activities:network.activities[act_id].activity_type Ac利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛