client: refactor url handling

Author: tossy310

client/livecli/commands/scrape.py

@@ -44,28 +44,27 @@ def _extract_filename_from_url(url: str) -> str:
     return components[-1] if components else 'resource'
 
 
-def _load_test_resources(url_list: list[str]) -> dict[str, bytes]:
+def _load_test_resources(url_dict: dict[str, str]) -> dict[str, bytes]:
     """Load test resources from local files.
 
     Args:
-        url_list: List of local file paths from scraper.get_urls()
+        url_dict: Dict mapping resource key to local file path from scraper.get_urls()
 
     Returns:
-        Dict mapping URL to bytes
+        Dict mapping resource key to bytes
     """
     resources = {}
 
-    for url in url_list:
+    for key, url in url_dict.items():
         # Strip query parameters (e.g., ?public=true) from file path
-        # but keep original URL as key in resources dict
         base_path = url.split('?')[0] if '?' in url else url
 
         # Try common extensions
         for ext in ['', '.json', '.html', '.txt']:
             filepath = f'{base_path}{ext}'
             if os.path.exists(filepath):
                 with open(filepath, 'rb') as f:
-                    resources[url] = f.read()
+                    resources[key] = f.read()
                 break
         else:
             raise FileNotFoundError(f'File not found: {base_path} (tried extensions: .json, .html, .txt, and no extension)')
@@ -87,37 +86,40 @@ def _wait_next_tick(interval_seconds: int) -> None:
     time.sleep(next_tick - now)
 
 
-def _fetch_resources(session: requests.Session, url_list: list[str], timeout: float) -> dict[str, bytes]:
+def _fetch_resources(session: requests.Session, url_dict: dict[str, str], timeout: float) -> dict[str, bytes]:
     """Fetch all resources from URLs.
 
     Args:
         session: requests.Session to use
-        url_list: List of URLs to fetch
+        url_dict: Dict mapping resource key to URL
         timeout: Request timeout in seconds
 
     Returns:
-        Dict mapping URL to response bytes
+        Dict mapping resource key to response bytes
     """
     resources = {}
-    for url in url_list:
+    for key, url in url_dict.items():
         r = session.get(url, timeout=timeout)
         r.raise_for_status()
-        resources[url] = r.content
+        resources[key] = r.content
     return resources
 
 
-def _save_resources(resources: dict[str, bytes], log_dir: str, timestamp: int) -> None:
+def _save_resources(resources: dict[str, bytes], url_dict: dict[str, str], log_dir: str, timestamp: int) -> None:
     """Save fetched resources to archive files.
 
     Args:
-        resources: Dict mapping URL to bytes
+        resources: Dict mapping resource key to bytes
+        url_dict: Dict mapping resource key to URL (for determining file extension)
         log_dir: Directory to save files
         timestamp: Unix timestamp for filenames
     """
-    for url, content in resources.items():
-        filename_base = _extract_filename_from_url(url)
+    for key, content in resources.items():
+        # Use the resource key as filename base (e.g., 'problems', 'scoreboard', 'teams')
+        filename_base = key
 
         # Determine file extension from URL
+        url = url_dict[key]
         if '/api/' in url or 'json' in url.lower():
             ext = '.json'
         else:
@@ -158,7 +160,7 @@ def scrape_main(options: argparse.Namespace) -> None:
     last_standings = init_feeds[types.FeedType.STANDINGS]
 
     session = requests.Session()
-    url_list = scraper.get_urls(scoreboard_url)
+    url_dict = scraper.get_urls(scoreboard_url)
 
     # Pre-configure authentication if credentials are available
     if scraper.has_credentials():
@@ -167,7 +169,7 @@ def scrape_main(options: argparse.Namespace) -> None:
 
     logging.info('Attempting an initial scrape...')
     try:
-        resources = _fetch_resources(session, url_list, options.interval_seconds * 0.9)
+        resources = _fetch_resources(session, url_dict, options.interval_seconds * 0.9)
         scraper.scrape(resources)
     except Exception:
         logging.exception('Unhandled exception')
@@ -198,8 +200,8 @@ def scrape_main(options: argparse.Namespace) -> None:
         try:
             logging.info('Scraping...%s' % ('' if options.upload and upload else ' (dry-run)'))
             timestamp = int(time.time())
-            resources = _fetch_resources(session, url_list, options.interval_seconds * 0.9)
-            _save_resources(resources, log_dir, timestamp)
+            resources = _fetch_resources(session, url_dict, options.interval_seconds * 0.9)
+            _save_resources(resources, url_dict, log_dir, timestamp)
 
             try:
                 standings = scraper.scrape(resources)

client/livecli/scrapers/base.py

@@ -48,25 +48,25 @@ class NeedLoginException(Exception):
 
 
 class Scraper(abc.ABC):
-    def get_urls(self, base_url: str) -> list[str]:
-        """Return list of URLs to fetch.
+    def get_urls(self, base_url: str) -> dict[str, str]:
+        """Return dict of resource keys to URLs to fetch.
 
-        Default implementation returns single URL (base_url).
+        Default implementation returns single URL with key 'standings'.
         Override for scrapers that need multiple URLs.
 
         Args:
             base_url: Base URL from --scoreboard-url
 
         Returns:
-            List of URLs to fetch
+            Dict mapping resource keys to URLs to fetch
         """
-        return [base_url]
+        return {'standings': base_url}
 
     def scrape(self, resources: dict[str, bytes]) -> dict[str, Any]:
         """Parse resources into standings format.
 
         Args:
-            resources: Dict mapping URL to raw bytes
+            resources: Dict mapping resource key to raw bytes
         """
         standings = self.scrape_impl(resources)
         if not standings['problems']:
@@ -80,7 +80,7 @@ def scrape_impl(self, resources: dict[str, bytes]) -> dict[str, Any]:
         """Parse resources into standings.
 
         Args:
-            resources: Dict mapping URL to raw response bytes
+            resources: Dict mapping resource key to raw response bytes
 
         Returns:
             Dict with 'problems' and 'entries' keys

client/livecli/scrapers/domestic.py

@@ -71,7 +71,7 @@ def has_credentials(self) -> bool:
                 self._options.login_user != '')
 
     def scrape_impl(self, resources: dict[str, bytes]) -> dict[str, Any]:
-        html = next(iter(resources.values())).decode('utf-8')
+        html = resources['standings'].decode('utf-8')
         standings = {'problems': [], 'entries': []}
         if 'rehearsal' in html and not self._options.allow_rehearsal:
             logging.info('Contest has not started yet.')

client/livecli/scrapers/domjudge.py

@@ -49,7 +49,7 @@ def has_credentials(self) -> bool:
                 self._options.login_user != '')
 
     def scrape_impl(self, resources: dict[str, bytes]) -> dict[str, Any]:
-        html = next(iter(resources.values())).decode('utf-8')
+        html = resources['standings'].decode('utf-8')
         doc = bs4.BeautifulSoup(html, 'html5lib')
 
         if doc.select('#loginform'):

client/livecli/scrapers/domjudge_api.py

@@ -40,38 +40,38 @@ class DomjudgeApiScraper(base.Scraper):
     def __init__(self, options: argparse.Namespace):
         self._options = options
 
-    def get_urls(self, base_url: str) -> list[str]:
-        """Declare the three API endpoints needed."""
+    def get_urls(self, base_url: str) -> dict[str, str]:
+        """Declare the three API endpoints needed.
+
+        Returns:
+            Dict mapping resource keys to URLs:
+            - 'problems': problems endpoint
+            - 'scoreboard': scoreboard endpoint (with ?public=true if --public flag set)
+            - 'teams': teams endpoint
+        """
         # Add ?public=true to scoreboard URL if --public flag is set
         scoreboard_url = f'{base_url}/scoreboard'
         if hasattr(self._options, 'public') and self._options.public:
             scoreboard_url += '?public=true'
 
-        return [
-            f'{base_url}/problems',
-            scoreboard_url,
-            f'{base_url}/teams',
-        ]
+        return {
+            'problems': f'{base_url}/problems',
+            'scoreboard': scoreboard_url,
+            'teams': f'{base_url}/teams',
+        }
 
     def scrape_impl(self, resources: dict[str, bytes]) -> dict[str, Any]:
         """Scrape DOMjudge contest data using the REST API.
 
         Args:
-            resources: Dict mapping URL to raw bytes
+            resources: Dict mapping resource key to raw bytes
 
         Returns:
             Dictionary with 'problems' and 'entries' keys in LiveSite format
         """
-        from urllib.parse import urlparse
-
-        # Match URLs by path component (ignoring query parameters like ?public=true)
-        problems_url = next(url for url in resources.keys() if urlparse(url).path.endswith('/problems'))
-        scoreboard_url = next(url for url in resources.keys() if urlparse(url).path.endswith('/scoreboard'))
-        teams_url = next(url for url in resources.keys() if urlparse(url).path.endswith('/teams'))
-
-        problems_data = json.loads(resources[problems_url])
-        scoreboard_data = json.loads(resources[scoreboard_url])
-        teams_data = json.loads(resources[teams_url])
+        problems_data = json.loads(resources['problems'])
+        scoreboard_data = json.loads(resources['scoreboard'])
+        teams_data = json.loads(resources['teams'])
 
         team_id_to_name = {team['id']: team['name'] for team in teams_data}