ruff + submissions validation
This commit is contained in:
@@ -4,6 +4,7 @@ Utilities for fetching Steam Workshop collection data using Steam Web API
|
||||
|
||||
import re
|
||||
import requests
|
||||
from submissions.models import SteamCollection
|
||||
from datetime import datetime
|
||||
from django.utils import timezone
|
||||
from django.conf import settings
|
||||
@@ -20,22 +21,28 @@ class SteamAPIClient:
|
||||
|
||||
def __init__(self, api_key: Optional[str] = None):
|
||||
# Priority: parameter > database > settings > environment
|
||||
self.api_key = api_key or self._get_api_key_from_db() or getattr(settings, "STEAM_API_KEY", None)
|
||||
self.api_key = (
|
||||
api_key
|
||||
or self._get_api_key_from_db()
|
||||
or getattr(settings, "STEAM_API_KEY", None)
|
||||
)
|
||||
self.session = requests.Session()
|
||||
|
||||
if not self.api_key:
|
||||
logger.warning("No Steam API key provided. Some features may be limited.")
|
||||
|
||||
|
||||
def _get_api_key_from_db(self) -> Optional[str]:
|
||||
"""Get active API key from database"""
|
||||
try:
|
||||
from .models import SteamAPIKey
|
||||
|
||||
api_key_obj = SteamAPIKey.get_active_key()
|
||||
if api_key_obj:
|
||||
# Update last_used timestamp
|
||||
from django.utils import timezone
|
||||
|
||||
api_key_obj.last_used = timezone.now()
|
||||
api_key_obj.save(update_fields=['last_used'])
|
||||
api_key_obj.save(update_fields=["last_used"])
|
||||
return api_key_obj.api_key
|
||||
except Exception as e:
|
||||
logger.debug(f"Could not fetch API key from database: {e}")
|
||||
@@ -234,55 +241,59 @@ class SteamCollectionFetcher:
|
||||
def _fetch_collection_items_via_api(self, collection_id: str) -> List[Dict]:
|
||||
"""
|
||||
Fetch collection items using GetCollectionDetails API
|
||||
|
||||
|
||||
Args:
|
||||
collection_id: Steam collection ID
|
||||
|
||||
|
||||
Returns:
|
||||
List of item dictionaries
|
||||
"""
|
||||
items = []
|
||||
|
||||
|
||||
try:
|
||||
# Use GetCollectionDetails API to get collection items
|
||||
url = f"{self.api_client.BASE_URL}/ISteamRemoteStorage/GetCollectionDetails/v1/"
|
||||
data = {
|
||||
'collectioncount': 1,
|
||||
'publishedfileids[0]': collection_id
|
||||
}
|
||||
|
||||
data = {"collectioncount": 1, "publishedfileids[0]": collection_id}
|
||||
|
||||
response = self.api_client.session.post(url, data=data, timeout=30)
|
||||
if response.status_code == 200:
|
||||
collection_response = response.json()
|
||||
|
||||
if 'response' in collection_response and 'collectiondetails' in collection_response['response']:
|
||||
for collection in collection_response['response']['collectiondetails']:
|
||||
if collection.get('result') == 1 and 'children' in collection:
|
||||
|
||||
if (
|
||||
"response" in collection_response
|
||||
and "collectiondetails" in collection_response["response"]
|
||||
):
|
||||
for collection in collection_response["response"][
|
||||
"collectiondetails"
|
||||
]:
|
||||
if collection.get("result") == 1 and "children" in collection:
|
||||
# Extract item IDs with their sort order
|
||||
child_items = []
|
||||
for child in collection['children']:
|
||||
if 'publishedfileid' in child:
|
||||
child_items.append({
|
||||
'id': str(child['publishedfileid']),
|
||||
'sort_order': child.get('sortorder', 0)
|
||||
})
|
||||
|
||||
for child in collection["children"]:
|
||||
if "publishedfileid" in child:
|
||||
child_items.append(
|
||||
{
|
||||
"id": str(child["publishedfileid"]),
|
||||
"sort_order": child.get("sortorder", 0),
|
||||
}
|
||||
)
|
||||
|
||||
# Sort by sort order to maintain collection order
|
||||
child_items.sort(key=lambda x: x['sort_order'])
|
||||
item_ids = [item['id'] for item in child_items]
|
||||
|
||||
child_items.sort(key=lambda x: x["sort_order"])
|
||||
item_ids = [item["id"] for item in child_items]
|
||||
|
||||
if item_ids:
|
||||
items = self._fetch_items_by_ids(item_ids)
|
||||
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to fetch collection items via API: {e}")
|
||||
|
||||
|
||||
return items
|
||||
|
||||
|
||||
def _fetch_items_by_ids(self, item_ids: List[str]) -> List[Dict]:
|
||||
"""Fetch item details by their IDs"""
|
||||
items = []
|
||||
|
||||
|
||||
# Fetch details for all items in batches (Steam API has limits)
|
||||
batch_size = 20 # Conservative batch size
|
||||
for i in range(0, len(item_ids), batch_size):
|
||||
@@ -300,7 +311,7 @@ class SteamCollectionFetcher:
|
||||
):
|
||||
item_id = item_data.get("publishedfileid", "unknown")
|
||||
result = item_data.get("result", 0)
|
||||
|
||||
|
||||
if result == 1: # Success
|
||||
item_info = {
|
||||
"steam_item_id": str(item_id),
|
||||
@@ -333,14 +344,15 @@ class SteamCollectionFetcher:
|
||||
items.append(item_info)
|
||||
else:
|
||||
# Log failed items
|
||||
logger.warning(f"Failed to fetch item {item_id}: result={result}, ban_reason={item_data.get('ban_reason', 'N/A')}")
|
||||
logger.warning(
|
||||
f"Failed to fetch item {item_id}: result={result}, ban_reason={item_data.get('ban_reason', 'N/A')}"
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to fetch batch of collection items: {e}")
|
||||
continue
|
||||
|
||||
return items
|
||||
|
||||
|
||||
|
||||
def fetch_steam_collection(url: str) -> Dict:
|
||||
@@ -357,7 +369,7 @@ def fetch_steam_collection(url: str) -> Dict:
|
||||
return fetcher.fetch_collection_data(url)
|
||||
|
||||
|
||||
def create_or_update_collection(url: str) -> Tuple["SteamCollection", bool]:
|
||||
def create_or_update_collection(url: str) -> Tuple[SteamCollection, bool]:
|
||||
"""
|
||||
Create or update a Steam collection in the database
|
||||
|
||||
|
||||
Reference in New Issue
Block a user