TraderAI/traderai/desktop.py

from __future__ import annotations

import os
from pathlib import Path
import shutil
import socket
import subprocess
import sys
import threading
import time
import traceback
from typing import NoReturn

import httpx
import uvicorn

from traderai.config import edge_profile_dir, log_path


def resource_path(*parts: str) -> Path:
    base = Path(getattr(sys, "_MEIPASS", Path(__file__).resolve().parent.parent))
    return base.joinpath(*parts)


def main() -> None:
    try:
        _chdir_to_app_dir()
        _log("TraderAI desktop starting")
        _log(f"cwd={Path.cwd()}")
        _log(f"executable={sys.executable}")
        _log(f"frozen={getattr(sys, 'frozen', False)} meipass={getattr(sys, '_MEIPASS', '')}")
        port = _select_port()
        url = f"http://127.0.0.1:{port}"
        _log(f"selected_url={url}")
        if _existing_server_ready(url):
            _log("existing TraderAI backend found; opening window")
            _open_window(url)
            return
        server_thread = threading.Thread(target=_run_server, args=(port,), daemon=True)
        server_thread.start()
        _log("backend thread started")
        _wait_for_server(url)
        _log("backend health check passed")
        _open_window(url)
        _log("webview closed")
    except Exception:
        _log("fatal startup error")
        _log(traceback.format_exc())
        raise


def _chdir_to_app_dir() -> None:
    if getattr(sys, "frozen", False):
        os.chdir(Path(sys.executable).resolve().parent)


def _select_port() -> int:
    preferred = int(os.getenv("TRADERAI_PORT", "8765"))
    if _port_available(preferred):
        return preferred
    _log(f"preferred port {preferred} is in use")
    return _free_port()


def _port_available(port: int) -> bool:
    try:
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
            sock.bind(("127.0.0.1", port))
        return True
    except OSError:
        return False


def _free_port() -> int:
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
        sock.bind(("127.0.0.1", 0))
        return int(sock.getsockname()[1])


def _existing_server_ready(url: str) -> bool:
    try:
        response = httpx.get(f"{url}/api/health", timeout=1)
        return response.status_code < 500 and response.headers.get("content-type", "").startswith("application/json")
    except httpx.HTTPError:
        return False


def _run_server(port: int) -> NoReturn:
    try:
        _log(f"backend starting on port {port}")
        from traderai.server import app

        config = uvicorn.Config(
            app,
            host="127.0.0.1",
            port=port,
            log_level="info",
            log_config=None,
            lifespan="on",
        )
        server = uvicorn.Server(config)
        server.run()
        _log("backend server stopped")
        raise SystemExit(0)
    except BaseException:
        _log("backend thread crashed")
        _log(traceback.format_exc())
        raise


def _wait_for_server(url: str) -> None:
    deadline = time.monotonic() + 30
    last_error = ""
    while time.monotonic() < deadline:
        try:
            response = httpx.get(f"{url}/api/health", timeout=1)
            _log(f"health probe status={response.status_code}")
            if response.status_code < 500:
                return
        except httpx.HTTPError as exc:
            last_error = str(exc)
            _log(f"health probe failed: {last_error}")
        time.sleep(0.25)
    raise RuntimeError(f"TraderAI backend did not start within 30 seconds. {last_error}")


def _open_window(url: str) -> None:
    mode = os.getenv("TRADERAI_DESKTOP_UI", "edge").casefold()
    _log(f"ui_mode={mode}")
    if mode == "webview":
        _open_webview(url)
        return
    if _open_edge_app(url):
        return
    _open_browser(url)


def _open_webview(url: str) -> None:
    _log("importing pywebview")
    import webview

    _log("creating pywebview window")
    webview.create_window(
        "TraderAI",
        url,
        width=1320,
        height=860,
        min_size=(980, 680),
        text_select=True,
        icon=str(resource_path("web", "art", "LBC_Logo.ico")),
    )
    _log("starting pywebview")
    webview.start(gui="edgechromium", debug=False)


def _open_edge_app(url: str) -> bool:
    edge = _edge_path()
    if not edge:
        _log("msedge not found; falling back to default browser")
        return False
    profile_dir = edge_profile_dir()
    profile_dir.mkdir(parents=True, exist_ok=True)
    command = [
        str(edge),
        f"--app={url}",
        f"--user-data-dir={profile_dir}",
        "--new-window",
        "--no-first-run",
        "--disable-features=Translate",
        f"--app-icon={resource_path('web', 'art', 'LBC_Logo.ico')}",
    ]
    _log(f"launching edge app: {' '.join(command)}")
    process = subprocess.Popen(command)
    _log(f"edge process id={process.pid}")
    time.sleep(2)
    if process.poll() is None:
        process.wait()
        _log("edge app process exited")
        return True
    _log(f"edge app process exited early code={process.returncode}; keeping backend alive")
    _keep_alive()
    return True


def _open_browser(url: str) -> None:
    import webbrowser

    _log(f"opening default browser at {url}")
    webbrowser.open(url)
    _keep_alive()


def _keep_alive() -> None:
    _log("backend staying alive; close TraderAI from Task Manager if no app window owns this process")
    while True:
        time.sleep(60)


def _edge_path() -> Path | None:
    edge = shutil.which("msedge")
    if edge:
        return Path(edge)
    candidates = [
        Path(os.environ.get("ProgramFiles", "")) / "Microsoft" / "Edge" / "Application" / "msedge.exe",
        Path(os.environ.get("ProgramFiles(x86)", "")) / "Microsoft" / "Edge" / "Application" / "msedge.exe",
        Path(os.environ.get("LocalAppData", "")) / "Microsoft" / "Edge" / "Application" / "msedge.exe",
    ]
    for candidate in candidates:
        if candidate.exists():
            return candidate
    return None


def _log(message: str) -> None:
    try:
        log_path = _log_path()
        timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
        with log_path.open("a", encoding="utf-8") as file:
            file.write(f"[{timestamp}] {message}\n")
    except Exception:
        pass


def _log_path() -> Path:
    return log_path()


if __name__ == "__main__":
    main()