Stan Runge
c2ba8febb4
Some checks failed
Test and deploy staging / staging (push) Failing after 5m29s
61 lines
2.4 KiB
Python
61 lines
2.4 KiB
Python
#!/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()
|