From baefa105a9734a4ff64541832189ef72d8a4f486 Mon Sep 17 00:00:00 2001 From: saphid <4596216+saphid@users.noreply.github.com> Date: Mon, 28 Sep 2026 23:06:15 +1000 Subject: [PATCH] Artwork sources: every optional source falls back, within limits - Any failure of a source image or SteamGridDB (HTTPException, odd JSON) now becomes a warning and generated art, never an aborted install; refresh-art --all reports each app and carries on. - URL artwork goes through apk_sources._images: public addresses only, at most three redirects, and one overall deadline for all of an install's fetches. - PNGs are checked from their header only (any depth or interlace; Steam's Chromium decodes them), JPEGs may have trailing padding, and 4K screenshots are within limits. The slow pure-Python decoder is gone. - SteamGridDB title matching keeps letters of every script and never matches on an empty name. One warning per source slot, not per candidate. Co-Authored-By: Claude Opus 5.5 (1M context) --- tests/test_frame_android_library.py | 99 +++++++-------- tests/test_library_entrypoints.py | 17 ++- ui/apk_sources/_images.py | 27 ++++- ui/frame_android.py | 4 +- ui/frame_artwork.py | 180 ++++++++-------------------- ui/frame_steamgriddb.py | 33 +++-- 6 files changed, 158 insertions(+), 202 deletions(-) diff --git a/tests/test_frame_android_library.py b/tests/test_frame_android_library.py index c34046f..2db3527 100644 --- a/tests/test_frame_android_library.py +++ b/tests/test_frame_android_library.py @@ -117,29 +117,17 @@ FIXTURES = ROOT / 'tests/fixtures/library' class ArtworkTests(unittest.TestCase): - def test_icon_roundtrip_and_transparency(self): - w, h, pixels = art.decode((FIXTURES / 'icon.png').read_bytes()) - self.assertEqual((w, h), (2, 2)) - self.assertEqual(pixels, bytes([255, 0, 0, 255, 0, 255, 0, 255, - 0, 0, 255, 128, 0, 0, 0, 0])) - background = bytearray([10, 20, 30, 255] * 4) - art.stamp(background, 2, (w, h, pixels), 0, 0, 2) - self.assertEqual(background[:8], pixels[:8]) - self.assertEqual(background[8:12], bytes([4, 9, 142, 255])) - self.assertEqual(background[12:], bytes([10, 20, 30, 255])) - def test_source_inputs_and_url(self): - import io + from apk_sources import _images data = (FIXTURES / 'icon.png').read_bytes() - response = io.BytesIO(data) - response.geturl = lambda: 'https://example.org/icon.png' with patch('frame_steamgriddb.lookup', return_value=({}, [])), \ - patch.object(art.urllib.request, 'urlopen', return_value=response) as fetch: + patch.object(_images, 'fetch', return_value=(data, 'image/png')) as fetch: images, warnings = art.prepare('Game', artwork={'banner': data, 'icon': 'https://example.org/icon.png'}) self.assertEqual(images['banner'], ('png', data)) self.assertEqual(images['icon'], ('png', data)) self.assertEqual(warnings, []) - self.assertEqual(fetch.call_count, 1) + self.assertEqual(fetch.call_args.args[0], 'https://example.org/icon.png') + self.assertIsNotNone(fetch.call_args.kwargs['deadline']) def test_provider_precedence_and_bad_source_fallback(self): data = (FIXTURES / 'icon.png').read_bytes() @@ -150,48 +138,63 @@ class ArtworkTests(unittest.TestCase): self.assertEqual(images['icon'], ('png', data)) self.assertEqual(images['screenshot'], ('png', data)) self.assertNotIn('wide', images) - self.assertEqual(len(warnings), 2) + self.assertEqual(warnings, ['Source wide unavailable; using fallback art']) # one per slot, not per candidate + + def test_any_source_failure_falls_back_to_generated_art(self): + import http.client + from apk_sources import _images + data = (FIXTURES / 'icon.png').read_bytes() + for error in (http.client.RemoteDisconnected('gone'), http.client.IncompleteRead(b''), AttributeError('x')): + with self.subTest(error=type(error).__name__), \ + patch.object(_images, 'fetch', side_effect=error), \ + patch('frame_steamgriddb.lookup', side_effect=error): + images, warnings = art.prepare('Game', data, {'banner': 'https://example.org/b.png'}) + self.assertEqual(set(images), {'icon'}) + self.assertEqual(len(warnings), 2) + + def test_url_fetch_refuses_private_hosts_and_honours_deadline(self): + from apk_sources import _images, SourceError + local = [(2, 1, 6, '', ('127.0.0.1', 443))] + with patch.object(_images.socket, 'getaddrinfo', return_value=local), \ + self.assertRaisesRegex(SourceError, 'Private'): + art.fetch('https://example.org/icon.png') + public = [(2, 1, 6, '', ('93.184.216.34', 443))] + with patch.object(_images.socket, 'getaddrinfo', return_value=public), \ + patch.object(_images.socket, 'create_connection') as connect, \ + self.assertRaisesRegex(SourceError, 'too long'): + art.fetch('https://example.org/icon.png', deadline=time.monotonic() - 1) + connect.assert_not_called() + with self.assertRaises(SourceError): + art.fetch('file:///etc/passwd') def test_supplied_jpeg(self): data = (FIXTURES / 'icon.jpg').read_bytes() self.assertEqual(art.image_type(data), 'jpg') + self.assertEqual(art.image_type(data + b'\0' * 64), 'jpg') # trailing padding after EOI with self.assertRaises(ValueError): art.image_type(data[:30]) - def test_bad_artwork_and_expansion_limits(self): - import zlib + def test_png_variants_left_to_chromium_and_limits(self): + def png(w, h, depth, color, interlace): + return art.PNG + art.chunk(b'IHDR', struct.pack('>IIBBBBB', w, h, depth, color, 0, 0, interlace)) + \ + art.chunk(b'IEND', b'') + self.assertEqual(art.image_type(png(3840, 1240, 16, 6, 0)), 'png') + self.assertEqual(art.image_type(png(3840, 2160, 8, 2, 1)), 'png') + for bad, message in ((png(10000, 10, 8, 6, 0), 'dimensions'), (png(5000, 5000, 8, 6, 0), 'dimensions'), + (png(10, 10, 3, 6, 0), 'encoding'), (png(10, 10, 8, 5, 0), 'encoding')): + with self.subTest(message=message), self.assertRaisesRegex(ValueError, message): + art.image_type(bad) + broken = bytearray(png(10, 10, 8, 6, 0)) + broken[20] ^= 1 + with self.assertRaisesRegex(ValueError, 'checksum'): + art.image_type(bytes(broken)) + with self.assertRaises(ValueError): + art.image_type(art.PNG + b'junk') + + def test_bad_artwork_arguments(self): for value in ({'bad': b'bad'}, ['hero']): with self.subTest(value=value), self.assertRaises(ValueError): art.prepare('Game', artwork=value) - data = bytearray((FIXTURES / 'icon.png').read_bytes()) - data[45] ^= 1 - with self.assertRaisesRegex(ValueError, 'checksum'): - art.decode(bytes(data)) - bomb = art.PNG + art.chunk(b'IHDR', struct.pack('>IIBBBBB', 1, 1, 8, 6, 0, 0, 0)) + \ - art.chunk(b'IDAT', zlib.compress(b'\0' * 1_000_000)) + art.chunk(b'IEND', b'') - with self.assertRaisesRegex(ValueError, 'pixels'): - art.decode(bomb) - huge = art.PNG + art.chunk(b'IHDR', struct.pack('>IIBBBBB', 10000, 10000, 8, 6, 0, 0, 0)) + art.chunk(b'IEND', b'') - with self.assertRaisesRegex(ValueError, 'dimensions'): - art.decode(huge) - - def test_palette_and_filters(self): - import zlib - header = art.chunk(b'IHDR', struct.pack('>IIBBBBB', 2, 1, 1, 3, 0, 0, 0)) - data = art.PNG + header + art.chunk(b'PLTE', b'\xff\0\0\0\xff\0') + art.chunk(b'tRNS', b'\xff\x80') + \ - art.chunk(b'IDAT', zlib.compress(b'\0\x40')) + art.chunk(b'IEND', b'') - self.assertEqual(art.decode(data)[2], bytes([255, 0, 0, 255, 0, 255, 0, 128])) - for method in range(5): - # Two identical RGBA rows: exercise each predictor with known filtered bytes. - first = bytes([10, 20, 30, 255] * 2) - filtered = bytearray() - for x, value in enumerate(first): - a, b, c = first[x-4] if x >= 4 else 0, first[x], first[x-4] if x >= 4 else 0 - predictor = (0, a, b, (a+b)//2, b)[method] - filtered.append((value - predictor) & 255) - data = art.PNG + art.chunk(b'IHDR', struct.pack('>IIBBBBB', 2, 2, 8, 6, 0, 0, 0)) + \ - art.chunk(b'IDAT', zlib.compress(b'\0' + first + bytes([method]) + filtered)) + art.chunk(b'IEND', b'') - self.assertEqual(art.decode(data)[2], first * 2) def test_godot_project_icon(self): import io diff --git a/tests/test_library_entrypoints.py b/tests/test_library_entrypoints.py index e49104b..1560633 100644 --- a/tests/test_library_entrypoints.py +++ b/tests/test_library_entrypoints.py @@ -140,8 +140,10 @@ class EntryPoints(unittest.TestCase): stop.assert_not_called(); copy.assert_not_called(); self.assert_art() def test_all_refresh_reports_partial_failures(self): + import http.client with patch.object(android, 'list_apps', return_value=[{'package':'org.a.game'},{'package':'org.b.game'}]), \ - patch.object(android, '_meta_or_fail', side_effect=android.FrameError('missing')): + patch.object(android, '_meta_or_fail', side_effect=[http.client.RemoteDisconnected('gone'), + AttributeError('odd')]): result=android.refresh_art() self.assertEqual(len(result),2) self.assertTrue(all('error' in a for a in result)) @@ -173,7 +175,7 @@ class SteamGridDB(unittest.TestCase): def test_exact_match_and_top_votes_per_slot(self): calls=[] - def get(path,key): + def get(path,key,deadline=None): calls.append(path) if 'search' in path: return [{'id':1,'name':'Other Game'},{'id':2,'name':'Game'}] dims=(600,900) if '600x900' in path else (920,430) @@ -186,6 +188,17 @@ class SteamGridDB(unittest.TestCase): self.assertTrue(all(url.endswith('/top.png') for url in images.values())) self.assertTrue(all('/game/2?' in p for p in calls[1:])) + def test_unicode_titles_match_exactly_and_symbols_never_match_all(self): + self.assertEqual(sgdb._name('ビートセイバー VR!'), 'ビートセイバーvr') + with patch.object(sgdb,'api_key',return_value='test-key'), \ + patch.object(sgdb,'_get',return_value=[{'id':1,'name':'Unrelated'},{'id':2,'name':'!!!'}]) as get: + self.assertEqual(sgdb.lookup('★★★'),({},[])) + get.assert_not_called() + self.assertEqual(sgdb.lookup('ビートセイバー'),({},[])) + with patch.object(sgdb,'api_key',return_value='test-key'), \ + patch.object(sgdb,'_get',side_effect=AttributeError("'list' object has no attribute 'get'")): + self.assertEqual(sgdb.lookup('Game')[0],{}) + def test_wrong_title_and_failed_lookup_fall_back(self): with patch.object(sgdb,'api_key',return_value='test-key'),patch.object(sgdb,'_get',return_value=[{'id':1,'name':'Unrelated'}]): self.assertEqual(sgdb.lookup('Game'),({},[])) diff --git a/ui/apk_sources/_images.py b/ui/apk_sources/_images.py index df71483..5b91081 100644 --- a/ui/apk_sources/_images.py +++ b/ui/apk_sources/_images.py @@ -6,6 +6,7 @@ import secrets import socket import ssl import threading +import time from urllib.parse import urljoin, urlsplit from apk_sources import SourceError @@ -62,16 +63,25 @@ def image_type(data): raise SourceError('Artwork is not a supported image') -def fetch(url, redirects=3): +def fetch(url, redirects=3, deadline=None, limit=MAX_IMAGE): + """deadline: time.monotonic() value by which the whole fetch, redirects included, must finish.""" if not valid_url(url): raise SourceError('Artwork URL is not allowed') + + def remaining(): + if deadline is None: + return 10 + left = deadline - time.monotonic() + if left <= 0: + raise SourceError('Artwork download took too long') + return min(10, left) p = urlsplit(url) port = p.port or (443 if p.scheme == 'https' else 80) addresses = socket.getaddrinfo(p.hostname, port, type=socket.SOCK_STREAM) if not addresses or any(not ipaddress.ip_address(a[4][0]).is_global for a in addresses): raise SourceError('Private network artwork is not allowed') # Connect to the checked IP, never resolve again between validation and use. - sock = socket.create_connection((addresses[0][4][0], port), timeout=10) + sock = socket.create_connection((addresses[0][4][0], port), timeout=remaining()) conn = http.client.HTTPConnection(p.hostname, port, timeout=10) try: if p.scheme == 'https': @@ -81,15 +91,22 @@ def fetch(url, redirects=3): if p.query: path += '?' + p.query conn.request('GET', path, headers={'User-Agent': 'FrameControl/0.3.1', 'Accept': 'image/png,image/jpeg,image/webp,image/gif'}) + sock.settimeout(remaining()) response = conn.getresponse() if response.status in (301, 302, 303, 307, 308) and redirects: target = urljoin(url, response.getheader('Location', '')) conn.close() - return fetch(target, redirects - 1) + return fetch(target, redirects - 1, deadline, limit) if response.status != 200: raise SourceError('Artwork is unavailable') - data = response.read(MAX_IMAGE + 1) - if len(data) > MAX_IMAGE: + data = b'' + while len(data) <= limit: + sock.settimeout(remaining()) + chunk = response.read(min(65536, limit + 1 - len(data))) + if not chunk: + break + data += chunk + if len(data) > limit: raise SourceError('Artwork is too large') return data, image_type(data) finally: diff --git a/ui/frame_android.py b/ui/frame_android.py index e029644..25d43eb 100644 --- a/ui/frame_android.py +++ b/ui/frame_android.py @@ -256,8 +256,8 @@ def refresh_art(pkg=None, artwork=None): for app in list_apps(): try: results.append(refresh_art(app['package'], artwork)) - except (FrameError, OSError, ValueError) as e: - results.append({'package': app['package'], 'error': str(e)}) + except Exception as e: # one app's failure must not stop the others + results.append({'package': app['package'], 'label': app.get('label'), 'error': str(e) or type(e).__name__}) return results with _install_lock: m = _meta_or_fail(pkg) diff --git a/ui/frame_artwork.py b/ui/frame_artwork.py index 41ed93f..13db919 100644 --- a/ui/frame_artwork.py +++ b/ui/frame_artwork.py @@ -1,13 +1,12 @@ """Bounded artwork inputs for Steam library canvas rendering.""" import struct -import urllib.parse -import urllib.request +import time import zlib SLOTS = {'grid': (600, 900), 'wide': (920, 430), 'hero': (3840, 1240), 'logo': (1280, 480), 'icon': (256, 256)} MAX_IMAGE = 12 * 1024 * 1024 -MAX_PIXELS = 8_000_000 +MAX_PIXELS = 4096 * 4096 # a 4K screenshot; Chromium decodes it on the Frame PNG = b'\x89PNG\r\n\x1a\n' @@ -15,107 +14,28 @@ def chunk(kind, data): return struct.pack('>I', len(data)) + kind + data + struct.pack('>I', zlib.crc32(kind + data)) -def png(width, height, pixels): - stride = width * 4 - raw = b''.join(b'\0' + pixels[y * stride:(y + 1) * stride] for y in range(height)) - return PNG + chunk(b'IHDR', struct.pack('>IIBBBBB', width, height, 8, 6, 0, 0, 0)) + \ - chunk(b'IDAT', zlib.compress(raw, 6)) + chunk(b'IEND', b'') - - -def decode(data): - """Non-interlaced PNG, including packed palette/grayscale APK icons.""" - if not isinstance(data, bytes) or not data.startswith(PNG) or len(data) > MAX_IMAGE: - raise ValueError('expected a PNG image (at most 12 MiB)') - pos, packed, palette, alpha, header = 8, bytearray(), b'', b'', None - while pos + 12 <= len(data): - size = struct.unpack_from('>I', data, pos)[0] - kind, body = data[pos + 4:pos + 8], data[pos + 8:pos + 8 + size] - if pos + size + 12 > len(data): - raise ValueError('truncated PNG') - crc = struct.unpack_from('>I', data, pos + 8 + size)[0] - if zlib.crc32(kind + body) != crc: - raise ValueError('invalid PNG checksum') - if kind == b'IHDR': - if header is not None or size != 13: - raise ValueError('invalid PNG header') - header = struct.unpack('>IIBBBBB', body) - elif kind == b'PLTE': - palette = body - elif kind == b'tRNS': - alpha = body - elif kind == b'IDAT': - packed.extend(body) - elif kind == b'IEND': - break - pos += size + 12 - else: - raise ValueError('incomplete PNG') - if header is None: - raise ValueError('missing PNG header') - w, h, depth, color, compression, filtering, interlace = header - channels = {0: 1, 2: 3, 3: 1, 4: 2, 6: 4}.get(color) +def png_size(data): + """IHDR dimensions of a PNG of any bit depth or interlace; Steam's Chromium decodes the pixels.""" + if len(data) < 33 or data[12:16] != b'IHDR' or struct.unpack_from('>I', data, 8)[0] != 13: + raise ValueError('invalid PNG header') + if zlib.crc32(data[12:29]) != struct.unpack_from('>I', data, 29)[0]: + raise ValueError('invalid PNG checksum') + w, h, depth, color = struct.unpack_from('>IIBB', data, 16) + if color not in (0, 2, 3, 4, 6) or depth not in (1, 2, 4, 8, 16): + raise ValueError('unsupported PNG encoding') if not w or not h or w * h > MAX_PIXELS or w > 8192 or h > 8192: raise ValueError('PNG dimensions exceed limits') - if not channels or compression or filtering or interlace or depth not in (1, 2, 4, 8) or (depth != 8 and color not in (0, 3)): - raise ValueError('unsupported PNG encoding') - stride, bpp = (w * channels * depth + 7) // 8, max(1, channels * depth // 8) - expected = h * (stride + 1) - decoder = zlib.decompressobj() - raw = decoder.decompress(bytes(packed), expected + 1) - if len(raw) != expected or not decoder.eof: - raise ValueError('invalid PNG pixels') - pixels, prev = bytearray(), bytearray(stride) - for y in range(h): - start = y * (stride + 1) - method, row = raw[start], bytearray(raw[start + 1:start + 1 + stride]) - if method > 4: - raise ValueError('invalid PNG filter') - for x in range(stride): - a, b, c = row[x - bpp] if x >= bpp else 0, prev[x], prev[x - bpp] if x >= bpp else 0 - if method == 1: - row[x] = (row[x] + a) & 255 - elif method == 2: - row[x] = (row[x] + b) & 255 - elif method == 3: - row[x] = (row[x] + (a + b) // 2) & 255 - elif method == 4: - p = a + b - c - distances = (abs(p - a), abs(p - b), abs(p - c)) - row[x] = (row[x] + (a, b, c)[distances.index(min(distances))]) & 255 - for x in range(w): - if depth < 8: - value = (row[x * depth // 8] >> (8 - depth - x * depth % 8)) & ((1 << depth) - 1) - values = [value] - else: - values = row[x * channels:(x + 1) * channels] - if color == 3: - i = values[0] - if i * 3 + 3 > len(palette): - raise ValueError('invalid PNG palette') - rgba = palette[i * 3:i * 3 + 3] + bytes([alpha[i] if i < len(alpha) else 255]) - elif color in (0, 4): - gray = values[0] * 255 // ((1 << depth) - 1) - opacity = values[1] if color == 4 else (0 if alpha == struct.pack('>H', values[0]) else 255) - rgba = bytes([gray, gray, gray, opacity]) - else: - opacity = values[3] if color == 6 else (0 if alpha == struct.pack('>HHH', *values) else 255) - rgba = bytes(values[:3]) + bytes([opacity]) - pixels.extend(rgba) - prev = row - return w, h, pixels + return w, h def image_type(data): if not isinstance(data, bytes) or len(data) > MAX_IMAGE: raise ValueError('artwork must be image bytes or an HTTP(S) URL, at most 12 MiB') if data.startswith(PNG): - try: - decode(data) - except (zlib.error, struct.error) as e: - raise ValueError('invalid PNG image') from e + png_size(data) return 'png' - if data.startswith(b'\xff\xd8') and data.endswith(b'\xff\xd9'): - # Check JPEG SOF dimensions without depending on an image library. + if data.startswith(b'\xff\xd8'): + # Check JPEG SOF dimensions without depending on an image library; trailing padding is fine. pos = 2 while pos + 4 <= len(data) and data[pos] == 255: marker = data[pos + 1] @@ -126,7 +46,7 @@ def image_type(data): size = struct.unpack_from('>H', data, pos)[0] if size < 2 or pos + size > len(data): break - if marker in (0xc0, 0xc1, 0xc2) and size >= 8: + if 0xc0 <= marker <= 0xcf and marker not in (0xc4, 0xc8, 0xcc) and size >= 8: h, w = struct.unpack_from('>HH', data, pos + 3) if w and h and w * h <= MAX_PIXELS and max(w, h) <= 8192: return 'jpg' @@ -135,59 +55,53 @@ def image_type(data): raise ValueError('artwork must be a supported PNG or JPEG') -def stamp(pixels, width, icon, x, y, size): - iw, ih, source = icon - dw, dh = max(1, size * iw // max(iw, ih)), max(1, size * ih // max(iw, ih)) - x, y = x + (size - dw) // 2, y + (size - dh) // 2 - for yy in range(dh): - for xx in range(dw): - src = ((yy * ih // dh) * iw + xx * iw // dw) * 4 - dst = ((y + yy) * width + x + xx) * 4 - a = source[src + 3] - for c in range(3): - pixels[dst + c] = (source[src + c] * a + pixels[dst + c] * (255 - a)) // 255 - pixels[dst + 3] = a + pixels[dst + 3] * (255 - a) // 255 - - - -def fetch(value): +def fetch(value, deadline=None): if isinstance(value, str): - if urllib.parse.urlsplit(value).scheme not in ('http', 'https'): - raise ValueError('artwork URLs must use HTTP(S)') - request = urllib.request.Request(value, headers={'User-Agent': 'FrameControl/1.0'}) - with urllib.request.urlopen(request, timeout=20) as response: - if urllib.parse.urlsplit(response.geturl()).scheme not in ('http', 'https'): - raise ValueError('artwork redirect must use HTTP(S)') - value = response.read(MAX_IMAGE + 1) + from apk_sources import _images + # Public addresses only, at most three redirects, within the overall deadline. + value = _images.fetch(value, deadline=deadline, limit=MAX_IMAGE)[0] return image_type(value), value -def prepare(label, icon_png=None, artwork=None): - """Gather inputs; the Frame's Chromium canvas renders every final slot.""" +def prepare(label, icon_png=None, artwork=None, budget=90): + """Gather inputs; the Frame's Chromium canvas renders every final slot. + + Every source is optional: any failure falls back to generated art, within budget seconds overall.""" import frame_steamgriddb artwork = artwork or {} allowed = set(SLOTS) | {'banner', 'feature_graphic', 'screenshots', 'screenshot'} if not isinstance(artwork, dict) or set(artwork) - allowed: raise ValueError('unknown artwork slot') + deadline = time.monotonic() + budget supplied, warnings = {}, [] + + def get(value): + try: + return fetch(value, deadline) + except Exception: # an optional source never blocks the install; generated art covers it + return None for slot, value in artwork.items(): values = value if slot == 'screenshots' and isinstance(value, (list, tuple)) else [value] for candidate in values[:4]: - try: - supplied['screenshot' if slot == 'screenshots' else slot] = fetch(candidate) + image = get(candidate) + if image: + supplied['screenshot' if slot == 'screenshots' else slot] = image break - except (ValueError, OSError): - warnings.append('Source ' + slot + ' unavailable; using fallback art') + else: + warnings.append('Source ' + slot + ' unavailable; using fallback art') if 'icon' not in supplied and icon_png: - try: - supplied['icon'] = fetch(icon_png) - except (ValueError, OSError): - pass - provider, provider_warnings = frame_steamgriddb.lookup(label) + image = get(icon_png) + if image: + supplied['icon'] = image + try: + provider, provider_warnings = frame_steamgriddb.lookup(label, deadline) + except Exception: + provider, provider_warnings = {}, ['SteamGridDB unavailable; using source or generated art'] warnings.extend(provider_warnings) for slot, value in provider.items(): - try: - supplied[slot] = fetch(value) - except (ValueError, OSError): + image = get(value) + if image: + supplied[slot] = image + else: warnings.append('SteamGridDB ' + slot + ' download failed; using fallback art') return supplied, warnings diff --git a/ui/frame_steamgriddb.py b/ui/frame_steamgriddb.py index 2a580c9..6384d3d 100644 --- a/ui/frame_steamgriddb.py +++ b/ui/frame_steamgriddb.py @@ -3,6 +3,8 @@ import json import os import re import tempfile +import time +import unicodedata import urllib.parse import urllib.request @@ -48,36 +50,42 @@ def save_settings(body): return settings() -def _get(path, key): +def _get(path, key, deadline=None): + timeout = 12 if deadline is None else min(12, deadline - time.monotonic()) + if timeout <= 0: + raise ValueError('SteamGridDB lookup took too long') request = urllib.request.Request(API + path, headers={'Authorization': 'Bearer ' + key, 'User-Agent': 'FrameControl/1.0'}) # Do not carry the credential to redirects or include it in error messages. class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, *args, **kwargs): return None - with urllib.request.build_opener(NoRedirect()).open(request, timeout=12) as response: + with urllib.request.build_opener(NoRedirect()).open(request, timeout=timeout) as response: data = response.read(MAX_JSON + 1) if len(data) > MAX_JSON: raise ValueError('SteamGridDB response too large') result = json.loads(data) - if not result.get('success') or not isinstance(result.get('data'), list): + if not isinstance(result, dict) or not result.get('success') or not isinstance(result.get('data'), list): raise ValueError('SteamGridDB lookup failed') return result['data'] def _name(value): - return re.sub(r'[^a-z0-9]', '', str(value).casefold()) + # Letters and digits of any script, so a CJK title never normalises to ''. + return ''.join(c for c in unicodedata.normalize('NFKC', str(value or '')).casefold() if c.isalnum()) -def lookup(name): +def lookup(name, deadline=None): """Best-voted art per slot for an exact title match; unrelated games are never guessed.""" key = api_key() if not key: return {}, [] try: - matches = _get('/search/autocomplete/' + urllib.parse.quote(name, safe=''), key) - names = {_name(name), _name(re.sub(r'\s+VR$', '', name, flags=re.I))} - game = next((g for g in matches if _name(g.get('name')) in names), None) + names = {_name(name), _name(re.sub(r'\s+VR$', '', name, flags=re.I))} - {''} + if not names: + return {}, [] + matches = _get('/search/autocomplete/' + urllib.parse.quote(name, safe=''), key, deadline) + game = next((g for g in matches if isinstance(g, dict) and _name(g.get('name')) in names), None) if not game: return {}, [] gid = int(game['id']) @@ -88,16 +96,17 @@ def lookup(name): query = {'types': 'static', 'nsfw': 'false', 'humor': 'false', 'mimes': 'image/png,image/jpeg'} if dimensions: query['dimensions'] = dimensions - records = _get('/' + kind + '/game/' + str(gid) + '?' + urllib.parse.urlencode(query), key) - records = [r for r in records if r.get('url', '').startswith('https://') and not r.get('nsfw')] + records = _get('/' + kind + '/game/' + str(gid) + '?' + urllib.parse.urlencode(query), key, deadline) + records = [r for r in records if isinstance(r, dict) and str(r.get('url', '')).startswith('https://') + and not r.get('nsfw')] if dimensions: w, h = map(int, dimensions.split('x')) records = [r for r in records if (r.get('width'), r.get('height')) == (w, h)] records.sort(key=lambda r: (int(r.get('score') or 0), int(r.get('upvotes') or 0)), reverse=True) if records: result[slot] = records[0]['url'] - except (OSError, ValueError, KeyError, TypeError): + except Exception: # HTTPException, odd JSON: this slot falls back warnings.append('SteamGridDB ' + slot + ' unavailable; using source or generated art') return result, warnings - except (OSError, ValueError, KeyError, TypeError): + except Exception: return {}, ['SteamGridDB unavailable; using source or generated art']