You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何实现异常与403状态码次数统计及达到阈值后终止脚本(解决代码重复与计数器重置问题)

Hey there! Let's fix up your code to be cleaner, eliminate repetition, and get that counter reset working properly.

First, let's recap what we need:

  • Track consecutive 403 status codes and consecutive connection errors separately (since you want to trigger a stop when either hits 10 in a row)
  • Reset all counters whenever a request succeeds
  • Cut down on duplicated try/except and sleep logic

Here's a polished, maintainable implementation:

import requests
import time
import random
from requests.exceptions import ConnectionError, ReadTimeout, Timeout, RequestException

class FailureTracker:
    def __init__(self, threshold=10):
        self.threshold = threshold
        # Separate counters for the two failure types we care about
        self.consecutive_403 = 0
        self.consecutive_connection_errors = 0

    def reset_all(self):
        """Reset all counters when a request succeeds"""
        self.consecutive_403 = 0
        self.consecutive_connection_errors = 0

    def check_403_limit(self):
        """Increment 403 counter and return True if threshold is hit"""
        self.consecutive_403 += 1
        return self.consecutive_403 >= self.threshold

    def check_connection_error_limit(self):
        """Increment connection error counter and return True if threshold is hit"""
        self.consecutive_connection_errors += 1
        return self.consecutive_connection_errors >= self.threshold

def make_request(url, timeout=12):
    """Encapsulate request logic to avoid repeating try/except blocks"""
    try:
        response = requests.get(url, timeout=timeout)
        return response, None
    except ConnectionError as err:
        return None, ("connection_error", err)
    except (ReadTimeout, Timeout) as err:
        return None, ("timeout", err)
    except RequestException as err:
        return None, ("request_exception", err)
    except Exception as err:
        return None, ("generic_exception", err)

def main():
    failure_tracker = FailureTracker(threshold=10)
    target_url = "https://stackoverflow.com/"
    
    while True:
        response, error = make_request(target_url)
        
        if response:
            if response.ok:
                print("✅ Request successful!")
                failure_tracker.reset_all()  # Reset all counters on success
                time.sleep(60)
            else:
                print(f"❌ Response status: {response.status_code} | URL: {response.url}")
                if response.status_code == 403:
                    if failure_tracker.check_403_limit():
                        print("⚠️ 10 consecutive 403 errors hit! Pausing indefinitely...")
                        time.sleep(4294968)
                        failure_tracker.reset_all()  # Reset after long pause
                else:
                    # Non-403 errors break the 403 streak, reset that counter
                    failure_tracker.consecutive_403 = 0
                time.sleep(60)
        else:
            error_type, err = error
            print(f"❌ Error: {err}")
            
            if error_type == "connection_error":
                if failure_tracker.check_connection_error_limit():
                    print(f"⚠️ 10 consecutive connection errors hit! Pausing indefinitely...")
                    time.sleep(4294968)
                    failure_tracker.reset_all()
                time.sleep(random.randint(1, 3))
            else:
                # Other errors (timeouts, etc.) break the connection error streak
                failure_tracker.consecutive_connection_errors = 0
                time.sleep(random.randint(1, 3))

if __name__ == "__main__":
    main()

Why this works better:

  1. Clean separation of concerns:

    • The FailureTracker class handles all counter logic—resetting, incrementing, and checking thresholds. No more scattered counter code cluttering your main loop.
    • The make_request function wraps all request and exception handling, so your main loop stays focused on business logic instead of repetitive error catching.
  2. Proper counter reset:

    • Anytime a request succeeds (response.ok), we call reset_all() to clear all consecutive failure counters—exactly what you needed to break the streak of errors.
    • We also reset individual counters when a different type of failure happens (e.g., a timeout breaks a connection error streak), so we only track true consecutive failures.
  3. Targeted failure tracking:

    • We track 403s and connection errors separately, so only consecutive instances of each will trigger the stop logic (matches your original requirement perfectly).
    • Other errors (like timeouts or 500s) don't interfere with these counters, so you won't get false triggers from mixed failure types.
  4. Less repetition:

    • No more copying and pasting sleep calls or exception handling blocks—everything is centralized where it makes sense, making the code easier to update later.

If you actually want to track any consecutive failure (not just 403s or connection errors), you can simplify the FailureTracker to use a single counter instead of two—just adjust the logic to increment on any failure and reset on success.

内容的提问来源于stack exchange,提问作者PythonNewbie

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.04.29 09:28:12