alpinux-static/app/app.py
Alpinux 130a901be7 feat: gestion des conflits lors de l'upload de fichiers
Ajoute un sélecteur de stratégie dans le formulaire de dépôt CDN,
identique à celui du redimensionnement :
- Écraser (défaut) : comportement précédent, écrase silencieusement
- Backup : renomme l'existant en {stem}_bak_{timestamp}{ext} avant dépôt
- Renommer : auto-incrémente le nom du fichier uploadé ({stem}_1, _2…)
- Ignorer : ne dépose pas si le fichier existe déjà

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-06 09:29:29 +02:00

728 lines
24 KiB
Python

import io
import json
import os
import shutil
import subprocess
import threading
from pathlib import Path
from datetime import datetime
from urllib.parse import urlencode
from PIL import Image
from flask import (Flask, redirect, url_for, session, request,
render_template, abort, send_from_directory, jsonify)
from authlib.integrations.flask_client import OAuth
from werkzeug.middleware.proxy_fix import ProxyFix
from werkzeug.utils import secure_filename
app = Flask(__name__)
app.secret_key = os.environ["SECRET_KEY"]
app.wsgi_app = ProxyFix(app.wsgi_app, x_proto=1)
oauth = OAuth(app)
oauth.register(
name="alpid",
server_metadata_url=os.environ["ALPID_DISCOVERY_URL"],
client_id=os.environ["ALPID_CLIENT_ID"],
client_secret=os.environ["ALPID_CLIENT_SECRET"],
client_kwargs={"scope": "openid profile email"},
)
ADMIN_GROUPS = set(os.environ.get("ADMIN_GROUPS", "admins").split(","))
ADMIN_EMAILS = set(e.strip() for e in os.environ.get("ADMIN_EMAILS", "").split(",") if e.strip())
ASSETS_ROOT = Path(os.environ.get("ASSETS_ROOT", ".")).resolve()
_alpid_base = os.environ["ALPID_DISCOVERY_URL"].split("/.well-known/")[0]
ALPID_LOGOUT_URL = _alpid_base + "/protocol/openid-connect/logout"
STATS_FILE = Path(os.environ.get("STATS_FILE", "/opt/static-cdn/goaccess.html"))
STATS_JSON = Path(os.environ.get("STATS_JSON", "/opt/static-cdn/goaccess.json"))
STATS_LOG_FILE = os.environ.get("STATS_LOG_FILE", "")
STATS_GENERATE_CMD = os.environ.get("STATS_GENERATE_CMD", "")
_gen_lock = threading.Lock()
_gen_state = {"generating": False}
_HIDDEN = frozenset({
".git", "scripts", "app",
".env", ".env.example", ".gitignore",
".htaccess", ".htpasswd_stats",
"standard_index.html",
})
try:
import cairosvg as _cairosvg
HAS_CAIROSVG = True
except ImportError:
HAS_CAIROSVG = False
IMAGE_EXT = frozenset({".png", ".jpg", ".jpeg", ".gif", ".ico", ".svg", ".webp"})
RESIZE_SIZES = frozenset({32, 64, 100, 128, 200, 300, 500, 600, 1024})
RESIZE_FORMATS = frozenset({"png", "jpg", "ico", "svg"})
TEXT_EXT = frozenset({".txt", ".html", ".htm", ".md", ".css", ".js", ".json",
".xml", ".conf", ".sh", ".robots", ".php"})
PDF_EXT = frozenset({".pdf"})
# ── Helpers ───────────────────────────────────────────────────────────
def _user():
return session.get("user")
def _require_admin():
u = _user()
if not u:
session["next_url"] = request.url
return redirect(url_for("login"))
if not u.get("is_admin"):
abort(403)
return None
def _safe_path(subpath: str) -> Path:
target = (ASSETS_ROOT / subpath).resolve()
if not target.is_relative_to(ASSETS_ROOT):
abort(400)
return target
def _humansize(n: int) -> str:
for unit in ("o", "Ko", "Mo", "Go"):
if n < 1024:
return f"{n:.0f} {unit}"
n /= 1024
return f"{n:.1f} To"
def _backup_path(p: Path) -> Path:
ts = datetime.now().strftime("%Y%m%d%H%M%S")
return p.parent / f"{p.stem}_bak_{ts}{p.suffix}"
def _auto_rename(p: Path) -> Path:
i = 1
while True:
candidate = p.parent / f"{p.stem}_{i}{p.suffix}"
if not candidate.exists():
return candidate
i += 1
def _folder_stats(path: Path) -> dict:
files, size = 0, 0
for f in path.rglob("*"):
if not f.is_file():
continue
if any(p in _HIDDEN or p.startswith(".") for p in f.relative_to(path).parts):
continue
files += 1
size += f.stat().st_size
return {"files": files, "size": size}
def _parse_date(s: str):
if not s:
return None
try:
return datetime.strptime(s, "%Y-%m-%d")
except ValueError:
return None
def _load_hits() -> dict:
"""Parse GoAccess JSON report → {relative_path: hit_count}."""
if not STATS_JSON.is_file():
return {}
try:
data = json.loads(STATS_JSON.read_text())
hits = {}
for item in data.get("requests", {}).get("data", []):
url = item.get("data", "").lstrip("/")
count = item.get("hits", {}).get("count", 0)
if url and count:
hits[url] = count
return hits
except Exception:
return {}
def _entry(item: Path) -> dict:
stat = item.stat()
ext = item.suffix.lower()
return {
"name": item.name,
"path": str(item.relative_to(ASSETS_ROOT)),
"is_dir": item.is_dir(),
"size": stat.st_size if item.is_file() else None,
"mtime": datetime.fromtimestamp(stat.st_mtime),
"ext": ext,
"is_image": ext in IMAGE_EXT,
"is_text": ext in TEXT_EXT,
"is_pdf": ext in PDF_EXT,
}
# ── Auth ──────────────────────────────────────────────────────────────
@app.route("/auth/login")
def login():
return oauth.alpid.authorize_redirect(url_for("callback", _external=True))
@app.route("/auth/callback")
def callback():
token = oauth.alpid.authorize_access_token()
info = token.get("userinfo") or oauth.alpid.userinfo(token=token)
email = info.get("email", "")
groups = set(info.get("groups", []))
if groups:
is_admin = bool(groups & ADMIN_GROUPS)
elif ADMIN_EMAILS:
is_admin = email in ADMIN_EMAILS
else:
is_admin = True
session["user"] = {
"sub": info["sub"],
"name": info.get("name") or info.get("preferred_username", ""),
"email": email,
"is_admin": is_admin,
}
session["id_token"] = token.get("id_token", "")
return redirect(session.pop("next_url", url_for("dashboard")))
@app.route("/auth/logout")
def logout():
id_token = session.get("id_token")
session.clear()
params = {"post_logout_redirect_uri": url_for("dashboard", _external=True)}
if id_token:
params["id_token_hint"] = id_token
return redirect(ALPID_LOGOUT_URL + "?" + urlencode(params))
# ── Dashboard ─────────────────────────────────────────────────────────
@app.route("/")
def dashboard():
redir = _require_admin()
if redir:
return redir
folders = {}
for item in sorted(ASSETS_ROOT.iterdir()):
if item.name in _HIDDEN or item.name.startswith("."):
continue
if item.is_dir():
folders[item.name] = _folder_stats(item)
return render_template("dashboard.html",
user=_user(),
folders=folders,
total_files=sum(v["files"] for v in folders.values()),
total_size=sum(v["size"] for v in folders.values()),
humansize=_humansize,
)
# ── Navigateur de fichiers ────────────────────────────────────────────
@app.route("/browse/")
@app.route("/browse/<path:subpath>")
def browse(subpath=""):
redir = _require_admin()
if redir:
return redir
target = _safe_path(subpath)
if not target.exists():
abort(404)
if subpath:
top = Path(subpath).parts[0]
if top in _HIDDEN or top.startswith("."):
abort(403)
if target.is_file():
return _file_preview(target, subpath)
hits = _load_hits()
entries = []
for item in sorted(target.iterdir()):
if item.name in _HIDDEN or item.name.startswith("."):
continue
e = _entry(item)
e["hits"] = hits.get(e["path"], 0) if not e["is_dir"] else None
entries.append(e)
entries.sort(key=lambda e: (0 if e["is_dir"] else 1, e["name"].lower()))
parts = Path(subpath).parts if subpath else ()
breadcrumb = [
{"name": p, "path": str(Path(*parts[:i + 1]))}
for i, p in enumerate(parts)
]
return render_template("browse.html",
user=_user(),
entries=entries,
subpath=subpath,
breadcrumb=breadcrumb,
humansize=_humansize,
has_hits=STATS_JSON.is_file(),
)
def _file_preview(path: Path, subpath: str):
ext = path.suffix.lower()
stat = path.stat()
parent = str(Path(subpath).parent) if Path(subpath).parent != Path(".") else ""
siblings = sorted(
[f for f in path.parent.iterdir()
if f.is_file() and f.name not in _HIDDEN and not f.name.startswith(".")],
key=lambda f: f.name.lower(),
)
idx = next((i for i, f in enumerate(siblings) if f == path), None)
prev_path = str(siblings[idx - 1].relative_to(ASSETS_ROOT)) if idx else None
next_path = str(siblings[idx + 1].relative_to(ASSETS_ROOT)) if idx is not None and idx < len(siblings) - 1 else None
ctx = dict(
user=_user(),
subpath=subpath,
filename=path.name,
filesize=_humansize(stat.st_size),
mtime=datetime.fromtimestamp(stat.st_mtime),
raw_url=url_for("raw_file", subpath=subpath),
parent_path=parent,
prev_path=prev_path,
next_path=next_path,
sibling_pos=f"{idx + 1}/{len(siblings)}" if idx is not None else "",
ext=ext,
)
if ext in IMAGE_EXT:
return render_template("preview_image.html", **ctx)
if ext in TEXT_EXT:
content = path.read_text(errors="replace")
return render_template("preview_text.html", content=content, lang=ext.lstrip("."), **ctx)
return render_template("preview_other.html", **ctx)
# ── Recherche ─────────────────────────────────────────────────────────
@app.route("/search")
def search():
redir = _require_admin()
if redir:
return redir
q = request.args.get("q", "").strip()
after_s = request.args.get("after", "").strip()
before_s = request.args.get("before", "").strip()
in_content = request.args.get("content") == "1"
dt_after = _parse_date(after_s)
dt_before = _parse_date(before_s)
results = []
searched = bool(q or dt_after or dt_before)
if searched:
q_low = q.lower()
for path in sorted(ASSETS_ROOT.rglob("*")):
if not path.is_file():
continue
parts = path.relative_to(ASSETS_ROOT).parts
if any(p in _HIDDEN or p.startswith(".") for p in parts):
continue
stat = path.stat()
mtime = datetime.fromtimestamp(stat.st_mtime)
if dt_after and mtime.date() < dt_after.date():
continue
if dt_before and mtime.date() > dt_before.date():
continue
name_match = (not q) or q_low in path.name.lower()
content_match = None
if in_content and q and not name_match and path.suffix.lower() in TEXT_EXT:
try:
text = path.read_text(errors="replace")
idx = text.lower().find(q_low)
if idx != -1:
start = max(0, idx - 60)
end = min(len(text), idx + len(q) + 60)
snippet = text[start:end].replace("\n", " ")
content_match = ("" if start else "") + snippet + ("" if end < len(text) else "")
name_match = True
except Exception:
pass
if not name_match:
continue
e = _entry(path)
e["content_match"] = content_match
results.append(e)
return render_template("search.html",
user=_user(),
q=q,
after=after_s,
before=before_s,
in_content=in_content,
results=results,
searched=searched,
humansize=_humansize,
)
# ── Statistiques GoAccess ─────────────────────────────────────────────
@app.route("/stats/")
def stats():
redir = _require_admin()
if redir:
return redir
return render_template("stats.html",
user=_user(),
has_report=STATS_FILE.is_file(),
)
@app.route("/stats/report")
def stats_report():
redir = _require_admin()
if redir:
return redir
if not STATS_FILE.is_file():
abort(404)
return send_from_directory(STATS_FILE.parent, STATS_FILE.name)
def _do_generate():
try:
if STATS_GENERATE_CMD:
subprocess.run(STATS_GENERATE_CMD, shell=True, timeout=600)
else:
cmd = ["goaccess", STATS_LOG_FILE, "--log-format=COMBINED",
f"--output={STATS_FILE}"]
if STATS_JSON:
cmd.append(f"--output={STATS_JSON}")
subprocess.run(cmd, timeout=600, check=False)
finally:
with _gen_lock:
_gen_state["generating"] = False
@app.route("/stats/generate", methods=["POST"])
def stats_generate():
redir = _require_admin()
if redir:
return redir
if STATS_FILE.is_file():
return jsonify({"status": "exists"})
with _gen_lock:
if _gen_state["generating"]:
return jsonify({"status": "generating"})
if not STATS_LOG_FILE and not STATS_GENERATE_CMD:
return jsonify({"status": "error",
"message": "STATS_LOG_FILE non configuré"}), 500
_gen_state["generating"] = True
threading.Thread(target=_do_generate, daemon=True).start()
return jsonify({"status": "started"})
@app.route("/stats/status")
def stats_status():
redir = _require_admin()
if redir:
return redir
return jsonify({"ready": STATS_FILE.is_file(),
"generating": _gen_state["generating"]})
# ── Fichiers bruts (aperçu / téléchargement) ──────────────────────────
@app.route("/raw/<path:subpath>")
def raw_file(subpath):
redir = _require_admin()
if redir:
return redir
target = _safe_path(subpath)
if not target.is_file():
abort(404)
top = Path(subpath).parts[0]
if top in _HIDDEN or top.startswith("."):
abort(403)
return send_from_directory(ASSETS_ROOT, subpath)
# ── Suppression de fichiers ───────────────────────────────────────────
@app.route("/delete", methods=["POST"])
def delete_file():
redir = _require_admin()
if redir:
return redir
subpath = request.form.get("path", "").strip()
if not subpath:
abort(400)
target = _safe_path(subpath)
if not target.exists():
abort(404)
if target.is_dir():
abort(400)
top = Path(subpath).parts[0]
if top in _HIDDEN or top.startswith("."):
abort(403)
parent = str(Path(subpath).parent)
target.unlink()
return redirect(url_for("browse", subpath=parent) if parent != "." else url_for("browse"))
# ── Renommage de fichiers ─────────────────────────────────────────────
@app.route("/rename", methods=["POST"])
def rename_file():
redir = _require_admin()
if redir:
return redir
subpath = request.form.get("path", "").strip()
new_name = request.form.get("new_name", "").strip()
if not subpath or not new_name:
return jsonify({"error": "Paramètres manquants"}), 400
if "/" in new_name or "\\" in new_name or new_name in (".", ".."):
return jsonify({"error": "Nom invalide"}), 400
new_name = secure_filename(new_name)
if not new_name:
return jsonify({"error": "Nom invalide après nettoyage"}), 400
target = _safe_path(subpath)
if not target.is_file():
return jsonify({"error": "Fichier introuvable"}), 404
top = Path(subpath).parts[0]
if top in _HIDDEN or top.startswith("."):
return jsonify({"error": "Accès refusé"}), 403
dest = target.parent / new_name
if not dest.is_relative_to(ASSETS_ROOT):
return jsonify({"error": "Destination invalide"}), 400
if dest.exists():
return jsonify({"error": f"« {new_name} » existe déjà"}), 409
target.rename(dest)
new_path = str(dest.relative_to(ASSETS_ROOT))
return jsonify({
"name": new_name,
"path": new_path,
"browse_url": url_for("browse", subpath=new_path),
})
# ── Upload de fichiers ────────────────────────────────────────────────
@app.route("/upload", methods=["POST"])
def upload_file():
redir = _require_admin()
if redir:
return redir
subpath = request.form.get("path", "").strip()
files = request.files.getlist("files")
if not files or all(f.filename == "" for f in files):
abort(400)
if subpath:
parts = Path(subpath).parts
if parts and (parts[0] in _HIDDEN or parts[0].startswith(".")):
abort(403)
dest = _safe_path(subpath) if subpath else ASSETS_ROOT
if not dest.is_dir():
abort(400)
conflict = request.form.get("conflict", "overwrite")
if conflict not in ("backup", "overwrite", "rename", "skip"):
conflict = "overwrite"
for f in files:
name = secure_filename(f.filename or "")
if not name:
continue
out_path = dest / name
if out_path.exists():
if conflict == "skip":
continue
elif conflict == "backup":
try:
out_path.rename(_backup_path(out_path))
except Exception:
continue
elif conflict == "rename":
out_path = _auto_rename(out_path)
f.save(out_path)
return redirect(url_for("browse", subpath=subpath) if subpath else url_for("browse"))
# ── Redimensionnement d'images ────────────────────────────────────────
@app.route("/resize", methods=["POST"])
def resize_image():
redir = _require_admin()
if redir:
return redir
subpath = request.form.get("path", "").strip()
raw_sizes = request.form.getlist("sizes")
raw_formats = request.form.getlist("formats")
if not subpath or not raw_sizes or not raw_formats:
return jsonify({"error": "Paramètres manquants"}), 400
target = _safe_path(subpath)
if not target.is_file():
abort(404)
ext = target.suffix.lower()
if ext not in IMAGE_EXT:
abort(400)
top = Path(subpath).parts[0]
if top in _HIDDEN or top.startswith("."):
abort(403)
try:
sizes = sorted({int(s) for s in raw_sizes if int(s) in RESIZE_SIZES})
except (ValueError, TypeError):
return jsonify({"error": "Tailles invalides"}), 400
formats = [f for f in raw_formats if f in RESIZE_FORMATS]
if not sizes or not formats:
return jsonify({"error": "Tailles ou formats invalides"}), 400
conflict = request.form.get("conflict", "skip")
if conflict not in ("backup", "overwrite", "rename", "skip"):
conflict = "skip"
is_svg = ext == ".svg"
stem = target.stem
parent = target.parent
created, errors = [], []
for fmt in formats:
for size in sizes:
out_name = f"{stem}_{size}x{size}.{fmt}"
out_path = parent / out_name
out_rel = str(out_path.relative_to(ASSETS_ROOT))
if out_path.exists():
if conflict == "skip":
errors.append({"name": out_name, "reason": "Fichier existant, ignoré"})
continue
elif conflict == "backup":
try:
bak = _backup_path(out_path)
out_path.rename(bak)
except Exception as exc:
errors.append({"name": out_name, "reason": f"Backup impossible : {exc}"})
continue
elif conflict == "rename":
out_path = _auto_rename(out_path)
out_name = out_path.name
out_rel = str(out_path.relative_to(ASSETS_ROOT))
# conflict == "overwrite" : on continue sans rien faire
try:
if fmt == "svg" and is_svg:
shutil.copy2(target, out_path)
created.append({"name": out_name, "path": out_rel})
elif fmt == "svg" and not is_svg:
errors.append({"name": out_name, "reason": "Conversion raster→SVG non supportée"})
elif is_svg:
if not HAS_CAIROSVG:
errors.append({"name": out_name, "reason": "cairosvg non disponible sur ce serveur"})
continue
if fmt in ("png", "ico"):
buf = io.BytesIO()
_cairosvg.svg2png(url=str(target), write_to=buf,
output_width=size, output_height=size)
buf.seek(0)
img = Image.open(buf).convert("RGBA")
if fmt == "ico":
img.save(out_path, format="ICO", sizes=[(size, size)])
else:
img.save(out_path, format="PNG")
else: # jpg
buf = io.BytesIO()
_cairosvg.svg2png(url=str(target), write_to=buf,
output_width=size, output_height=size)
buf.seek(0)
img = Image.open(buf).convert("RGB")
img.save(out_path, format="JPEG", quality=90)
created.append({"name": out_name, "path": out_rel})
else:
img = Image.open(target)
img = img.resize((size, size), Image.LANCZOS)
if fmt == "png":
if img.mode not in ("RGBA", "RGB", "L"):
img = img.convert("RGBA")
img.save(out_path, format="PNG")
elif fmt == "jpg":
img = img.convert("RGB")
img.save(out_path, format="JPEG", quality=90)
elif fmt == "ico":
img = img.convert("RGBA")
img.save(out_path, format="ICO", sizes=[(size, size)])
created.append({"name": out_name, "path": out_rel})
except Exception as exc:
errors.append({"name": out_name, "reason": str(exc)})
return jsonify({"created": created, "errors": errors})
# ── Fichiers CDN publics (dev local uniquement) ───────────────────────
_PUBLIC_TOP = frozenset({"logo", "wiki", "error"})
@app.route("/favicon.ico")
def favicon():
return send_from_directory(ASSETS_ROOT, "favicon.ico")
@app.route("/robots.txt")
def robots():
return send_from_directory(ASSETS_ROOT, "robots.txt")
@app.route("/<path:subpath>")
def cdn_file(subpath):
top = Path(subpath).parts[0]
if top not in _PUBLIC_TOP:
abort(404)
target = _safe_path(subpath)
if not target.is_file():
abort(404)
return send_from_directory(ASSETS_ROOT, subpath)
if __name__ == "__main__":
app.run(debug=True, port=5003)