求助：编写Python脚本对比告警与工单CSV并生成更新工单

需求说明 #

从活跃告警CSV文件（CSV1）生成工单，生成前需与存储所有告警对应工单的事件工单CSV文件（CSV2）对比，满足以下任一条件则生成新工单并更新CSV2：

• 告警ID未在CSV2中出现过
• 该告警ID对应的最近一次工单日期距离当前时间不足3天

CSV示例 #

CSV1（活跃告警） #

Alert_ID    Country     Severity    Timestamp
1234        US          Critical    2023-03-23 00:12:13     
1235        UK          Critical    2023-03-23 00:11:14 
1236        SG          Critical    2023-03-23 00:11:13 
1237        HK          Critical    2023-03-23 00:16:13    

CSV2（历史工单） #

Alert_ID    Incident_No     Timestamp
1234        INC01           2023-02-17  
1235        INC02           2023-02-18
1236        INC03           2023-02-19
1237        INC04           2023-02-19 
1234        INC01           2023-02-20  
1235        INC02           2023-02-23 
1236        INC03           2023-02-23 
1237        INC04           2023-02-23 

原代码存在的问题 #

1. 拼写错误：cav1应为csv1，eleif应为elif
2. 路径错误：Windows路径需用双反斜杠(\\)或原始字符串，.c:\temp格式错误
3. 语法错误：isin后缺少括号，if/elif语句末尾缺少冒号，datetime.datetime重复引用（已导入datetime）
4. 日期处理错误：CSV中的Timestamp未转换为日期类型，直接计算会报错
5. 逻辑错误：未针对每个告警ID获取最近一次工单日期，而是直接对比整个CSV2的日期，无法准确判断是否满足3天条件

修正后的完整代码 #

import pandas as pd
from datetime import datetime, timedelta

# 读取CSV文件（注意Windows路径用双反斜杠或原始字符串）
csv1_path = r'C:\temp\csv1.csv'
csv2_path = r'C:\temp\csv2.csv'

# 读取文件，指定分隔符为任意空白字符（适配示例中的空格分隔）
df_alerts = pd.read_csv(csv1_path, sep=r'\s+', parse_dates=['Timestamp'])
df_incidents = pd.read_csv(csv2_path, sep=r'\s+', parse_dates=['Timestamp'])

# 获取每个告警ID的最近工单日期
latest_incidents = df_incidents.groupby('Alert_ID')['Timestamp'].max().reset_index()
latest_incidents.rename(columns={'Timestamp': 'Latest_Incident_Time'}, inplace=True)

# 合并告警数据与最近工单数据
merged_df = pd.merge(df_alerts, latest_incidents, on='Alert_ID', how='left')

# 计算当前时间，判断是否需要生成工单
current_time = datetime.now()
three_days_ago = current_time - timedelta(days=3)

# 生成工单条件：1. 无历史工单；2. 最近工单时间在3天内
need_ticket = merged_df[
    merged_df['Latest_Incident_Time'].isna() | 
    (merged_df['Latest_Incident_Time'] >= three_days_ago)
]

# 生成新工单数据（模拟生成Incident_No，这里用INC+自增数字）
if not need_ticket.empty:
    # 获取当前最大的Incident编号，生成新编号
    if not df_incidents.empty:
        max_inc_num = max(df_incidents['Incident_No'].apply(lambda x: int(x[3:])))
        new_inc_nums = [f'INC{str(max_inc_num + i + 1).zfill(2)}' for i in range(len(need_ticket))]
    else:
        new_inc_nums = [f'INC{str(i+1).zfill(2)}' for i in range(len(need_ticket))]
    
    # 构造新工单记录
    new_incidents = need_ticket[['Alert_ID', 'Timestamp']].copy()
    new_incidents['Incident_No'] = new_inc_nums
    # 调整列顺序匹配CSV2
    new_incidents = new_incidents[['Alert_ID', 'Incident_No', 'Timestamp']]
    
    # 更新CSV2：追加新工单
    df_incidents = pd.concat([df_incidents, new_incidents], ignore_index=True)
    df_incidents.to_csv(csv2_path, index=False, sep='\t')  # 用制表符分隔，保持格式清晰
    
    print(f"成功生成{len(new_incidents)}条新工单，已更新CSV2")
else:
    print("无需要生成的新工单")

代码说明 #

• 用sep=r'\s+'适配CSV中的空格分隔格式，避免因列间距不一致读取错误
• 读取时直接解析Timestamp为日期类型，方便后续时间计算
• 通过groupby获取每个告警ID的最近工单日期，确保判断逻辑准确
• 模拟生成工单编号，可根据实际需求替换为真实工单系统的调用逻辑
• 用concat追加新工单到CSV2，保存时用制表符分隔保持格式可读性

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