基于API时段配置实现Python脚本定时启停与用户活动记录

看起来你现在的代码思路方向是对的，但有个关键问题：schedule.every().week.at(start_time).do(record_activity(idle_time=300)) 会在指定时间只执行一次record_activity，没法在整个工作时段内持续记录，也不会在work_end_at自动停止，更没法自动衔接下一个时段。下面我给你一套完整的实现方案，刚好匹配你的需求：

第一步：整理时段数据，建立星期映射 #

先把API返回的JSON数据转换成更易处理的结构，把星期缩写映射成datetime.isoweekday()返回的数字（1=周一，7=周日），方便后续匹配当前日期：

import datetime
import time

# 星期缩写到数字的映射（isoweekday()返回1=周一，7=周日）
WEEKDAY_MAP = {
    "mon": 1,
    "tue": 2,
    "wed": 3,
    "thu": 4,
    "fri": 5,
    "sat": 6,
    "sun": 7
}

def parse_working_periods(api_response):
    """解析API返回的时段数据，返回{星期数字: (开始时间, 结束时间)}的字典"""
    period_dict = {}
    for period in api_response:
        weekday_num = WEEKDAY_MAP[period["working_day"]]
        # 把字符串时间转换成datetime.time对象
        start_time = datetime.datetime.strptime(period["work_start_at"], "%H:%M").time()
        end_time = datetime.datetime.strptime(period["work_end_at"], "%H:%M").time()
        period_dict[weekday_num] = (start_time, end_time)
    return period_dict

# 示例API响应
api_response = [
    {"working_day": "mon", "work_start_at": "09:00", "work_end_at": "18:00"},
    {"working_day": "tue", "work_start_at": "09:00", "work_end_at": "15:00"}
]
working_periods = parse_working_periods(api_response)

第二步：实现时段内持续记录的核心函数 #

这个函数会在指定的起止时间窗口内，每隔idle_time秒执行一次记录，直到当前时间超过结束时间，完美解决“时段内持续运行、到点停止”的需求：

def record_activity(idle_time=300):
    """你的记录逻辑，这里用打印代替实际操作，替换成你的业务代码即可"""
    print(f"记录用户活动：{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")

def run_recording_session(start_time: datetime.time, end_time: datetime.time, idle_time=300):
    """在指定时段内持续执行记录活动"""
    now = datetime.datetime.now()
    # 把今天的日期和时段时间组合成完整的datetime对象
    session_start = datetime.datetime.combine(now.date(), start_time)
    session_end = datetime.datetime.combine(now.date(), end_time)

    # 处理当前时间早于时段开始的情况：先等待到开始时间
    if now < session_start:
        wait_seconds = (session_start - now).total_seconds()
        print(f"等待到时段开始：{session_start.strftime('%H:%M')}，需等待{int(wait_seconds/60)}分钟")
        time.sleep(wait_seconds)

    # 在时段内循环执行记录
    current_time = datetime.datetime.now()
    while current_time <= session_end:
        record_activity(idle_time)
        # 计算下一次记录的时间，避免最后一次等待超时
        next_run = current_time + datetime.timedelta(seconds=idle_time)
        if next_run > session_end:
            break
        time.sleep(idle_time)
        current_time = datetime.datetime.now()

    print(f"时段结束：{session_end.strftime('%H:%M')}，停止记录")

第三步：实现自动衔接下一个时段的调度逻辑 #

这个函数会自动找到未来最近的工作时段，处理跨天、跨周的情况，让脚本无需人工干预就能持续循环执行：

def find_next_session():
    """找到下一个要执行的记录时段，返回(开始datetime, 结束datetime)"""
    now = datetime.datetime.now()
    current_weekday = now.isoweekday()
    current_time = now.time()

    # 先检查今天是否还有未开始/正在进行的时段
    if current_weekday in working_periods:
        start_time, end_time = working_periods[current_weekday]
        if current_time < end_time:
            return datetime.datetime.combine(now.date(), start_time), datetime.datetime.combine(now.date(), end_time)

    # 遍历未来7天，找到最近的工作时段
    for days_ahead in range(1, 8):
        target_date = now.date() + datetime.timedelta(days=days_ahead)
        target_weekday = target_date.isoweekday()
        if target_weekday in working_periods:
            start_time, end_time = working_periods[target_weekday]
            return datetime.datetime.combine(target_date, start_time), datetime.datetime.combine(target_date, end_time)

    # 理论上不会走到这里，除非API返回空时段
    raise ValueError("未找到任何有效工作时段，请检查API响应")

def main_loop():
    """主循环，持续调度下一个记录时段"""
    print("=== 用户活动记录脚本启动 ===")
    while True:
        try:
            next_start, next_end = find_next_session()
            date_str = next_start.date().strftime('%Y-%m-%d')
            print(f"\n下一个记录时段：{date_str} {next_start.time()} 至 {next_end.time()}")
            # 等待到下一个时段开始
            wait_seconds = (next_start - datetime.datetime.now()).total_seconds()
            if wait_seconds > 0:
                time.sleep(wait_seconds)
            # 运行当前时段的记录任务
            run_recording_session(next_start.time(), next_end.time())
        except Exception as e:
            print(f"脚本出错：{str(e)}，10秒后重试...")
            time.sleep(10)

if __name__ == "__main__":
    main_loop()

关键细节说明 #

• 时段内持续运行：run_recording_session会严格卡在时段窗口内执行记录，最后一次记录会自动判断是否超时，避免在时段结束后额外执行。
• 自动跨时段衔接：find_next_session会优先检查当天剩余时段，再遍历未来7天找最近的时段，不管是周内还是跨周的情况都能处理。
• 容错机制：主循环加入了异常捕获，即使某次记录任务出错，脚本也会自动重试，不会直接崩溃。
• 可扩展性：如果API的时段数据会动态更新，你可以在main_loop里定期重新调用parse_working_periods来刷新配置。

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