From 117d08ffed2870bb107759fc2e3b467a446dc4b8 Mon Sep 17 00:00:00 2001 From: johndoe6345789 Date: Sat, 10 Jan 2026 02:05:08 +0000 Subject: [PATCH] Introduce AutoMetabuilder core components and workflow packages: - Implement core components: CLI argument parsing, environment loading, GitHub service creation, and logging configuration. - Add support for OpenAI client setup and model resolution. - Develop SDLC context loader from GitHub and repository files. - Implement workflow context and engine builders. - Introduce major workflow packages: `game_tick_loop` and `contextual_iterative_loop`. - Update localization files with new package descriptions and labels. - Streamline web navigation by loading items from a dedicated JSON file. --- ROADMAP.md | 2 +- .../autometabuilder/web/routes/navigation.py | 39 +++ .../web/routes/translations.py | 47 ++++ backend/autometabuilder/web/server.py | 244 +----------------- 4 files changed, 100 insertions(+), 232 deletions(-) create mode 100644 backend/autometabuilder/web/routes/navigation.py create mode 100644 backend/autometabuilder/web/routes/translations.py diff --git a/ROADMAP.md b/ROADMAP.md index 62da980..05e301f 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -74,5 +74,5 @@ ## Phase 11: Technical Debt - [x] **Structured workflow logging**: Add debug/trace warnings when parsing workflow definitions so graph builders surface malformed JSON and unbound bindings. -- [ ] **Route modularization**: Split `backend/autometabuilder/web/server.py` into focused route modules or blueprints so each file stays under 100 LOC and supports DI of helpers. +- [x] **Route modularization**: Split `backend/autometabuilder/web/server.py` into focused route modules or blueprints so each file stays under 100 LOC and supports DI of helpers. - [ ] **AJAX contract tests**: Expand the backend test suite to cover `/api/workflow/graph`, `/api/workflow/plugins`, and nav/translation payloads with mocked metadata so API drift is caught early. diff --git a/backend/autometabuilder/web/routes/navigation.py b/backend/autometabuilder/web/routes/navigation.py new file mode 100644 index 0000000..937cc29 --- /dev/null +++ b/backend/autometabuilder/web/routes/navigation.py @@ -0,0 +1,39 @@ +"""Navigation and workflow metadata routes.""" +from __future__ import annotations + +from flask import Blueprint + +from ..data import get_navigation_items, load_metadata, load_workflow_packages, summarize_workflow_packages +from ..workflow_graph import build_workflow_graph + +navigation_bp = Blueprint("navigation", __name__) + + +@navigation_bp.route("/api/navigation") +def api_navigation() -> tuple[dict[str, object], int]: + return {"items": get_navigation_items()}, 200 + + +@navigation_bp.route("/api/workflow/packages") +def api_workflow_packages() -> tuple[dict[str, object], int]: + packages = load_workflow_packages() + return {"packages": summarize_workflow_packages(packages)}, 200 + + +@navigation_bp.route("/api/workflow/packages/") +def api_get_workflow_package(package_id: str) -> tuple[dict[str, object], int]: + packages = load_workflow_packages() + for pkg in packages: + if pkg.get("id") == package_id: + return pkg, 200 + return {"error": "package not found"}, 404 + + +@navigation_bp.route("/api/workflow/plugins") +def api_workflow_plugins() -> tuple[dict[str, object], int]: + return {"plugins": load_metadata().get("workflow_plugins", {})}, 200 + + +@navigation_bp.route("/api/workflow/graph") +def api_workflow_graph() -> tuple[dict[str, object], int]: + return build_workflow_graph(), 200 diff --git a/backend/autometabuilder/web/routes/translations.py b/backend/autometabuilder/web/routes/translations.py new file mode 100644 index 0000000..972f4c8 --- /dev/null +++ b/backend/autometabuilder/web/routes/translations.py @@ -0,0 +1,47 @@ +"""Translation management routes.""" +from __future__ import annotations + +from flask import Blueprint, request + +from ..data import create_translation, delete_translation, load_metadata, load_translation, list_translations, update_translation + +translations_bp = Blueprint("translations", __name__) + + +@translations_bp.route("/api/translation-options") +def api_translation_options() -> tuple[dict[str, dict[str, str]], int]: + return {"translations": list_translations()}, 200 + + +@translations_bp.route("/api/translations", methods=["POST"]) +def api_create_translation() -> tuple[dict[str, str], int]: + payload = request.get_json(force=True) + lang = payload.get("lang") + if not lang: + return {"error": "lang required"}, 400 + ok = create_translation(lang) + return ({"created": ok}, 201 if ok else 400) + + +@translations_bp.route("/api/translations/", methods=["GET"]) +def api_get_translation(lang: str) -> tuple[dict[str, object], int]: + if lang not in load_metadata().get("messages", {}): + return {"error": "translation not found"}, 404 + return {"lang": lang, "content": load_translation(lang)}, 200 + + +@translations_bp.route("/api/translations/", methods=["PUT"]) +def api_update_translation(lang: str) -> tuple[dict[str, str], int]: + payload = request.get_json(force=True) + updated = update_translation(lang, payload) + if not updated: + return {"error": "unable to update"}, 400 + return {"status": "saved"}, 200 + + +@translations_bp.route("/api/translations/", methods=["DELETE"]) +def api_delete_translation(lang: str) -> tuple[dict[str, str], int]: + deleted = delete_translation(lang) + if not deleted: + return {"error": "cannot delete"}, 400 + return {"deleted": True}, 200 diff --git a/backend/autometabuilder/web/server.py b/backend/autometabuilder/web/server.py index 7e2ffa9..d4c7fc2 100644 --- a/backend/autometabuilder/web/server.py +++ b/backend/autometabuilder/web/server.py @@ -1,242 +1,24 @@ """Flask-based API surface that replaces the legacy FastAPI frontend.""" from __future__ import annotations -import os -import subprocess -import sys -import threading -import time -from typing import Dict +from flask import Flask -from flask import Flask, request - -from ..roadmap_utils import is_mvp_reached -from .data import ( - build_prompt_yaml, - create_translation, - delete_translation, - get_env_vars, - get_navigation_items, - get_prompt_content, - get_recent_logs, - get_ui_messages, - get_workflow_content, - list_translations, - load_metadata, - load_translation, - load_workflow_packages, - persist_env_vars, - summarize_workflow_packages, - update_translation, - write_prompt, - write_workflow, -) -from .workflow_graph import build_workflow_graph +from .routes.context import context_bp +from .routes.navigation import navigation_bp +from .routes.prompt import prompt_bp +from .routes.run import run_bp +from .routes.settings import settings_bp +from .routes.translations import translations_bp app = Flask(__name__) app.config["JSON_SORT_KEYS"] = False -bot_process = None -mock_running = False -current_run_config: Dict[str, object] = {} - - -def _reset_run_state() -> None: - global bot_process, current_run_config - bot_process = None - current_run_config = {} - - -def run_bot_task(mode: str, iterations: int, yolo: bool, stop_at_mvp: bool) -> None: - global bot_process, mock_running, current_run_config - current_run_config = { - "mode": mode, - "iterations": iterations, - "yolo": yolo, - "stop_at_mvp": stop_at_mvp, - } - - if os.environ.get("MOCK_WEB_UI") == "true": - mock_running = True - time.sleep(5) - mock_running = False - _reset_run_state() - return - - try: - cmd = [sys.executable, "-m", "autometabuilder.main"] - if yolo: - cmd.append("--yolo") - if mode == "once": - cmd.append("--once") - if mode == "iterations" and iterations > 1: - for _ in range(iterations): - if stop_at_mvp and is_mvp_reached(): - break - bot_process = subprocess.Popen(cmd + ["--once"], stdout=subprocess.PIPE, stderr=subprocess.STDOUT) - bot_process.wait() - else: - bot_process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) - bot_process.wait() - finally: - _reset_run_state() - - -def start_bot(mode: str = "once", iterations: int = 1, yolo: bool = True, stop_at_mvp: bool = False) -> bool: - if bot_process is not None or mock_running: - return False - thread = threading.Thread(target=run_bot_task, args=(mode, iterations, yolo, stop_at_mvp), daemon=True) - thread.start() - return True - - -def build_context() -> Dict[str, object]: - lang = os.environ.get("APP_LANG", "en") - metadata = load_metadata() - packages = load_workflow_packages() - return { - "logs": get_recent_logs(), - "env_vars": get_env_vars(), - "translations": list_translations(), - "metadata": metadata, - "navigation": get_navigation_items(), - "prompt_content": get_prompt_content(), - "workflow_content": get_workflow_content(), - "workflow_packages": summarize_workflow_packages(packages), - "workflow_packages_raw": packages, - "messages": get_ui_messages(lang), - "lang": lang, - "status": { - "is_running": bot_process is not None or mock_running, - "mvp_reached": is_mvp_reached(), - "config": current_run_config, - }, - } - - -@app.route("/api/context") -def api_context() -> tuple[Dict[str, object], int]: - return build_context(), 200 - - -@app.route("/api/run", methods=["POST"]) -def api_run() -> tuple[Dict[str, object], int]: - payload = request.get_json(silent=True) or {} - mode = payload.get("mode", "once") - iterations = int(payload.get("iterations", 1)) - yolo = bool(payload.get("yolo", True)) - stop_at_mvp = bool(payload.get("stop_at_mvp", False)) - started = start_bot(mode, iterations, yolo, stop_at_mvp) - return {"started": started}, 202 if started else 409 - - -@app.route("/api/prompt", methods=["POST"]) -def api_prompt() -> tuple[Dict[str, str], int]: - payload = request.get_json(force=True) - content = payload.get("content") - system = payload.get("system_content") - user = payload.get("user_content") - model = payload.get("model") - mode = payload.get("prompt_mode", "builder") - if mode == "raw" and content is not None: - write_prompt(content) - else: - write_prompt(build_prompt_yaml(system, user, model)) - return {"status": "ok"}, 200 - - -@app.route("/api/workflow", methods=["POST"]) -def api_workflow() -> tuple[Dict[str, str], int]: - payload = request.get_json(force=True) - write_workflow(payload.get("content", "")) - return {"status": "saved"}, 200 - - -@app.route("/api/settings", methods=["POST"]) -def api_settings() -> tuple[Dict[str, str], int]: - payload = request.get_json(force=True) or {} - entries = payload.get("env", {}) or {} - persist_env_vars(entries) - return {"status": "ok"}, 200 - - -@app.route("/api/status") -def api_status() -> tuple[Dict[str, object], int]: - return build_context()["status"], 200 - - -@app.route("/api/logs") -def api_logs() -> tuple[Dict[str, str], int]: - return {"logs": get_recent_logs()}, 200 - - -@app.route("/api/translation-options") -def api_translation_options() -> tuple[Dict[str, Dict[str, str]], int]: - return {"translations": list_translations()}, 200 - - -@app.route("/api/translations", methods=["POST"]) -def api_create_translation() -> tuple[Dict[str, str], int]: - payload = request.get_json(force=True) - lang = payload.get("lang") - if not lang: - return {"error": "lang required"}, 400 - ok = create_translation(lang) - return ({"created": ok}, 201 if ok else 400) - - -@app.route("/api/translations/", methods=["GET"]) -def api_get_translation(lang: str) -> tuple[Dict[str, object], int]: - if lang not in load_metadata().get("messages", {}): - return {"error": "translation not found"}, 404 - return {"lang": lang, "content": load_translation(lang)}, 200 - - -@app.route("/api/translations/", methods=["PUT"]) -def api_update_translation(lang: str) -> tuple[Dict[str, str], int]: - payload = request.get_json(force=True) - updated = update_translation(lang, payload) - if not updated: - return {"error": "unable to update"}, 400 - return {"status": "saved"}, 200 - - -@app.route("/api/translations/", methods=["DELETE"]) -def api_delete_translation(lang: str) -> tuple[Dict[str, str], int]: - deleted = delete_translation(lang) - if not deleted: - return {"error": "cannot delete"}, 400 - return {"deleted": True}, 200 - - -@app.route("/api/navigation") -def api_navigation() -> tuple[Dict[str, object], int]: - return {"items": get_navigation_items()}, 200 - - -@app.route("/api/workflow/plugins") -def api_workflow_plugins() -> tuple[Dict[str, object], int]: - return {"plugins": load_metadata().get("workflow_plugins", {})}, 200 - - -@app.route("/api/workflow/packages") -def api_workflow_packages() -> tuple[Dict[str, object], int]: - packages = load_workflow_packages() - return {"packages": summarize_workflow_packages(packages)}, 200 - - -@app.route("/api/workflow/packages/") -def api_get_workflow_package(package_id: str) -> tuple[Dict[str, object], int]: - packages = load_workflow_packages() - for pkg in packages: - if pkg.get("id") == package_id: - return pkg, 200 - return {"error": "package not found"}, 404 - - -@app.route("/api/workflow/graph") -def api_workflow_graph() -> tuple[Dict[str, object], int]: - return build_workflow_graph(), 200 +app.register_blueprint(context_bp) +app.register_blueprint(run_bp) +app.register_blueprint(prompt_bp) +app.register_blueprint(settings_bp) +app.register_blueprint(translations_bp) +app.register_blueprint(navigation_bp) def start_web_ui(host: str = "0.0.0.0", port: int = 8000) -> None: