"""Flask routes for the alt text generator.""" from __future__ import annotations import base64 from datetime import datetime import UniAuth.auth from flask import ( Blueprint, Response, abort, current_app, flash, redirect, render_template, request, url_for, ) from .immich import ImmichAsset, ImmichError from .mastodon import MastodonError from .openai_client import AltTextGenerationError, TextImprovementError bp = Blueprint("main", __name__) def _parse_timestamp(raw: str | None) -> datetime | None: if not raw: return None try: normalized = raw.replace("Z", "+00:00") return datetime.fromisoformat(normalized) except ValueError: return None def _humanize_timestamp(raw: str | None) -> str | None: dt = _parse_timestamp(raw) if not dt: return raw return dt.strftime("%d %b %Y %H:%M") def _timestamp_with_timezone(raw: str | None) -> str | None: dt = _parse_timestamp(raw) if not dt: return raw label = dt.strftime("%d %b %Y %H:%M") tzname = dt.tzname() if tzname: return f"{label} {tzname}" offset = dt.utcoffset() if offset is not None: total_minutes = int(offset.total_seconds() // 60) hours, minutes = divmod(abs(total_minutes), 60) sign = "+" if total_minutes >= 0 else "-" return f"{label} UTC{sign}{hours:02d}:{minutes:02d}" return f"{label} (timezone unknown)" def _coordinates_text(asset: ImmichAsset) -> str | None: if asset.latitude is None or asset.longitude is None: return None return f"{asset.latitude:.5f}, {asset.longitude:.5f}" def _generate_alt_text_for_asset(asset: ImmichAsset, notes: str | None = None) -> str: immich_client = current_app.immich_client generator = current_app.alt_text_generator content, mime_type = immich_client.fetch_asset_content(asset.id, "preview") data_url = "data:{};base64,{}".format( mime_type, base64.b64encode(content).decode("ascii") ) return generator.generate_alt_text( data_url, notes, captured_at=_timestamp_with_timezone(asset.captured_at), location=asset.location_label, coordinates=_coordinates_text(asset), ) def _ensure_alt_text(asset: ImmichAsset) -> str: cache = current_app.alt_text_cache cached = cache.get(asset.id) if cached: return cached generated = _generate_alt_text_for_asset(asset) cache.set(asset.id, generated) return generated def _unique_asset_ids(values: list[str]) -> list[str]: seen: set[str] = set() ordered: list[str] = [] for value in values: value = (value or "").strip() if not value or value in seen: continue seen.add(value) ordered.append(value) return ordered MAX_MEDIA_ATTACHMENTS = 4 bp.before_request(UniAuth.auth.require_authentication) @bp.route("/callback") def auth_callback(): """Process the authentication callback.""" return UniAuth.auth.auth_callback() @bp.route("/") def index(): return redirect(url_for("main.compose_select")) @bp.route("/alt-text-helper") def alt_helper(): immich_client = current_app.immich_client alt_cache = current_app.alt_text_cache days = current_app.config.get("RECENT_DAYS", 3) assets = [] error_message = None try: for asset in immich_client.get_recent_assets(days=days): assets.append( { "id": asset.id, "file_name": asset.file_name, "captured_at": asset.captured_at, "captured_display": _humanize_timestamp(asset.captured_at), "thumbnail_url": asset.thumbnail_url, "alt_text": alt_cache.get(asset.id), } ) except ImmichError as exc: error_message = str(exc) return render_template("index.html", assets=assets, error_message=error_message) @bp.route("/assets/", methods=["GET", "POST"]) def asset_detail(asset_id: str): immich_client = current_app.immich_client alt_cache = current_app.alt_text_cache error_message = None notes = "" try: asset = immich_client.get_asset(asset_id) except ImmichError as exc: return render_template( "detail.html", asset=None, alt_text=None, error_message=str(exc), notes=notes, ) alt_text = alt_cache.get(asset_id) if request.method == "POST": notes = request.form.get("notes", "") try: generated = _generate_alt_text_for_asset(asset, notes) alt_cache.set(asset_id, generated) flash("Alt text generated.") return redirect(url_for("main.asset_detail", asset_id=asset_id)) except ImmichError as exc: error_message = f"Failed to fetch image: {exc}" except AltTextGenerationError as exc: error_message = str(exc) formatted_asset = { "id": asset.id, "file_name": asset.file_name, "captured_at": asset.captured_at, "captured_display": _humanize_timestamp(asset.captured_at), "thumbnail_url": asset.thumbnail_url, "preview_url": asset.preview_url, "original_url": asset.original_url, "web_url": asset.web_url, "latitude": asset.latitude, "longitude": asset.longitude, "location": asset.location_label, } return render_template( "detail.html", asset=formatted_asset, alt_text=alt_cache.get(asset_id), error_message=error_message, notes=notes, ) @bp.route("/compose/select", methods=["GET", "POST"]) def compose_select(): immich_client = current_app.immich_client alt_cache = current_app.alt_text_cache days = current_app.config.get("RECENT_DAYS", 3) asset_objs: list[ImmichAsset] = [] error_message = None try: asset_objs = immich_client.get_recent_assets(days=days, limit=200) except ImmichError as exc: error_message = str(exc) usage_map = alt_cache.get_usage_map(asset.id for asset in asset_objs) assets = [] for asset in asset_objs: usage = usage_map.get(asset.id) assets.append( { "id": asset.id, "file_name": asset.file_name, "captured_display": _humanize_timestamp(asset.captured_at), "alt_text": alt_cache.get(asset.id), "posted": bool(usage), "last_posted_display": _humanize_timestamp( usage["last_posted_at"] if usage else None ), } ) selected_set: set[str] = set() if request.method == "POST": selected = _unique_asset_ids(request.form.getlist("asset_ids")) selected_set = set(selected) if not selected: error_message = "Select at least one photo." elif len(selected) > MAX_MEDIA_ATTACHMENTS: error_message = f"Select up to {MAX_MEDIA_ATTACHMENTS} photos." else: return redirect(url_for("main.compose_draft", ids=",".join(selected))) return render_template( "compose_select.html", assets=assets, error_message=error_message, max_photos=MAX_MEDIA_ATTACHMENTS, selected_ids=selected_set, ) @bp.route("/compose/draft", methods=["GET", "POST"]) def compose_draft(): mastodon_client = getattr(current_app, "mastodon_client", None) immich_client = current_app.immich_client alt_cache = current_app.alt_text_cache generator = current_app.alt_text_generator hashtag_counts = current_app.config.get("HASHTAG_COUNTS_TEXT") error_message = None post_text = "" instructions = "" reply_to_latest = False if request.method == "POST": asset_ids = _unique_asset_ids(request.form.getlist("asset_ids")) post_text = request.form.get("post_text", "") instructions = request.form.get("post_instructions", "") reply_values = request.form.getlist("reply_to_latest") reply_to_latest = any(value in {"1", "on", "true"} for value in reply_values) latest_status_id_form = request.form.get("latest_status_id") else: ids_param = request.args.get("ids", "") asset_ids = _unique_asset_ids(ids_param.split(",") if ids_param else []) latest_status_id_form = None if not asset_ids: flash("Choose photos before composing a post.") return redirect(url_for("main.compose_select")) if len(asset_ids) > MAX_MEDIA_ATTACHMENTS: flash(f"Select at most {MAX_MEDIA_ATTACHMENTS} photos.") return redirect(url_for("main.compose_select")) latest_status_raw = None latest_status_error = None if mastodon_client: try: latest_status_raw = mastodon_client.get_latest_status() except MastodonError as exc: latest_status_error = str(exc) assets: list[ImmichAsset] = [] try: for asset_id in asset_ids: assets.append(immich_client.get_asset(asset_id)) except ImmichError as exc: return render_template( "compose_draft.html", assets=[], error_message=str(exc), post_text=post_text, instructions=instructions, mastodon_ready=bool(mastodon_client), ) latest_status = None if latest_status_raw: latest_status = { "id": latest_status_raw.get("id"), "content": latest_status_raw.get("content", ""), "created_display": _humanize_timestamp(latest_status_raw.get("created_at")), "url": latest_status_raw.get("url"), } elif latest_status_id_form: latest_status = { "id": latest_status_id_form, "content": "", "created_display": None, "url": None, } asset_entries = [] for asset in assets: if request.method == "POST": alt_value = request.form.get(f"alt_text_{asset.id}", "") else: try: alt_value = _ensure_alt_text(asset) except (ImmichError, AltTextGenerationError) as exc: alt_value = "" if not error_message: error_message = str(exc) asset_entries.append( { "id": asset.id, "file_name": asset.file_name, "captured_display": _humanize_timestamp(asset.captured_at), "preview_url": url_for( "main.asset_proxy", asset_id=asset.id, variant="preview" ), "alt_text": alt_value, } ) if request.method == "POST": action = request.form.get("action") if action == "refine": try: post_text = generator.improve_post_text( post_text, instructions, hashtag_counts ) flash("Post refined with ChatGPT.") except TextImprovementError as exc: error_message = str(exc) elif action == "post": if not mastodon_client: error_message = "Configure Mastodon access before posting." elif not post_text.strip(): error_message = "Enter post text before posting." else: try: media_ids: list[str] = [] for entry, asset in zip(asset_entries, assets): alt_value = entry["alt_text"].strip() if not alt_value: raise ValueError( f"Alt text missing for {entry['file_name']}" ) entry["alt_text"] = alt_value alt_cache.set(asset.id, alt_value) content, mime_type = immich_client.fetch_asset_content( asset.id, "original" ) media_id = mastodon_client.upload_media( asset.file_name, content, mime_type, alt_value ) media_ids.append(media_id) reply_target_id = latest_status_id_form or ( latest_status["id"] if latest_status else None ) reply_id = reply_target_id if reply_to_latest else None status = mastodon_client.create_status( post_text.strip(), media_ids, in_reply_to_id=reply_id ) status_url = status.get("url") for asset in assets: alt_cache.mark_posted(asset.id, status_url) if status_url: flash( f'Post sent to Mastodon: {status_url}', "success", ) else: flash("Post sent to Mastodon.", "success") return redirect(url_for("main.index")) except ValueError as exc: error_message = str(exc) except ImmichError as exc: error_message = f"Failed to fetch media: {exc}" except MastodonError as exc: error_message = str(exc) elif action: error_message = "Unknown action." return render_template( "compose_draft.html", assets=asset_entries, error_message=error_message, post_text=post_text, instructions=instructions, mastodon_ready=bool(mastodon_client), max_photos=MAX_MEDIA_ATTACHMENTS, latest_status=latest_status, latest_status_error=latest_status_error, reply_to_latest=reply_to_latest, ) @bp.route("/proxy/assets//") def asset_proxy(asset_id: str, variant: str): immich_client = current_app.immich_client try: content, mimetype = immich_client.fetch_asset_content(asset_id, variant) except ImmichError as exc: abort(404, description=str(exc)) response = Response(content, mimetype=mimetype) response.headers.setdefault("Cache-Control", "public, max-age=3600") return response