m038
7dc7caee26
Fix D: apply _sanitise_slug() to grav_trip_slug in POST /select before storing in TripState, preventing path traversal via ../sequences. Fix E: add _yaml_str() helper that doubles single quotes; apply to title, location_city, and location_country in both run_export and overwrite_export frontmatter blocks, preventing invalid YAML for values like Xi'an. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
80 lines
2.4 KiB
Python
80 lines
2.4 KiB
Python
import re
|
|
from pathlib import Path
|
|
|
|
from flask import Blueprint, current_app, redirect, render_template, request
|
|
from app.immich import ImmichClient
|
|
from app.state import TripState, Photo, load_state, save_state
|
|
|
|
bp = Blueprint("albums", __name__)
|
|
|
|
|
|
def _sanitise_slug(s: str) -> str:
|
|
s = s.strip().lower()
|
|
s = re.sub(r'[^a-z0-9-]+', '-', s)
|
|
return s.strip('-')
|
|
|
|
|
|
def _client():
|
|
return ImmichClient(current_app.config["IMMICH_URL"],
|
|
current_app.config["IMMICH_API_KEY"])
|
|
|
|
|
|
@bp.get("/")
|
|
def index():
|
|
try:
|
|
albums = _client().list_albums()
|
|
error = None
|
|
except ConnectionError as e:
|
|
albums = []
|
|
error = str(e)
|
|
state_dir = Path(current_app.config["STATE_DIR"])
|
|
for album in albums:
|
|
album["has_state"] = (state_dir / f"{album['id']}.json").exists()
|
|
return render_template("phase1.html", albums=albums, error=error,
|
|
current_phase="", album_id=None,
|
|
phase_stale=[], notes_content="")
|
|
|
|
|
|
@bp.post("/select")
|
|
def select():
|
|
album_ids = request.form.getlist("album_ids[]")
|
|
grav_trip_slug = _sanitise_slug(request.form["grav_trip_slug"])
|
|
start_over = request.form.get("start_over") == "1"
|
|
|
|
if len(album_ids) == 1:
|
|
primary_id = album_ids[0]
|
|
else:
|
|
primary_id = "__merged__" + "_".join(sorted(album_ids))
|
|
|
|
existing = load_state(primary_id, current_app)
|
|
if existing and not start_over:
|
|
return redirect(f"/{existing.phase}?album_id={primary_id}")
|
|
|
|
# Fetch and merge assets, deduplicating by asset ID
|
|
all_assets = {}
|
|
album_name_parts = []
|
|
for aid in album_ids:
|
|
album = _client().get_album(aid)
|
|
album_name_parts.append(album["albumName"])
|
|
for asset in album["assets"]:
|
|
if asset["id"] not in all_assets:
|
|
all_assets[asset["id"]] = asset
|
|
|
|
photos = [
|
|
Photo(id=a["id"], original_filename=a["originalFileName"],
|
|
local_datetime=a["localDateTime"])
|
|
for a in sorted(all_assets.values(), key=lambda x: x["localDateTime"])
|
|
]
|
|
for i, p in enumerate(photos):
|
|
p.order = i
|
|
|
|
state = TripState(
|
|
album_id=primary_id,
|
|
album_name=", ".join(album_name_parts),
|
|
grav_trip_slug=grav_trip_slug,
|
|
photos=photos,
|
|
)
|
|
save_state(state, current_app)
|
|
return redirect(f"/triage?album_id={primary_id}")
|
|
|