Pandas DataFrame日期时间列过滤脚本异常：结果固定为当日21:00

问题描述 #

编写的Python定时脚本需在指定时间列表["20:00", "20:30", "22:15", "22:45", "23:30", "00:00", "01:45", "02:30", "03:00", "03:30"]运行，过滤DataFrame中Expected Ship Date等于脚本运行时间+1小时的数据（示例：2023年6月4日23:30运行时，需过滤2023年6月5日00:30的数据），但当前无论何时运行，均返回当日21:00的数据，无法得到预期结果。

错误原因分析 #

• 变量作用域错误：roster函数中使用的job_times变量仅在主程序的if __name__ == "__main__"块内定义，函数无法访问该变量，会导致运行时NameError。若为规避错误在函数内错误定义job_times（如仅包含["20:00"]），则会始终计算出21:00的过滤时间。
• 时间逻辑完全错误：脚本通过APScheduler定时触发，每次触发时间就是目标运行时间，但函数内却通过datetime.now()获取实时时间，再寻找下一个job时间计算过滤条件，而非直接使用定时任务的触发时间。例如20:00定时触发时，函数可能因实时时间略晚于20:00，找到20:30作为下一个job时间，或因遍历不到符合条件的时间触发默认值0（即20:00），最终得到21:00的过滤时间。
• 日期处理逻辑不完善：单独判断adjusted_job_time == "00:00"来添加日期的方式存在漏洞，例如23:30运行时，加1小时会跨天，但当前逻辑中adjusted_date仍为当日，导致计算出错误的日期。
• 字符串格式比较时间易出错：将Expected Ship Date转成字符串后再比较，若数据中时间格式与生成的adjusted_expected_ship_date格式不一致（如时区、日期分隔符、位数差异），会导致过滤失效或错误匹配。

解决方案 #

修复步骤 #

1. 修正变量作用域，确保roster函数可访问job_times。
2. 获取定时任务的触发时间，而非依赖实时时间计算。
3. 简化日期时间计算，直接对触发时间加1小时，自动处理跨天情况。
4. 直接比较datetime对象，避免字符串转换带来的格式匹配问题。

修复后的代码 #

__author__ = "studorie"
__version__ = "1.0.1"

from datetime import datetime, timedelta
from io import StringIO

import pandas as pd
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger

from function.requests_retryer import requests_retry_session

# 全局定义job_times，确保roster函数可访问
job_times = ["20:00", "20:30", "22:15", "22:45", "23:30", "00:00", "01:45", "02:30", "03:00", "03:30"]

def roster(fc, trigger_time):
    url = f"https://link.com/{fc}"

    with requests_retry_session() as request:
        response = request.get(url)

    if response.status_code != 200:
        print(response.raise_for_status())

    df = pd.read_csv(StringIO(response.text), sep=",")
    df.drop(['FN SKU', 'Scannable ID', 'Condition', 'Ship Method', 'Ship Option', 'Pick Priority'], axis=1,
            inplace=True)

    df["Expected Ship Date"] = pd.to_datetime(df["Expected Ship Date"])

    # 直接使用触发时间加1小时作为过滤目标时间，自动处理跨天
    target_datetime = trigger_time + timedelta(hours=1)

    # 统一时间精度到分钟，直接比较datetime对象
    filter_condition = df["Expected Ship Date"].dt.floor('min') == target_datetime.floor('min')
    filtered_df = df[filter_condition]

    timestamp = datetime.now().strftime("%d-%m-%Y-%H-%M")
    filename = f"Y:/Public/L&D/Reports for ops/Flow risk/Pick_SLA_{timestamp}.csv"

    filtered_df.to_csv(filename, index=False)
    print(f"Filtered data saved to {filename}")


if __name__ == "__main__":
    schedule = BlockingScheduler(timezone="Europe/London")
    for job_time in job_times:
        hour, minute = map(int, job_time.split(':'))
        # 生成触发时间的datetime对象
        trigger_datetime = datetime.now().replace(hour=hour, minute=minute, second=0, microsecond=0)
        # 创建Cron触发器并添加任务，传递触发时间参数
        schedule.add_job(
            lambda fc="EMA2", t=trigger_datetime: roster(fc, t),
            trigger=CronTrigger(hour=hour, minute=minute, timezone="Europe/London")
        )
    schedule.start()

关键修复说明 #

• 将job_times设为全局变量，解决函数访问不到变量的问题。
• 通过参数传递定时任务的触发时间，避免依赖datetime.now()的实时时间误差。
• 直接对触发时间加1小时，自动处理跨天场景（如23:30+1小时自动变为次日00:30）。
• 使用dt.floor('min')统一时间精度，直接比较datetime对象，消除字符串格式匹配风险。

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