Refactor exception handling and remove unused file

Replaced bare except clauses with 'except Exception' for better error handling in several protocol files. Removed the unused 'opengsq/responses/flatout2.py' file. Minor formatting and whitespace adjustments were made for consistency.

Author: BattlefieldDuck

opengsq/binary_reader.py

@@ -73,4 +73,3 @@ def read_pascal_string(self, encoding='utf-8', errors='ignore'):
         length = self.read_byte()
         pascal_string = str(self.read_bytes(length - 1), encoding=encoding, errors=errors)
         return pascal_string
-

opengsq/protocol_base.py

@@ -11,4 +11,4 @@ def __init__(self, host: str, port: int, timeout: float = 5.0):
         self._host = host
         self._port = port
         self._timeout = timeout
-        self._allow_broadcast = False
+        self._allow_broadcast = False

opengsq/protocol_socket.py

@@ -152,14 +152,14 @@ async def communicate(protocol: ProtocolBase, data: bytes, source_port: int = No
             if source_port:
                 udpClient.bind_port(source_port)
             udpClient.settimeout(protocol._timeout)
-            
+
             loop = asyncio.get_running_loop()
             transport, protocol_instance = await loop.create_datagram_endpoint(
                 lambda: Socket.Protocol(protocol._timeout),  # Use public Protocol class
                 local_addr=('0.0.0.0', source_port if source_port else 0),
                 allow_broadcast=protocol._allow_broadcast
             )
-            
+
             try:
                 transport.sendto(data, (protocol._host, protocol._port))
                 return await protocol_instance.recv()
@@ -194,4 +194,4 @@ async def test_socket_async():
 
     loop = asyncio.get_event_loop()
     loop.run_until_complete(test_socket_async())
-    loop.close()
+    loop.close()

opengsq/protocols/eldewrito.py

@@ -172,4 +172,4 @@ async def discover_servers(self, broadcast_address: str = "255.255.255.255") ->
         except Exception as e:
             self.logger.error(f"Error during server discovery: {e}")
 
-        return discovered_servers
+        return discovered_servers

opengsq/protocols/flatout2.py

@@ -610,7 +610,7 @@ def _parse_response(self, br: BinaryReader, original_data: bytes) -> Status:
                 try:
                     config_data = br.read_bytes(min(remaining_bytes, 12))
                     info["config"] = config_data.hex()
-                except:
+                except Exception:
                     info["config"] = ""
             else:
                 info["config"] = ""

opengsq/protocols/trackmania_nations.py

@@ -283,7 +283,7 @@ def _deserialize_string(self, data: bytes, offset: int) -> Tuple[Optional[str],
             string_data = data[offset+4:offset+4+length]
             string = string_data.decode('utf-8', errors='replace')
             return string, 4 + length
-        except:
+        except Exception:
             return None, 0
 
     def _extract_all_strings(self, data: bytes) -> List[str]:
@@ -307,7 +307,7 @@ def _extract_all_strings(self, data: bytes) -> List[str]:
                     try:
                         string = current_string.decode('ascii')
                         strings.append(string)
-                    except:
+                    except Exception:
                         pass
                 current_string = bytearray()
 
@@ -316,7 +316,7 @@ def _extract_all_strings(self, data: bytes) -> List[str]:
             try:
                 string = current_string.decode('ascii')
                 strings.append(string)
-            except:
+            except Exception:
                 pass
 
         return strings
@@ -634,7 +634,7 @@ def _extract_challenge_without_prefix(self, data: bytes, srv_pos: int) -> Option
                         if self._is_valid_challenge_name(string):
                             return string
                         found_strings.append(string)
-                    except:
+                    except Exception:
                         pass
                 current_string = bytearray()
 
@@ -645,7 +645,7 @@ def _extract_challenge_without_prefix(self, data: bytes, srv_pos: int) -> Option
                 if self._is_valid_challenge_name(string):
                     return string
                 found_strings.append(string)
-            except:
+            except Exception:
                 pass
 
         # Fallback: Erste gültige Challenge aus gefundenen Strings

opengsq/protocols/vcmp.py

@@ -79,7 +79,7 @@ async def __send_and_receive(self, data: bytes):
                 f"Packet header mismatch. Received: {header}. Expected: {self._response_header}."
             )
 
-        return response[len(self._response_header) + len(packet_header) :]
+        return response[len(self._response_header) + len(packet_header):]
 
     def __read_string(self, br: BinaryReader, read_offset=1):
         """

opengsq/protocols/w40kdow.py

@@ -226,8 +226,6 @@ def _parse_broadcast(self, data: bytes, addr: tuple) -> Status:
 
 
 if __name__ == "__main__":
-    import asyncio
-
     async def main_async():
         # Test with the provided server
         w4kdow = W40kDow(host="172.29.100.29", port=6112, timeout=10.0)
@@ -236,7 +234,7 @@ async def main_async():
             print("Listening for Dawn of War server broadcasts...")
             status = await w4kdow.get_status()
             print(f"\n{'='*60}")
-            print(f"Server Status:")
+            print("Server Status:")
             print(f"{'='*60}")
             print(f"GUID: {status.guid}")
             print(f"Hostname: {status.hostname}")
@@ -249,7 +247,7 @@ async def main_async():
             print(f"Build: {status.build_number}")
             print(f"Magic: {status.magic_marker}")
             print(f"\nFaction Codes: {', '.join(status.faction_codes)}")
-            print(f"\nMap Features:")
+            print("\nMap Features:")
             for i, feature in enumerate(status.map_features, 1):
                 print(f"  {i}. {feature}")
 

opengsq/rcon_protocols/source_rcon.py

@@ -151,7 +151,7 @@ def __get_packets(self, packet_bytes: bytes):
             size = br.read_long()
 
             if br.stream_position + size > len(packet_bytes):
-                return packets, packet_bytes[br.stream_position - 4 :]
+                return packets, packet_bytes[br.stream_position - 4:]
 
             id = br.read_long()
             type = br.read_long()