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) <noreply@anthropic.com>
This commit is contained in:
saphidandClaude Opus 5.5 committed 2026-09-28 23:06:15 +10:00
1 parent 70897cfd1d
commit baefa105a9
6 files changed
+154 -198

No files matched your search

+50 -47
View File
@@ -117,29 +117,17 @@ FIXTURES = ROOT / 'tests/fixtures/library'
class ArtworkTests(unittest.TestCase): 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): def test_source_inputs_and_url(self):
import io from apk_sources import _images
data = (FIXTURES / 'icon.png').read_bytes() 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=({}, [])), \ 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'}) images, warnings = art.prepare('Game', artwork={'banner': data, 'icon': 'https://example.org/icon.png'})
self.assertEqual(images['banner'], ('png', data)) self.assertEqual(images['banner'], ('png', data))
self.assertEqual(images['icon'], ('png', data)) self.assertEqual(images['icon'], ('png', data))
self.assertEqual(warnings, []) 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): def test_provider_precedence_and_bad_source_fallback(self):
data = (FIXTURES / 'icon.png').read_bytes() data = (FIXTURES / 'icon.png').read_bytes()
@@ -150,48 +138,63 @@ class ArtworkTests(unittest.TestCase):
self.assertEqual(images['icon'], ('png', data)) self.assertEqual(images['icon'], ('png', data))
self.assertEqual(images['screenshot'], ('png', data)) self.assertEqual(images['screenshot'], ('png', data))
self.assertNotIn('wide', images) self.assertNotIn('wide', images)
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) 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): def test_supplied_jpeg(self):
data = (FIXTURES / 'icon.jpg').read_bytes() data = (FIXTURES / 'icon.jpg').read_bytes()
self.assertEqual(art.image_type(data), 'jpg') 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): with self.assertRaises(ValueError):
art.image_type(data[:30]) art.image_type(data[:30])
def test_bad_artwork_and_expansion_limits(self): def test_png_variants_left_to_chromium_and_limits(self):
import zlib 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']): for value in ({'bad': b'bad'}, ['hero']):
with self.subTest(value=value), self.assertRaises(ValueError): with self.subTest(value=value), self.assertRaises(ValueError):
art.prepare('Game', artwork=value) 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): def test_godot_project_icon(self):
import io import io
+15 -2
View File
@@ -140,8 +140,10 @@ class EntryPoints(unittest.TestCase):
stop.assert_not_called(); copy.assert_not_called(); self.assert_art() stop.assert_not_called(); copy.assert_not_called(); self.assert_art()
def test_all_refresh_reports_partial_failures(self): 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'}]), \ 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() result=android.refresh_art()
self.assertEqual(len(result),2) self.assertEqual(len(result),2)
self.assertTrue(all('error' in a for a in result)) 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): def test_exact_match_and_top_votes_per_slot(self):
calls=[] calls=[]
def get(path,key): def get(path,key,deadline=None):
calls.append(path) calls.append(path)
if 'search' in path: return [{'id':1,'name':'Other Game'},{'id':2,'name':'Game'}] if 'search' in path: return [{'id':1,'name':'Other Game'},{'id':2,'name':'Game'}]
dims=(600,900) if '600x900' in path else (920,430) 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(url.endswith('/top.png') for url in images.values()))
self.assertTrue(all('/game/2?' in p for p in calls[1:])) 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): 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'}]): 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'),({},[])) self.assertEqual(sgdb.lookup('Game'),({},[]))
+22 -5
View File
@@ -6,6 +6,7 @@ import secrets
import socket import socket
import ssl import ssl
import threading import threading
import time
from urllib.parse import urljoin, urlsplit from urllib.parse import urljoin, urlsplit
from apk_sources import SourceError from apk_sources import SourceError
@@ -62,16 +63,25 @@ def image_type(data):
raise SourceError('Artwork is not a supported image') 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): if not valid_url(url):
raise SourceError('Artwork URL is not allowed') 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) p = urlsplit(url)
port = p.port or (443 if p.scheme == 'https' else 80) port = p.port or (443 if p.scheme == 'https' else 80)
addresses = socket.getaddrinfo(p.hostname, port, type=socket.SOCK_STREAM) 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): 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') raise SourceError('Private network artwork is not allowed')
# Connect to the checked IP, never resolve again between validation and use. # 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) conn = http.client.HTTPConnection(p.hostname, port, timeout=10)
try: try:
if p.scheme == 'https': if p.scheme == 'https':
@@ -81,15 +91,22 @@ def fetch(url, redirects=3):
if p.query: if p.query:
path += '?' + 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'}) 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() response = conn.getresponse()
if response.status in (301, 302, 303, 307, 308) and redirects: if response.status in (301, 302, 303, 307, 308) and redirects:
target = urljoin(url, response.getheader('Location', '')) target = urljoin(url, response.getheader('Location', ''))
conn.close() conn.close()
return fetch(target, redirects - 1) return fetch(target, redirects - 1, deadline, limit)
if response.status != 200: if response.status != 200:
raise SourceError('Artwork is unavailable') raise SourceError('Artwork is unavailable')
data = response.read(MAX_IMAGE + 1) data = b''
if len(data) > MAX_IMAGE: 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') raise SourceError('Artwork is too large')
return data, image_type(data) return data, image_type(data)
finally: finally:
+2 -2
View File
@@ -256,8 +256,8 @@ def refresh_art(pkg=None, artwork=None):
for app in list_apps(): for app in list_apps():
try: try:
results.append(refresh_art(app['package'], artwork)) results.append(refresh_art(app['package'], artwork))
except (FrameError, OSError, ValueError) as e: except Exception as e: # one app's failure must not stop the others
results.append({'package': app['package'], 'error': str(e)}) results.append({'package': app['package'], 'label': app.get('label'), 'error': str(e) or type(e).__name__})
return results return results
with _install_lock: with _install_lock:
m = _meta_or_fail(pkg) m = _meta_or_fail(pkg)
+44 -130
View File
@@ -1,13 +1,12 @@
"""Bounded artwork inputs for Steam library canvas rendering.""" """Bounded artwork inputs for Steam library canvas rendering."""
import struct import struct
import urllib.parse import time
import urllib.request
import zlib import zlib
SLOTS = {'grid': (600, 900), 'wide': (920, 430), 'hero': (3840, 1240), SLOTS = {'grid': (600, 900), 'wide': (920, 430), 'hero': (3840, 1240),
'logo': (1280, 480), 'icon': (256, 256)} 'logo': (1280, 480), 'icon': (256, 256)}
MAX_IMAGE = 12 * 1024 * 1024 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' 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)) return struct.pack('>I', len(data)) + kind + data + struct.pack('>I', zlib.crc32(kind + data))
def png(width, height, pixels): def png_size(data):
stride = width * 4 """IHDR dimensions of a PNG of any bit depth or interlace; Steam's Chromium decodes the pixels."""
raw = b''.join(b'\0' + pixels[y * stride:(y + 1) * stride] for y in range(height)) if len(data) < 33 or data[12:16] != b'IHDR' or struct.unpack_from('>I', data, 8)[0] != 13:
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') raise ValueError('invalid PNG header')
header = struct.unpack('>IIBBBBB', body) if zlib.crc32(data[12:29]) != struct.unpack_from('>I', data, 29)[0]:
elif kind == b'PLTE': raise ValueError('invalid PNG checksum')
palette = body w, h, depth, color = struct.unpack_from('>IIBB', data, 16)
elif kind == b'tRNS': if color not in (0, 2, 3, 4, 6) or depth not in (1, 2, 4, 8, 16):
alpha = body raise ValueError('unsupported PNG encoding')
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)
if not w or not h or w * h > MAX_PIXELS or w > 8192 or h > 8192: if not w or not h or w * h > MAX_PIXELS or w > 8192 or h > 8192:
raise ValueError('PNG dimensions exceed limits') 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)): return w, h
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
def image_type(data): def image_type(data):
if not isinstance(data, bytes) or len(data) > MAX_IMAGE: 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') raise ValueError('artwork must be image bytes or an HTTP(S) URL, at most 12 MiB')
if data.startswith(PNG): if data.startswith(PNG):
try: png_size(data)
decode(data)
except (zlib.error, struct.error) as e:
raise ValueError('invalid PNG image') from e
return 'png' return 'png'
if data.startswith(b'\xff\xd8') and data.endswith(b'\xff\xd9'): if data.startswith(b'\xff\xd8'):
# Check JPEG SOF dimensions without depending on an image library. # Check JPEG SOF dimensions without depending on an image library; trailing padding is fine.
pos = 2 pos = 2
while pos + 4 <= len(data) and data[pos] == 255: while pos + 4 <= len(data) and data[pos] == 255:
marker = data[pos + 1] marker = data[pos + 1]
@@ -126,7 +46,7 @@ def image_type(data):
size = struct.unpack_from('>H', data, pos)[0] size = struct.unpack_from('>H', data, pos)[0]
if size < 2 or pos + size > len(data): if size < 2 or pos + size > len(data):
break 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) h, w = struct.unpack_from('>HH', data, pos + 3)
if w and h and w * h <= MAX_PIXELS and max(w, h) <= 8192: if w and h and w * h <= MAX_PIXELS and max(w, h) <= 8192:
return 'jpg' return 'jpg'
@@ -135,59 +55,53 @@ def image_type(data):
raise ValueError('artwork must be a supported PNG or JPEG') raise ValueError('artwork must be a supported PNG or JPEG')
def stamp(pixels, width, icon, x, y, size): def fetch(value, deadline=None):
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):
if isinstance(value, str): if isinstance(value, str):
if urllib.parse.urlsplit(value).scheme not in ('http', 'https'): from apk_sources import _images
raise ValueError('artwork URLs must use HTTP(S)') # Public addresses only, at most three redirects, within the overall deadline.
request = urllib.request.Request(value, headers={'User-Agent': 'FrameControl/1.0'}) value = _images.fetch(value, deadline=deadline, limit=MAX_IMAGE)[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)
return image_type(value), value return image_type(value), value
def prepare(label, icon_png=None, artwork=None): def prepare(label, icon_png=None, artwork=None, budget=90):
"""Gather inputs; the Frame's Chromium canvas renders every final slot.""" """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 import frame_steamgriddb
artwork = artwork or {} artwork = artwork or {}
allowed = set(SLOTS) | {'banner', 'feature_graphic', 'screenshots', 'screenshot'} allowed = set(SLOTS) | {'banner', 'feature_graphic', 'screenshots', 'screenshot'}
if not isinstance(artwork, dict) or set(artwork) - allowed: if not isinstance(artwork, dict) or set(artwork) - allowed:
raise ValueError('unknown artwork slot') raise ValueError('unknown artwork slot')
deadline = time.monotonic() + budget
supplied, warnings = {}, [] 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(): for slot, value in artwork.items():
values = value if slot == 'screenshots' and isinstance(value, (list, tuple)) else [value] values = value if slot == 'screenshots' and isinstance(value, (list, tuple)) else [value]
for candidate in values[:4]: for candidate in values[:4]:
try: image = get(candidate)
supplied['screenshot' if slot == 'screenshots' else slot] = fetch(candidate) if image:
supplied['screenshot' if slot == 'screenshots' else slot] = image
break break
except (ValueError, OSError): else:
warnings.append('Source ' + slot + ' unavailable; using fallback art') warnings.append('Source ' + slot + ' unavailable; using fallback art')
if 'icon' not in supplied and icon_png: if 'icon' not in supplied and icon_png:
image = get(icon_png)
if image:
supplied['icon'] = image
try: try:
supplied['icon'] = fetch(icon_png) provider, provider_warnings = frame_steamgriddb.lookup(label, deadline)
except (ValueError, OSError): except Exception:
pass provider, provider_warnings = {}, ['SteamGridDB unavailable; using source or generated art']
provider, provider_warnings = frame_steamgriddb.lookup(label)
warnings.extend(provider_warnings) warnings.extend(provider_warnings)
for slot, value in provider.items(): for slot, value in provider.items():
try: image = get(value)
supplied[slot] = fetch(value) if image:
except (ValueError, OSError): supplied[slot] = image
else:
warnings.append('SteamGridDB ' + slot + ' download failed; using fallback art') warnings.append('SteamGridDB ' + slot + ' download failed; using fallback art')
return supplied, warnings return supplied, warnings
+21 -12
View File
@@ -3,6 +3,8 @@ import json
import os import os
import re import re
import tempfile import tempfile
import time
import unicodedata
import urllib.parse import urllib.parse
import urllib.request import urllib.request
@@ -48,36 +50,42 @@ def save_settings(body):
return settings() 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, request = urllib.request.Request(API + path, headers={'Authorization': 'Bearer ' + key,
'User-Agent': 'FrameControl/1.0'}) 'User-Agent': 'FrameControl/1.0'})
# Do not carry the credential to redirects or include it in error messages. # Do not carry the credential to redirects or include it in error messages.
class NoRedirect(urllib.request.HTTPRedirectHandler): class NoRedirect(urllib.request.HTTPRedirectHandler):
def redirect_request(self, *args, **kwargs): def redirect_request(self, *args, **kwargs):
return None 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) data = response.read(MAX_JSON + 1)
if len(data) > MAX_JSON: if len(data) > MAX_JSON:
raise ValueError('SteamGridDB response too large') raise ValueError('SteamGridDB response too large')
result = json.loads(data) 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') raise ValueError('SteamGridDB lookup failed')
return result['data'] return result['data']
def _name(value): 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.""" """Best-voted art per slot for an exact title match; unrelated games are never guessed."""
key = api_key() key = api_key()
if not key: if not key:
return {}, [] return {}, []
try: try:
matches = _get('/search/autocomplete/' + urllib.parse.quote(name, safe=''), key) names = {_name(name), _name(re.sub(r'\s+VR$', '', name, flags=re.I))} - {''}
names = {_name(name), _name(re.sub(r'\s+VR$', '', name, flags=re.I))} if not names:
game = next((g for g in matches if _name(g.get('name')) in names), None) 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: if not game:
return {}, [] return {}, []
gid = int(game['id']) gid = int(game['id'])
@@ -88,16 +96,17 @@ def lookup(name):
query = {'types': 'static', 'nsfw': 'false', 'humor': 'false', 'mimes': 'image/png,image/jpeg'} query = {'types': 'static', 'nsfw': 'false', 'humor': 'false', 'mimes': 'image/png,image/jpeg'}
if dimensions: if dimensions:
query['dimensions'] = dimensions query['dimensions'] = dimensions
records = _get('/' + kind + '/game/' + str(gid) + '?' + urllib.parse.urlencode(query), key) records = _get('/' + kind + '/game/' + str(gid) + '?' + urllib.parse.urlencode(query), key, deadline)
records = [r for r in records if r.get('url', '').startswith('https://') and not r.get('nsfw')] records = [r for r in records if isinstance(r, dict) and str(r.get('url', '')).startswith('https://')
and not r.get('nsfw')]
if dimensions: if dimensions:
w, h = map(int, dimensions.split('x')) w, h = map(int, dimensions.split('x'))
records = [r for r in records if (r.get('width'), r.get('height')) == (w, h)] 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) records.sort(key=lambda r: (int(r.get('score') or 0), int(r.get('upvotes') or 0)), reverse=True)
if records: if records:
result[slot] = records[0]['url'] 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') warnings.append('SteamGridDB ' + slot + ' unavailable; using source or generated art')
return result, warnings return result, warnings
except (OSError, ValueError, KeyError, TypeError): except Exception:
return {}, ['SteamGridDB unavailable; using source or generated art'] return {}, ['SteamGridDB unavailable; using source or generated art']