routes/fal.py
Rama main · réplica de lectura del repositorio original.
"""
fal.ai image generation proxy — keeps FAL_KEY server-side and instruments every call
so the JamFlow Watch board lights the prov:fal node live (Mike 2026-06-20).
POST /api/fal/image
body: { prompt, model?, image_size?, num_images?, num_inference_steps? }
returns: { images: [{url, width, height, content_type}], seed, timings }
NOTE: every generated image is downloaded and saved to UPLOADS_DIR on the server
immediately before the response is returned (CLAUDE.md: AI output is saved the
moment it is produced). The returned `url` is the permanent server path; the
upstream fal.media URL is ALSO included as `fal_url` for reference.
Mirrors routes/suno.py for telemetry: a per-call provider receipt is written two ways
(file-drop receipt + books queue) so the prov:fal node fires per call and books.db gets
the api_call row. Best-effort — telemetry NEVER fails a generation.
"""
import json
import logging
import os
import threading
import time
from datetime import datetime, timezone
from pathlib import Path
import requests as http
from flask import Blueprint, jsonify, request
from services.paths import UPLOADS_DIR
logger = logging.getLogger(__name__)
fal_bp = Blueprint('fal', __name__)
FAL_KEY = os.getenv('FAL_KEY', '')
FAL_BASE = 'https://fal.run'
DEFAULT_MODEL = 'fal-ai/flux/schnell'
# Allow-list of supported fal image models (prefix is the fal.run path).
FAL_MODELS = {
'fal-ai/flux/schnell',
'fal-ai/flux/dev',
'fal-ai/flux-pro/v1.1',
'fal-ai/flux-pro/v1.1-ultra',
'fal-ai/recraft-v3',
}
# ---------------------------------------------------------------------------
# Per-call provider receipt (JamFlow Watch live pulse) — same pattern as suno.py.
# ---------------------------------------------------------------------------
PROVIDER_RECEIPTS_DIR = Path(os.environ.get(
'PROVIDER_RECEIPTS_DIR', '/app/runtime/monitoring-events/provider-calls'))
_RECEIPT_TENANT = os.environ.get('JAMBOT_TENANT') or os.environ.get('HOST_TENANT', '') or 'unknown'
def _provider_receipt(op: str, units: str = '1') -> None:
"""Append one {ts,provider,tenant,op,units} receipt row — fire-and-forget."""
try:
PROVIDER_RECEIPTS_DIR.mkdir(parents=True, exist_ok=True)
now = datetime.now(timezone.utc)
row = json.dumps({'ts': now.isoformat(timespec='seconds'),
'provider': 'fal', 'tenant': _RECEIPT_TENANT,
'op': op, 'units': units})
with open(PROVIDER_RECEIPTS_DIR / f"fal-{now.strftime('%Y-%m-%d')}.jsonl", 'a') as f:
f.write(row + '\n')
except Exception: # never let telemetry touch the generation path
pass
# JamBot Books: also record as an api_call to the host-tailed books queue
# (the receipt path above is unmounted on most containers; this one isn't).
try:
from services.jambot_books_hook import record_provider_call
record_provider_call('fal', endpoint='/' + DEFAULT_MODEL, op=op, units=units)
except Exception:
pass
def _save_generated_image(url: str, ext_hint: str = 'jpg') -> str:
"""Download a generated image from a fal.media URL into UPLOADS_DIR.
Returns the permanent /uploads/... server URL. Raises on download failure."""
UPLOADS_DIR.mkdir(parents=True, exist_ok=True)
ext = url.rsplit('.', 1)[-1].split('?')[0][:4] if '.' in url.rsplit('/', 1)[-1] else ext_hint
filename = f'ai-gen-{int(time.time() * 1000)}.{ext}'
path = UPLOADS_DIR / filename
r = http.get(url, timeout=60)
r.raise_for_status()
path.write_bytes(r.content)
logger.info('fal: saved generated image → %s (%d bytes)', path, path.stat().st_size)
return f'/uploads/{filename}'
@fal_bp.route('/api/fal/image', methods=['POST'])
def fal_image():
if not FAL_KEY:
return jsonify({'error': 'FAL_KEY not configured on server'}), 500
body = request.get_json(silent=True) or {}
prompt = (body.get('prompt') or '').strip()
if not prompt:
return jsonify({'error': 'prompt is required'}), 400
model = body.get('model') or DEFAULT_MODEL
if model not in FAL_MODELS:
return jsonify({'error': f'unsupported model {model}',
'supported': sorted(FAL_MODELS)}), 400
num_images = int(body.get('num_images') or 1)
payload = {
'prompt': prompt,
'image_size': body.get('image_size') or 'square',
'num_images': num_images,
}
# schnell/dev accept num_inference_steps; pro models ignore it harmlessly.
if body.get('num_inference_steps'):
payload['num_inference_steps'] = int(body['num_inference_steps'])
logger.info('fal: model=%s prompt_len=%d num=%d', model, len(prompt), num_images)
try:
resp = http.post(f'{FAL_BASE}/{model}',
headers={'Authorization': f'Key {FAL_KEY}',
'Content-Type': 'application/json'},
json=payload, timeout=120)
except Exception as e: # noqa: BLE001
logger.exception('fal: request failed')
return jsonify({'error': f'fal request failed: {e}'}), 502
if resp.status_code != 200:
logger.warning('fal: upstream %d — %s', resp.status_code, resp.text[:300])
return jsonify({'error': 'fal upstream error', 'status': resp.status_code,
'detail': resp.text[:500]}), 502
result = resp.json()
# Save every returned image to the server immediately (permanent URLs).
out_images = []
for img in (result.get('images') or []):
u = img.get('url')
if not u:
continue
ext = (img.get('content_type') or 'image/jpeg').split('/')[-1]
try:
server_url = _save_generated_image(u, ext_hint=ext)
except Exception: # noqa: BLE001
logger.exception('fal: failed to persist %s', u)
server_url = None
out_images.append({'url': server_url, 'fal_url': u,
'width': img.get('width'), 'height': img.get('height'),
'content_type': img.get('content_type')})
# Telemetry: one receipt per successful call (units = images saved).
_provider_receipt('image', units=str(len(out_images) or num_images))
return jsonify({'images': out_images, 'seed': result.get('seed'),
'timings': result.get('timings'), 'model': model})