如何在Python脚本中让特定函数触发错误后停用一天并切换替代函数？

核心思路 #

通过记录函数触发错误的时间，每次调用前检查是否处于停用窗口：

• 用持久化存储（文件或轻量数据库）保存错误时间，避免脚本重启后丢失状态
• 调用时先判断是否在停用期内，是则执行替代函数，否则尝试原函数；若原函数再次触发错误，更新停用时间

代码实现示例 #

1. 基础文件存储版本 #

import time
from datetime import datetime, timedelta
import os

# 配置停用时长（1天）
DISABLE_PERIOD = timedelta(days=1)
# 存储错误时间的文件路径
STATUS_PATH = "func_disable_status.txt"

def get_disable_timestamp():
    """读取存储的错误触发时间"""
    try:
        with open(STATUS_PATH, "r") as f:
            return float(f.read().strip())
    except (FileNotFoundError, ValueError):
        return None

def save_disable_timestamp():
    """记录当前错误触发时间"""
    with open(STATUS_PATH, "w") as f:
        f.write(str(time.time()))

def clear_disable_status():
    """重置停用状态"""
    try:
        os.remove(STATUS_PATH)
    except FileNotFoundError:
        pass

def target_func():
    """需要监控的目标函数"""
    # 替换为实际的业务逻辑与错误触发条件
    if True:
        raise ValueError("所需条件不满足")
    return "目标函数执行完成"

def fallback_func():
    """替代函数"""
    return "替代函数执行中"

def safe_invoke():
    """安全调用入口：自动判断执行目标/替代函数"""
    disable_ts = get_disable_timestamp()
    now = time.time()

    # 检查是否在停用期内
    if disable_ts and (now - disable_ts) < DISABLE_PERIOD.total_seconds():
        return fallback_func()
    else:
        try:
            result = target_func()
            # 执行成功则清除停用状态（若之前的停用时间已过期）
            clear_disable_status()
            return result
        except Exception as e:
            # 触发错误，记录停用时间
            save_disable_timestamp()
            return fallback_func()

# 测试调用
if __name__ == "__main__":
    print(safe_invoke())

2. 进阶内存缓存版本（适合长期运行的脚本） #

减少重复文件IO，用内存缓存停用状态：

import time
from datetime import datetime, timedelta
import os

DISABLE_PERIOD = timedelta(days=1)
STATUS_PATH = "func_disable_status.txt"
# 内存缓存停用时间，避免重复读文件
_cached_disable_ts = None

def _load_disable_status():
    """加载并缓存停用时间"""
    global _cached_disable_ts
    if _cached_disable_ts is None:
        try:
            with open(STATUS_PATH, "r") as f:
                _cached_disable_ts = float(f.read().strip())
        except (FileNotFoundError, ValueError):
            _cached_disable_ts = None

def _save_disable_status():
    """保存停用时间并更新缓存"""
    global _cached_disable_ts
    _cached_disable_ts = time.time()
    with open(STATUS_PATH, "w") as f:
        f.write(str(_cached_disable_ts))

def clear_disable_status():
    global _cached_disable_ts
    _cached_disable_ts = None
    try:
        os.remove(STATUS_PATH)
    except FileNotFoundError:
        pass

def target_func():
    if True:
        raise ValueError("所需条件不满足")
    return "目标函数执行完成"

def fallback_func():
    return "替代函数执行中"

def safe_invoke():
    _load_disable_status()
    now = time.time()

    if _cached_disable_ts and (now - _cached_disable_ts) < DISABLE_PERIOD.total_seconds():
        return fallback_func()
    else:
        try:
            result = target_func()
            clear_disable_status()
            return result
        except Exception as e:
            _save_disable_status()
            return fallback_func()

关键注意事项 #

• 持久化必要性：如果脚本会频繁重启，必须用文件、SQLite等持久化方式保存状态，内存缓存仅适用于长期运行的进程
• 错误捕获范围：根据实际需求调整except的捕获类型，避免捕获无关错误导致误停用
• 时间准确性：依赖系统时间，确保运行脚本的设备时间同步正常
• 手动恢复：clear_disable_status函数可用于手动重置状态，方便调试或紧急恢复

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