API Reference

Documentation for the propongo Python modules.

Main Flask application for Propongo.

create_app()

Create and configure the Flask application.

Returns:

Type	Description
Flask	Configured Flask application

Source code in app/main.py

def create_app() -> Flask:
    """Create and configure the Flask application.

    Returns:
        Configured Flask application
    """
    app = Flask(__name__)
    # Trust Render's proxy so request.host/scheme (used by url_for(_external=True)
    # for password-reset links) reflect the public URL, not an internal one.
    app.wsgi_app = ProxyFix(app.wsgi_app, x_proto=1, x_host=1)
    app.secret_key = Config.SECRET_KEY
    app.config.update(
        SESSION_COOKIE_HTTPONLY=True,
        SESSION_COOKIE_SAMESITE="Lax",
        SESSION_COOKIE_SECURE=os.environ.get("PROPONGO_SESSION_COOKIE_SECURE", "false").lower() in ("true", "1", "yes"),
    )

    logger.info(f"Starting Propongo v{__version__}")

    ensure_admin_user()
    init_login_manager(app)

    app.register_blueprint(export_bp)
    app.register_blueprint(snippets_bp)
    app.register_blueprint(results_bp)
    app.register_blueprint(rfp_bp)
    app.register_blueprint(auth_bp)

    PUBLIC_ENDPOINTS = {"static", "set_language_route", "healthz"}

    @app.before_request
    def require_login():
        """Redirect unauthenticated requests to the login page when auth is enabled."""
        if not auth_enabled():
            return
        if request.endpoint is None:
            return
        if request.endpoint.startswith("auth.") or request.endpoint in PUBLIC_ENDPOINTS:
            return
        if current_user.is_authenticated:
            return
        return redirect(url_for("auth.login", next=request.path))

    @app.route("/healthz")
    def healthz():
        """Health check endpoint for the hosting platform."""
        return Response("ok", status=200)

    @app.before_request
    def set_language():
        """Read the selected language from the cookie into flask.g."""
        lang = request.cookies.get(LANG_COOKIE, DEFAULT_LANG)
        if lang not in LANGUAGES:
            lang = DEFAULT_LANG
        g.lang = lang

    @app.context_processor
    def inject_version():
        """Inject application version into all templates."""
        return {"app_version": __version__}

    @app.context_processor
    def inject_i18n():
        """Inject translation helpers and language info into all templates."""
        lang = getattr(g, "lang", DEFAULT_LANG)

        def _(text: str) -> str:
            return translate(text, lang)

        return {
            "_": _,
            "current_lang": lang,
            "LANGUAGES": LANGUAGES,
            "js_translations": TRANSLATIONS.get(lang, {}),
        }

    @app.context_processor
    def inject_auth():
        """Inject authentication state into all templates."""
        return {
            "auth_enabled": auth_enabled(),
            "allow_registration": allow_registration(),
            "current_user": current_user,
        }

    @app.route("/set-language/<lang>")
    def set_language_route(lang):
        """Persist the selected app language in a cookie and redirect back."""
        if lang not in LANGUAGES:
            lang = DEFAULT_LANG
        resp = redirect(request.referrer or url_for("index"))
        resp.set_cookie(LANG_COOKIE, lang, max_age=60 * 60 * 24 * 365, samesite="Lax")
        return resp

    app.jinja_env.filters["md"] = lambda text: Markup(markdown_to_html(text))
    app.jinja_env.filters["currency"] = lambda value: f"{value:,.0f}"

    @app.route("/")
    def index():
        """List all proposals on the homepage."""
        proposals = Proposal.list_all()
        return render_template("index.html", proposals=proposals)

    @app.route("/new", methods=["GET", "POST"])
    def new_proposal():
        """Create a proposal (POST with a title) or bounce GET to the index."""
        if request.method == "POST":
            data = request.get_json(silent=True) or {}
            title = (data.get("title") or "").strip()
            if not title:
                return jsonify({"error": "Title required"}), 400
            proposal = Proposal(title=title)
            proposal.save()
            logger.info(f"Created new proposal: {proposal.id}")
            return jsonify({"id": proposal.id}), 201
        return redirect(url_for("index"))

    @app.route("/editor/<proposal_id>")
    def editor(proposal_id):
        """Render the proposal editor page."""
        proposal = Proposal.load(proposal_id)
        if not proposal:
            return redirect(url_for("index"))
        return render_template(
            "base.html",
            proposal=proposal,
            tasks=proposal.tasks,
            budget_items=proposal.budget_items,
        )

    @app.route("/api/proposal/<proposal_id>", methods=["GET"])
    def get_proposal(proposal_id: str) -> Tuple[Response, int]:
        """Return a proposal as JSON."""
        proposal = Proposal.load(proposal_id)
        if not proposal:
            logger.warning(f"Proposal not found: {proposal_id}")
            return jsonify(ERROR_MESSAGES['PROPOSAL_NOT_FOUND']), 404
        return jsonify(proposal.to_dict()), 200

    @app.route("/api/proposal/<proposal_id>", methods=["PUT"])
    def save_proposal(proposal_id: str) -> Tuple[Response, int]:
        """Update and save an existing proposal."""
        data = request.get_json()
        if not data:
            return jsonify(ERROR_MESSAGES['NO_DATA']), 400

        with _proposal_locks_lock:
            if proposal_id not in _proposal_locks:
                _proposal_locks[proposal_id] = threading.Lock()
            lock = _proposal_locks[proposal_id]

        with lock:
            proposal = Proposal.load(proposal_id)
            if not proposal:
                logger.warning(f"Proposal not found for update: {proposal_id}")
                return jsonify(ERROR_MESSAGES['PROPOSAL_NOT_FOUND']), 404

            if "tasks" in data:
                incoming_tasks = data.pop("tasks")
                existing_by_id = {t.get("id"): t for t in proposal.tasks}
                merged = []
                for t in incoming_tasks:
                    tid = t.get("id", "")
                    if tid in existing_by_id:
                        merged_task = dict(existing_by_id[tid])
                        merged_task.update(t)
                    else:
                        merged_task = t
                    merged.append(merged_task)
                proposal.tasks = merged

            skip_fields = {"id", "title", "created_at"}
            for key, value in data.items():
                if key in skip_fields:
                    continue
                if hasattr(proposal, key):
                    setattr(proposal, key, value)

            proposal.save()
            return jsonify(proposal.to_dict())

    @app.route("/api/proposal/<proposal_id>", methods=["DELETE"])
    def delete_proposal(proposal_id):
        """Delete a proposal by ID."""
        Proposal.delete(proposal_id)
        return jsonify({"ok": True})

    @app.route("/api/proposal/<proposal_id>/save-as", methods=["POST"])
    def save_proposal_as(proposal_id):
        """Create a copy of a proposal with a new title."""
        data = request.get_json()
        title = data.get("title", "").strip()
        if not title:
            return jsonify({"error": "Title required"}), 400

        proposal = Proposal.load(proposal_id)
        if not proposal:
            return jsonify({"error": "Not found"}), 404

        new_id = re.sub(r"[^a-z0-9_-]", "_", title.lower())
        new_id = re.sub(r"_+", "_", new_id).strip("_")
        if not new_id:
            new_id = uuid.uuid4().hex[:8]

        original_id = new_id
        counter = 1
        while Proposal.load(new_id):
            new_id = f"{original_id}_{counter}"
            counter += 1

        new_proposal = Proposal(id=new_id, title=title)
        new_proposal.client_name = proposal.client_name
        new_proposal.subtitle = getattr(proposal, 'subtitle', '') or ''
        new_proposal.project_summary = proposal.project_summary
        new_proposal.scope = getattr(proposal, 'scope', '') or ''
        new_proposal.tasks = list(proposal.tasks)
        new_proposal.qualifications = proposal.qualifications
        new_proposal.budget_items = list(proposal.budget_items)
        new_proposal.budget_item_timings = dict(proposal.budget_item_timings) if proposal.budget_item_timings else {}
        new_proposal.start_date = proposal.start_date
        new_proposal.indirect_percent = getattr(proposal, 'indirect_percent', 0) or 0
        new_proposal.show_budget_description = getattr(proposal, 'show_budget_description', False)
        new_proposal.budget_description = getattr(proposal, 'budget_description', '') or ''
        new_proposal.custom_sections = list(proposal.custom_sections) if proposal.custom_sections else []
        new_proposal.timeline_use_days = proposal.timeline_use_days
        new_proposal.timeline_show_budget = proposal.timeline_show_budget
        new_proposal.end_date = getattr(proposal, 'end_date', '') or ''
        new_proposal.milestones = list(proposal.milestones) if proposal.milestones else []
        new_proposal.reports = list(proposal.reports) if proposal.reports else []
        new_proposal.save()

        return jsonify({"id": new_id}), 201

    @app.route("/api/proposals", methods=["GET"])
    def list_proposals():
        """Return all proposals as JSON."""
        return jsonify(Proposal.list_all())

    @app.route("/templates")
    def templates_page():
        """Render the templates listing page."""
        templates = Proposal.list_templates()
        return render_template("templates.html", templates=templates)

    @app.route("/api/templates", methods=["GET"])
    def list_templates():
        """Return all templates as JSON."""
        return jsonify(Proposal.list_templates())

    @app.route("/api/template/<template_id>", methods=["DELETE"])
    def delete_template(template_id):
        """Delete a template by ID."""
        Proposal.delete(template_id, is_template=True)
        return jsonify({"ok": True})

    @app.route("/api/proposal/<proposal_id>/save-as-template", methods=["POST"])
    def save_as_template(proposal_id):
        """Save a proposal as a reusable template."""
        data = request.get_json()
        template_name = data.get("template_name", "").strip()
        template_category = data.get("template_category", "").strip()
        if not template_name:
            return jsonify({"error": "Template name required"}), 400

        proposal = Proposal.load(proposal_id)
        if not proposal:
            return jsonify({"error": "Not found"}), 404

        new_id = re.sub(r"[^a-z0-9_-]", "_", template_name.lower())
        new_id = re.sub(r"_+", "_", new_id).strip("_")
        if not new_id:
            new_id = uuid.uuid4().hex[:8]

        original_id = new_id
        counter = 1
        while Proposal.load(new_id, is_template=True):
            new_id = f"{original_id}_{counter}"
            counter += 1

        tmpl = Proposal(id=new_id, title=proposal.title)
        tmpl.client_name = proposal.client_name
        tmpl.subtitle = getattr(proposal, 'subtitle', '') or ''
        tmpl.project_summary = proposal.project_summary
        tmpl.scope = getattr(proposal, 'scope', '') or ''
        tmpl.tasks = list(proposal.tasks)
        tmpl.qualifications = proposal.qualifications
        tmpl.budget_items = list(proposal.budget_items)
        tmpl.budget_item_timings = dict(proposal.budget_item_timings) if proposal.budget_item_timings else {}
        tmpl.indirect_percent = getattr(proposal, 'indirect_percent', 0) or 0
        tmpl.show_budget_description = getattr(proposal, 'show_budget_description', False)
        tmpl.budget_description = getattr(proposal, 'budget_description', '') or ''
        tmpl.custom_sections = list(proposal.custom_sections) if proposal.custom_sections else []
        tmpl.timeline_use_days = proposal.timeline_use_days
        tmpl.timeline_show_budget = proposal.timeline_show_budget
        tmpl.is_template = True
        tmpl.template_name = template_name
        tmpl.template_category = template_category
        tmpl.save()

        return jsonify({"id": new_id}), 201

    @app.route("/templates/new-from/<template_id>")
    def new_from_template(template_id):
        """Create a new proposal from a template and redirect to editor."""
        tmpl = Proposal.load(template_id, is_template=True)
        if not tmpl:
            return redirect(url_for("templates_page"))

        new_id = uuid.uuid4().hex[:8]
        proposal = Proposal(id=new_id, title=tmpl.title)
        proposal.client_name = tmpl.client_name
        proposal.subtitle = getattr(tmpl, 'subtitle', '') or ''
        proposal.project_summary = tmpl.project_summary
        proposal.scope = getattr(tmpl, 'scope', '') or ''
        proposal.tasks = list(tmpl.tasks)
        proposal.qualifications = tmpl.qualifications
        proposal.budget_items = list(tmpl.budget_items)
        proposal.budget_item_timings = dict(tmpl.budget_item_timings) if tmpl.budget_item_timings else {}
        proposal.indirect_percent = getattr(tmpl, 'indirect_percent', 0) or 0
        proposal.show_budget_description = getattr(tmpl, 'show_budget_description', False)
        proposal.budget_description = getattr(tmpl, 'budget_description', '') or ''
        proposal.custom_sections = list(tmpl.custom_sections) if tmpl.custom_sections else []
        proposal.timeline_use_days = tmpl.timeline_use_days
        proposal.timeline_show_budget = tmpl.timeline_show_budget
        proposal.save()

        return redirect(url_for("editor", proposal_id=proposal.id))

    @app.route("/scope/<proposal_id>")
    def scope_tab(proposal_id: str) -> Tuple[str, int] | Tuple[Response, int]:
        """Render the scope/editing tab for a proposal."""
        proposal = Proposal.load(proposal_id)
        if not proposal:
            return jsonify(ERROR_MESSAGES['PROPOSAL_NOT_FOUND']), 404
        return render_template("scope.html", proposal=proposal, tasks=proposal.tasks)

    @app.route("/map/<proposal_id>")
    def map_tab(proposal_id):
        """Render the map tab with an embedded GeoLibre map."""
        proposal = Proposal.load(proposal_id)
        if not proposal:
            return jsonify(ERROR_MESSAGES['PROPOSAL_NOT_FOUND']), 404
        map_config = getattr(proposal, 'map_config', {}) or {}
        return render_template(
            "map.html",
            proposal=proposal,
            map_config=map_config,
            geolibre_src=build_geolibre_embed_src(map_config),
        )

    @app.route("/api/map/<proposal_id>/upload", methods=["POST"])
    def upload_map_image(proposal_id):
        """Upload a static map image (PNG/JPG) for a proposal.

        Saves it under the data dir as ``<proposal_id>.<ext>`` and switches the
        map to ``static_image`` mode so the image is used in preview and the
        WeasyPrint PDF (which cannot run the live GeoLibre iframe).
        """
        proposal = Proposal.load(proposal_id)
        if not proposal:
            return jsonify(ERROR_MESSAGES['PROPOSAL_NOT_FOUND']), 404
        if 'file' not in request.files:
            return jsonify(ERROR_MESSAGES['NO_FILE']), 400
        file = request.files['file']
        if file.filename == '':
            return jsonify({"error": "No file selected"}), 400
        ext = os.path.splitext(file.filename)[1].lower()
        if ext not in ('.png', '.jpg', '.jpeg'):
            return jsonify(ERROR_MESSAGES['INVALID_FILE_TYPE']), 400

        os.makedirs(Config.MAPS_DIR, exist_ok=True)
        filename = f"{proposal.id}{ext}"
        file.save(os.path.join(Config.MAPS_DIR, filename))

        map_config = dict(getattr(proposal, 'map_config', None) or {})
        map_config.update({
            "mode": "static_image",
            "image_path": filename,
            "show_in_preview": True,
        })
        proposal.map_config = map_config
        proposal.save()
        return jsonify({"ok": True, "url": f"/map-image/{proposal.id}"}), 200

    @app.route("/map-image/<proposal_id>")
    def serve_map_image(proposal_id):
        """Serve the proposal's uploaded static map image."""
        proposal = Proposal.load(proposal_id)
        if not proposal:
            return jsonify(ERROR_MESSAGES['PROPOSAL_NOT_FOUND']), 404
        map_config = getattr(proposal, 'map_config', None) or {}
        image_path = (map_config.get('image_path') or '').strip()
        if map_config.get('mode') != 'static_image' or not image_path:
            return jsonify({"error": "No map image"}), 404
        full = os.path.join(Config.MAPS_DIR, image_path)
        if not os.path.isfile(full):
            return jsonify({"error": "Map image not found"}), 404
        return send_file(full)

    @app.route("/api/map/<proposal_id>/remove-image", methods=["POST"])
    def remove_map_image(proposal_id):
        """Remove the proposal's uploaded static map image."""
        proposal = Proposal.load(proposal_id)
        if not proposal:
            return jsonify(ERROR_MESSAGES['PROPOSAL_NOT_FOUND']), 404
        map_config = dict(getattr(proposal, 'map_config', None) or {})
        image_path = (map_config.get('image_path') or '').strip()
        if image_path:
            full = os.path.join(Config.MAPS_DIR, image_path)
            if os.path.isfile(full):
                try:
                    os.remove(full)
                except OSError:
                    pass
        map_config.pop('image_path', None)
        if map_config.get('mode') == 'static_image':
            map_config['mode'] = 'basemap'
        proposal.map_config = map_config
        proposal.save()
        return jsonify({"ok": True}), 200

    @app.route("/budget/<proposal_id>")
    def budget_tab(proposal_id):
        """Render the budget tab with cost breakdowns per task."""
        proposal = Proposal.load(proposal_id)
        if not proposal:
            return jsonify(ERROR_MESSAGES['PROPOSAL_NOT_FOUND']), 404

        task_budgets = {}
        for task in proposal.tasks:
            items = [b for b in proposal.budget_items if b.get("task_id") == task["id"]]
            subtotal = sum(i.get("cost_per_unit", 0) * i.get("units", 0) for i in items)
            task_budgets[task["id"]] = {
                "task": task,
                "items": items,
                "subtotal": subtotal,
            }

        indirect_percent = getattr(proposal, 'indirect_percent', 0) or 0
        indirect_amount = proposal.total_budget * (indirect_percent / 100)
        total_with_indirect = proposal.total_budget + indirect_amount

        try:
            start_date_year = int(str(proposal.start_date)[:4])
        except (ValueError, TypeError):
            start_date_year = 2025

        return render_template(
            "budget.html",
            proposal=proposal,
            tasks=proposal.tasks,
            budget_items=proposal.budget_items,
            total_budget=proposal.total_budget,
            task_budgets=task_budgets,
            indirect_percent=indirect_percent,
            indirect_amount=indirect_amount,
            total_with_indirect=total_with_indirect,
            budget_by_year=build_budget_by_year(proposal),
            start_date_year=start_date_year,
        )

    @app.route("/qualifications/<proposal_id>")
    def qualifications_tab(proposal_id):
        """Render the qualifications tab for a proposal."""
        proposal = Proposal.load(proposal_id)
        if not proposal:
            return jsonify(ERROR_MESSAGES['PROPOSAL_NOT_FOUND']), 404
        return render_template("qualifications.html", proposal=proposal)

    @app.route("/timeline/<proposal_id>")
    def timeline_tab(proposal_id):
        """Render the timeline tab with scheduling information."""
        proposal = Proposal.load(proposal_id)
        if not proposal:
            return jsonify(ERROR_MESSAGES['PROPOSAL_NOT_FOUND']), 404

        task_budgets = {}
        timings = proposal.budget_item_timings or {}
        for task in proposal.tasks:
            items = [b for b in proposal.budget_items if b.get("task_id") == task["id"]]
            for item in items:
                t = timings.get(item.get("id", ""), {})
                if t:
                    item["start_month"] = t.get("start_month")
                    item["start_year"] = t.get("start_year")
                    item["duration_months"] = t.get("duration_months", 1)
                    if t.get("lead_entity"):
                        item["lead_entity"] = t["lead_entity"]
                    item["recurring"] = t.get("recurring", False)
                    item["recurring_interval"] = t.get("recurring_interval", 3)
            if items:
                task_budgets[task["id"]] = {
                    "task": task,
                    "items": items,
                }

        try:
            sd = datetime.strptime(proposal.start_date, "%Y-%m-%d")
            start_date_month = sd.month
            start_date_year = sd.year
        except (ValueError, TypeError):
            now = datetime.now()
            start_date_month = now.month
            start_date_year = now.year

        try:
            ed = datetime.strptime(proposal.end_date, "%Y-%m-%d") if proposal.end_date else None
            end_date_month = ed.month if ed else start_date_month
            end_date_year = ed.year if ed else start_date_year + 1
        except (ValueError, TypeError):
            end_date_month = start_date_month
            end_date_year = start_date_year + 1

        return render_template(
            "timeline.html",
            proposal=proposal,
            tasks=proposal.tasks,
            start_date=proposal.start_date,
            start_date_month=start_date_month,
            start_date_year=start_date_year,
            end_date=proposal.end_date,
            end_date_month=end_date_month,
            end_date_year=end_date_year,
            task_budgets=task_budgets,
        )

    @app.route("/custom-sections/<proposal_id>")
    def custom_sections_tab(proposal_id):
        """Render the custom sections tab for a proposal."""
        proposal = Proposal.load(proposal_id)
        if not proposal:
            return jsonify(ERROR_MESSAGES['PROPOSAL_NOT_FOUND']), 404
        custom_sections = getattr(proposal, 'custom_sections', [])
        sections = sorted(custom_sections, key=lambda s: s.get("order", 0))
        return render_template(
            "custom_sections.html",
            proposal=proposal,
            sections=sections
        )

    @app.route("/api/section/<proposal_id>", methods=["POST"])
    def add_section(proposal_id):
        """Add a new custom section to a proposal."""
        proposal = Proposal.load(proposal_id)
        if not proposal:
            return jsonify(ERROR_MESSAGES['PROPOSAL_NOT_FOUND']), 404

        data = request.get_json()
        custom_sections = getattr(proposal, 'custom_sections', [])
        new_section = {
            "id": str(uuid.uuid4()),
            "title": data.get("title", "New Section"),
            "content": data.get("content", ""),
            "order": len(custom_sections)
        }
        if not hasattr(proposal, 'custom_sections') or proposal.custom_sections is None:
            proposal.custom_sections = []
        proposal.custom_sections.append(new_section)
        proposal.save()
        return jsonify(new_section), 201

    @app.route("/api/section/<proposal_id>/<section_id>", methods=["PUT"])
    def update_section(proposal_id, section_id):
        """Update an existing custom section."""
        proposal = Proposal.load(proposal_id)
        if not proposal:
            return jsonify(ERROR_MESSAGES['PROPOSAL_NOT_FOUND']), 404

        data = request.get_json()
        sections = getattr(proposal, 'custom_sections', [])
        for section in sections:
            if section["id"] == section_id:
                section.update({
                    "title": data.get("title", section["title"]),
                    "content": data.get("content", section["content"]),
                    "order": data.get("order", section.get("order", 0))
                })
                break

        proposal.save()
        return jsonify({"ok": True})

    @app.route("/api/section/<proposal_id>/<section_id>", methods=["DELETE"])
    def delete_section(proposal_id, section_id):
        """Delete a custom section from a proposal."""
        proposal = Proposal.load(proposal_id)
        if not proposal:
            return jsonify(ERROR_MESSAGES['PROPOSAL_NOT_FOUND']), 404

        sections = getattr(proposal, 'custom_sections', [])
        proposal.custom_sections = [
            s for s in sections if s["id"] != section_id
        ]
        proposal.save()
        return jsonify({"ok": True})

    @app.route("/api/section/<proposal_id>/import-excel", methods=["POST"])
    def import_excel_section(proposal_id):
        """Import an Excel file as a markdown table custom section."""
        proposal = Proposal.load(proposal_id)
        if not proposal:
            return jsonify(ERROR_MESSAGES['PROPOSAL_NOT_FOUND']), 404

        if 'file' not in request.files:
            return jsonify({"error": "No file provided"}), 400

        file = request.files['file']
        if file.filename == '':
            return jsonify({"error": "No file selected"}), 400

        if not file.filename.endswith(('.xlsx', '.xls')):
            return jsonify({"error": "Only Excel files (.xlsx, .xls) are supported"}), 400

        try:
            import pandas as pd
            import io
        except ImportError:
            logger.error("Pandas/openpyxl not installed")
            return jsonify(ERROR_MESSAGES['EXCEL_NOT_INSTALLED']), 500

        try:
            # Read Excel file
            excel_data = file.read()
            excel_file = io.BytesIO(excel_data)
            df = pd.read_excel(excel_file, engine='openpyxl' if file.filename.endswith('.xlsx') else 'xlrd')

            # Convert DataFrame to markdown table
            markdown_content = df.to_markdown(index=False)

            # Create new section with Excel data
            custom_sections = getattr(proposal, 'custom_sections', [])
            new_section = {
                "id": str(uuid.uuid4()),
                "title": request.form.get('title', file.filename),
                "content": markdown_content,
                "order": len(custom_sections)
            }

            if not hasattr(proposal, 'custom_sections') or proposal.custom_sections is None:
                proposal.custom_sections = []
            proposal.custom_sections.append(new_section)
            proposal.save()

            logger.info(f"Imported Excel file as section: {new_section['id']}")
            return jsonify(new_section), 201

        except (ValueError, KeyError) as e:
            logger.warning(f"Invalid Excel file: {e}")
            return jsonify({**ERROR_MESSAGES['EXCEL_INVALID_FILE'], 'details': str(e)}), 400
        except Exception as e:
            logger.error(f"Excel import failed: {e}")
            return jsonify(ERROR_MESSAGES['EXCEL_PROCESSING_ERROR']), 500

    @app.route("/api/section/<proposal_id>/reorder", methods=["PUT"])
    def reorder_sections(proposal_id):
        """Reorder custom sections within a proposal."""
        proposal = Proposal.load(proposal_id)
        if not proposal:
            return jsonify(ERROR_MESSAGES['PROPOSAL_NOT_FOUND']), 404

        data = request.get_json()
        section_order = data.get('section_order', [])

        sections = getattr(proposal, 'custom_sections', [])
        section_dict = {s['id']: s for s in sections}

        reordered = []
        for idx, section_id in enumerate(section_order):
            if section_id in section_dict:
                section = section_dict[section_id]
                section['order'] = idx
                reordered.append(section)

        proposal.custom_sections = reordered
        proposal.save()
        return jsonify({"ok": True})

    @app.route("/preview/<proposal_id>")
    def preview_tab(proposal_id):
        """Render the proposal preview page."""
        proposal = Proposal.load(proposal_id)
        if not proposal:
            return jsonify(ERROR_MESSAGES['PROPOSAL_NOT_FOUND']), 404

        ctx = build_export_context(proposal)
        return render_template("preview.html", **ctx)

    @app.route("/api/map-image/<proposal_id>")
    def map_image(proposal_id):
        """Return a static PNG map image for print preview."""
        proposal = Proposal.load(proposal_id)
        if not proposal:
            return jsonify(ERROR_MESSAGES['PROPOSAL_NOT_FOUND']), 404
        map_png = build_map_export_image(proposal)
        if not map_png:
            return "", 204
        from flask import Response
        return Response(map_png, mimetype="image/png")

    @app.route("/api/task/<proposal_id>", methods=["POST"])
    def add_task(proposal_id):
        """Add a new task to a proposal."""
        proposal = Proposal.load(proposal_id)
        if not proposal:
            return jsonify(ERROR_MESSAGES['PROPOSAL_NOT_FOUND']), 404

        data = request.get_json()
        task = {
            "id": uuid.uuid4().hex[:8],
            "name": data.get("name", ""),
            "description": data.get("description", ""),
            "start_month": data.get("start_month", 1),
            "start_year": data.get("start_year", datetime.now().year),
            "duration_months": data.get("duration_months", 1),
            "lead_months": data.get("lead_months", 0),
            "lead_entity": data.get("lead_entity", ""),
            "recurring": data.get("recurring", False),
            "recurring_interval": data.get("recurring_interval", 3),
        }
        proposal.tasks.append(task)
        proposal.save()
        return jsonify(task), 201

    @app.route("/api/task/<proposal_id>/<task_id>", methods=["DELETE"])
    def delete_task(proposal_id, task_id):
        """Delete a task and its associated budget items."""
        proposal = Proposal.load(proposal_id)
        if not proposal:
            return jsonify(ERROR_MESSAGES['PROPOSAL_NOT_FOUND']), 404

        proposal.tasks = [t for t in proposal.tasks if t.get("id") != task_id]
        proposal.budget_items = [b for b in proposal.budget_items if b.get("task_id") != task_id]
        proposal.save()
        return "", 200

    @app.route("/api/budget/<proposal_id>", methods=["POST"])
    def add_budget_item(proposal_id):
        """Add a new budget item to a proposal."""
        proposal = Proposal.load(proposal_id)
        if not proposal:
            return jsonify(ERROR_MESSAGES['PROPOSAL_NOT_FOUND']), 404

        data = request.get_json()
        try:
            item = {
                "id": uuid.uuid4().hex[:8],
                "task_id": data.get("task_id", ""),
                "name": data.get("name", ""),
                "cost_per_unit": validate_numeric(data.get("cost_per_unit", 0), "cost_per_unit", min_val=0.0),
                "units": validate_numeric(data.get("units", 1), "units", min_val=0.0),
            }
        except ValueError as e:
            logger.warning(f"Invalid budget item data: {e}")
            return jsonify({**ERROR_MESSAGES['INVALID_NUMERIC'], 'details': str(e)}), 400

        if data.get("start_month") and data.get("start_year"):
            timings = proposal.budget_item_timings or {}
            timings[item["id"]] = {
                "start_month": int(data["start_month"]),
                "start_year": int(data["start_year"]),
                "duration_months": max(int(data.get("duration_months", 1) or 1), 1),
            }
            proposal.budget_item_timings = timings

        proposal.budget_items.append(item)
        proposal.save()
        logger.debug(f"Added budget item {item['id']} to proposal {proposal_id}")
        return jsonify(item), 201

    @app.route("/api/budget/<proposal_id>/<item_id>", methods=["DELETE"])
    def delete_budget_item(proposal_id, item_id):
        """Delete a budget item from a proposal."""
        proposal = Proposal.load(proposal_id)
        if not proposal:
            return jsonify(ERROR_MESSAGES['PROPOSAL_NOT_FOUND']), 404

        proposal.budget_items = [b for b in proposal.budget_items if b.get("id") != item_id]
        if proposal.budget_item_timings:
            proposal.budget_item_timings.pop(item_id, None)
        proposal.save()
        return jsonify({"ok": True})

    @app.route("/api/budget/<proposal_id>/<item_id>", methods=["PUT"])
    def update_budget_item(proposal_id: str, item_id: str) -> Tuple[Response, int]:
        """Update an existing budget item."""
        proposal = Proposal.load(proposal_id)
        if not proposal:
            return jsonify(ERROR_MESSAGES['PROPOSAL_NOT_FOUND']), 404

        data = request.get_json()
        if not data:
            return jsonify(ERROR_MESSAGES['NO_DATA']), 400

        for item in proposal.budget_items:
            if item.get("id") == item_id:
                try:
                    item["task_id"] = data.get("task_id", item.get("task_id", ""))
                    item["name"] = data.get("name", item.get("name", ""))
                    item["cost_per_unit"] = validate_numeric(
                        data.get("cost_per_unit", item.get("cost_per_unit", 0)),
                        "cost_per_unit",
                        min_val=0.0
                    )
                    item["units"] = validate_numeric(
                        data.get("units", item.get("units", 1)),
                        "units",
                        min_val=0.0
                    )
                except ValueError as e:
                    logger.warning(f"Invalid budget item data: {e}")
                    return jsonify({**ERROR_MESSAGES['INVALID_NUMERIC'], 'details': str(e)}), 400

                timings = proposal.budget_item_timings or {}
                sm, sy = data.get("start_month"), data.get("start_year")
                if sm and sy:
                    timings[item_id] = {
                        "start_month": int(sm),
                        "start_year": int(sy),
                        "duration_months": max(int(data.get("duration_months", 1) or 1), 1),
                    }
                elif sm == "" and sy == "":
                    timings.pop(item_id, None)
                proposal.budget_item_timings = timings
                break
        else:
            return jsonify(ERROR_MESSAGES['BUDGET_ITEM_NOT_FOUND']), 404

        proposal.save()
        logger.debug(f"Updated budget item {item_id} in proposal {proposal_id}")
        return jsonify({"ok": True}), 200

    @app.route("/api/proposal/<proposal_id>/import-budget", methods=["POST"])
    def import_budget(proposal_id):
        """Import budget items and tasks from an Excel file."""
        proposal = Proposal.load(proposal_id)
        if not proposal:
            return jsonify(ERROR_MESSAGES['PROPOSAL_NOT_FOUND']), 404

        if 'file' not in request.files:
            return jsonify({"error": "No file provided"}), 400

        file = request.files['file']
        if file.filename == '':
            return jsonify({"error": "No file selected"}), 400

        if not file.filename.endswith(('.xlsx', '.xls')):
            return jsonify({"error": "Only Excel files (.xlsx, .xls) are supported"}), 400

        try:
            from openpyxl import load_workbook
            import io
        except ImportError:
            return jsonify({"error": "openpyxl not installed"}), 500

        try:
            excel_data = file.read()
            wb = load_workbook(io.BytesIO(excel_data), read_only=True)
            ws = wb.active

            rows = list(ws.iter_rows(values_only=True))
            if len(rows) < 2:
                return jsonify({"error": "File has no data rows"}), 400

            header = [str(c).strip().lower() if c else "" for c in rows[0]]
            if not all(h in header for h in ["task", "item", "cost/unit", "units"]):
                return jsonify({"error": "Required columns: Task, Item, Cost/Unit, Units"}), 400

            task_idx = header.index("task")
            item_idx = header.index("item")
            cost_idx = header.index("cost/unit")
            units_idx = header.index("units")

            existing_tasks = {t["name"].strip().lower(): t for t in proposal.tasks}
            created_tasks = 0
            created_items = 0

            for row in rows[1:]:
                task_name = str(row[task_idx]).strip() if row[task_idx] else ""
                item_name = str(row[item_idx]).strip() if row[item_idx] else ""

                if not task_name or not item_name:
                    continue

                task_key = task_name.lower()
                if task_key not in existing_tasks:
                    new_task = {
                        "id": uuid.uuid4().hex[:8],
                        "name": task_name,
                        "description": "",
                        "start_month": 1,
                        "start_year": datetime.now().year,
                        "duration_months": 12,
                    }
                    proposal.tasks.append(new_task)
                    existing_tasks[task_key] = new_task
                    created_tasks += 1

                try:
                    cost = float(row[cost_idx]) if row[cost_idx] else 0
                    units = float(row[units_idx]) if row[units_idx] else 1
                except (ValueError, TypeError):
                    cost = 0
                    units = 1

                budget_item = {
                    "id": uuid.uuid4().hex[:8],
                    "task_id": existing_tasks[task_key]["id"],
                    "name": item_name,
                    "cost_per_unit": cost,
                    "units": units,
                }
                proposal.budget_items.append(budget_item)
                created_items += 1

            wb.close()
            proposal.save()

            return jsonify({
                "ok": True,
                "created_tasks": created_tasks,
                "created_items": created_items,
            }), 200

        except Exception as e:
            logger.error(f"Budget import failed: {e}")
            return jsonify({"error": f"Failed to process file: {str(e)}"}), 500

    @app.route("/api/budget-template", methods=["GET"])
    def download_budget_template():
        """Download a sample Excel budget template."""
        from openpyxl import Workbook
        import io

        wb = Workbook()
        ws = wb.active
        ws.title = "Budget Template"
        ws.append(["Task", "Item", "Cost/Unit", "Units"])
        ws.append(["Scoping", "Initial meeting", 200, 6])
        ws.append(["Scoping", "Data collection", 200, 20])
        ws.append(["Analysis", "Forest valuation", 300, 40])
        ws.append(["Results", "Report writing", 250, 30])

        buf = io.BytesIO()
        wb.save(buf)
        buf.seek(0)

        return Response(
            buf.getvalue(),
            mimetype="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
            headers={"Content-Disposition": "attachment; filename=budget_template.xlsx"}
        )

    @app.route("/tracker/<proposal_id>")
    def tracker(proposal_id):
        """Render the project tracker page with progress and milestones."""
        proposal = Proposal.load(proposal_id)
        if not proposal:
            return redirect(url_for("index"))

        indirect_percent = getattr(proposal, 'indirect_percent', 0) or 0
        indirect_amount = proposal.total_budget * (indirect_percent / 100)
        total_with_indirect = proposal.total_budget + indirect_amount

        task_budgets = {}
        timings = proposal.budget_item_timings or {}
        for task in proposal.tasks:
            items = [b for b in proposal.budget_items if b.get("task_id") == task["id"]]
            for item in items:
                t = timings.get(item.get("id", ""), {})
                if t:
                    item["actual_cost"] = t.get("actual_cost", 0)
            subtotal = sum(i.get("cost_per_unit", 0) * i.get("units", 0) for i in items)
            actual_total = sum(i.get("actual_cost", 0) for i in items)
            task_budgets[task["id"]] = {
                "task": task,
                "items": items,
                "subtotal": subtotal,
                "actual_total": actual_total,
            }

        milestones = getattr(proposal, 'milestones', []) or []
        reports = getattr(proposal, 'reports', []) or []

        completed_tasks = sum(1 for t in proposal.tasks if t.get("status") == "completed")
        total_tasks = len(proposal.tasks)
        overall_pct = round(completed_tasks / total_tasks * 100) if total_tasks else 0

        return render_template(
            "tracker.html",
            proposal=proposal,
            tasks=proposal.tasks,
            task_budgets=task_budgets,
            total_budget=proposal.total_budget,
            indirect_percent=indirect_percent,
            indirect_amount=indirect_amount,
            total_with_indirect=total_with_indirect,
            milestones=milestones,
            reports=reports,
            overall_pct=overall_pct,
        )

    @app.route("/api/tracker/<proposal_id>/task/<task_id>", methods=["PUT"])
    def update_tracker_task(proposal_id, task_id):
        """Update task progress, status, or notes in the tracker."""
        proposal = Proposal.load(proposal_id)
        if not proposal:
            return jsonify({"error": "Not found"}), 404

        data = request.get_json()
        for task in proposal.tasks:
            if task.get("id") == task_id:
                if "status" in data:
                    task["status"] = data["status"]
                if "progress_pct" in data:
                    task["progress_pct"] = int(data["progress_pct"])
                if "actual_start" in data:
                    task["actual_start"] = data["actual_start"]
                if "actual_end" in data:
                    task["actual_end"] = data["actual_end"]
                if "notes" in data:
                    task["notes"] = data["notes"]
                break
        else:
            return jsonify({"error": "Task not found"}), 404

        proposal.save()
        return jsonify({"ok": True})

    @app.route("/api/tracker/<proposal_id>/budget/<item_id>", methods=["PUT"])
    def update_tracker_budget(proposal_id, item_id):
        """Update the actual cost for a budget item in the tracker."""
        proposal = Proposal.load(proposal_id)
        if not proposal:
            return jsonify({"error": "Not found"}), 404

        data = request.get_json()
        timings = proposal.budget_item_timings or {}
        if item_id not in timings:
            timings[item_id] = {}
        if "actual_cost" in data:
            timings[item_id]["actual_cost"] = float(data["actual_cost"])
        proposal.budget_item_timings = timings
        proposal.save()
        return jsonify({"ok": True})

    @app.route("/api/tracker/<proposal_id>/milestone", methods=["POST"])
    def add_milestone(proposal_id):
        """Add a new milestone to the tracker."""
        proposal = Proposal.load(proposal_id)
        if not proposal:
            return jsonify({"error": "Not found"}), 404

        data = request.get_json()
        milestones = getattr(proposal, 'milestones', []) or []
        milestone = {
            "id": uuid.uuid4().hex[:8],
            "name": data.get("name", ""),
            "date": data.get("date", ""),
            "completed": False,
        }
        milestones.append(milestone)
        proposal.milestones = milestones
        proposal.save()
        return jsonify(milestone), 201

    @app.route("/api/tracker/<proposal_id>/milestone/<milestone_id>", methods=["PUT"])
    def update_milestone(proposal_id, milestone_id):
        """Update a milestone's name, date, or completion status."""
        proposal = Proposal.load(proposal_id)
        if not proposal:
            return jsonify({"error": "Not found"}), 404

        data = request.get_json()
        milestones = getattr(proposal, 'milestones', []) or []
        for m in milestones:
            if m["id"] == milestone_id:
                if "name" in data:
                    m["name"] = data["name"]
                if "date" in data:
                    m["date"] = data["date"]
                if "completed" in data:
                    m["completed"] = data["completed"]
                break
        else:
            return jsonify({"error": "Milestone not found"}), 404

        proposal.milestones = milestones
        proposal.save()
        return jsonify({"ok": True})

    @app.route("/api/tracker/<proposal_id>/milestone/<milestone_id>", methods=["DELETE"])
    def delete_milestone(proposal_id, milestone_id):
        """Delete a milestone from the tracker."""
        proposal = Proposal.load(proposal_id)
        if not proposal:
            return jsonify({"error": "Not found"}), 404

        milestones = getattr(proposal, 'milestones', []) or []
        proposal.milestones = [m for m in milestones if m["id"] != milestone_id]
        proposal.save()
        return jsonify({"ok": True})

    @app.route("/api/tracker/<proposal_id>/report", methods=["POST"])
    def add_report(proposal_id):
        """Add a new progress report to the tracker."""
        proposal = Proposal.load(proposal_id)
        if not proposal:
            return jsonify({"error": "Not found"}), 404

        data = request.get_json()
        reports = getattr(proposal, 'reports', []) or []
        report = {
            "id": uuid.uuid4().hex[:8],
            "date": data.get("date", datetime.now().strftime("%Y-%m-%d")),
            "title": data.get("title", ""),
            "content": data.get("content", ""),
        }
        reports.append(report)
        proposal.reports = reports
        proposal.save()
        return jsonify(report), 201

    @app.route("/api/tracker/<proposal_id>/report/<report_id>", methods=["PUT"])
    def update_report(proposal_id, report_id):
        """Update a progress report's content, title, or date."""
        proposal = Proposal.load(proposal_id)
        if not proposal:
            return jsonify({"error": "Not found"}), 404

        data = request.get_json()
        reports = getattr(proposal, 'reports', []) or []
        for r in reports:
            if r["id"] == report_id:
                if "title" in data:
                    r["title"] = data["title"]
                if "date" in data:
                    r["date"] = data["date"]
                if "content" in data:
                    r["content"] = data["content"]
                break
        else:
            return jsonify({"error": "Report not found"}), 404

        proposal.reports = reports
        proposal.save()
        return jsonify({"ok": True})

    @app.route("/api/tracker/<proposal_id>/report/<report_id>", methods=["DELETE"])
    def delete_report(proposal_id, report_id):
        """Delete a progress report from the tracker."""
        proposal = Proposal.load(proposal_id)
        if not proposal:
            return jsonify({"error": "Not found"}), 404

        reports = getattr(proposal, 'reports', []) or []
        proposal.reports = [r for r in reports if r["id"] != report_id]
        proposal.save()
        return jsonify({"ok": True})

    return app

markdown_to_html(text)

Convert Markdown text to HTML using the markdown library.

Parameters:

Name	Type	Description	Default
text	str	Markdown formatted text	required

Returns:

Type	Description
str	HTML formatted string

Source code in app/main.py

def markdown_to_html(text: str) -> str:
    """Convert Markdown text to HTML using the markdown library.

    Args:
        text: Markdown formatted text

    Returns:
        HTML formatted string
    """
    if not text:
        return ""
    return markdown.markdown(
        text,
        extensions=['tables', 'nl2br', 'fenced_code', 'sane_lists']
    )

run_server()

Run the development server.

Source code in app/main.py

def run_server() -> None:
    """Run the development server."""
    import logging as _logging
    _logging.getLogger("werkzeug").setLevel(_logging.WARNING)
    app = create_app()
    print("\n ✨✨✨ 🔌 Server started ✨✨✨")
    print(f"  👉  http://localhost:{Config.PORT}  👈\n")
    app.run(debug=Config.DEBUG, host=Config.HOST, port=Config.PORT)

validate_numeric(value, name, min_val=0.0)

Validate and convert a numeric value.

Parameters:

Name	Type	Description	Default
value	Any	Value to validate	required
name	str	Name of the field (for error messages)	required
min_val	float	Minimum allowed value	0.0

Returns:

Type	Description
float	Validated float value

Raises:

Type	Description
ValueError	If value is invalid

Source code in app/main.py

def validate_numeric(value: Any, name: str, min_val: float = 0.0) -> float:
    """Validate and convert a numeric value.

    Args:
        value: Value to validate
        name: Name of the field (for error messages)
        min_val: Minimum allowed value

    Returns:
        Validated float value

    Raises:
        ValueError: If value is invalid
    """
    try:
        num = float(value)
        if not math.isfinite(num):
            raise ValueError(f"{name} must be a finite number")
        if num < min_val:
            raise ValueError(f"{name} must be >= {min_val}")
        return num
    except (ValueError, TypeError) as e:
        raise ValueError(f"Invalid {name}: {str(e)}")

Data models for Propongo.

Tasks and budget items are stored as dictionaries with the following structure:

Task dict

{ 'id': str, # UUID 'name': str, # Task name 'description': str, # Task description 'lead_entity': str, # Organization responsible 'start_month': int, # Start month (1-12) 'start_year': int, # Start year 'duration_months': int, # Duration in months }

BudgetItem dict

{ 'id': str, # UUID 'task_id': str, # Associated task UUID 'name': str, # Item name 'cost_per_unit': float, # Unit cost 'units': float, # Number of units }

Custom Section dict

{ 'id': str, # UUID 'title': str, # Section title 'content': str, # Markdown content 'order': int, # Display order }

map_config dict (Map tab): which context layer the embedded GeoLibre map opens with. One of: {'mode': 'basemap'} # bare map, pick basemap in-app {'mode': 'data_url', 'url': ''} # GeoJSON/COG/PMTiles/etc.

Proposal dataclass

A project proposal containing tasks, budget items, and custom sections.

Source code in app/models.py

@dataclass
class Proposal:
    """A project proposal containing tasks, budget items, and custom sections."""
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    title: str = "Untitled Proposal"
    client_name: str = ""
    subtitle: str = ""
    created_at: str = field(default_factory=lambda: datetime.now().isoformat())
    updated_at: str = field(default_factory=lambda: datetime.now().isoformat())

    project_summary: str = ""
    scope: str = ""
    tasks: list = field(default_factory=list)
    qualifications: str = ""

    budget_items: list = field(default_factory=list)
    budget_item_timings: dict = field(default_factory=dict)
    start_date: str = field(default_factory=lambda: datetime.now().strftime("%Y-%m-%d"))
    end_date: str = ""
    indirect_percent: float = 0.0
    show_budget_description: bool = False
    budget_description: str = ""
    timeline_use_days: bool = False
    timeline_show_budget: bool = False
    custom_sections: list = field(default_factory=list)
    map_config: dict = field(default_factory=dict)

    is_template: bool = False
    template_name: str = ""
    template_category: str = ""

    milestones: list = field(default_factory=list)
    reports: list = field(default_factory=list)

    def to_dict(self) -> dict:
        """Serialize the proposal to a dictionary."""
        return asdict(self)

    @classmethod
    def from_dict(cls, data: dict) -> "Proposal":
        """Create a Proposal from a dictionary, ignoring unknown fields."""
        return cls(**{k: v for k, v in data.items() if k in cls.__dataclass_fields__})

    def save(self):
        """Save the proposal to disk as a JSON file."""
        self.updated_at = datetime.now().isoformat()
        target_dir = _scoped_dir(TEMPLATES_DIR if self.is_template else PROPOSALS_DIR)
        filepath = os.path.join(target_dir, f"{self.id}.json")
        tmp = filepath + ".tmp"
        with open(tmp, "w") as f:
            json.dump(self.to_dict(), f, indent=2)
            f.write("\n")
        os.replace(tmp, filepath)

    @classmethod
    def load(cls, proposal_id: str, is_template: bool = False) -> Optional["Proposal"]:
        """Load a proposal or template by ID from disk."""
        target_dir = _scoped_dir(TEMPLATES_DIR if is_template else PROPOSALS_DIR)
        filepath = os.path.join(target_dir, f"{proposal_id}.json")
        if not os.path.exists(filepath):
            return None
        try:
            with open(filepath, "r") as f:
                return cls.from_dict(json.load(f))
        except (json.JSONDecodeError, OSError):
            return None

    @classmethod
    def list_all(cls) -> list:
        """List all proposals, returning summaries sorted by most recent."""
        target_dir = _scoped_dir(PROPOSALS_DIR)
        _seed_demo_proposals(target_dir)
        proposals = []
        for filename in sorted(os.listdir(target_dir)):
            if filename.endswith(".json"):
                try:
                    with open(os.path.join(target_dir, filename), "r") as f:
                        data = json.load(f)
                        proposals.append({
                            "id": data.get("id", filename.replace(".json", "")),
                            "title": data.get("title", "Untitled"),
                            "client_name": data.get("client_name", ""),
                            "subtitle": data.get("subtitle", ""),
                            "updated_at": data.get("updated_at", ""),
                        })
                except (json.JSONDecodeError, OSError):
                    continue
        return sorted(proposals, key=lambda x: x["updated_at"], reverse=True)

    @classmethod
    def list_templates(cls) -> list:
        """List all templates, returning summaries sorted by most recent."""
        target_dir = _scoped_dir(TEMPLATES_DIR)
        templates = []
        for filename in sorted(os.listdir(target_dir)):
            if filename.endswith(".json"):
                try:
                    with open(os.path.join(target_dir, filename), "r") as f:
                        data = json.load(f)
                        templates.append({
                            "id": data.get("id", filename.replace(".json", "")),
                            "title": data.get("title", "Untitled"),
                            "template_name": data.get("template_name", ""),
                            "template_category": data.get("template_category", ""),
                            "updated_at": data.get("updated_at", ""),
                        })
                except (json.JSONDecodeError, OSError):
                    continue
        return sorted(templates, key=lambda x: x["updated_at"], reverse=True)

    @classmethod
    def delete(cls, proposal_id: str, is_template: bool = False) -> bool:
        """Delete a proposal or template by ID. Returns True if deleted."""
        target_dir = _scoped_dir(TEMPLATES_DIR if is_template else PROPOSALS_DIR)
        filepath = os.path.join(target_dir, f"{proposal_id}.json")
        if os.path.exists(filepath):
            os.remove(filepath)
            return True
        return False

    @property
    def total_budget(self) -> float:
        """Compute total budget as the sum of cost_per_unit * units for all items."""
        return sum(
            item.get("cost_per_unit", 0) * item.get("units", 0)
            for item in self.budget_items
        )

total_budget property

Compute total budget as the sum of cost_per_unit * units for all items.

delete(proposal_id, is_template=False) classmethod

Delete a proposal or template by ID. Returns True if deleted.

Source code in app/models.py

@classmethod
def delete(cls, proposal_id: str, is_template: bool = False) -> bool:
    """Delete a proposal or template by ID. Returns True if deleted."""
    target_dir = _scoped_dir(TEMPLATES_DIR if is_template else PROPOSALS_DIR)
    filepath = os.path.join(target_dir, f"{proposal_id}.json")
    if os.path.exists(filepath):
        os.remove(filepath)
        return True
    return False

from_dict(data) classmethod

Create a Proposal from a dictionary, ignoring unknown fields.

Source code in app/models.py

@classmethod
def from_dict(cls, data: dict) -> "Proposal":
    """Create a Proposal from a dictionary, ignoring unknown fields."""
    return cls(**{k: v for k, v in data.items() if k in cls.__dataclass_fields__})

list_all() classmethod

List all proposals, returning summaries sorted by most recent.

Source code in app/models.py

@classmethod
def list_all(cls) -> list:
    """List all proposals, returning summaries sorted by most recent."""
    target_dir = _scoped_dir(PROPOSALS_DIR)
    _seed_demo_proposals(target_dir)
    proposals = []
    for filename in sorted(os.listdir(target_dir)):
        if filename.endswith(".json"):
            try:
                with open(os.path.join(target_dir, filename), "r") as f:
                    data = json.load(f)
                    proposals.append({
                        "id": data.get("id", filename.replace(".json", "")),
                        "title": data.get("title", "Untitled"),
                        "client_name": data.get("client_name", ""),
                        "subtitle": data.get("subtitle", ""),
                        "updated_at": data.get("updated_at", ""),
                    })
            except (json.JSONDecodeError, OSError):
                continue
    return sorted(proposals, key=lambda x: x["updated_at"], reverse=True)

list_templates() classmethod

List all templates, returning summaries sorted by most recent.

Source code in app/models.py

@classmethod
def list_templates(cls) -> list:
    """List all templates, returning summaries sorted by most recent."""
    target_dir = _scoped_dir(TEMPLATES_DIR)
    templates = []
    for filename in sorted(os.listdir(target_dir)):
        if filename.endswith(".json"):
            try:
                with open(os.path.join(target_dir, filename), "r") as f:
                    data = json.load(f)
                    templates.append({
                        "id": data.get("id", filename.replace(".json", "")),
                        "title": data.get("title", "Untitled"),
                        "template_name": data.get("template_name", ""),
                        "template_category": data.get("template_category", ""),
                        "updated_at": data.get("updated_at", ""),
                    })
            except (json.JSONDecodeError, OSError):
                continue
    return sorted(templates, key=lambda x: x["updated_at"], reverse=True)

load(proposal_id, is_template=False) classmethod

Load a proposal or template by ID from disk.

Source code in app/models.py

@classmethod
def load(cls, proposal_id: str, is_template: bool = False) -> Optional["Proposal"]:
    """Load a proposal or template by ID from disk."""
    target_dir = _scoped_dir(TEMPLATES_DIR if is_template else PROPOSALS_DIR)
    filepath = os.path.join(target_dir, f"{proposal_id}.json")
    if not os.path.exists(filepath):
        return None
    try:
        with open(filepath, "r") as f:
            return cls.from_dict(json.load(f))
    except (json.JSONDecodeError, OSError):
        return None

save()

Save the proposal to disk as a JSON file.

Source code in app/models.py

def save(self):
    """Save the proposal to disk as a JSON file."""
    self.updated_at = datetime.now().isoformat()
    target_dir = _scoped_dir(TEMPLATES_DIR if self.is_template else PROPOSALS_DIR)
    filepath = os.path.join(target_dir, f"{self.id}.json")
    tmp = filepath + ".tmp"
    with open(tmp, "w") as f:
        json.dump(self.to_dict(), f, indent=2)
        f.write("\n")
    os.replace(tmp, filepath)

to_dict()

Serialize the proposal to a dictionary.

Source code in app/models.py

def to_dict(self) -> dict:
    """Serialize the proposal to a dictionary."""
    return asdict(self)

ensure_dirs()

Ensure data directories exist.

Source code in app/models.py

def ensure_dirs() -> None:
    """Ensure data directories exist."""
    os.makedirs(PROPOSALS_DIR, exist_ok=True)
    os.makedirs(TEMPLATES_DIR, exist_ok=True)
    os.makedirs(MAPS_DIR, exist_ok=True)

Export functionality for HTML, PDF, and Markdown generation.

PDF exports are produced by printing the export HTML document with WeasyPrint. WeasyPrint is lightweight and self-contained (no headless browser), so PDF export stays within memory limits on constrained instances. Because it does not execute JavaScript, the live GeoLibre map iframe cannot be printed — the export template shows the uploaded static map image instead, and omits the map otherwise.

export_html(proposal_id)

Export proposal as HTML file.

Uses the live GeoLibre iframe (same as preview) — no static image needed since HTML renders in a browser.

Source code in app/export.py

@export_bp.route("/export/html/<proposal_id>")
def export_html(proposal_id: str) -> Tuple[Response, int] | Response:
    """Export proposal as HTML file.

    Uses the live GeoLibre iframe (same as preview) — no static image
    needed since HTML renders in a browser.
    """
    proposal = Proposal.load(proposal_id)
    if not proposal:
        logger.warning(f"Proposal not found for HTML export: {proposal_id}")
        return jsonify(ERROR_MESSAGES['PROPOSAL_NOT_FOUND']), 404

    ctx = build_export_context(proposal)
    ctx["map_static_data_uri"] = None

    html_content = render_template("export_proposal.html", **ctx)

    return Response(
        html_content,
        mimetype="text/html",
        headers={"Content-Disposition": "inline"},
    )

export_markdown(proposal_id)

Export proposal as a Markdown document.

Replaces the former DOCX export as the editable, plain-text format. The same HTML export template is rendered, then converted to Markdown.

Source code in app/export.py

@export_bp.route("/export/markdown/<proposal_id>")
def export_markdown(proposal_id: str) -> Tuple[Response, int] | Response:
    """Export proposal as a Markdown document.

    Replaces the former DOCX export as the editable, plain-text format. The
    same HTML export template is rendered, then converted to Markdown.
    """
    proposal = Proposal.load(proposal_id)
    if not proposal:
        logger.warning(f"Proposal not found for Markdown export: {proposal_id}")
        return jsonify(ERROR_MESSAGES['PROPOSAL_NOT_FOUND']), 404

    ctx = build_export_context(proposal)
    ctx["map_static_data_uri"] = None

    html_content = render_template("export_proposal.html", **ctx)
    html_content = _strip_timeline_chart(html_content)
    md_content = _html_to_markdown(html_content)

    timeline_table = _timeline_markdown_table(proposal, ctx)
    if timeline_table:
        md_content = md_content.replace(
            "## Timeline\n", f"## Timeline\n\n{timeline_table}\n"
        )

    return Response(
        md_content,
        mimetype="text/markdown",
        headers={"Content-Disposition": f"attachment; filename={proposal.title or 'proposal'}.md"},
    )

export_pdf(proposal_id)

Export proposal as PDF by printing the HTML export with WeasyPrint.

The map is included only when a static image is available (uploaded PNG/JPG or a remote image_url); the live GeoLibre iframe cannot be rendered by WeasyPrint and is omitted with a note.

Source code in app/export.py

@export_bp.route("/export/pdf/<proposal_id>")
def export_pdf(proposal_id: str) -> Tuple[Response, int] | Response:
    """Export proposal as PDF by printing the HTML export with WeasyPrint.

    The map is included only when a static image is available (uploaded PNG/JPG
    or a remote ``image_url``); the live GeoLibre iframe cannot be rendered by
    WeasyPrint and is omitted with a note.
    """
    proposal = Proposal.load(proposal_id)
    if not proposal:
        logger.warning(f"Proposal not found for PDF export: {proposal_id}")
        return jsonify(ERROR_MESSAGES['PROPOSAL_NOT_FOUND']), 404

    ctx = build_export_context(proposal)
    ctx["map_static_data_uri"] = _local_map_data_uri(proposal)
    ctx["for_pdf"] = True
    html_content = render_template("export_proposal.html", **ctx)

    try:
        pdf_bytes = _render_pdf(html_content, base_url=request.host_url)
    except Exception as e:
        logger.error(f"PDF export failed: {e}")
        return jsonify({"error": f"PDF export failed: {str(e)}"}), 500

    return send_file(
        BytesIO(pdf_bytes),
        mimetype="application/pdf",
        as_attachment=True,
        download_name=f"{proposal.title or 'proposal'}.pdf",
    )

export_timeline_png(proposal_id)

Export just the timeline chart as a PNG image.

Renders the chart with WeasyPrint onto a content-sized page, then rasterizes it with pypdfium2 so the image is a tight crop rather than a full page.

Source code in app/export.py

@export_bp.route("/export/timeline/png/<proposal_id>")
def export_timeline_png(proposal_id: str) -> Tuple[Response, int] | Response:
    """Export just the timeline chart as a PNG image.

    Renders the chart with WeasyPrint onto a content-sized page, then
    rasterizes it with pypdfium2 so the image is a tight crop rather than a
    full page.
    """
    try:
        from weasyprint import HTML
    except (OSError, ImportError):
        return jsonify({"error": GTK3_MISSING_MSG}), 500

    try:
        import pypdfium2 as pdfium
    except ImportError:
        return jsonify({"error": "pypdfium2 not installed"}), 500

    proposal = Proposal.load(proposal_id)
    if not proposal:
        logger.warning(f"Proposal not found for timeline PNG export: {proposal_id}")
        return jsonify(ERROR_MESSAGES['PROPOSAL_NOT_FOUND']), 404

    ctx = build_export_context(proposal)
    html_content = render_template("timeline_export.html", **ctx)

    os.makedirs(Config.EXPORTS_DIR, exist_ok=True)
    pdf_buf = BytesIO()
    HTML(string=html_content, base_url=request.host_url).write_pdf(pdf_buf)
    pdf_buf.seek(0)

    pdf = pdfium.PdfDocument(pdf_buf)
    page = pdf[0]
    bitmap = page.render(scale=2)
    pil = bitmap.to_pil()
    png_path = os.path.join(Config.EXPORTS_DIR, f"timeline_{proposal_id}.png")
    pil.save(png_path, "PNG")

    return send_file(
        png_path,
        mimetype="image/png",
        as_attachment=True,
        download_name=f"{proposal.title or 'proposal'}_timeline.png",
    )

export_tracker_html(proposal_id)

Export tracker as HTML.

Source code in app/export.py

@export_bp.route("/export/tracker/html/<proposal_id>")
def export_tracker_html(proposal_id: str) -> Tuple[Response, int] | Response:
    """Export tracker as HTML."""
    proposal = Proposal.load(proposal_id)
    if not proposal:
        return jsonify(ERROR_MESSAGES['PROPOSAL_NOT_FOUND']), 404

    ctx = build_tracker_export_context(proposal)
    html_content = render_template("export_tracker.html", **ctx)

    return Response(
        html_content,
        mimetype="text/html",
        headers={"Content-Disposition": "inline"},
    )

export_tracker_markdown(proposal_id)

Export tracker as a Markdown document.

Source code in app/export.py

@export_bp.route("/export/tracker/markdown/<proposal_id>")
def export_tracker_markdown(proposal_id: str) -> Tuple[Response, int] | Response:
    """Export tracker as a Markdown document."""
    proposal = Proposal.load(proposal_id)
    if not proposal:
        return jsonify(ERROR_MESSAGES['PROPOSAL_NOT_FOUND']), 404

    ctx = build_tracker_export_context(proposal)
    md_content = _html_to_markdown(render_template("export_tracker.html", **ctx))

    return Response(
        md_content,
        mimetype="text/markdown",
        headers={"Content-Disposition": f"attachment; filename={proposal.title or 'project'}_tracker.md"},
    )

export_tracker_pdf(proposal_id)

Export tracker as PDF.

Source code in app/export.py

@export_bp.route("/export/tracker/pdf/<proposal_id>")
def export_tracker_pdf(proposal_id: str) -> Tuple[Response, int] | Response:
    """Export tracker as PDF."""
    proposal = Proposal.load(proposal_id)
    if not proposal:
        return jsonify(ERROR_MESSAGES['PROPOSAL_NOT_FOUND']), 404

    ctx = build_tracker_export_context(proposal)
    html_content = render_template("export_tracker.html", **ctx)

    try:
        pdf_bytes = _render_pdf(html_content, base_url=request.host_url)
    except Exception as e:
        logger.error(f"Tracker PDF export failed: {e}")
        return jsonify({"error": f"PDF export failed: {str(e)}"}), 500

    return send_file(
        BytesIO(pdf_bytes),
        mimetype="application/pdf",
        as_attachment=True,
        download_name=f"{proposal.title or 'project'}_tracker.pdf",
    )

Snippet management for reusable text blocks.

add_snippet(category)

Add a new snippet to the given category.

Stock categories (organization, deliverables) append to their list files; anything else — including free-form labels sent in the body's category field — is stored under custom/ so it stays user-editable.

Source code in app/snippets.py

@snippets_bp.route("/snippets/<category>", methods=["POST"])
def add_snippet(category):
    """Add a new snippet to the given category.

    Stock categories (`organization`, `deliverables`) append to their list
    files; anything else — including free-form labels sent in the body's
    `category` field — is stored under custom/ so it stays user-editable.
    """
    data = request.get_json()
    if not data or "title" not in data or "content" not in data:
        return jsonify({"error": "title and content required"}), 400

    if category in _STOCK_SOURCES:
        label = category
    else:
        label = str(data.get("category") or "").strip() or category

    snippet = {
        "id": data.get("id", uuid.uuid4().hex[:8]),
        "title": data["title"],
        "content": data["content"],
        "category": label,
    }

    if category in _STOCK_SOURCES:
        snippets = load_snippets(f"{category}.json")
        snippets.append(snippet)
        save_snippets(f"{category}.json", snippets)
    else:
        ensure_dirs()
        filepath = os.path.join(_custom_dir(), f"{snippet['id']}.json")
        with open(filepath, "w") as f:
            json.dump(snippet, f, indent=2)

    return jsonify(snippet), 201

delete_snippet(category, snippet_id)

Delete a snippet by category and ID.

Source code in app/snippets.py

@snippets_bp.route("/snippets/<category>/<snippet_id>", methods=["DELETE"])
def delete_snippet(category, snippet_id):
    """Delete a snippet by category and ID."""
    if category == "custom":
        filepath = os.path.join(_custom_dir(), f"{snippet_id}.json")
        if os.path.exists(filepath):
            os.remove(filepath)
            return jsonify({"ok": True})
    elif category in ("organization", "deliverables"):
        snippets = load_snippets(f"{category}.json")
        snippets = [s for s in snippets if s.get("id") != snippet_id]
        save_snippets(f"{category}.json", snippets)
        return jsonify({"ok": True})

    return jsonify(ERROR_MESSAGES['SECTION_NOT_FOUND']), 404

ensure_dirs()

Ensure per-user snippet directories exist.

Source code in app/snippets.py

def ensure_dirs():
    """Ensure per-user snippet directories exist."""
    root = _snippets_root()
    os.makedirs(root, exist_ok=True)
    os.makedirs(_custom_dir(), exist_ok=True)

get_all_snippets()

Return all snippets grouped by category, tagged with their source.

Source code in app/snippets.py

@snippets_bp.route("/snippets")
def get_all_snippets():
    """Return all snippets grouped by category, tagged with their source."""
    return jsonify({
        "organization": _with_source(load_snippets("organization.json"), "organization"),
        "deliverables": _with_source(load_snippets("deliverables.json"), "deliverables"),
        "custom": _with_source(load_custom_snippets(), "custom"),
    })

import_snippet()

Import a snippet from a .md, .txt, or .docx file.

Source code in app/snippets.py

@snippets_bp.route("/snippets/import", methods=["POST"])
def import_snippet():
    """Import a snippet from a .md, .txt, or .docx file."""
    if "file" not in request.files:
        return jsonify(ERROR_MESSAGES['NO_FILE']), 400

    file = request.files["file"]
    if not file.filename:
        return jsonify(ERROR_MESSAGES['NO_FILE']), 400

    filename = file.filename.lower()
    title = request.form.get("title", "").strip()
    if not title:
        title = os.path.splitext(file.filename)[0]

    try:
        if filename.endswith(".md") or filename.endswith(".markdown") or filename.endswith(".txt"):
            content = file.read().decode("utf-8")
        elif filename.endswith(".docx"):
            from docx import Document
            doc = Document(file)
            content = "\n\n".join(p.text for p in doc.paragraphs if p.text.strip())
        else:
            return jsonify({"error": "Unsupported file type. Use .md, .txt, or .docx"}), 400
    except Exception as e:
        return jsonify({"error": f"Failed to read file: {str(e)}"}), 400

    if not content.strip():
        return jsonify({"error": "File is empty"}), 400

    snippet = {
        "id": uuid.uuid4().hex[:8],
        "title": title,
        "content": content,
        "category": "custom",
    }

    ensure_dirs()
    filepath = os.path.join(_custom_dir(), f"{snippet['id']}.json")
    with open(filepath, "w") as f:
        json.dump(snippet, f, indent=2)

    return jsonify(snippet), 201

load_custom_snippets()

Load all custom snippets for the current user.

Source code in app/snippets.py

def load_custom_snippets():
    """Load all custom snippets for the current user."""
    ensure_dirs()
    snippets = []
    for filename in sorted(os.listdir(_custom_dir())):
        if filename.endswith(".json"):
            with open(os.path.join(_custom_dir(), filename), "r") as f:
                snippets.append(json.load(f))
    return snippets

load_snippets(filename)

Load snippets from the current user's data directory.

Source code in app/snippets.py

def load_snippets(filename):
    """Load snippets from the current user's data directory."""
    ensure_dirs()
    filepath = os.path.join(_snippets_root(), filename)
    if os.path.exists(filepath):
        with open(filepath, "r") as f:
            return json.load(f)
    return []

save_snippets(filename, data)

Save snippets to the current user's data directory.

Source code in app/snippets.py

def save_snippets(filename, data):
    """Save snippets to the current user's data directory."""
    ensure_dirs()
    filepath = os.path.join(_snippets_root(), filename)
    tmp = filepath + ".tmp"
    with open(tmp, "w") as f:
        json.dump(data, f, indent=2)
    os.replace(tmp, filepath)

update_snippet(category, snippet_id)

Update a snippet's title, content, and/or category.

Only the display fields change; the snippet stays in its current storage location (list file or custom directory), which is identified by the URL.

Source code in app/snippets.py

@snippets_bp.route("/snippets/<category>/<snippet_id>", methods=["PUT"])
def update_snippet(category, snippet_id):
    """Update a snippet's title, content, and/or category.

    Only the display fields change; the snippet stays in its current storage
    location (list file or custom directory), which is identified by the URL.
    """
    data = request.get_json()
    if not data:
        return jsonify({"error": "no fields to update"}), 400

    updates = {}
    for field in ("title", "content", "category"):
        if field in data and str(data[field]).strip():
            updates[field] = str(data[field])
    if not updates:
        return jsonify({"error": "no fields to update"}), 400

    if category == "custom":
        filepath = os.path.join(_custom_dir(), f"{snippet_id}.json")
        if not os.path.exists(filepath):
            return jsonify(ERROR_MESSAGES['SECTION_NOT_FOUND']), 404
        with open(filepath, "r") as f:
            snippet = json.load(f)
        snippet.update(updates)
        tmp = filepath + ".tmp"
        with open(tmp, "w") as f:
            json.dump(snippet, f, indent=2)
        os.replace(tmp, filepath)
        return jsonify(snippet)

    if category in _STOCK_SOURCES:
        snippets = load_snippets(f"{category}.json")
        for i, s in enumerate(snippets):
            if s.get("id") == snippet_id:
                updated = dict(s)
                updated.update(updates)
                snippets[i] = updated
                save_snippets(f"{category}.json", snippets)
                return jsonify(updated), 200

    return jsonify(ERROR_MESSAGES['SECTION_NOT_FOUND']), 404

Configuration settings for Propongo.

Config

Application configuration.

Source code in app/config.py

class Config:
    """Application configuration."""

    # Server settings
    HOST = os.environ.get('HOST', '0.0.0.0')
    PORT = int(os.environ.get('PORT', 5000))
    DEBUG = os.environ.get('DEBUG', 'True').lower() in ('true', '1', 'yes')

    # Security
    SECRET_KEY = os.environ.get('FLASK_SECRET_KEY', os.urandom(24).hex())

    # File paths
    DATA_DIR = DATA_ROOT
    PROPOSALS_DIR = os.path.join(DATA_DIR, 'proposals')
    EXPORTS_DIR = os.path.join(DATA_DIR, 'exports')
    MAPS_DIR = os.path.join(DATA_DIR, 'maps')

    # GeoLibre embed base URL. Points at the hosted app by default; set to a
    # self-hosted instance (e.g. http://localhost:8080) when running one.
    GEOLIBRE_EMBED_URL = os.environ.get('GEOLIBRE_EMBED_URL', 'https://web.geolibre.app')

    # Logging
    LOG_LEVEL = os.environ.get('LOG_LEVEL', 'INFO')

    # Outbound email (SMTP) for password reset
    SMTP_HOST = os.environ.get('SMTP_HOST', '')
    SMTP_PORT = int(os.environ.get('SMTP_PORT', '587'))
    SMTP_USER = os.environ.get('SMTP_USER', '')
    SMTP_PASS = os.environ.get('SMTP_PASS', '')
    SMTP_FROM = os.environ.get('SMTP_FROM', '') or os.environ.get('SMTP_USER', '')

Utility functions for Propongo.

build_budget_by_year(proposal)

Allocate budget items across calendar years based on their spend dates.

Each item's cost is spread evenly over the months of its user-defined spend window (budget_item_timings). Items with no dates set are reported as unscheduled so totals still reconcile.

Source code in app/utils.py

def build_budget_by_year(proposal) -> Dict[str, Any]:
    """Allocate budget items across calendar years based on their spend dates.

    Each item's cost is spread evenly over the months of its user-defined
    spend window (``budget_item_timings``). Items with no dates set are
    reported as unscheduled so totals still reconcile.
    """
    timings = getattr(proposal, "budget_item_timings", None) or {}
    by_year = {}
    unscheduled = []

    for item in proposal.budget_items:
        total = float(item.get("cost_per_unit", 0) * item.get("units", 0))
        if total <= 0:
            continue
        timing = timings.get(item.get("id", ""), {})
        sm = timing.get("start_month")
        sy = timing.get("start_year")
        dur = timing.get("duration_months")
        if not (sm and sy and dur):
            unscheduled.append({"name": item.get("name", ""), "amount": total})
            continue

        start = int(sy) * 12 + (int(sm) - 1)
        dur = max(int(dur), 1)
        monthly = total / dur
        for m in range(dur):
            y = (start + m) // 12
            by_year[y] = by_year.get(y, 0.0) + monthly

    years = [{"year": y, "amount": round(by_year[y], 2)} for y in sorted(by_year)]
    return {
        "years": years,
        "unscheduled": unscheduled,
        "total_scheduled": round(sum(r["amount"] for r in years), 2),
        "total_unscheduled": round(sum(u["amount"] for u in unscheduled), 2),
    }

build_export_context(proposal)

Build context dictionary for export and preview templates.

Parameters:

Name	Type	Description	Default
proposal		Proposal object	required

Returns:

Name	Type	Description
dict	Dict[str, Any]	Context dictionary with all necessary template variables

Source code in app/utils.py

def build_export_context(proposal) -> Dict[str, Any]:
    """Build context dictionary for export and preview templates.

    Args:
        proposal: Proposal object

    Returns:
        dict: Context dictionary with all necessary template variables
    """
    indirect_percent = getattr(proposal, 'indirect_percent', 0) or 0
    indirect_amount = proposal.total_budget * (indirect_percent / 100)
    total_with_indirect = proposal.total_budget + indirect_amount

    tasks_with_timing = []
    for t in proposal.tasks:
        tasks_with_timing.append({
            "id": t.get("id", ""),
            "name": t.get("name", ""),
            "description": t.get("description", ""),
            "lead_entity": t.get("lead_entity", ""),
            "start_month": t.get("start_month"),
            "start_year": t.get("start_year"),
            "duration_months": t.get("duration_months", 1),
            "recurring": t.get("recurring", False),
            "recurring_interval": t.get("recurring_interval", 3),
        })

    budget_with_timing = []
    timings = proposal.budget_item_timings or {}
    for item in proposal.budget_items:
        item_id = item.get("id", "")
        timing = timings.get(item_id, {})
        budget_with_timing.append({
            **item,
            "start_month": timing.get("start_month"),
            "start_year": timing.get("start_year"),
            "duration_months": timing.get("duration_months", 1),
            "task_id": item.get("task_id", ""),
            "recurring": timing.get("recurring", False),
            "recurring_interval": timing.get("recurring_interval", 3),
        })

    from datetime import datetime as _dt
    try:
        sd = _dt.strptime(proposal.start_date, "%Y-%m-%d")
        proj_start_month = sd.month
        proj_start_year = sd.year
    except (ValueError, TypeError):
        proj_start_month = 1
        proj_start_year = 2025

    try:
        ed = _dt.strptime(proposal.end_date, "%Y-%m-%d") if proposal.end_date else None
        proj_end_month = ed.month if ed else proj_start_month
        proj_end_year = ed.year if ed else proj_start_year + 1
    except (ValueError, TypeError):
        proj_end_month = proj_start_month
        proj_end_year = proj_start_year + 1

    if proposal.end_date:
        timeline_total_months = max((proj_end_year - proj_start_year) * 12 + (proj_end_month - proj_start_month) + 1, 1)
    else:
        max_end = 0
        for t in tasks_with_timing:
            sm = t.get("start_month") or proj_start_month
            sy = t.get("start_year") or proj_start_year
            offset = (sy - proj_start_year) * 12 + (sm - proj_start_month)
            dur = t.get("duration_months") or 1
            if t.get("recurring"):
                interval = t.get("recurring_interval") or 3
                last = offset
                while last < 120:
                    end = last + dur
                    if end > max_end:
                        max_end = end
                    last += interval
            else:
                end = offset + dur
                if end > max_end:
                    max_end = end
        for bi in budget_with_timing:
            sm = bi.get("start_month") or proj_start_month
            sy = bi.get("start_year") or proj_start_year
            offset = (sy - proj_start_year) * 12 + (sm - proj_start_month)
            dur = bi.get("duration_months") or 1
            if bi.get("recurring"):
                interval = bi.get("recurring_interval") or 3
                last = offset
                while last < 120:
                    end = last + dur
                    if end > max_end:
                        max_end = end
                    last += interval
            else:
                end = offset + dur
                if end > max_end:
                    max_end = end
        timeline_total_months = max(max_end, 1)
    if timeline_total_months <= 12:
        timeline_granularity = "months"
    elif timeline_total_months <= 36:
        timeline_granularity = "quarters"
    else:
        timeline_granularity = "years"

    task_bi_data = {}
    for bi in budget_with_timing:
        tid = bi.get("task_id", "")
        if not tid:
            continue
        sm = bi.get("start_month") or proj_start_month
        sy = bi.get("start_year") or proj_start_year
        bi_offset = (sy - proj_start_year) * 12 + (sm - proj_start_month)
        bi_dur = bi.get("duration_months") or 1
        if tid not in task_bi_data:
            task_bi_data[tid] = {"min_offset": bi_offset, "max_end": bi_offset + bi_dur}
        else:
            task_bi_data[tid]["min_offset"] = min(task_bi_data[tid]["min_offset"], bi_offset)
            task_bi_data[tid]["max_end"] = max(task_bi_data[tid]["max_end"], bi_offset + bi_dur)

    all_rows = []
    for t in tasks_with_timing:
        sm = t.get("start_month") or proj_start_month
        sy = t.get("start_year") or proj_start_year
        offset = (sy - proj_start_year) * 12 + (sm - proj_start_month)
        dur = t.get("duration_months") or 1
        tid = t.get("id", "")
        recurring = t.get("recurring", False)
        interval = t.get("recurring_interval") or 3

        if tid in task_bi_data:
            offset = task_bi_data[tid]["min_offset"]
            dur = task_bi_data[tid]["max_end"] - task_bi_data[tid]["min_offset"]
            if dur < 1:
                dur = 1
            recurring = False

        if recurring:
            bars = []
            r_offset = offset
            while r_offset < timeline_total_months:
                bars.append({"offset": r_offset, "duration": dur})
                r_offset += interval
            all_rows.append({
                "name": t.get("name", ""),
                "bars": bars,
                "is_indent": False,
                "lead_entity": t.get("lead_entity", ""),
            })
        else:
            all_rows.append({
                "name": t.get("name", ""),
                "bars": [{"offset": offset, "duration": dur}],
                "is_indent": False,
                "lead_entity": t.get("lead_entity", ""),
            })

        for bi in budget_with_timing:
            if bi.get("task_id") == tid:
                bi_sm = bi.get("start_month") or sm
                bi_sy = bi.get("start_year") or sy
                bi_offset = (bi_sy - proj_start_year) * 12 + (bi_sm - proj_start_month)
                bi_dur = bi.get("duration_months") or 1
                bi_recurring = bi.get("recurring", False)
                bi_interval = bi.get("recurring_interval") or 3
                if bi_recurring:
                    bars = []
                    br_offset = bi_offset
                    while br_offset < timeline_total_months:
                        bars.append({"offset": br_offset, "duration": bi_dur})
                        br_offset += bi_interval
                    all_rows.append({
                        "name": bi.get("name", ""),
                        "bars": bars,
                        "is_indent": True,
                        "lead_entity": "",
                    })
                else:
                    all_rows.append({
                        "name": bi.get("name", ""),
                        "bars": [{"offset": bi_offset, "duration": bi_dur}],
                        "is_indent": True,
                        "lead_entity": "",
                    })

    return {
        "proposal": proposal,
        "tasks": tasks_with_timing,
        "budget_items": proposal.budget_items,
        "budget_with_timing": budget_with_timing,
        "total_budget": proposal.total_budget,
        "indirect_percent": indirect_percent,
        "indirect_amount": indirect_amount,
        "total_with_indirect": total_with_indirect,
        "timeline_granularity": timeline_granularity,
        "timeline_total_months": timeline_total_months,
        "all_rows": all_rows,
        "budget_by_year": build_budget_by_year(proposal),
        "map_ctx": build_map_export_context(proposal),
    }

build_geolibre_embed_src(config, layout='viewer')

Build the GeoLibre iframe URL from a proposal's map_config.

Parameters:

Name	Type	Description	Default
config	dict	The proposal's map_config dict (mode + optional url).	required
layout	str	GeoLibre layout mode — "viewer" (interactive), "print" (clean map-only export), etc.	'viewer'

Returns:

Type	Description
str	A URL to the configured GeoLibre embed. Nested URLs are
str	percent-encoded so their query strings are not parsed by GeoLibre.

Source code in app/utils.py

def build_geolibre_embed_src(config: dict, layout: str = "viewer") -> str:
    """Build the GeoLibre iframe URL from a proposal's map_config.

    Args:
        config: The proposal's map_config dict (mode + optional url).
        layout: GeoLibre layout mode — ``"viewer"`` (interactive),
            ``"print"`` (clean map-only export), etc.

    Returns:
        A URL to the configured GeoLibre embed. Nested URLs are
        percent-encoded so their query strings are not parsed by GeoLibre.
    """
    config = config or {}
    params = {"layout": layout, "welcome": "0"}
    url = (config.get("url") or "").strip()
    if config.get("mode") == "data_url" and url:
        params["data"] = url
    elif config.get("mode") == "project_url" and url:
        params["url"] = normalize_geolibre_project_url(url)
    return f"{Config.GEOLIBRE_EMBED_URL}?{urlencode(params)}"

build_map_export_context(proposal)

Build the Map figure context for preview and export templates.

Returns None unless the proposal opts in via map_config.show_in_preview. Resolution for the figure source, in order:

• static_image mode (uploaded local PNG/JPG) → local_image_url.
• A remote image_url in map_config → used directly.
• Otherwise the live GeoLibre embed URL is provided (renders in browsers).

share_url is the human-facing GeoLibre project link used in the Markdown export.

Source code in app/utils.py

def build_map_export_context(proposal) -> Dict[str, Any]:
    """Build the Map figure context for preview and export templates.

    Returns None unless the proposal opts in via map_config.show_in_preview.
    Resolution for the figure source, in order:

    * ``static_image`` mode (uploaded local PNG/JPG) → ``local_image_url``.
    * A remote ``image_url`` in map_config → used directly.
    * Otherwise the live GeoLibre embed URL is provided (renders in browsers).

    ``share_url`` is the human-facing GeoLibre project link used in the
    Markdown export.
    """
    map_config = getattr(proposal, "map_config", None) or {}
    if not map_config.get("show_in_preview"):
        return None
    url = (map_config.get("url") or "").strip()
    share_url = None
    if url and map_config.get("mode") in ("project_url", "data_url"):
        share_url = url.rstrip("/")
        if share_url.endswith(".geolibre.json"):
            share_url = share_url[: -len(".geolibre.json")]
    return {
        "embed_src": build_geolibre_embed_src(map_config),
        "image_url": (map_config.get("image_url") or "").strip(),
        "local_image_url": _local_map_image_url(proposal),
        "share_url": share_url,
        "caption": (map_config.get("caption") or "").strip().rstrip("."),
    }

build_map_export_image(proposal)

Try to produce a raster PNG image of the map.

Used by the on-screen print preview so a static image can be swapped in for printing. Returns a bytes PNG payload or None. Checks, in order: 1. A user-supplied image_url in map_config (fetched over HTTP). 2. A Playwright screenshot of the GeoLibre embed (all modes). 3. An auto-generated tile map from a GeoJSON data_url. 4. For project_url mode, fetches the .geolibre.json project file, looks for embedded data URLs, and generates a tile map from the first one. 5. For basemap-only mode (no data URL), stitches a default basemap.

Source code in app/utils.py

def build_map_export_image(proposal):
    """Try to produce a raster PNG image of the map.

    Used by the on-screen print preview so a static image can be swapped in
    for printing. Returns a ``bytes`` PNG payload or *None*.  Checks, in
    order:
    1. A user-supplied ``image_url`` in map_config (fetched over HTTP).
    2. A Playwright screenshot of the GeoLibre embed (all modes).
    3. An auto-generated tile map from a GeoJSON ``data_url``.
    4. For project_url mode, fetches the .geolibre.json project file,
       looks for embedded data URLs, and generates a tile map from the first one.
    5. For basemap-only mode (no data URL), stitches a default basemap.
    """
    map_config = getattr(proposal, "map_config", None) or {}
    if not map_config.get("show_in_preview"):
        return None

    image_url = (map_config.get("image_url") or "").strip()
    if image_url:
        try:
            req = urllib.request.Request(
                image_url,
                headers={"User-Agent": "Propongo/1.0 (proposal-generator)"},
            )
            with urllib.request.urlopen(req, timeout=15) as resp:
                return resp.read()
        except Exception as exc:
            logger.warning("Could not fetch map image_url %s: %s", image_url, exc)

    embed_src = build_geolibre_embed_src(map_config)
    screenshot = _screenshot_embed(embed_src)
    if screenshot:
        return screenshot

    url = (map_config.get("url") or "").strip()
    mode = map_config.get("mode", "")

    if url and mode == "data_url":
        return generate_static_map_image(url)

    if url and mode == "project_url":
        return _generate_map_from_project_tiles(
            normalize_geolibre_project_url(url)
        )

    return _generate_basemap_image()

build_tracker_export_context(proposal)

Build context dictionary for tracker export templates.

Parameters:

Name	Type	Description	Default
proposal		Proposal object	required

Returns:

Name	Type	Description
dict	Dict[str, Any]	Context dictionary with all necessary template variables

Source code in app/utils.py

def build_tracker_export_context(proposal) -> Dict[str, Any]:
    """Build context dictionary for tracker export templates.

    Args:
        proposal: Proposal object

    Returns:
        dict: Context dictionary with all necessary template variables
    """
    indirect_percent = getattr(proposal, 'indirect_percent', 0) or 0
    indirect_amount = proposal.total_budget * (indirect_percent / 100)
    total_with_indirect = proposal.total_budget + indirect_amount

    timings = proposal.budget_item_timings or {}
    task_budgets = {}
    for task in proposal.tasks:
        items = [b for b in proposal.budget_items if b.get("task_id") == task["id"]]
        for item in items:
            t = timings.get(item.get("id", ""), {})
            if t:
                item["actual_cost"] = t.get("actual_cost", 0)
        subtotal = sum(i.get("cost_per_unit", 0) * i.get("units", 0) for i in items)
        actual_total = sum(i.get("actual_cost", 0) for i in items)
        task_budgets[task["id"]] = {
            "task": task,
            "items": items,
            "subtotal": subtotal,
            "actual_total": actual_total,
        }

    total_actual = sum(tb["actual_total"] for tb in task_budgets.values())

    milestones = getattr(proposal, 'milestones', []) or []
    reports = getattr(proposal, 'reports', []) or []

    completed_tasks = sum(1 for t in proposal.tasks if t.get("status") == "completed")
    total_tasks = len(proposal.tasks)
    overall_pct = round(completed_tasks / total_tasks * 100) if total_tasks else 0

    return {
        "proposal": proposal,
        "tasks": proposal.tasks,
        "task_budgets": task_budgets,
        "total_budget": proposal.total_budget,
        "indirect_percent": indirect_percent,
        "indirect_amount": indirect_amount,
        "total_with_indirect": total_with_indirect,
        "total_actual": total_actual,
        "milestones": milestones,
        "reports": reports,
        "overall_pct": overall_pct,
    }

generate_static_map_image(data_url, width=800, height=500)

Generate a static map PNG from a GeoJSON data URL.

Downloads the GeoJSON, computes the bounding box, fetches OSM tiles, and stitches them into a single image. Returns PNG bytes or None on failure (wrong format, network error, etc.).

Source code in app/utils.py

def generate_static_map_image(data_url, width=800, height=500):
    """Generate a static map PNG from a GeoJSON data URL.

    Downloads the GeoJSON, computes the bounding box, fetches OSM tiles,
    and stitches them into a single image.  Returns PNG bytes or *None*
    on failure (wrong format, network error, etc.).
    """
    try:
        req = urllib.request.Request(
            data_url,
            headers={"User-Agent": "Propongo/1.0 (proposal-generator)"},
        )
        with urllib.request.urlopen(req, timeout=15) as resp:
            raw = resp.read()
        geojson = json.loads(raw)
    except Exception as exc:
        logger.debug("Could not fetch GeoJSON for static map: %s", exc)
        return None

    bbox = _bbox_from_geojson(geojson)
    if bbox is None:
        return None

    return _stitch_basemap_tiles(bbox, width, height)

normalize_geolibre_project_url(url)

Return a fetchable .geolibre.json URL for a shared GeoLibre project.

GeoLibre's Share dialog hands out extension-less page links (https://share.geolibre.app/user/project), but the embed's url= param needs the raw project file. Appends .geolibre.json when the host is share.geolibre.app and the extension is missing.

Source code in app/utils.py

def normalize_geolibre_project_url(url: str) -> str:
    """Return a fetchable .geolibre.json URL for a shared GeoLibre project.

    GeoLibre's Share dialog hands out extension-less page links
    (https://share.geolibre.app/user/project), but the embed's `url=` param
    needs the raw project file. Appends `.geolibre.json` when the host is
    share.geolibre.app and the extension is missing.
    """
    url = (url or "").strip()
    if not url:
        return url
    if "share.geolibre.app" in url and not url.rstrip("/").endswith(".geolibre.json"):
        url = url.rstrip("/") + ".geolibre.json"
    return url