import html
import subprocess
from pathlib import Path
from fastapi import FastAPI, Form
from fastapi.responses import HTMLResponse, RedirectResponse
NEW_APP_SCRIPT = "/tools/new-python-app.sh"
DEPLOY_SCRIPT = "/tools/deploy-app.sh"
DELETE_APP_SCRIPT = "/tools/delete-app.sh"
BACKUP_SCRIPT = "/tools/backup-appfactory.sh"
CATALOG_FILE = "/opt/appfactory/apps/catalog.yml"
APPFACTORY_ENV = "/opt/appfactory/config/appfactory.env"
app = FastAPI(title="AppFactory Portal")
def read_env_value(key: str, default: str = "") -> str:
try:
with open(APPFACTORY_ENV, "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if line.startswith(f"{key}="):
return line.split("=", 1)[1].strip().strip('"')
except Exception:
pass
return default
def load_apps():
apps = []
current = {}
path = Path(CATALOG_FILE)
if not path.exists():
return apps
for raw_line in path.read_text(encoding="utf-8").splitlines():
line = raw_line.strip()
if line.startswith("- id:"):
if current:
apps.append(current)
current = {"id": line.split(":", 1)[1].strip()}
elif ":" in line and current:
key, value = line.split(":", 1)
current[key.strip()] = value.strip().strip('"')
if current:
apps.append(current)
return apps
def save_apps(apps):
lines = ["apps:"]
for item in apps:
app_id = item.get("id", "")
lines.append("")
lines.append(f" - id: {app_id}")
lines.append(f" name: {item.get('name', app_id)}")
lines.append(f" language: {item.get('language', 'python')}")
lines.append(f" version: {item.get('version', '1.0.0')}")
lines.append(f" base_path: /apps/{app_id}")
lines.append(f" docs: /apps/{app_id}/docs")
lines.append(f" health: /apps/{app_id}/health")
lines.append(f" status: {item.get('status', 'deployed')}")
if item.get("memory"):
lines.append(f" memory: {item.get('memory')}")
if item.get("cpus"):
lines.append(f" cpus: \"{item.get('cpus')}\"")
Path(CATALOG_FILE).write_text("\n".join(lines) + "\n", encoding="utf-8")
def list_backups():
backup_dir = Path(read_env_value("BACKUP_DIR", "/opt/appfactory/backups"))
if not backup_dir.exists():
return []
return sorted(
backup_dir.glob("*.tar.gz"),
key=lambda p: p.stat().st_mtime,
reverse=True,
)
def page(title: str, body: str) -> str:
return f"""
{html.escape(title)}
{body}
"""
def render_result(title, back_url, sections, extra_link=None, extra_label=None):
rendered_sections = ""
for section_title, content in sections:
rendered_sections += f"""
{html.escape(section_title)}
{html.escape(content)}
"""
extra = ""
if extra_link and extra_label:
extra = f'{html.escape(extra_label)}
'
return page(
title,
f"""
{html.escape(title)}
{extra}
← Back
{rendered_sections}
""",
)
@app.get("/health")
def health():
return {"status": "ok"}
@app.get("/", response_class=HTMLResponse)
def index():
apps = load_apps()
gitea_url = read_env_value("GITEA_URL", "")
gitea_org = read_env_value("GITEA_ORG", "appfactory")
host = read_env_value("APPFACTORY_HOST", "192.168.66.130")
rows = ""
for item in apps:
app_id = html.escape(item.get("id", ""))
status = html.escape(item.get("status", ""))
docs = html.escape(item.get("docs", f"/apps/{app_id}/docs"))
memory = item.get("memory", "")
cpus = item.get("cpus", "")
http_clone = html.escape(f"git clone {gitea_url}/{gitea_org}/{app_id}.git")
ssh_clone = html.escape(f"git clone ssh://git@{host}:2222/{gitea_org}/{app_id}.git")
memory_options = ""
for value, label in [
("", "Default"),
("256m", "256 MB - small service"),
("512m", "512 MB - normal service"),
("1g", "1 GB - larger service"),
("2g", "2 GB - heavy service"),
]:
selected = "selected" if memory == value else ""
memory_options += f'{label} '
cpu_options = ""
for value, label in [
("", "Default"),
("0.25", "0.25 CPU - very small"),
("0.50", "0.50 CPU - normal"),
("1.00", "1 CPU - full core"),
("2.00", "2 CPU - heavy"),
]:
selected = "selected" if cpus == value else ""
cpu_options += f'{label} '
rows += f"""
{app_id} /apps/{app_id}
{status} Swagger
Clone commands
HTTP
Copy
SSH
Copy
"""
if not rows:
rows = 'No apps deployed yet.
Apps
Create, deploy, clone, tune resources and delete services.
+ New App
Backups
Create backups and copy restore commands. Restore is intentionally manual and guarded.
Manage Backups
Deployed Apps
App
Status
Docs
Resources
Git
Actions
{rows}
""",
)
@app.get("/new-app", response_class=HTMLResponse)
def new_app_form():
return page(
"New App",
"""
Create New App
This creates Gitea repo, webhook, local workspace, initial commit and deploys the service.
← Back
""",
)
@app.post("/new-app", response_class=HTMLResponse)
def create_app(app_id: str = Form(...), app_name: str = Form(...), template: str = Form(...)):
if template != "python-fastapi":
return HTMLResponse("Unsupported template", status_code=400)
create_result = subprocess.run([NEW_APP_SCRIPT, app_id, app_name], capture_output=True, text=True)
deploy_result = subprocess.run([DEPLOY_SCRIPT, app_id], capture_output=True, text=True)
status = "OK" if create_result.returncode == 0 and deploy_result.returncode == 0 else "FAILED"
return render_result(
title=f"Create App: {status}",
back_url="/portal",
sections=[
("Create Output", create_result.stdout),
("Create Error", create_result.stderr),
("Deploy Output", deploy_result.stdout),
("Deploy Error", deploy_result.stderr),
],
extra_link=f"/apps/{html.escape(app_id)}/docs",
extra_label="Open Swagger",
)
@app.post("/delete-app", response_class=HTMLResponse)
def delete_app(app_id: str = Form(...)):
result = subprocess.run([DELETE_APP_SCRIPT, app_id], capture_output=True, text=True)
status = "OK" if result.returncode == 0 else "FAILED"
return render_result(
title=f"Delete App: {status}",
back_url="/portal",
sections=[("Output", result.stdout), ("Error", result.stderr)],
)
@app.post("/update-resources", response_class=HTMLResponse)
def update_resources(app_id: str = Form(...), memory: str = Form(""), cpus: str = Form("")):
apps = load_apps()
for item in apps:
if item.get("id") == app_id:
memory = memory.strip()
cpus = cpus.strip()
if memory:
item["memory"] = memory
else:
item.pop("memory", None)
if cpus:
item["cpus"] = cpus
else:
item.pop("cpus", None)
save_apps(apps)
compose_result = subprocess.run(["/tools/generate-apps-compose.sh"], capture_output=True, text=True)
deploy_result = subprocess.run([DEPLOY_SCRIPT, app_id], capture_output=True, text=True)
status = "OK" if compose_result.returncode == 0 and deploy_result.returncode == 0 else "FAILED"
return render_result(
title=f"Update Resources: {status}",
back_url="/portal",
sections=[
("Compose Output", compose_result.stdout),
("Compose Error", compose_result.stderr),
("Deploy Output", deploy_result.stdout),
("Deploy Error", deploy_result.stderr),
],
)
@app.get("/backups", response_class=HTMLResponse)
def backups_page():
rows = ""
for path in list_backups():
size_mb = path.stat().st_size / 1024 / 1024
backup_name = html.escape(path.name)
backup_path = html.escape(str(path))
restore_cmd = html.escape(
f"sudo /home/jiri/workspace/appfactory-tools/scripts/restore-appfactory.sh --force {path}"
)
rows += f"""
{backup_name} {size_mb:.2f} MB
Copy
Copy
Restore is dangerous and must be run manually over SSH.
"""
if not rows:
rows = 'No backups found.
Backup Management
Backups do not include the backup directory itself.
Restoring an older backup should not delete newer .tar.gz files.
Restore remains manual to avoid accidental overwrite.
Available Backups
Backup
Path
Restore Command
Actions
{rows}
""",
)
@app.post("/backups/create", response_class=HTMLResponse)
def create_backup():
result = subprocess.run([BACKUP_SCRIPT], capture_output=True, text=True)
status = "OK" if result.returncode == 0 else "FAILED"
return render_result(
title=f"Create Backup: {status}",
back_url="/portal/backups",
sections=[("Output", result.stdout), ("Error", result.stderr)],
)
@app.post("/backups/delete")
def delete_backup(backup_path: str = Form(...)):
backup_dir = Path(read_env_value("BACKUP_DIR", "/opt/appfactory/backups")).resolve()
target = Path(backup_path).resolve()
if backup_dir not in target.parents:
return HTMLResponse("Invalid backup path", status_code=400)
if target.exists() and target.suffixes[-2:] == [".tar", ".gz"]:
target.unlink()
return RedirectResponse(url="/portal/backups", status_code=303)