app.py

#!/usr/bin/env python3
"""
Flask web application for the City Map Poster Generator.
Wraps create_map_poster.py in a web interface with background job processing.
"""

import io
import json
import os
import sys
import threading
import uuid
from pathlib import Path

from flask import Flask, jsonify, render_template, request, send_file

import create_map_poster as cmp
from font_management import load_fonts

app = Flask(__name__)

# Job tracking: job_id -> {status, output_file, error, logs, filename}
jobs: dict = {}
jobs_lock = threading.Lock()

# Only one generation at a time (OSM data fetching + matplotlib rendering)
generation_lock = threading.Lock()


def _load_all_themes():
    """Load all theme data for rendering the UI."""
    themes = cmp.get_available_themes()
    theme_data = {}
    for name in themes:
        theme_data[name] = cmp.load_theme(name)
    return theme_data


def run_generation(job_id: str, params: dict):
    """Run poster generation in a background thread."""
    log_lines: list[str] = []

    class LogCapture(io.TextIOWrapper):
        """Captures writes to stdout/stderr into log_lines."""

        def __init__(self):
            # Wrap a raw bytes buffer so the base class is happy
            super().__init__(io.BytesIO(), encoding="utf-8")

        def write(self, text: str) -> int:
            if text and text.strip():
                log_lines.append(text.rstrip())
                with jobs_lock:
                    if job_id in jobs:
                        jobs[job_id]["logs"] = list(log_lines)
            return len(text)

        def flush(self):
            pass

    old_stdout = sys.stdout
    old_stderr = sys.stderr
    capturer = LogCapture()

    try:
        with jobs_lock:
            jobs[job_id]["status"] = "waiting"

        with generation_lock:
            with jobs_lock:
                jobs[job_id]["status"] = "running"

            sys.stdout = capturer
            sys.stderr = capturer

            try:
                city = params["city"].strip()
                country = params["country"].strip()
                theme_name = params.get("theme", "terracotta")
                distance = max(4000, min(int(params.get("distance", 18000)), 50000))
                width = min(float(params.get("width", 12)), 20)
                height = min(float(params.get("height", 16)), 20)
                output_format = params.get("format", "png")
                display_city = params.get("display_city") or None
                display_country = params.get("display_country") or None
                font_family = params.get("font_family") or None
                lat_override = params.get("latitude") or None
                lon_override = params.get("longitude") or None
                country_label = params.get("country_label") or None

                # Load theme (set global used by rendering functions)
                cmp.THEME = cmp.load_theme(theme_name)

                # Load fonts
                fonts = None
                if font_family:
                    fonts = load_fonts(font_family)
                if not fonts:
                    fonts = cmp.FONTS

                # Get coordinates
                if lat_override and lon_override:
                    from lat_lon_parser import parse
                    coords = (parse(lat_override), parse(lon_override))
                else:
                    coords = cmp.get_coordinates(city, country)

                # Generate poster
                output_file = cmp.generate_output_filename(city, theme_name, output_format)
                cmp.create_poster(
                    city,
                    country,
                    coords,
                    distance,
                    output_file,
                    output_format,
                    width,
                    height,
                    country_label=country_label,
                    display_city=display_city,
                    display_country=display_country,
                    fonts=fonts,
                )

                with jobs_lock:
                    jobs[job_id]["status"] = "done"
                    jobs[job_id]["output_file"] = output_file
                    jobs[job_id]["filename"] = os.path.basename(output_file)
                    jobs[job_id]["logs"] = list(log_lines)

            except Exception as e:
                import traceback
                tb = traceback.format_exc()
                log_lines.append(f"Error: {e}")
                log_lines.append(tb)
                with jobs_lock:
                    jobs[job_id]["status"] = "error"
                    jobs[job_id]["error"] = str(e)
                    jobs[job_id]["logs"] = list(log_lines)

    finally:
        sys.stdout = old_stdout
        sys.stderr = old_stderr


@app.route("/")
def index():
    theme_data = _load_all_themes()
    return render_template("index.html", theme_data=theme_data)


@app.route("/generate", methods=["POST"])
def generate():
    params = request.get_json()
    if not params:
        return jsonify({"error": "No parameters provided"}), 400

    city = (params.get("city") or "").strip()
    country = (params.get("country") or "").strip()
    if not city or not country:
        return jsonify({"error": "City and country are required"}), 400

    job_id = str(uuid.uuid4())
    with jobs_lock:
        jobs[job_id] = {
            "status": "queued",
            "logs": [],
            "output_file": None,
            "filename": None,
            "error": None,
        }

    thread = threading.Thread(target=run_generation, args=(job_id, params), daemon=True)
    thread.start()

    return jsonify({"job_id": job_id})


@app.route("/status/<job_id>")
def status(job_id: str):
    with jobs_lock:
        job = jobs.get(job_id)
    if not job:
        return jsonify({"error": "Job not found"}), 404
    return jsonify(job)


@app.route("/download/<job_id>")
def download(job_id: str):
    with jobs_lock:
        job = jobs.get(job_id)
    if not job or job["status"] != "done":
        return jsonify({"error": "Poster not ready"}), 404
    output_file = job["output_file"]
    if not output_file or not os.path.exists(output_file):
        return jsonify({"error": "File not found"}), 404
    return send_file(output_file, as_attachment=True, download_name=job["filename"])


@app.route("/preview/<job_id>")
def preview(job_id: str):
    with jobs_lock:
        job = jobs.get(job_id)
    if not job or job["status"] != "done":
        return jsonify({"error": "Poster not ready"}), 404
    output_file = job["output_file"]
    if not output_file or not os.path.exists(output_file):
        return jsonify({"error": "File not found"}), 404
    return send_file(output_file)


if __name__ == "__main__":
    port = int(os.environ.get("PORT", 5000))
    app.run(debug=True, host="0.0.0.0", port=port)