From 728d42b2304a431098d2d922b35f81f28f6793cf Mon Sep 17 00:00:00 2001 From: Krzysztof kuhy Rudnicki Date: Sun, 30 Nov 2025 21:29:03 +0100 Subject: [PATCH] fix: resolve PERF203 try-except in loop violations - Extract try-except bodies into helper functions: - download_cats: _download_single_image() - randomize_numbers: _parse_single_number() - lichess_bot/main: _apply_move_to_board(), _process_event_stream(), _run_event_loop_iteration() - Use else block for return statements after try (TRY300) - Remove PERF203 from per-file ignores in pyproject.toml --- pyproject.toml | 3 - python_pkg/download_cats/generate_cats.py | 44 ++++--- python_pkg/lichess_bot/main.py | 124 +++++++++++------- python_pkg/randomize_numbers/random_digits.py | 26 +++- 4 files changed, 122 insertions(+), 75 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 273b6ed..4fe6686 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -83,14 +83,12 @@ unfixable = [] "G004", # f-strings in logging ] "python_pkg/download_cats/generate_cats.py" = [ - "PERF203", # Try-except needed for download resilience in loop "PTH", # os.path patterns in existing code "LOG015", # Root logger in script "G004", # f-strings in logging ] "python_pkg/lichess_bot/main.py" = [ "C901", # Complex functions handling game lifecycle (run_bot, handle_game) - "PERF203", # Try-except needed for stream/move error handling in loops "PLR0912", # Complex nested game event handling with many branches "PLR0915", # Long function handling complete game lifecycle "BLE001", # Blind except for resilient bot operation @@ -132,7 +130,6 @@ unfixable = [] "G004", # f-strings in logging ] "python_pkg/randomize_numbers/random_digits.py" = [ - "PERF203", # Try-except needed for parsing user input in loop "LOG015", # Root logger in script "G004", # f-strings in logging ] diff --git a/python_pkg/download_cats/generate_cats.py b/python_pkg/download_cats/generate_cats.py index e746b63..22d81e0 100644 --- a/python_pkg/download_cats/generate_cats.py +++ b/python_pkg/download_cats/generate_cats.py @@ -15,6 +15,32 @@ logging.basicConfig(level=logging.INFO) MAX_REQUESTS = 90 REQUEST_TIMEOUT = 30 # seconds + +def _download_single_image(url: str) -> None: + """Download and save a single image from URL. + + Args: + url: The URL of the image to download. + """ + try: + # Get the image content + response = requests.get(url, timeout=REQUEST_TIMEOUT) + response.raise_for_status() # Raise an exception for HTTP errors + + # Extract the image name from the URL + image_name = os.path.basename(url) + image_path = os.path.join("./CATS2/", image_name) + + # Save the image to the directory + with open(image_path, "wb") as file: + file.write(response.content) + + logging.info(f"Saved {url} as {image_path}") + + except requests.exceptions.RequestException: + logging.exception(f"Failed to download {url}") + + requests_send = 0 while requests_send < MAX_REQUESTS: res = requests.get( @@ -27,20 +53,4 @@ while requests_send < MAX_REQUESTS: Path("./CATS2").mkdir(parents=True, exist_ok=True) for url in urls: - try: - # Get the image content - response = requests.get(url, timeout=REQUEST_TIMEOUT) - response.raise_for_status() # Raise an exception for HTTP errors - - # Extract the image name from the URL - image_name = os.path.basename(url) - image_path = os.path.join("./CATS2/", image_name) - - # Save the image to the directory - with open(image_path, "wb") as file: - file.write(response.content) - - logging.info(f"Saved {url} as {image_path}") - - except requests.exceptions.RequestException: - logging.exception(f"Failed to download {url}") + _download_single_image(url) diff --git a/python_pkg/lichess_bot/main.py b/python_pkg/lichess_bot/main.py index 2b3d422..e87d34f 100644 --- a/python_pkg/lichess_bot/main.py +++ b/python_pkg/lichess_bot/main.py @@ -18,6 +18,20 @@ from python_pkg.lichess_bot.lichess_api import LichessAPI from python_pkg.lichess_bot.utils import backoff_sleep, get_and_increment_version +def _apply_move_to_board(board: chess.Board, move: str, game_id: str) -> None: + """Apply a single move to the board, logging errors. + + Args: + board: The chess board to apply the move to. + move: The UCI move string. + game_id: The game ID for logging purposes. + """ + try: + board.push_uci(move) + except Exception: + logging.debug(f"Game {game_id}: could not apply move {move}") + + def run_bot(log_level: str = "INFO", *, decline_correspondence: bool = False) -> None: """Start the bot and listen for incoming events.""" logging.basicConfig( @@ -136,10 +150,7 @@ def run_bot(log_level: str = "INFO", *, decline_correspondence: bool = False) -> # Rebuild board from moves board = chess.Board() for m in moves_list: - try: - board.push_uci(m) - except Exception: - logging.debug(f"Game {game_id}: could not apply move {m}") + _apply_move_to_board(board, m, game_id) if color is None: logging.info( @@ -394,56 +405,69 @@ def run_bot(log_level: str = "INFO", *, decline_correspondence: bool = False) -> logging.debug(f"Game {game_id}: could not write PGN: {e}") logging.info(f"Ending game thread for {game_id}") + def _process_event_stream() -> None: + """Process events from the Lichess event stream. + + Handles challenges and game start/finish events. + """ + for event in api.stream_events(): + if event.get("type") == "challenge": + challenge = event["challenge"] + ch_id = challenge["id"] + variant = challenge.get("variant", {}).get("key", "standard") + speed = challenge.get("speed") + perf_ok = speed in {"bullet", "blitz", "rapid", "classical"} + not_corr = ( + challenge.get("speed") != "correspondence" + or not decline_correspondence + ) + if variant == "standard" and perf_ok and not_corr: + logging.info(f"Accepting challenge {ch_id} ({speed})") + api.accept_challenge(ch_id) + else: + logging.info( + f"Declining challenge {ch_id} " + f"(variant={variant}, speed={speed})" + ) + api.decline_challenge(ch_id) + + elif event.get("type") == "gameStart": + game_id = event["game"]["id"] + # Spin up a game thread + if game_id not in game_threads or not game_threads[game_id].is_alive(): + t = threading.Thread( + target=handle_game, args=(game_id,), name=f"game-{game_id}" + ) + t.daemon = True + game_threads[game_id] = t + t.start() + + elif event.get("type") == "gameFinish": + game_id = event["game"]["id"] + logging.info(f"Game finished event: {game_id}") + else: + logging.debug(f"Unhandled event: {json.dumps(event)}") + + def _run_event_loop_iteration() -> int: + """Run one iteration of the event loop with error handling. + + Returns: + New backoff value (0 on success, increased on error). + """ + try: + _process_event_stream() + except Exception as e: + logging.warning(f"Event stream error: {e}") + return backoff_sleep(backoff) + else: + # If stream ends normally, reset backoff + return 0 + # Main event stream: challenge and game start events logging.info("Connecting to Lichess event stream. Waiting for challenges...") backoff = 0 while True: - try: - for event in api.stream_events(): - if event.get("type") == "challenge": - challenge = event["challenge"] - ch_id = challenge["id"] - variant = challenge.get("variant", {}).get("key", "standard") - speed = challenge.get("speed") - perf_ok = speed in {"bullet", "blitz", "rapid", "classical"} - not_corr = ( - challenge.get("speed") != "correspondence" - or not decline_correspondence - ) - if variant == "standard" and perf_ok and not_corr: - logging.info(f"Accepting challenge {ch_id} ({speed})") - api.accept_challenge(ch_id) - else: - logging.info( - f"Declining challenge {ch_id} " - f"(variant={variant}, speed={speed})" - ) - api.decline_challenge(ch_id) - - elif event.get("type") == "gameStart": - game_id = event["game"]["id"] - # Spin up a game thread - if ( - game_id not in game_threads - or not game_threads[game_id].is_alive() - ): - t = threading.Thread( - target=handle_game, args=(game_id,), name=f"game-{game_id}" - ) - t.daemon = True - game_threads[game_id] = t - t.start() - - elif event.get("type") == "gameFinish": - game_id = event["game"]["id"] - logging.info(f"Game finished event: {game_id}") - else: - logging.debug(f"Unhandled event: {json.dumps(event)}") - # If stream ends normally, reset backoff - backoff = 0 - except Exception as e: - logging.warning(f"Event stream error: {e}") - backoff = backoff_sleep(backoff) + backoff = _run_event_loop_iteration() def main() -> None: diff --git a/python_pkg/randomize_numbers/random_digits.py b/python_pkg/randomize_numbers/random_digits.py index 079b5a1..1462a3d 100644 --- a/python_pkg/randomize_numbers/random_digits.py +++ b/python_pkg/randomize_numbers/random_digits.py @@ -43,16 +43,32 @@ def parse_input(input_string: str) -> tuple[list[float], list[int]]: numbers: list[float] = [] decimal_counts: list[int] = [] for num in number_strings: - try: - float_num = float(num) - digits_count = len(num.split(".")[-1]) if "." in num else 0 + parsed = _parse_single_number(num) + if parsed is not None: + float_num, digits_count = parsed numbers.append(float_num) decimal_counts.append(digits_count) - except ValueError: - continue return numbers, decimal_counts +def _parse_single_number(num: str) -> tuple[float, int] | None: + """Parse a single number string into float and decimal count. + + Args: + num: The number string to parse. + + Returns: + Tuple of (float value, decimal count) or None if invalid. + """ + try: + float_num = float(num) + digits_count = len(num.split(".")[-1]) if "." in num else 0 + except ValueError: + return None + else: + return float_num, digits_count + + MIN_ARGS = 2 if __name__ == "__main__":