咨询任务业务工时与实际工作时长的计算方法（含多状态切换场景）

嘿，这个问题我之前帮同事处理过类似的，其实核心就是把任务的时间线拆成有效工作时段和阻塞时段，只累加有效部分的时长就行。我给你一步步理清楚思路，再附个代码示例参考：

你要做的本质是追踪「计时启动/停止」的时间节点，把连续的有效工作时段（启动→停止）找出来，然后把这些时段的时长加总。关键规则对应到时间线就是：

• 第一个activity_tag='Add_Tag'是初始启动点
• 每次progress='Blocked'是停止点，暂停计时
• 每次progress='In Progress'（且前一个状态是Blocked）是重启点，继续计时
• 任务最终结束的状态（比如Done/Completed）也作为最后一个停止点

假设你的数据结构大概是这样（我模拟了一组典型数据，你可以对应自己的表结构调整）：

第一步：整理时间线，标记关键节点 #

先把任务的所有事件按event_time严格排序，然后给每个节点标记类型：

• 第一个Add_Tag → 标记为「START」
• 每个Blocked → 标记为「STOP」
• 每个在Blocked之后的In Progress → 标记为「RESTART」

第二步：配对启动/停止时段，计算时长 #

把连续的「启动→停止」配对，计算每段的时长再累加：

1. START(09:00) → STOP(10:30) → 1.5小时
2. RESTART(11:15) → STOP(14:45) → 3.5小时
3. 如果任务最后是In Progress到Done，那RESTART到Done的时段也要算（示例里没有这种情况，但你可以自己补充）

我给你写两个常用场景的代码：SQL（适合数据库直接计算）和Python/Pandas（适合本地处理数据）

1. SQL实现（用窗口函数追踪状态） #

WITH task_events AS (
    -- 筛选任务事件并标记计时信号
    SELECT 
        task_id,
        event_time,
        progress,
        activity_tag,
        -- 标记启动/停止/重启信号
        CASE 
            WHEN activity_tag = 'Add_Tag' THEN 'START'
            WHEN progress = 'Blocked' THEN 'STOP'
            WHEN progress = 'In Progress' AND LAG(progress) OVER (PARTITION BY task_id ORDER BY event_time) = 'Blocked' THEN 'RESTART'
            ELSE NULL
        END AS timer_signal,
        -- 获取上一个事件的时间
        LAG(event_time) OVER (PARTITION BY task_id ORDER BY event_time) AS prev_time
    FROM your_task_table
    -- 可选：过滤单个任务，去掉则计算所有任务
    WHERE task_id = '101'
    ORDER BY event_time
),
valid_intervals AS (
    -- 提取有效工作的时段：启动/重启点 到 下一个停止点/结束点
    SELECT 
        task_id,
        event_time AS interval_start,
        LEAD(event_time) OVER (PARTITION BY task_id ORDER BY event_time) AS interval_end
    FROM task_events
    WHERE timer_signal IN ('START', 'RESTART')
)
-- 计算总有效工时
SELECT 
    task_id,
    -- 转成小时，保留两位小数
    ROUND(SUM(TIMESTAMPDIFF(MINUTE, interval_start, interval_end)) / 60, 2) AS total_effort_hours
FROM valid_intervals
WHERE interval_end IS NOT NULL -- 排除未结束的任务时段
GROUP BY task_id;

2. Python/Pandas实现（本地数据处理） #

import pandas as pd

# 示例数据，替换成你的实际数据
data = {
    'task_id': ['101']*5,
    'activity_tag': ['Add_Tag', None, None, None, None],
    'progress': ['In Progress', 'Blocked', 'In Progress', 'Blocked', 'Done'],
    'event_time': ['2024-05-01 09:00:00', '2024-05-01 10:30:00', 
                   '2024-05-01 11:15:00', '2024-05-01 14:45:00', '2024-05-01 16:00:00']
}

df = pd.DataFrame(data)
df['event_time'] = pd.to_datetime(df['event_time'])
df = df.sort_values('event_time').reset_index(drop=True)

total_hours = 0
current_start = None  # 追踪当前计时的启动时间

for _, row in df.iterrows():
    # 触发启动计时
    if row['activity_tag'] == 'Add_Tag':
        current_start = row['event_time']
    # 触发停止计时，计算时长
    elif row['progress'] == 'Blocked' and current_start is not None:
        duration = (row['event_time'] - current_start).total_seconds() / 3600
        total_hours += duration
        current_start = None
    # 触发重启计时
    elif row['progress'] == 'In Progress' and current_start is None:
        current_start = row['event_time']
    # 任务结束，计算最后一段有效时长
    elif row['progress'] == 'Done' and current_start is not None:
        duration = (row['event_time'] - current_start).total_seconds() / 3600
        total_hours += duration
        current_start = None

print(f"任务总有效工时：{total_hours:.2f}小时")

• 如果任务最后停在In Progress状态，你可以根据业务规则用当前时间作为结束点，或者标记为“未完成”
• 一定要确保所有事件按时间严格排序，顺序错乱会导致计算完全错误
• 如果存在多个Add_Tag标签，通常取第一个作为启动点，你可以根据自己的业务规则调整

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