manzoloandClaude Opus 5.5
33e2598718
Lavoro della sessione precedente: colonna sonora (titolo, paese, ricordi) con muto e volume, vecchietta e vecchietto sulla panchina, scene di Pieve, oratorio, Via de' Medici, Villa Adami, salita, campeggio, Bosco ai Frati e viale dei cipressi, panorama dell'introduzione, tools/genera.py per gli sfondi dalle foto. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01UdeKd62opy9PKgpQXtVctF
303 lines
14 KiB
Python
303 lines
14 KiB
Python
#!/usr/bin/env python3
|
||
"""Trasforma le foto del gioco con Images Edit; --dry-run non usa rete né scrive file."""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import base64
|
||
import hashlib
|
||
import html
|
||
import io
|
||
import json
|
||
import os
|
||
from pathlib import Path
|
||
import sys
|
||
import tempfile
|
||
import time
|
||
from datetime import datetime, timezone
|
||
from urllib.parse import quote
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
EXTENSIONS = {'.jpg', '.jpeg', '.png', '.webp'}
|
||
SIZE = (1536, 864)
|
||
DEFAULT_PROMPT = """Use case: style-transfer.
|
||
Create a clean background for La Leggenda del Regolo, a point-and-click adventure
|
||
set in San Piero a Sieve, Mugello, Tuscany, in late May 2007.
|
||
The input photograph is the architectural and composition reference.
|
||
Transform it into early-1990s (1993) VGA adventure pixel art: carefully placed
|
||
pixel clusters, limited warm palette of about 64 colors, crisp edges, no blur,
|
||
no smooth gradients, no photographic texture, no anti-aliasing.
|
||
Preserve the actual architecture, perspective, relative building positions,
|
||
historic details, openings, road layout and ground slope. Do not mirror the view.
|
||
Warm late-May afternoon light, consistent across the game's outdoor scenes.
|
||
Output one opaque 1536x864 image, 16:9, full bleed, no frame or letterboxing.
|
||
For a non-16:9 source, extend the framing conservatively without stretching
|
||
buildings or cropping essential architecture. Do not invent landmarks.
|
||
Keep existing foreground ground readable for walking without flattening slopes
|
||
or moving roads. Remove people and vehicles to leave an empty background; game
|
||
characters will be rendered separately. Remove readable lettering from signs.
|
||
No new characters, animals, text, logos, UI, inventory, cursor or watermark.
|
||
"""
|
||
|
||
|
||
def positive(value):
|
||
number = int(value)
|
||
if number <= 0:
|
||
raise argparse.ArgumentTypeError('deve essere maggiore di zero')
|
||
return number
|
||
|
||
|
||
def atomic_write(path, data):
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
with tempfile.NamedTemporaryFile(dir=path.parent, delete=False) as stream:
|
||
temporary = Path(stream.name)
|
||
stream.write(data)
|
||
try:
|
||
temporary.replace(path)
|
||
finally:
|
||
temporary.unlink(missing_ok=True)
|
||
|
||
|
||
def save_log(path, records):
|
||
atomic_write(path, (json.dumps(records, indent=2, ensure_ascii=False) + '\n').encode())
|
||
|
||
|
||
def read_log(path):
|
||
if not path.exists():
|
||
return {}
|
||
records = json.loads(path.read_text(encoding='utf-8'))
|
||
if not isinstance(records, dict):
|
||
raise ValueError(f'Registro non valido: {path}')
|
||
return records
|
||
|
||
|
||
def photos_in(folder):
|
||
if not folder.is_dir():
|
||
raise ValueError(f'Cartella originali non trovata: {folder}')
|
||
photos = sorted((p for p in folder.iterdir() if p.is_file() and
|
||
p.suffix.lower() in EXTENSIONS), key=lambda p: p.name.casefold())
|
||
names = set()
|
||
for photo in photos:
|
||
key = photo.stem.casefold()
|
||
if key in names:
|
||
raise ValueError(f'Nome duplicato: {photo.stem}. Rinominare le foto con lo stesso nome base.')
|
||
names.add(key)
|
||
return photos
|
||
|
||
|
||
def prompt_for(photo, folder):
|
||
common = folder / 'base.txt'
|
||
custom = folder / f'{photo.stem}.txt'
|
||
text = common.read_text(encoding='utf-8') if common.exists() else DEFAULT_PROMPT
|
||
if custom.exists():
|
||
text += '\n\nSCENE-SPECIFIC INSTRUCTIONS:\n' + custom.read_text(encoding='utf-8')
|
||
if not text.strip() or len(text) > 32000:
|
||
raise ValueError(f'Prompt vuoto o troppo lungo per {photo.name}')
|
||
return text
|
||
|
||
|
||
def validate_image(path, expected_size=None):
|
||
from PIL import Image
|
||
with Image.open(path) as image:
|
||
image.load()
|
||
if expected_size and (image.size != expected_size or image.format != 'PNG'):
|
||
raise ValueError(f'PNG {expected_size[0]}x{expected_size[1]} atteso: {path}')
|
||
|
||
|
||
def normalize_response(raw, destination):
|
||
from PIL import Image
|
||
with Image.open(raw) as image:
|
||
image.load()
|
||
if abs(image.width / image.height - 16 / 9) > 0.02:
|
||
raise ValueError('Risposta non 16:9: conservata in .responses; serve revisione, nessun ritaglio automatico.')
|
||
image = image.convert('RGB')
|
||
if image.size != SIZE:
|
||
image = image.resize(SIZE, Image.Resampling.NEAREST)
|
||
data = io.BytesIO()
|
||
image.save(data, format='PNG')
|
||
atomic_write(destination, data.getvalue())
|
||
|
||
|
||
def gallery(photos, output, records):
|
||
def link(path):
|
||
return html.escape(quote(os.path.relpath(path, output), safe='/'), quote=True)
|
||
|
||
cards = []
|
||
for photo in photos:
|
||
result = output / f'{photo.stem}.png'
|
||
state = records.get(photo.name, {}).get('status', 'da generare')
|
||
if result.exists() and state != 'completed':
|
||
state = 'file presente (verificare)'
|
||
after = (f'<img loading="lazy" src="{link(result)}" alt="{html.escape(photo.stem)} — risultato">'
|
||
if result.exists() else '<p class="pending">Non ancora generata</p>')
|
||
cards.append(f'<section><h2>{html.escape(photo.stem)}</h2><p>{html.escape(state)}</p>'
|
||
f'<div class="pair"><figure><img loading="lazy" src="{link(photo)}" '
|
||
f'alt="{html.escape(photo.stem)} — originale"><figcaption>Fotografia originale</figcaption>'
|
||
f'</figure><figure>{after}<figcaption>Pixel art · 1536×864</figcaption></figure></div></section>')
|
||
page = '''<!doctype html><html lang="it"><meta charset="utf-8">
|
||
<meta name="viewport" content="width=device-width, initial-scale=1">
|
||
<title>La Leggenda del Regolo — confronto sfondi</title>
|
||
<style>body{font:16px system-ui;background:#182027;color:#f6ebd8;margin:2rem auto;padding:0 1rem;max-width:1400px}
|
||
.pair{display:grid;grid-template-columns:1fr 1fr;gap:1rem}figure{margin:0;background:#26323b;padding:1rem}
|
||
img{width:100%;height:360px;object-fit:contain}figure+figure img{image-rendering:pixelated}
|
||
figcaption{padding-top:.5rem}.pending{height:328px;display:grid;place-items:center}
|
||
section{margin:2rem 0}a{color:#f4c47f}@media(max-width:700px){.pair{grid-template-columns:1fr}img{height:auto}}</style>
|
||
<h1>La Leggenda del Regolo — sfondi dalle fotografie</h1>
|
||
<p>San Piero a Sieve · originali e risultati. Le immagini mancanti sono indicate senza anteprime simulate.</p>
|
||
''' + '\n'.join(cards) + '</html>\n'
|
||
atomic_write(output / 'index.html', page.encode())
|
||
|
||
|
||
def run(args):
|
||
photos = photos_in(args.input_dir)
|
||
if args.input_dir.resolve() == args.output_dir.resolve():
|
||
raise ValueError('Originali e risultati devono avere cartelle diverse.')
|
||
log_path = args.output_dir / 'registro.json'
|
||
records = read_log(log_path)
|
||
if args.gallery_only:
|
||
gallery(photos, args.output_dir, records)
|
||
print(f'Confronto: {args.output_dir / "index.html"}')
|
||
return 0
|
||
|
||
pending = [p for p in photos if not (args.output_dir / f'{p.stem}.png').exists()]
|
||
limit = 1 if args.test else args.limit
|
||
selected = pending[:limit] if limit else pending
|
||
prompts = {p.name: prompt_for(p, args.prompt_dir) for p in selected}
|
||
cached = {p.name for p in selected if (args.output_dir / '.responses' / f'{p.stem}.png').exists()}
|
||
api_count = len(selected) - len(cached)
|
||
print(f'Foto trovate: {len(photos)}; PNG già presenti: {len(photos) - len(pending)}')
|
||
print(f'Da elaborare: {len(selected)}; da cache: {len(cached)}; immagini da inviare all’API: {api_count}')
|
||
print(f'Modello: {args.model}; formato: 1536x864; qualità: {args.quality}')
|
||
for photo in selected:
|
||
print(f' {photo.name} -> {args.output_dir / (photo.stem + ".png")} '
|
||
f'[{"cache locale" if photo.name in cached else "Images Edit"}]')
|
||
if args.dry_run:
|
||
print('DRY RUN: nessuna chiamata API e nessun file scritto.')
|
||
return 0
|
||
|
||
# Validate all existing results and selected sources before any paid request.
|
||
for photo in photos:
|
||
result = args.output_dir / f'{photo.stem}.png'
|
||
if result.exists():
|
||
validate_image(result, SIZE)
|
||
for photo in selected:
|
||
validate_image(photo)
|
||
|
||
client = None
|
||
if api_count:
|
||
if not os.environ.get('OPENAI_API_KEY'):
|
||
raise ValueError('Imposta OPENAI_API_KEY nell’ambiente. Non inserirla nel codice o nei prompt.')
|
||
try:
|
||
import openai
|
||
except ImportError as exc:
|
||
raise ValueError('Installa le dipendenze: python3 -m pip install -r requirements.txt') from exc
|
||
print(f'Possibili costi: {api_count} immagini, fino a {api_count * args.attempts} richieste con i retry.')
|
||
print('Un timeout può avvenire dopo l’elaborazione: un retry potrebbe comportare un altro addebito.')
|
||
try:
|
||
answer = input('Per procedere scrivi GENERA: ')
|
||
except EOFError:
|
||
answer = ''
|
||
if answer.strip() != 'GENERA':
|
||
print('Annullato: nessuna chiamata API.')
|
||
return 0
|
||
client = openai.OpenAI(timeout=args.timeout, max_retries=0)
|
||
|
||
failures = 0
|
||
try:
|
||
for photo in selected:
|
||
target = args.output_dir / f'{photo.stem}.png'
|
||
raw = args.output_dir / '.responses' / f'{photo.stem}.png'
|
||
previous = records.get(photo.name, {})
|
||
record = dict(previous) if raw.exists() else {
|
||
'source': str(photo.resolve()), 'output': str(target.resolve()),
|
||
'source_sha256': hashlib.sha256(photo.read_bytes()).hexdigest(),
|
||
'prompt': prompts[photo.name], 'model': args.model,
|
||
'size': '1536x864', 'quality': args.quality, 'attempts': 0,
|
||
}
|
||
records[photo.name] = record
|
||
try:
|
||
if not raw.exists():
|
||
for attempt in range(args.attempts):
|
||
record.update(status='processing', attempts=record['attempts'] + 1,
|
||
updated_at=datetime.now(timezone.utc).isoformat())
|
||
save_log(log_path, records)
|
||
try:
|
||
from PIL import Image, ImageOps
|
||
# Honor EXIF rotation and supply PNG regardless of source format.
|
||
with Image.open(photo) as source:
|
||
upload = io.BytesIO()
|
||
ImageOps.exif_transpose(source).convert('RGB').save(upload, format='PNG')
|
||
if upload.tell() >= 50 * 1024 * 1024:
|
||
raise ValueError('Immagine convertita oltre il limite di 50 MB.')
|
||
response = client.images.edit(
|
||
model=args.model, image=(photo.stem + '.png', upload.getvalue(), 'image/png'),
|
||
prompt=prompts[photo.name], size='1536x864', quality=args.quality,
|
||
output_format='png', n=1,
|
||
)
|
||
if not response.data or not response.data[0].b64_json:
|
||
raise ValueError('Risposta senza immagine base64.')
|
||
# Persist before post-processing, so local errors never require regeneration.
|
||
atomic_write(raw, base64.b64decode(response.data[0].b64_json, validate=True))
|
||
record['request_id'] = getattr(response, '_request_id', None)
|
||
usage = getattr(response, 'usage', None)
|
||
if usage is not None:
|
||
record['usage'] = usage.model_dump(mode='json')
|
||
break
|
||
except (openai.APIConnectionError, openai.APIStatusError) as exc:
|
||
status = getattr(exc, 'status_code', None)
|
||
transient = status is None or status in (408, 409, 429) or status >= 500
|
||
record['last_api_error'] = {'type': type(exc).__name__, 'status_code': status}
|
||
save_log(log_path, records)
|
||
if not transient or attempt + 1 == args.attempts:
|
||
raise
|
||
delay = min(2 ** (attempt + 1), 60)
|
||
print(f' {photo.name}: {type(exc).__name__}, nuovo tentativo tra {delay}s.')
|
||
time.sleep(delay)
|
||
normalize_response(raw, target)
|
||
record.update(status='completed', updated_at=datetime.now(timezone.utc).isoformat())
|
||
record.pop('error', None)
|
||
print(f' Completata: {target}')
|
||
except Exception as exc:
|
||
failures += 1
|
||
# Do not log SDK exception bodies, headers, credentials or base64 payloads.
|
||
message = str(exc) if isinstance(exc, (ValueError, OSError)) else type(exc).__name__
|
||
record.update(status='error', error=message, updated_at=datetime.now(timezone.utc).isoformat())
|
||
print(f' Errore {photo.name}: {message}', file=sys.stderr)
|
||
finally:
|
||
save_log(log_path, records)
|
||
gallery(photos, args.output_dir, records)
|
||
gallery(photos, args.output_dir, records)
|
||
finally:
|
||
if client is not None:
|
||
client.close()
|
||
return 1 if failures else 0
|
||
|
||
|
||
def main(argv=None):
|
||
parser = argparse.ArgumentParser(description=__doc__)
|
||
parser.add_argument('--input-dir', type=Path, default=ROOT / 'assets/raw/foto')
|
||
parser.add_argument('--output-dir', type=Path, default=ROOT / 'assets/generated')
|
||
parser.add_argument('--prompt-dir', type=Path, default=ROOT / 'prompts/backgrounds')
|
||
parser.add_argument('--model', default='gpt-image-2')
|
||
parser.add_argument('--quality', choices=['low', 'medium', 'high', 'auto'], default='high')
|
||
parser.add_argument('--timeout', type=positive, default=180)
|
||
parser.add_argument('--attempts', type=positive, default=3, help='numero massimo di richieste per foto')
|
||
limit = parser.add_mutually_exclusive_group()
|
||
limit.add_argument('--test', action='store_true', help='prima foto non ancora completata')
|
||
limit.add_argument('--limit', type=positive)
|
||
mode = parser.add_mutually_exclusive_group()
|
||
mode.add_argument('--dry-run', action='store_true')
|
||
mode.add_argument('--gallery-only', action='store_true', help='scrive solo il confronto HTML, senza API')
|
||
args = parser.parse_args(argv)
|
||
try:
|
||
return run(args)
|
||
except (ValueError, OSError, ImportError) as exc:
|
||
print(f'Errore: {exc}', file=sys.stderr)
|
||
return 1
|
||
except KeyboardInterrupt:
|
||
print('\nInterrotto. I risultati già salvati verranno riutilizzati.', file=sys.stderr)
|
||
return 130
|
||
|
||
|
||
if __name__ == '__main__':
|
||
sys.exit(main())
|