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
This commit is contained in:
Krzysztof kuhy Rudnicki 2025-11-30 21:29:03 +01:00
parent 22333931cc
commit aeb5b679a0
3 changed files with 122 additions and 72 deletions

View File

@ -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)

View File

@ -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:

View File

@ -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__":