app.py

#!/usr/bin/env python3
"""
LocalSpace Deployer — Hugging Face Space (Vanilla)

Drag a DMG. It gets OPENED on the server: extracted, inspected, and its
app bundle metadata is displayed. The Space itself hosts the DMG contents.

Pure FastAPI + HTML/JS. No Gradio, no Streamlit, no API keys.
Public by default.
"""
from __future__ import annotations

import hashlib
import json
import os
import plistlib
import re
import shutil
import subprocess
import time
import uuid
from pathlib import Path
from typing import Any

from fastapi import FastAPI, File, Request, UploadFile
from fastapi.responses import FileResponse, HTMLResponse, JSONResponse
from fastapi.staticfiles import StaticFiles

# Try biplist for binary plists, fall back to plistlib
try:
    import biplist
    HAS_BIPLIST = True
except ImportError:
    HAS_BIPLIST = False

DATA_DIR = Path("data")
DATA_DIR.mkdir(exist_ok=True)
UPLOAD_DIR = DATA_DIR / "uploads"
UPLOAD_DIR.mkdir(exist_ok=True)
EXTRACT_DIR = DATA_DIR / "extracted"
EXTRACT_DIR.mkdir(exist_ok=True)
DB_PATH = DATA_DIR / "apps.json"

app = FastAPI(title="LocalSpace Deployer")
app.mount("/static", StaticFiles(directory="static"), name="static")

_store: dict[str, dict[str, Any]] = {}


def _load_db() -> None:
    global _store
    if DB_PATH.exists():
        with open(DB_PATH) as f:
            _store = json.load(f)
    else:
        _store = {}


def _save_db() -> None:
    with open(DB_PATH, "w") as f:
        json.dump(_store, f, indent=2)


def _slugify(name: str) -> str:
    return re.sub(r"[^a-z0-9]+", "-", name.lower().replace(".dmg", "")).strip("-")


def _hash_file(path: Path) -> str:
    h = hashlib.sha256()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(65536), b""):
            h.update(chunk)
    return h.hexdigest()


def _read_plist(path: Path) -> dict[str, Any]:
    """Read a plist file (XML or binary)."""
    try:
        with open(path, "rb") as f:
            data = f.read()
        try:
            return plistlib.loads(data)
        except Exception:
            pass
        if HAS_BIPLIST:
            try:
                return biplist.readPlistFromString(data)
            except Exception:
                pass
    except Exception:
        pass
    return {}


def _find_app_bundles(root: Path) -> list[Path]:
    """Find all .app directories under root."""
    apps = []
    for p in root.rglob("*.app"):
        if p.is_dir():
            apps.append(p)
    return apps


def _extract_dmg(dmg_path: Path, dest: Path) -> bool:
    """Extract DMG using 7z. Returns True on success."""
    try:
        dest.mkdir(parents=True, exist_ok=True)
        result = subprocess.run(
            ["7z", "x", "-y", "-o" + str(dest), str(dmg_path)],
            capture_output=True,
            text=True,
            timeout=120,
        )
        return result.returncode == 0
    except Exception:
        return False


def _inspect_app_bundle(app_path: Path) -> dict[str, Any]:
    """Read Info.plist and extract metadata from an .app bundle."""
    info_plist = app_path / "Contents" / "Info.plist"
    if not info_plist.exists():
        info_plist = app_path / "Info.plist"

    info = _read_plist(info_plist) if info_plist.exists() else {}

    icon_name = info.get("CFBundleIconFile", "")
    icon_path = None
    if icon_name:
        icons_dir = app_path / "Contents" / "Resources"
        if icons_dir.exists():
            icns = icons_dir / (icon_name if icon_name.endswith(".icns") else icon_name + ".icns")
            if icns.exists():
                icon_path = str(icns)

    files = []
    if app_path.exists():
        for f in sorted(app_path.rglob("*")):
            if f.is_file():
                try:
                    rel = str(f.relative_to(app_path))
                    files.append(rel)
                except ValueError:
                    pass

    return {
        "bundle_name": app_path.name,
        "bundle_id": info.get("CFBundleIdentifier", ""),
        "display_name": info.get("CFBundleDisplayName", "") or info.get("CFBundleName", ""),
        "version": info.get("CFBundleShortVersionString", "") or info.get("CFBundleVersion", ""),
        "min_os_version": info.get("LSMinimumSystemVersion", ""),
        "executable": info.get("CFBundleExecutable", ""),
        "icon_path": icon_path,
        "info_plist": info,
        "file_count": len(files),
        "files": files[:200],
    }


def _open_dmg(app_id: str, dmg_path: Path) -> dict[str, Any]:
    """Open a DMG: extract it, find apps, inspect them."""
    extract_to = EXTRACT_DIR / app_id
    if extract_to.exists():
        shutil.rmtree(extract_to)

    success = _extract_dmg(dmg_path, extract_to)
    if not success:
        return {"opened": False, "error": "Extraction failed. DMG may be encrypted or use an unsupported format."}

    apps = _find_app_bundles(extract_to)
    inspected = [_inspect_app_bundle(a) for a in apps]

    # Build top-level tree
    tree = []
    for item in sorted(extract_to.iterdir()):
        tree.append({
            "name": item.name,
            "type": "directory" if item.is_dir() else "file",
            "size": item.stat().st_size if item.is_file() else 0,
        })

    # Discover all HTML files recursively for iframe preview
    html_files: list[dict[str, Any]] = []
    for f in extract_to.rglob("*.html"):
        rel = f.relative_to(extract_to)
        depth = len(rel.parts)
        html_files.append({"path": str(rel), "depth": depth, "size": f.stat().st_size})
    # Prefer shallowest HTML, then shortest path, then largest
    html_files.sort(key=lambda x: (x["depth"], len(x["path"]), -x["size"]))

    return {
        "opened": True,
        "extracted_path": str(extract_to),
        "apps_found": len(inspected),
        "apps": inspected,
        "tree": tree,
        "html_files": html_files[:50],
        "has_preview": bool(html_files),
        "preview_entry": html_files[0]["path"] if html_files else None,
    }


_load_db()


@app.get("/", response_class=HTMLResponse)
def index(request: Request) -> HTMLResponse:
    with open("static/index.html") as f:
        return HTMLResponse(content=f.read())


@app.post("/api/upload")
async def upload_file(file: UploadFile = File(...)) -> JSONResponse:
    if not file.filename or not file.filename.lower().endswith(".dmg"):
        return JSONResponse({"error": "Only .dmg files are accepted."}, status_code=400)

    app_id = str(uuid.uuid4())
    slug = _slugify(file.filename)
    base_slug = slug
    counter = 1
    while any(a.get("slug") == slug for a in _store.values()):
        slug = f"{base_slug}-{counter}"
        counter += 1

    dest = UPLOAD_DIR / f"{app_id}.dmg"
    with open(dest, "wb") as f:
        while True:
            chunk = await file.read(65536)
            if not chunk:
                break
            f.write(chunk)

    sha256 = _hash_file(dest)
    size = dest.stat().st_size

    opened = _open_dmg(app_id, dest)

    entry = {
        "app_id": app_id,
        "slug": slug,
        "filename": file.filename,
        "size": size,
        "sha256": sha256,
        "download_url": f"/api/download/{app_id}",
        "opened": opened.get("opened", False),
        "apps_found": opened.get("apps_found", 0),
        "apps": opened.get("apps", []),
        "tree": opened.get("tree", []),
        "html_files": opened.get("html_files", []),
        "has_preview": opened.get("has_preview", False),
        "preview_entry": opened.get("preview_entry", None),
        "created_at": time.time(),
    }
    _store[app_id] = entry
    _save_db()

    return JSONResponse({
        "app": entry,
        "message": "DMG uploaded and opened." if entry["opened"] else "DMG uploaded but could not be opened.",
    }, status_code=201)


@app.get("/api/apps")
def list_apps() -> JSONResponse:
    apps = sorted(_store.values(), key=lambda a: a["created_at"], reverse=True)
    return JSONResponse({"apps": apps})


@app.get("/app/{app_id}", response_class=HTMLResponse)
def serve_app(app_id: str) -> HTMLResponse:
    """Serve a published app as a full-page iframe viewer."""
    entry = _store.get(app_id)
    if not entry:
        return HTMLResponse("<h1>Not found</h1>", status_code=404)

    if not entry.get("has_preview"):
        return HTMLResponse("<h1>This app has no web preview</h1>", status_code=404)

    entry_path = entry.get("preview_entry", "")
    iframe_src = f"/api/preview/{app_id}/{entry_path}" if entry_path else f"/api/preview/{app_id}"
    app_name = entry.get("filename", "App")
    display_name = ""
    if entry.get("apps") and entry["apps"]:
        display_name = entry["apps"][0].get("display_name", "") or entry["apps"][0].get("bundle_name", "")
    if not display_name:
        display_name = app_name.replace(".dmg", "").replace(".zip", "")

    html = f'''<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>{display_name} — LocalSpace</title>
<style>
  * {{ box-sizing: border-box; margin: 0; }}
  body {{ background: #0a0a0f; font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, sans-serif; height: 100vh; display: flex; flex-direction: column; overflow: hidden; }}
  header {{ background: #14141a; border-bottom: 1px solid #2a2a35; padding: 12px 20px; display: flex; align-items: center; justify-content: space-between; flex-shrink: 0; }}
  header .title {{ color: #e2e2e8; font-weight: 600; font-size: 1rem; display: flex; align-items: center; gap: 10px; }}
  header .title .badge {{ background: rgba(34,197,94,0.15); color: #22c55e; padding: 2px 8px; border-radius: 999px; font-size: 0.72rem; font-weight: 500; }}
  header .actions {{ display: flex; gap: 10px; }}
  header .actions a, header .actions button {{ background: #1c1c24; border: 1px solid #2a2a35; color: #e2e2e8; padding: 6px 14px; border-radius: 8px; font-size: 0.85rem; text-decoration: none; cursor: pointer; transition: background 0.15s; }}
  header .actions a:hover, header .actions button:hover {{ background: #2a2a35; }}
  .iframe-wrap {{ flex: 1; position: relative; overflow: hidden; }}
  iframe {{ width: 100%; height: 100%; border: none; display: block; }}
  .loading {{ position: absolute; inset: 0; display: flex; align-items: center; justify-content: center; background: #0a0a0f; color: #888898; font-size: 0.95rem; z-index: 1; }}
</style>
</head>
<body>
<header>
  <div class="title">
    ⬡ {display_name}
    <span class="badge">Live</span>
  </div>
  <div class="actions">
    <a href="/" title="Back to deployer">← Back</a>
    <a href="{iframe_src}" target="_blank" title="Open in new tab">↗ Open</a>
    <button onclick="document.querySelector('iframe').requestFullscreen()">⛶ Fullscreen</button>
  </div>
</header>
<div class="iframe-wrap">
  <div class="loading" id="loader">Loading app...</div>
  <iframe src="{iframe_src}" onload="document.getElementById('loader').style.display='none'"></iframe>
</div>
</body>
</html>'''
    return HTMLResponse(content=html)


@app.get("/api/apps/{app_id}")
def get_app(app_id: str) -> JSONResponse:
    entry = _store.get(app_id)
    if not entry:
        return JSONResponse({"error": "Not found"}, status_code=404)
    return JSONResponse({"app": entry})


@app.get("/api/download/{app_id}")
def download_app(app_id: str):
    entry = _store.get(app_id)
    if not entry:
        return JSONResponse({"error": "Not found"}, status_code=404)

    path = UPLOAD_DIR / f"{app_id}.dmg"
    if not path.exists():
        return JSONResponse({
            "error": "File not found on disk",
            "detail": "This upload was stored in temporary storage that was cleared during a Space restart. Please re-upload the DMG."
        }, status_code=404)

    return FileResponse(
        path=path,
        filename=entry["filename"],
        media_type="application/x-apple-diskimage",
    )


@app.get("/api/browse/{app_id}/{path:path}")
def browse_extracted(app_id: str, path: str):
    entry = _store.get(app_id)
    if not entry:
        return JSONResponse({"error": "Not found"}, status_code=404)

    safe_path = Path(path).name if not path else path
    base = EXTRACT_DIR / app_id
    target = base / safe_path

    if not base.exists():
        return JSONResponse({
            "error": "Extracted files not found",
            "detail": "This upload was stored in temporary storage that was cleared during a Space restart. Please re-upload the DMG."
        }, status_code=404)

    try:
        target.resolve().relative_to(base.resolve())
    except ValueError:
        return JSONResponse({"error": "Access denied"}, status_code=403)

    if not target.exists():
        return JSONResponse({"error": "Not found"}, status_code=404)

    if target.is_dir():
        items = []
        for item in sorted(target.iterdir()):
            items.append({
                "name": item.name,
                "type": "directory" if item.is_dir() else "file",
                "size": item.stat().st_size if item.is_file() else 0,
            })
        return JSONResponse({"items": items})

    return FileResponse(path=target)


# ─── Preview extracted HTML content in iframe ──────────────────────────────

@app.get("/api/preview/{app_id}/{path:path}")
def preview_content(app_id: str, path: str):
    """Serve extracted DMG content for iframe preview."""
    entry = _store.get(app_id)
    if not entry:
        return JSONResponse({"error": "Not found"}, status_code=404)

    base = EXTRACT_DIR / app_id
    if not base.exists():
        return JSONResponse({
            "error": "Extracted files not found",
            "detail": "This upload was stored in temporary storage that was cleared during a Space restart. Please re-upload the DMG."
        }, status_code=404)

    if path:
        target = base / path
    else:
        # Default to index.html if no path
        target = base / "index.html"
        if not target.exists():
            # Find any HTML file
            for f in base.rglob("*.html"):
                target = f
                break

    if not target.exists():
        return JSONResponse({"error": "Not found"}, status_code=404)

    # Security check
    try:
        target.resolve().relative_to(base.resolve())
    except ValueError:
        return JSONResponse({"error": "Access denied"}, status_code=403)

    if target.is_dir():
        # Look for index.html in directory
        idx = target / "index.html"
        if idx.exists():
            target = idx
        else:
            return JSONResponse({"error": "No index.html"}, status_code=404)

    mime_types = {
        ".html": "text/html",
        ".htm": "text/html",
        ".js": "application/javascript",
        ".css": "text/css",
        ".png": "image/png",
        ".jpg": "image/jpeg",
        ".jpeg": "image/jpeg",
        ".gif": "image/gif",
        ".svg": "image/svg+xml",
        ".json": "application/json",
        ".woff2": "font/woff2",
        ".woff": "font/woff",
        ".ttf": "font/ttf",
    }
    suffix = target.suffix.lower()
    media_type = mime_types.get(suffix, "application/octet-stream")

    # For HTML, inject base tag to handle relative paths
    if media_type == "text/html":
        content = target.read_text(errors="replace")
        # Inject base tag after <head>
        if "<base" not in content:
            rel_dir = str(target.parent.relative_to(base)) if target.parent != base else ""
            base_tag = f'<base href="/api/preview/{app_id}/{rel_dir}/">' if rel_dir else f'<base href="/api/preview/{app_id}/">'
            content = content.replace("<head>", f"<head>{base_tag}", 1)
            content = content.replace("<HEAD>", f"<HEAD>{base_tag}", 1)
        return HTMLResponse(content=content, media_type=media_type)

    return FileResponse(path=target, media_type=media_type)


# ─── Create DMG from iframe URL (Web App → macOS App) ─────────────────────

WEBAPP_DIR = DATA_DIR / "webapps"
WEBAPP_DIR.mkdir(exist_ok=True)


def _create_app_bundle(url: str, app_name: str, bundle_id: str, version: str) -> Path:
    """Create a minimal macOS .app bundle that opens a URL."""
    bundle_root = WEBAPP_DIR / f"{app_name}.app"
    if bundle_root.exists():
        shutil.rmtree(bundle_root)

    contents = bundle_root / "Contents"
    macos = contents / "MacOS"
    resources = contents / "Resources"
    macos.mkdir(parents=True)
    resources.mkdir(parents=True)

    # Info.plist
    plist = {
        "CFBundleDevelopmentRegion": "en",
        "CFBundleExecutable": app_name.replace(" ", ""),
        "CFBundleIdentifier": bundle_id,
        "CFBundleInfoDictionaryVersion": "6.0",
        "CFBundleName": app_name,
        "CFBundlePackageType": "APPL",
        "CFBundleShortVersionString": version,
        "CFBundleVersion": version,
        "LSMinimumSystemVersion": "10.15",
        "LSUIElement": False,
    }
    with open(contents / "Info.plist", "wb") as f:
        plistlib.dump(plist, f)

    # Wrapper script that opens URL
    script_path = macos / app_name.replace(" ", "")
    script_content = f'''#!/bin/bash
# Auto-generated web app wrapper
URL="{url}"
if command -v open >/dev/null 2>&1; then
    open "$URL"
else
    # Fallback for Linux testing
    xdg-open "$URL" 2>/dev/null || python3 -m webbrowser "$URL"
fi
'''
    script_path.write_text(script_content)
    script_path.chmod(0o755)

    # Create a local HTML file as backup/embedded view
    html_path = resources / "index.html"
    html_path.write_text(f'''<!DOCTYPE html>
<html><head><meta charset="utf-8"><title>{app_name}</title>
<style>body{{margin:0;height:100vh;display:flex;flex-direction:column;align-items:center;justify-content:center;font-family:sans-serif;background:#0a0a0f;color:#e2e2e8}}iframe{{width:95%;height:85%;border:1px solid #333;border-radius:8px}}</style></head>
<body><h2>{app_name}</h2><p>Loading {url}...</p><iframe src="{url}" sandbox="allow-scripts allow-same-origin allow-popups allow-forms"></iframe></body></html>
''')

    return bundle_root


def _create_dmg_from_app(app_bundle: Path, output_name: str) -> Path | None:
    """Best-effort DMG creation. Returns path to DMG or None."""
    dmg_path = WEBAPP_DIR / f"{output_name}.dmg"

    # Try genisoimage + dmg if available (Linux)
    try:
        iso_path = WEBAPP_DIR / f"{output_name}.iso"
        result = subprocess.run(
            ["genisoimage", "-D", "-V", output_name, "-no-pad", "-r", "-apple",
             "-o", str(iso_path), str(app_bundle)],
            capture_output=True, text=True, timeout=30,
        )
        if result.returncode == 0:
            # Try converting ISO to DMG
            dmg_result = subprocess.run(
                ["dmg", "iso", str(iso_path), str(dmg_path)],
                capture_output=True, text=True, timeout=30,
            )
            if dmg_result.returncode == 0 and dmg_path.exists():
                iso_path.unlink(missing_ok=True)
                return dmg_path
    except FileNotFoundError:
        pass
    except Exception:
        pass

    # Fallback: create a ZIP that user can extract on Mac and run hdiutil
    zip_path = WEBAPP_DIR / f"{output_name}.zip"
    shutil.make_archive(
        base_name=str(WEBAPP_DIR / output_name),
        format="zip",
        root_dir=str(app_bundle.parent),
        base_dir=app_bundle.name,
    )
    if zip_path.exists():
        return zip_path

    return None


@app.post("/api/create-webapp")
async def create_webapp(request: Request) -> JSONResponse:
    """Create a macOS app bundle + DMG from a URL."""
    try:
        data = await request.json()
    except Exception:
        return JSONResponse({"error": "Invalid JSON"}, status_code=400)

    url = data.get("url", "").strip()
    app_name = data.get("app_name", "WebApp").strip()
    bundle_id = data.get("bundle_id", "app.localspace.webapp").strip()
    version = data.get("version", "1.0.0").strip()

    if not url:
        return JSONResponse({"error": "URL is required"}, status_code=400)
    if not app_name:
        return JSONResponse({"error": "App name is required"}, status_code=400)

    # Sanitize
    app_name_safe = re.sub(r'[^a-zA-Z0-9 ]+', '', app_name).strip()
    if not app_name_safe:
        app_name_safe = "WebApp"

    try:
        bundle = _create_app_bundle(url, app_name_safe, bundle_id, version)
        pkg = _create_dmg_from_app(bundle, app_name_safe.replace(" ", "-"))

        if pkg is None:
            return JSONResponse({"error": "Failed to create package"}, status_code=500)

        pkg_id = str(uuid.uuid4())
        dest = UPLOAD_DIR / f"{pkg_id}{pkg.suffix}"
        shutil.copy2(pkg, dest)

        entry = {
            "app_id": pkg_id,
            "slug": _slugify(app_name_safe),
            "filename": dest.name,
            "size": dest.stat().st_size,
            "sha256": _hash_file(dest),
            "download_url": f"/api/download/{pkg_id}",
            "source_url": url,
            "app_name": app_name_safe,
            "bundle_id": bundle_id,
            "version": version,
            "is_webapp": True,
            "created_at": time.time(),
        }
        _store[pkg_id] = entry
        _save_db()

        return JSONResponse({
            "app": entry,
            "message": f"'{app_name_safe}' packaged. Download and extract on macOS, then run 'hdiutil create -srcfolder {app_name_safe}.app {app_name_safe}.dmg' to convert to DMG." if pkg.suffix == ".zip" else f"'{app_name_safe}' DMG created.",
        }, status_code=201)
    except Exception as exc:
        return JSONResponse({"error": str(exc)}, status_code=500)


@app.delete("/api/apps/{app_id}")
def delete_app(app_id: str) -> JSONResponse:
    entry = _store.pop(app_id, None)
    if entry:
        dmg = UPLOAD_DIR / f"{app_id}.dmg"
        if dmg.exists():
            dmg.unlink()
        extracted = EXTRACT_DIR / app_id
        if extracted.exists():
            shutil.rmtree(extracted)
        _save_db()
    return JSONResponse({"ok": True})