如何用Python基于字段变更历史计算字段特定值累计时长

核心逻辑 #

处理带时间戳的状态变更历史时，不需要提前统计每个工单进出目标状态的次数，按时间线逐段匹配状态区间累加即可，自动适配任意进出次数的场景：

• 按工单维度拆分所有变更记录，单工单独立计算
• 单工单内的变更记录严格按时间戳从小到大排序，还原真实状态流转顺序
• 遍历流转记录时实时标记当前是否处于目标状态、记录进入目标状态的时间点
• 每检测到一次离开目标状态的动作，就计算刚结束的这段停留时长并累加
• 若遍历结束后工单仍停留在目标状态，取当前时间作为结束点补算最后一段时长

代码实现 #

依赖说明 #

如果是结构化表格类的变更数据，推荐用pandas做分组排序处理，效率更高；小批量数据也可以用纯Python原生逻辑实现。
安装pandas命令：

pip install pandas

完整代码示例 #

import pandas as pd
from datetime import datetime

# 配置项
TARGET_STATUS = "Termination in Progress"
# 替换为实际业务数据即可，要求字段包含：ticket_id(工单ID)、status(变更后状态)、change_time(变更时间，支持datetime类型或标准时间字符串)
# 以下为3次进出目标状态的工单示例数据
sample_data = [
    {"ticket_id": "T001", "status": "Open", "change_time": "2024-01-01 10:00:00"},
    {"ticket_id": "T001", "status": TARGET_STATUS, "change_time": "2024-01-01 11:00:00"},
    {"ticket_id": "T001", "status": "Processing", "change_time": "2024-01-01 12:30:00"},
    {"ticket_id": "T001", "status": TARGET_STATUS, "change_time": "2024-01-02 09:00:00"},
    {"ticket_id": "T001", "status": "Pending", "change_time": "2024-01-02 10:15:00"},
    {"ticket_id": "T001", "status": TARGET_STATUS, "change_time": "2024-01-03 14:00:00"},
    {"ticket_id": "T001", "status": "Closed", "change_time": "2024-01-03 15:45:00"},
]

def calc_target_status_duration(change_records: list[dict]) -> dict[str, float]:
    """
    计算所有工单在目标状态的总停留时长
    返回值：key为工单ID，value为总停留时长，单位为小时，可自行调整单位
    """
    # 统一转换时间字段格式
    df = pd.DataFrame(change_records)
    df["change_time"] = pd.to_datetime(df["change_time"])
    result = {}

    # 按工单维度分组处理
    for ticket_id, group in df.groupby("ticket_id", sort=False):
        # 单工单内按变更时间升序排序
        group = group.sort_values("change_time", ascending=True).reset_index(drop=True)
        total_duration = 0.0
        in_target = False
        enter_time = None

        for _, row in group.iterrows():
            current_status = row["status"]
            current_time = row["change_time"]

            # 匹配到进入目标状态的节点
            if not in_target and current_status == TARGET_STATUS:
                in_target = True
                enter_time = current_time
            # 匹配到离开目标状态的节点，计算本段时长累加
            elif in_target and current_status != TARGET_STATUS:
                duration = (current_time - enter_time).total_seconds() / 3600
                total_duration += duration
                in_target = False
                enter_time = None
        
        # 遍历完所有记录仍处于目标状态时，用当前时间补算最后一段未结束的时长
        if in_target:
            duration = (datetime.now() - enter_time).total_seconds() / 3600
            total_duration += duration
        
        result[ticket_id] = round(total_duration, 2)
    return result

# 测试运行
if __name__ == "__main__":
    durations = calc_target_status_duration(sample_data)
    for tid, dur in durations.items():
        print(f"工单{tid}在'{TARGET_STATUS}'状态总停留时长：{dur}小时")

注意事项 #

• 必须保证输入的变更记录时间字段格式可被正确解析，时间排序错误会导致时长计算完全失真
• 如果业务规则中工单不会停留在目标状态终止流转，可以删掉遍历结束后补算未结束时长的逻辑
• 时长单位可按需调整，代码默认输出小时，需要分钟则除以60，需要秒则直接取total_seconds()返回值即可
• 示例中3次进出目标状态的测试工单，计算总时长为4.5小时，和手动核算结果一致。

内容的提问来源于stack exchange，提问作者Elvyra Sabaliauskaite