feat(*): Country display

Author: dsafxP

app/fetcher.py

@@ -1,6 +1,9 @@
 import aiohttp
+
+from app.server import Server
 from app.xml_parser import parse_servers_from_xml
 
+
 async def fetch_game_servers(callback=None):
     # Define the SOAP request body
     soap_request = """<?xml version='1.0' encoding='utf-8'?>
@@ -28,6 +31,8 @@ async def fetch_game_servers(callback=None):
                 servers = parse_servers_from_xml(response_text)
 
                 servers = [server for server in servers if server.version_nr != 0]
+                
+                Server.get_country_codes(servers)
 
                 if callback:
                     # Call the provided callback function (GUI update)

app/server.py

@@ -1,14 +1,17 @@
+import json
 from dataclasses import dataclass
-from typing import Optional
+from typing import ClassVar, Dict, List
+
+import requests
 
 
 @dataclass
 class Server:
     # Network
     address_ipv4: str
     port: int
-    address_ipv6: Optional[str] = None
-    lip: Optional[str] = None
+    address_ipv6: str | None = None
+    lip: str | None = None
 
     # Game info
     game_name: str = ""
@@ -25,7 +28,7 @@ class Server:
     description: str = ""
     version: str = ""
     version_nr: int = 0
-    application_instance: Optional[str] = None
+    application_instance: str | None = None
 
     GAME_MODES = {
         1: "Versus",
@@ -34,6 +37,9 @@ class Server:
         4: "Survival"
     }
 
+    # Class-level cache for IP to country code mapping
+    _country_cache: ClassVar[Dict[str, str]] = {}
+
     def __str__(self):
         return f"Server({self.game_name}, {self.address_ipv4}, {self.port})"
 
@@ -50,4 +56,118 @@ def is_full(self):
     @property
     def connection_string(self):
         """Get connection string for the server."""
-        return f"{self.address_ipv4}:{self.port}"
+        return f"{self.address_ipv4}:{self.port}"
+
+    @property
+    def country_code(self) -> str:
+        """
+        Get the country code for this server's IP address.
+
+        If the IP is not in cache, this will make an API request and block
+        until the result is available. Uses single IP endpoint for efficiency.
+
+        Returns:
+            Country code string (e.g., 'US', 'GB') or 'Unknown' if lookup fails
+        """
+        # Check if this IP is already cached
+        if self.address_ipv4 in Server._country_cache:
+            return Server._country_cache[self.address_ipv4]
+
+        # IP not in cache, use single IP lookup for efficiency
+        return Server._get_single_country_code(self.address_ipv4)
+
+    @staticmethod
+    def get_country_codes(servers: List['Server']) -> Dict[str, str]:
+        """
+        Retrieve country codes for a collection of Server instances using ip-api batch endpoint.
+
+        Args:
+            servers: List of Server instances
+
+        Returns:
+            Dictionary mapping IP addresses to country codes
+        """
+        if not servers:
+            return {}
+
+        # Get unique IP addresses that aren't already cached
+        unique_ips = set(server.address_ipv4 for server in servers)
+        uncached_ips = [ip for ip in unique_ips if ip not in Server._country_cache]
+
+        # If all IPs are cached, return cached results
+        if not uncached_ips:
+            return {ip: Server._country_cache[ip] for ip in unique_ips}
+
+        # Prepare batch request for uncached IPs
+        # ip-api batch endpoint expects a JSON array of IP addresses or objects
+        batch_data = uncached_ips
+
+        try:
+            response = requests.post(
+                'http://ip-api.com/batch?fields=16386',
+                json=batch_data,
+                headers={'Content-Type': 'application/json'},
+                timeout=10
+            )
+            response.raise_for_status()
+
+            batch_results = response.json()
+
+            # Process results and update cache
+            for i, result in enumerate(batch_results):
+                ip = uncached_ips[i]
+                if result.get('status') == 'success':
+                    country_code = result.get('countryCode', 'Unknown')
+                    Server._country_cache[ip] = country_code
+                else:
+                    # Cache failed lookups as 'Unknown' to avoid repeated requests
+                    Server._country_cache[ip] = 'Unknown'
+
+        except (requests.RequestException, json.JSONDecodeError, KeyError) as e:
+            # On error, cache all uncached IPs as 'Unknown' to avoid repeated failed requests
+            for ip in uncached_ips:
+                Server._country_cache[ip] = 'Unknown'
+            print(f"Error fetching country codes: {e}")
+
+        # Return results for all requested IPs (both cached and newly fetched)
+        return {ip: Server._country_cache[ip] for ip in unique_ips}
+
+    @staticmethod
+    def _get_single_country_code(ip: str) -> str:
+        """
+        Fetch country code for a single IP address using the single query endpoint.
+
+        Args:
+            ip: IP address to lookup
+
+        Returns:
+            Country code string or 'Unknown' if lookup fails
+        """
+        try:
+            response = requests.get(
+                f'http://ip-api.com/json/{ip}?fields=16386',
+                timeout=10
+            )
+            response.raise_for_status()
+
+            result = response.json()
+
+            if result.get('status') == 'success':
+                country_code = result.get('countryCode', 'Unknown')
+                Server._country_cache[ip] = country_code
+                return country_code
+            else:
+                # Cache failed lookups as 'Unknown' to avoid repeated requests
+                Server._country_cache[ip] = 'Unknown'
+                return 'Unknown'
+
+        except (requests.RequestException, json.JSONDecodeError, KeyError) as e:
+            # On error, cache as 'Unknown' to avoid repeated failed requests
+            Server._country_cache[ip] = 'Unknown'
+            print(f"Error fetching country code for {ip}: {e}")
+            return 'Unknown'
+
+    @staticmethod
+    def clear_country_cache():
+        """Clear the country code cache."""
+        Server._country_cache.clear()

app/xml_parser.py

@@ -1,4 +1,5 @@
 import xml.etree.ElementTree as Et
+
 from app.server import Server
 
 

gui.pyw

@@ -1,7 +1,7 @@
 import flet as ft
 import asyncio
 import threading
-from typing import List, Optional, Dict
+from typing import List, Dict
 from app.fetcher import fetch_game_servers
 from app.server import Server
 
@@ -19,15 +19,15 @@ class GameServersApp:
         self.page = page
         self.servers: List[Server] = []
         self.filtered_servers: List[Server] = []
-        self.selected_server: Optional[Server] = None
+        self.selected_server: Server | None = None
         self.auto_update_enabled = True
         self.fetch_interval = 12  # seconds
         self.sort_reverse: Dict[str, bool] = {}  # Track sort direction for each column
-        self.current_sort_column: Optional[str] = None  # Track current sort column
+        self.current_sort_column: str | None = None  # Track current sort column
 
         # Add shutdown control
         self.shutdown_event = threading.Event()
-        self.auto_update_timer: Optional[threading.Timer] = None
+        self.auto_update_timer: threading.Timer | None = None
 
         # Register cleanup on page close
         self.page.on_window_event = self.on_window_event
@@ -112,6 +112,10 @@ class GameServersApp:
                     ft.Text("Game Mode", weight=ft.FontWeight.BOLD),
                     on_sort=lambda e: self.sort_table("game_mode")
                 ),
+                ft.DataColumn(
+                    ft.Text("Country", weight=ft.FontWeight.BOLD),
+                    on_sort=lambda e: self.sort_table("country_code")
+                ),
                 ft.DataColumn(
                     ft.Text("Players", weight=ft.FontWeight.BOLD),
                     on_sort=lambda e: self.sort_table("players")
@@ -311,6 +315,11 @@ class GameServersApp:
                     cells=[
                         ft.DataCell(ft.Text(server.game_name or "Unknown")),
                         ft.DataCell(ft.Text(server.game_mode_name)),
+                        ft.DataCell(ft.Image(
+                            src=f"https://flagcdn.com/w20/{server.country_code.lower()}.png",
+                            width=20,
+                            fit=ft.ImageFit.CONTAIN
+                        )),
                         ft.DataCell(ft.Text(f"{server.players}/{server.max_players}")),
                         ft.DataCell(ft.Text("Yes" if server.has_password else "No")),
                         ft.DataCell(ft.Text(server.version or "Unknown"))
@@ -439,6 +448,8 @@ Map Name: {server.map_name or 'Unknown'}"""
             self.filtered_servers.sort(key=lambda x: x.game_name or "", reverse=reverse)
         elif column == "game_mode":
             self.filtered_servers.sort(key=lambda x: x.game_mode_name, reverse=reverse)
+        elif column == "country_code":
+            self.filtered_servers.sort(key=lambda x: x.country_code, reverse=reverse)
         elif column == "players":
             self.filtered_servers.sort(key=lambda x: x.players, reverse=not reverse)  # Default desc for players
         elif column == "password":
@@ -463,7 +474,8 @@ Map Name: {server.map_name or 'Unknown'}"""
 Game Mode: {server.game_mode_name}
 Players: {server.players}/{server.max_players}
 Version: {server.version or 'Unknown'}
-Address: {server.connection_string}"""
+Address: {server.connection_string}
+Country: {server.country_code}"""
 
             try:
                 self.page.set_clipboard(game_info)
@@ -510,6 +522,16 @@ Address: {server.connection_string}"""
             self.auto_update_timer.daemon = True
             self.auto_update_timer.start()
 
+    @staticmethod
+    def country_code_to_flag(country_code: str) -> str:
+        """Convert country code to flag emoji."""
+        if not country_code or country_code == 'Unknown' or len(country_code) != 2:
+            return "🏳️"  # White flag for unknown
+
+        # Convert country code to flag emoji using regional indicator symbols
+        flag = ''.join(chr(ord(char) + 0x1F1A5) for char in country_code.upper())
+        return flag
+
 
 def run_gui():
     """Main entry point for the Flet application."""

main.py

@@ -1,7 +1,9 @@
 import asyncio
 import os
+
 from app.fetcher import fetch_game_servers
 
+
 def cls():
     os.system('cls' if os.name=='nt' else 'clear')
 
@@ -12,12 +14,13 @@ def display_server_details(servers):
     else:
         for i, server in enumerate(servers, start=1):
             print(f"Server {i}:")
-            print(f"  Game Name: {server.game_name}")
-            print(f"  Game Mode: {server.game_mode_name}")
             print(f"  IP Address (IPv4): {server.address_ipv4}")
             #print(f"  IP Address (IPv6): {server.address_ipv6}")
             #print(f"  LIP: {server.lip}")
             print(f"  Port: {server.port}")
+            print(f"  Country: {server.country_code}")
+            print(f"  Game Name: {server.game_name}")
+            print(f"  Game Mode: {server.game_mode_name}")
             print(f"  Map Name: {server.map_name}")
             print(f"  Players: {server.players}/{server.max_players}")
 
@@ -49,8 +52,8 @@ def main():
 
     display_server_details(servers)
 
-    # Wait for the user to press any key to continue
-    input("\nPress any key to continue...")
+    # Wait for the user to press Enter to continue
+    input("\nPress Enter to continue...")
 
 if __name__ == "__main__":
     main()