import re import aiohttp from loguru import logger async def fetch_profile(steam_id: str, steam=None) -> dict | None: """Fetch persona name and avatar URL from Steam community profile page. If *steam* session is provided, uses authenticated cookies. Returns {"nickname": str, "avatar_url": str, "steam_level": int|None} or None on failure. """ url = f"https://steamcommunity.com/profiles/{steam_id}/" try: if steam is not None: from app.services.steam_auth import raw_request resp = await raw_request(steam, url, method="GET", allow_redirects=True) if resp.status != 200: logger.warning(f"Profile fetch for {steam_id}: HTTP {resp.status}") return None html = await resp.text() else: headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"} async with aiohttp.ClientSession(headers=headers) as session: async with session.get(url, timeout=aiohttp.ClientTimeout(total=15)) as resp: if resp.status != 200: logger.warning(f"Profile fetch for {steam_id}: HTTP {resp.status}") return None html = await resp.text() except Exception as exc: logger.warning(f"Profile fetch error for {steam_id}: {exc}") return None nickname = _parse_persona_name(html) avatar_url = _parse_avatar(html) steam_level = _parse_level(html) last_online = _parse_last_online(html) if not nickname and not avatar_url: return None return {"nickname": nickname, "avatar_url": avatar_url, "steam_level": steam_level, "last_online": last_online} def _parse_persona_name(html: str) -> str | None: # g_rgProfileData = {"personaname":"jeffreyoelze1999",...} m = re.search(r'"personaname"\s*:\s*"([^"]*)"', html) if m: return m.group(1) # Fallback: ... m = re.search(r'([^<]+)', html) return m.group(1) if m else None def _parse_avatar(html: str) -> str | None: # 1) g_rgProfileData JSON: {"avatarfull":"https://avatars.fastly.steamstatic.com/..."} m = re.search(r'"avatarfull"\s*:\s*"(https://[^"]+)"', html) if m: return m.group(1).replace("_full.jpg", "_medium.jpg") # 2) m = re.search(r']+srcset="(https://[^"]+)"', html) if m: return m.group(1).replace("_full.jpg", "_medium.jpg") return None def _parse_level(html: str) -> int | None: # 100 m = re.search(r'(\d+)', html) return int(m.group(1)) if m else None def _parse_last_online(html: str) -> str | None: # "Last Online X days ago" → "Xd", "X hrs ago" → "Xh", "X min ago" → "Xm" m = re.search(r'Last\s+Online[^<]{0,150}?(\d+)\s+day', html, re.IGNORECASE) if m: return f"{m.group(1)}d" m = re.search(r'Last\s+Online[^<]{0,150}?(\d+)\s+hr', html, re.IGNORECASE) if m: return f"{m.group(1)}h" m = re.search(r'Last\s+Online[^<]{0,150}?(\d+)\s+min', html, re.IGNORECASE) if m: return f"{m.group(1)}m" if re.search(r'Currently\s+(In-Game|Online)', html, re.IGNORECASE): return "online" return None