481 lines
16 KiB
Python
481 lines
16 KiB
Python
"""Flask routes for the alt text generator."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import re
|
|
from datetime import datetime
|
|
from html import unescape
|
|
|
|
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__)
|
|
|
|
|
|
_TAG_RE = re.compile(r"<[^>]+>")
|
|
|
|
|
|
def _html_to_plain_text(raw: str | None) -> str:
|
|
if not raw:
|
|
return ""
|
|
text = _TAG_RE.sub(" ", raw)
|
|
text = unescape(text)
|
|
return " ".join(text.split()).strip()
|
|
|
|
|
|
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/<asset_id>", 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":
|
|
thread_context: list[str] | None = None
|
|
if (
|
|
reply_to_latest
|
|
and mastodon_client
|
|
and latest_status
|
|
and latest_status.get("id")
|
|
):
|
|
status_id = str(latest_status["id"])
|
|
context_entries: list[str] = []
|
|
try:
|
|
ancestors = mastodon_client.get_status_ancestors(status_id)
|
|
except MastodonError as exc:
|
|
if not error_message:
|
|
error_message = str(exc)
|
|
ancestors = []
|
|
for item in ancestors:
|
|
ancestor_text = _html_to_plain_text(item.get("content"))
|
|
if ancestor_text:
|
|
context_entries.append(ancestor_text)
|
|
latest_content = latest_status.get("content")
|
|
if not latest_content:
|
|
try:
|
|
latest_payload = mastodon_client.get_status(status_id)
|
|
latest_content = latest_payload.get("content")
|
|
except MastodonError as exc:
|
|
if not error_message:
|
|
error_message = str(exc)
|
|
latest_content = None
|
|
latest_text = _html_to_plain_text(latest_content)
|
|
if latest_text:
|
|
context_entries.append(latest_text)
|
|
if context_entries:
|
|
thread_context = context_entries
|
|
try:
|
|
post_text = generator.improve_post_text(
|
|
post_text, instructions, hashtag_counts, thread_context
|
|
)
|
|
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: <a href="{status_url}" target="_blank" rel="noopener">{status_url}</a>',
|
|
"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/<asset_id>/<variant>")
|
|
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
|