Expand website and game trackers
Some checks failed
Test and deploy staging / staging (push) Failing after 5m29s
Some checks failed
Test and deploy staging / staging (push) Failing after 5m29s
This commit is contained in:
60
scripts/download-game-images.py
Normal file
60
scripts/download-game-images.py
Normal file
@@ -0,0 +1,60 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Download the reviewed artwork manifest, preserving aspect ratio, as local WebP.
|
||||
Requires Pillow. Existing valid files are reused; failed downloads never replace files.
|
||||
"""
|
||||
import argparse
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
import io
|
||||
import json
|
||||
from pathlib import Path
|
||||
import time
|
||||
import urllib.request
|
||||
from PIL import Image, ImageOps
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
|
||||
def download(row):
|
||||
target = ROOT / 'static' / row['path'].lstrip('/')
|
||||
if target.exists():
|
||||
with Image.open(target) as img:
|
||||
if img.format == 'WEBP':
|
||||
return row['id'], None
|
||||
for attempt in range(3):
|
||||
try:
|
||||
request = urllib.request.Request(row['url'], headers={'User-Agent': 'PersonalGameJournal/1.0 (image catalog)'})
|
||||
with urllib.request.urlopen(request, timeout=45) as response:
|
||||
content = response.read()
|
||||
with Image.open(io.BytesIO(content)) as image:
|
||||
image = ImageOps.exif_transpose(image).convert('RGBA')
|
||||
image.thumbnail((row.get('maxSize', 800),) * 2, Image.Resampling.LANCZOS)
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
temporary = target.with_suffix('.tmp')
|
||||
image.save(temporary, format='WEBP', lossless=row.get('lossless', False), quality=88, method=6)
|
||||
temporary.replace(target)
|
||||
return row['id'], None
|
||||
except Exception as error:
|
||||
if attempt == 2:
|
||||
return row['id'], str(error)
|
||||
time.sleep(attempt + 1)
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument('--filter', default='')
|
||||
parser.add_argument('--workers', type=int, default=6)
|
||||
args = parser.parse_args()
|
||||
rows = json.loads((ROOT / 'data/game-images.json').read_text())
|
||||
rows = [r for r in rows if args.filter in r['id']]
|
||||
failed = []
|
||||
with ThreadPoolExecutor(max_workers=args.workers) as pool:
|
||||
for index, (name, error) in enumerate(pool.map(download, rows), 1):
|
||||
if error:
|
||||
failed.append((name, error))
|
||||
print('FAILED', name, error, flush=True)
|
||||
elif index % 100 == 0 or len(rows) < 100:
|
||||
print(f'{index}/{len(rows)} {name}', flush=True)
|
||||
print(f'{len(rows) - len(failed)}/{len(rows)} images ready', flush=True)
|
||||
if failed:
|
||||
raise SystemExit(1)
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
Reference in New Issue
Block a user