""" Atelier IA — page de classe pour générer des images via fal.ai La clé API reste côté serveur : les étudiants ne la voient jamais. Variables d'environnement : FAL_KEY (obligatoire) clé API fal.ai CLASS_CODE mot de passe commun des étudiants (défaut : iloi.ia) ADMIN_USER identifiant de l'enseignant (défaut : prof) ADMIN_CODE mot de passe de l'enseignant (obligatoire pour l'espace prof) QUOTA crédits par étudiant (défaut : 60) DATA_DIR dossier de stockage (défaut : ./data) HF_TOKEN + DATA_REPO (optionnel, Hugging Face) sauvegarde automatique des galeries dans un dataset privé, ex. "cedric/atelier-ia-data" """ import csv import io import json import os import re import threading import time import unicodedata import uuid import zipfile from datetime import datetime import fal_client import gradio as gr # Compatibilité si le Space est réglé sur du matériel "ZeroGPU" : # l'appli n'utilise pas de GPU (le calcul se fait chez fal.ai), mais ZeroGPU exige # au moins une fonction @spaces.GPU. Sur "CPU basic" ou en local, ce bloc est ignoré. try: import spaces @spaces.GPU def _zerogpu_placeholder(): return None except Exception: pass import requests # ---------------------------------------------------------------- configuration DATA_DIR = os.path.abspath(os.environ.get("DATA_DIR", "data")) CLASS_CODE = os.environ.get("CLASS_CODE", "iloi.ia") ADMIN_USER = os.environ.get("ADMIN_USER", "prof").strip().lower() ADMIN_CODE = os.environ.get("ADMIN_CODE", "") DEFAULT_QUOTA = int(os.environ.get("QUOTA", "60")) # "cout" = crédits décomptés par image générée (à ajuster selon vos tarifs fal.ai) T2I_MODELS = { "Krea 2 Turbo": {"id": "fal-ai/krea-2/turbo", "cout": 1}, "Z-Image Turbo": {"id": "fal-ai/z-image/turbo", "cout": 1}, } I2I_MODELS = { "Nano Banana 2 (édition)": {"id": "fal-ai/nano-banana-2/edit", "cout": 4}, "Seedream 5 Pro (édition)": {"id": "bytedance/seedream/v5/pro/edit", "cout": 4}, } # Format commun -> paramètres propres à chaque modèle FORMATS = ["1:1", "4:3", "3:4", "16:9", "9:16"] IMAGE_SIZE = { "auto": "auto_2K", "1:1": "square_hd", "4:3": "landscape_4_3", "3:4": "portrait_4_3", "16:9": "landscape_16_9", "9:16": "portrait_16_9", } MAX_REF_IMAGES = 4 os.makedirs(os.path.join(DATA_DIR, "users"), exist_ok=True) EXAMPLE_CSV = os.path.join(os.path.dirname(os.path.abspath(__file__)), "templates_exemple.csv") # ---------------------------------------------------------------- sauvegarde HF (optionnelle) _lock = threading.Lock() _scheduler = None if os.environ.get("HF_TOKEN") and os.environ.get("DATA_REPO"): try: from huggingface_hub import CommitScheduler, snapshot_download repo = os.environ["DATA_REPO"] try: snapshot_download(repo_id=repo, repo_type="dataset", local_dir=DATA_DIR, token=os.environ["HF_TOKEN"]) except Exception as e: # premier lancement : le dataset n'existe pas encore print("Pas de données à restaurer :", e) _scheduler = CommitScheduler(repo_id=repo, repo_type="dataset", folder_path=DATA_DIR, every=2, private=True, token=os.environ["HF_TOKEN"]) _lock = _scheduler.lock print("Sauvegarde automatique activée vers", repo) except Exception as e: print("Sauvegarde HF désactivée :", e) # ---------------------------------------------------------------- utilitaires def slug(name: str) -> str: name = unicodedata.normalize("NFKD", name or "").encode("ascii", "ignore").decode() name = re.sub(r"[^a-z0-9_-]+", "-", name.strip().lower()).strip("-") return name[:30] def read_json(path, default): try: with open(path, encoding="utf-8") as f: return json.load(f) except (FileNotFoundError, json.JSONDecodeError): return default def write_json(path, data): os.makedirs(os.path.dirname(path), exist_ok=True) tmp = path + ".tmp" with open(tmp, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=1) os.replace(tmp, path) def settings(): return read_json(os.path.join(DATA_DIR, "settings.json"), {"quota": DEFAULT_QUOTA}) def user_dir(user): return os.path.join(DATA_DIR, "users", user) def load_user(user): return read_json(os.path.join(user_dir(user), "gallery.json"), {"used": 0, "items": []}) def save_user(user, data): write_json(os.path.join(user_dir(user), "gallery.json"), data) def current_user(request: gr.Request): user = slug(getattr(request, "username", "") or "") if not user: raise gr.Error("Session expirée : rechargez la page et reconnectez-vous.") return user def remaining(user): return settings()["quota"] - load_user(user)["used"] def status_line(user): if user == ADMIN_USER: return f"**Connecté : {user}** (enseignant)" q = settings()["quota"] used = load_user(user)["used"] return f"**Connecté : {user}** · crédits restants : **{max(q - used, 0)} / {q}**" # ---------------------------------------------------------------- templates CSV def parse_templates_csv(path): """Lit un CSV (virgule ou point-virgule, UTF-8 ou Excel) avec au minimum les colonnes titre et prompt.""" raw = open(path, "rb").read() for enc in ("utf-8-sig", "cp1252", "latin-1"): try: text = raw.decode(enc) break except UnicodeDecodeError: continue first = text.splitlines()[0] if text else "" delim = ";" if first.count(";") > first.count(",") else "," rows = list(csv.DictReader(io.StringIO(text), delimiter=delim)) out = [] for r in rows: r = {(k or "").strip().lower(): (v or "").strip() for k, v in r.items()} prompt = r.get("prompt") or r.get("texte") or "" title = r.get("titre") or r.get("title") or r.get("nom") or prompt[:40] cat = r.get("categorie") or r.get("catégorie") or r.get("category") or "" if prompt: out.append({"categorie": cat, "titre": title, "prompt": prompt}) if not out: raise gr.Error("Aucun template trouvé. Le CSV doit contenir au moins les colonnes « titre » et « prompt ».") return out def all_templates(user): cls = read_json(os.path.join(DATA_DIR, "templates.json"), []) perso = read_json(os.path.join(user_dir(user), "templates.json"), []) if user else [] return [dict(t, source="classe") for t in cls] + [dict(t, source="perso") for t in perso] def template_label(t): tag = "★ " if t["source"] == "perso" else "" return f"{tag}{t['categorie'] + ' — ' if t['categorie'] else ''}{t['titre']}" def template_choices(user): return [template_label(t) for t in all_templates(user)] def seed_class_templates(): """Au premier lancement, charge templates_exemple.csv comme templates de classe.""" path = os.path.join(DATA_DIR, "templates.json") if not os.path.exists(path) and os.path.exists(EXAMPLE_CSV): try: write_json(path, parse_templates_csv(EXAMPLE_CSV)) except Exception as e: print("Templates d'exemple non chargés :", e) seed_class_templates() # ---------------------------------------------------------------- génération def _save_results(user, result, meta, cost_per_image): images = result.get("images") or [] nsfw = result.get("has_nsfw_concepts") or [False] * len(images) saved = [] for i, img in enumerate(images): if i < len(nsfw) and nsfw[i]: continue r = requests.get(img["url"], timeout=120) r.raise_for_status() ext = "png" if "png" in (img.get("content_type") or "") else ( "webp" if "webp" in (img.get("content_type") or "") else "jpg") iid = datetime.now().strftime("%Y%m%d-%H%M%S-") + uuid.uuid4().hex[:6] rel = f"images/{iid}.{ext}" path = os.path.join(user_dir(user), rel) os.makedirs(os.path.dirname(path), exist_ok=True) with _lock: with open(path, "wb") as f: f.write(r.content) saved.append({"id": iid, "file": rel, "created": time.strftime("%Y-%m-%d %H:%M"), "seed": result.get("seed"), **meta}) with _lock: data = load_user(user) data["items"].extend(saved) data["used"] += cost_per_image * len(saved) save_user(user, data) blocked = len(images) - len(saved) return saved, blocked def _check_quota(user, cost): if user == ADMIN_USER: return left = remaining(user) if cost > left: raise gr.Error(f"Crédits insuffisants : cette génération coûte {cost}, il vous reste {max(left, 0)}.") def _call_fal(endpoint, args): if not os.environ.get("FAL_KEY"): raise gr.Error("La clé FAL_KEY n'est pas configurée sur le serveur. Prévenez l'enseignant.") try: return fal_client.subscribe(endpoint, arguments=args) except Exception as e: raise gr.Error(f"Erreur fal.ai : {e}") def _result_message(saved, blocked, cost, user): msg = f"{len(saved)} image(s) générée(s) · {cost * len(saved)} crédit(s) utilisé(s)." if blocked: msg += f" {blocked} image(s) bloquée(s) par le filtre de sécurité (non décomptée(s))." return msg def generate_t2i(prompt, model_label, fmt, n, seed, request: gr.Request): user = current_user(request) prompt = (prompt or "").strip() if not prompt: raise gr.Error("Écrivez un prompt (ou choisissez un template).") model = T2I_MODELS[model_label] n = int(n) cost = model["cout"] * n _check_quota(user, cost) args = {"prompt": prompt, "num_images": n, "image_size": IMAGE_SIZE[fmt], "enable_safety_checker": True, "output_format": "png"} if seed and int(seed) > 0: args["seed"] = int(seed) result = _call_fal(model["id"], args) saved, blocked = _save_results(user, result, {"mode": "Texte → Image", "model": model_label, "prompt": prompt}, model["cout"]) gal = [(os.path.join(user_dir(user), s["file"]), s["model"]) for s in saved] return gal, _result_message(saved, blocked, model["cout"], user), status_line(user) def generate_i2i(refs, prompt, model_label, fmt, n, request: gr.Request): user = current_user(request) prompt = (prompt or "").strip() paths = [r[0] if isinstance(r, (list, tuple)) else r for r in (refs or [])] if not paths: raise gr.Error("Ajoutez au moins une image de référence.") if len(paths) > MAX_REF_IMAGES: raise gr.Error(f"{MAX_REF_IMAGES} images de référence maximum.") if not prompt: raise gr.Error("Décrivez la modification souhaitée.") model = I2I_MODELS[model_label] n = int(n) cost = model["cout"] * n _check_quota(user, cost) try: urls = [fal_client.upload_file(p) for p in paths] except Exception as e: raise gr.Error(f"Envoi des images impossible : {e}") args = {"prompt": prompt, "image_urls": urls, "num_images": n} if "nano-banana" in model["id"]: args.update({"aspect_ratio": fmt, "resolution": "1K", "output_format": "png"}) else: args.update({"image_size": IMAGE_SIZE[fmt], "enable_safety_checker": True}) result = _call_fal(model["id"], args) saved, blocked = _save_results(user, result, {"mode": "Image → Image", "model": model_label, "prompt": prompt, "refs": len(paths)}, model["cout"]) gal = [(os.path.join(user_dir(user), s["file"]), s["model"]) for s in saved] return gal, _result_message(saved, blocked, model["cout"], user), status_line(user) # ---------------------------------------------------------------- galerie def visible_items(user): return [it for it in reversed(load_user(user)["items"]) if not it.get("deleted")] def gallery_view(user): items = visible_items(user) gal = [(os.path.join(user_dir(user), it["file"]), f"{it['model']} · {it['prompt'][:80]}") for it in items] return gal, [it["id"] for it in items] def refresh_gallery(request: gr.Request): user = current_user(request) gal, ids = gallery_view(user) return gal, ids, None, "Cliquez sur une image pour voir son prompt.", status_line(user) def select_item(ids, evt: gr.SelectData, request: gr.Request): user = current_user(request) if evt.index is None or evt.index >= len(ids): return None, "" iid = ids[evt.index] it = next((x for x in load_user(user)["items"] if x["id"] == iid), None) if not it: return None, "" info = (f"**{it['mode']}** · {it['model']} · {it['created']}" + (f" · seed {it['seed']}" if it.get("seed") else "") + f"\n\n> {it['prompt']}") return iid, info def _item_path(user, iid): it = next((x for x in load_user(user)["items"] if x["id"] == iid), None) return os.path.join(user_dir(user), it["file"]) if it else None def send_to_i2i(iid, current_refs, request: gr.Request): user = current_user(request) if not iid: raise gr.Error("Sélectionnez d'abord une image dans votre galerie.") refs = [r[0] if isinstance(r, (list, tuple)) else r for r in (current_refs or [])] if len(refs) >= MAX_REF_IMAGES: raise gr.Error(f"{MAX_REF_IMAGES} images de référence maximum.") refs.append(_item_path(user, iid)) gr.Info("Image ajoutée dans l'onglet Image → Image.") return refs def reuse_prompt(iid, request: gr.Request): user = current_user(request) it = next((x for x in load_user(user)["items"] if x["id"] == iid), None) if not it: raise gr.Error("Sélectionnez d'abord une image.") gr.Info("Prompt copié dans l'onglet Texte → Image.") return it["prompt"] def delete_item(iid, request: gr.Request): user = current_user(request) if not iid: raise gr.Error("Sélectionnez d'abord une image.") with _lock: data = load_user(user) for it in data["items"]: if it["id"] == iid: it["deleted"] = True save_user(user, data) gal, ids = gallery_view(user) return gal, ids, None, "Image retirée de la galerie." def zip_gallery(request: gr.Request): user = current_user(request) items = visible_items(user) if not items: raise gr.Error("Votre galerie est vide.") zpath = os.path.join(DATA_DIR, "_exports", f"galerie-{user}.zip") os.makedirs(os.path.dirname(zpath), exist_ok=True) lines = ["fichier;date;mode;modele;prompt"] with zipfile.ZipFile(zpath, "w", zipfile.ZIP_DEFLATED) as z: for it in items: name = os.path.basename(it["file"]) z.write(os.path.join(user_dir(user), it["file"]), name) lines.append(f"{name};{it['created']};{it['mode']};{it['model']};\"{it['prompt'].replace(chr(34), chr(39))}\"") z.writestr("prompts.csv", "" + "\n".join(lines)) return zpath # ---------------------------------------------------------------- templates (UI) def on_template(label, request: gr.Request): user = current_user(request) for t in all_templates(user): if template_label(t) == label: return t["prompt"] return gr.update() def import_perso_templates(file, request: gr.Request): user = current_user(request) if not file: raise gr.Error("Choisissez un fichier CSV.") new = parse_templates_csv(file) path = os.path.join(user_dir(user), "templates.json") with _lock: write_json(path, read_json(path, []) + new) gr.Info(f"{len(new)} template(s) personnel(s) importé(s).") return gr.Dropdown(choices=template_choices(user), value=None), None # ---------------------------------------------------------------- espace enseignant def _require_admin(request): user = current_user(request) if user != ADMIN_USER: raise gr.Error("Réservé à l'enseignant.") return user def admin_overview(request: gr.Request): _require_admin(request) q = settings()["quota"] rows, total = [], 0 base = os.path.join(DATA_DIR, "users") for u in sorted(os.listdir(base)): if u == ADMIN_USER or not os.path.isdir(os.path.join(base, u)): continue d = load_user(u) n = len([i for i in d["items"] if not i.get("deleted")]) rows.append([u, n, d["used"], max(q - d["used"], 0)]) total += d["used"] admin = load_user(ADMIN_USER) summary = (f"**{len(rows)} étudiant(s)** · quota actuel : **{q} crédits** · " f"crédits consommés par les étudiants : **{total}** · par l'enseignant : **{admin['used']}**") students = [r[0] for r in rows] return rows, summary, gr.Dropdown(choices=students), gr.Dropdown(choices=students) def admin_set_quota(q, request: gr.Request): _require_admin(request) with _lock: s = settings() s["quota"] = int(q) write_json(os.path.join(DATA_DIR, "settings.json"), s) gr.Info(f"Quota fixé à {int(q)} crédits par étudiant.") return admin_overview(request) def admin_reset(student, request: gr.Request): _require_admin(request) if not student: raise gr.Error("Choisissez un étudiant.") with _lock: d = load_user(student) d["used"] = 0 save_user(student, d) gr.Info(f"Crédits de {student} remis à zéro.") return admin_overview(request) def admin_view_gallery(student, request: gr.Request): _require_admin(request) if not student: return [] return gallery_view(student)[0] def admin_import_templates(file, request: gr.Request): _require_admin(request) if not file: raise gr.Error("Choisissez un fichier CSV.") new = parse_templates_csv(file) with _lock: write_json(os.path.join(DATA_DIR, "templates.json"), new) gr.Info(f"{len(new)} template(s) de classe importé(s) (remplacent les précédents).") return None # ---------------------------------------------------------------- authentification def check_auth(username, password): u = slug(username) if not u: return False if u == ADMIN_USER: return bool(ADMIN_CODE) and password == ADMIN_CODE return password == CLASS_CODE # ---------------------------------------------------------------- interface CSS = """ .gradio-container {max-width: 1200px !important} #status {text-align:right} footer {display:none !important} """ with gr.Blocks(title="IA — Génération d'images", theme=gr.themes.Soft(primary_hue="orange"), css=CSS) as demo: with gr.Row(): gr.Markdown("## IA — Génération d'images") status = gr.Markdown("", elem_id="status") with gr.Tabs() as tabs: # ----------------------------------------------------- Texte → Image with gr.Tab("Texte → Image", id="t2i"): with gr.Row(): with gr.Column(scale=2): tpl = gr.Dropdown(label="Template de prompt", choices=[], value=None, info="Choisir un modèle de prompt, puis l'adapter (remplacer les [CROCHETS]).") t2i_prompt = gr.Textbox(label="Prompt", lines=5, placeholder="Décrivez l'image : sujet, style, cadrage, lumière…") with gr.Row(): t2i_model = gr.Dropdown(list(T2I_MODELS), value="Krea 2 Turbo", label="Modèle") t2i_fmt = gr.Dropdown(FORMATS, value="1:1", label="Format") with gr.Row(): t2i_n = gr.Slider(1, 4, value=1, step=1, label="Nombre d'images") t2i_seed = gr.Number(value=0, precision=0, label="Seed (0 = aléatoire)") t2i_btn = gr.Button("Générer", variant="primary") with gr.Accordion("Importer mes propres templates (CSV)", open=False): gr.Markdown("Colonnes : `categorie`, `titre`, `prompt` (séparateur virgule ou point-virgule). " "Vos templates apparaissent avec une ★.") perso_csv = gr.File(label="Fichier CSV", file_types=[".csv"], type="filepath") perso_btn = gr.Button("Importer") with gr.Column(scale=3): t2i_out = gr.Gallery(label="Résultat", columns=2, height=520, object_fit="contain") t2i_msg = gr.Markdown() # ----------------------------------------------------- Image → Image with gr.Tab("Image → Image", id="i2i"): with gr.Row(): with gr.Column(scale=2): refs = gr.Gallery(label=f"Images de référence (1 à {MAX_REF_IMAGES}) — croquis, rough, décor…", type="filepath", interactive=True, file_types=["image"], columns=4, height=220) i2i_prompt = gr.Textbox(label="Consigne", lines=4, placeholder="Ex. : Colorise ce rough en aplats façon anime, garde le trait original.") with gr.Row(): i2i_model = gr.Dropdown(list(I2I_MODELS), value="Nano Banana 2 (édition)", label="Modèle") i2i_fmt = gr.Dropdown(["auto"] + FORMATS, value="auto", label="Format") i2i_n = gr.Slider(1, 4, value=1, step=1, label="Nombre d'images") i2i_btn = gr.Button("Générer", variant="primary") gr.Markdown(f"Les modèles d'édition coûtent plus cher : " + ", ".join(f"{k} = {v['cout']} crédits/image" for k, v in I2I_MODELS.items()) + ". Texte → Image = 1 crédit/image.") with gr.Column(scale=3): i2i_out = gr.Gallery(label="Résultat", columns=2, height=520, object_fit="contain") i2i_msg = gr.Markdown() # ----------------------------------------------------- Galerie with gr.Tab("Ma galerie", id="gal") as gal_tab: gal_ids = gr.State([]) gal_sel = gr.State(None) with gr.Row(): gal_refresh = gr.Button("Actualiser") gal_zip = gr.Button("Tout télécharger (.zip)") my_gal = gr.Gallery(label="Mes images (les plus récentes d'abord)", columns=5, height=560, object_fit="cover", allow_preview=True) gal_info = gr.Markdown("Cliquez sur une image pour voir son prompt.") with gr.Row(): gal_to_i2i = gr.Button("Utiliser comme référence (Image → Image)") gal_reuse = gr.Button("Réutiliser le prompt") gal_del = gr.Button("Retirer de la galerie", variant="stop") gal_file = gr.File(label="Archive", visible=True) # ----------------------------------------------------- Espace enseignant with gr.Tab("Enseignant", id="admin", visible=False) as admin_tab: adm_summary = gr.Markdown() adm_table = gr.Dataframe(headers=["Étudiant", "Images", "Crédits utilisés", "Crédits restants"], interactive=False) adm_refresh = gr.Button("Actualiser") with gr.Row(): adm_quota = gr.Number(value=DEFAULT_QUOTA, precision=0, label="Quota par étudiant (crédits)") adm_quota_btn = gr.Button("Appliquer le quota") with gr.Row(): adm_student = gr.Dropdown(label="Étudiant", choices=[]) adm_reset_btn = gr.Button("Remettre ses crédits à zéro") with gr.Accordion("Templates de la classe (CSV)", open=True): gr.Markdown("Colonnes : `categorie`, `titre`, `prompt`. L'import **remplace** les templates de classe.") adm_csv = gr.File(label="Fichier CSV", file_types=[".csv"], type="filepath") adm_csv_btn = gr.Button("Importer les templates de classe") with gr.Accordion("Voir la galerie d'un étudiant", open=False): adm_gal_student = gr.Dropdown(label="Étudiant", choices=[]) adm_gal = gr.Gallery(columns=5, height=480, object_fit="cover") # ----------------------------------------------------- événements def on_load(request: gr.Request): user = current_user(request) s = settings() return (status_line(user), gr.Dropdown(choices=template_choices(user), value=None), gr.Tab(visible=(user == ADMIN_USER)), s["quota"]) demo.load(on_load, None, [status, tpl, admin_tab, adm_quota]) def refresh_templates(request: gr.Request): return gr.Dropdown(choices=template_choices(current_user(request))) tpl.focus(refresh_templates, None, tpl) tpl.input(on_template, tpl, t2i_prompt) perso_btn.click(import_perso_templates, perso_csv, [tpl, perso_csv]) t2i_btn.click(generate_t2i, [t2i_prompt, t2i_model, t2i_fmt, t2i_n, t2i_seed], [t2i_out, t2i_msg, status], concurrency_limit=10) i2i_btn.click(generate_i2i, [refs, i2i_prompt, i2i_model, i2i_fmt, i2i_n], [i2i_out, i2i_msg, status], concurrency_limit=10) gal_outputs = [my_gal, gal_ids, gal_sel, gal_info, status] gal_tab.select(refresh_gallery, None, gal_outputs) gal_refresh.click(refresh_gallery, None, gal_outputs) my_gal.select(select_item, gal_ids, [gal_sel, gal_info]) gal_to_i2i.click(send_to_i2i, [gal_sel, refs], refs) gal_reuse.click(reuse_prompt, gal_sel, t2i_prompt) gal_del.click(delete_item, gal_sel, [my_gal, gal_ids, gal_sel, gal_info]) gal_zip.click(zip_gallery, None, gal_file) adm_outputs = [adm_table, adm_summary, adm_student, adm_gal_student] admin_tab.select(admin_overview, None, adm_outputs) adm_refresh.click(admin_overview, None, adm_outputs) adm_quota_btn.click(admin_set_quota, adm_quota, adm_outputs) adm_reset_btn.click(admin_reset, adm_student, adm_outputs) adm_csv_btn.click(admin_import_templates, adm_csv, adm_csv) adm_gal_student.change(admin_view_gallery, adm_gal_student, adm_gal) if __name__ == "__main__": demo.queue(default_concurrency_limit=10).launch( auth=check_auth, auth_message="Prénom (ou pseudo) + code de la classe", server_name="0.0.0.0", server_port=int(os.environ.get("PORT", "7860")), allowed_paths=[DATA_DIR], ssr_mode=False, )