from __future__ import annotations import time import re import os import json import shutil import subprocess import sys import tempfile import plistlib from pathlib import Path from flask import Blueprint, jsonify, request, send_file, session from .auth_helpers import api_login_required, resolve_admin_policy from .context import with_terminal, attach_user_broadcast from .state import ( PROJECT_STORAGE_CACHE, PROJECT_STORAGE_CACHE_TTL_SECONDS, PROJECT_MAX_STORAGE_MB, container_manager, user_manager, ) from config import AGENT_VERSION, TERMINAL_SANDBOX_MODE from modules.host_workspace_manager import ( create_host_workspace, delete_host_workspace, load_host_workspace_catalog, rename_host_workspace, resolve_host_workspace, ) from utils.host_workspace_debug import write_host_workspace_debug from .utils_common import log_conn_diag from . import state status_bp = Blueprint('status', __name__) STATUS_DIAG_VERBOSE = os.environ.get("STATUS_DIAG_VERBOSE", "0") not in {"0", "false", "False"} STATUS_DIAG_SLOW_MS = int(os.environ.get("STATUS_DIAG_SLOW_MS", "1200") or 1200) def _close_terminal_for_key(term_key: str): terminal = state.user_terminals.pop(term_key, None) if terminal: try: if getattr(terminal, "terminal_manager", None): terminal.terminal_manager.close_all() except Exception: pass def _build_docker_projects_payload(username: str, current_id: str) -> list[dict]: running_by_workspace = _active_task_counts(username) projects = [] for ws_id, item in user_manager.list_user_workspaces(username).items(): projects.append({ "workspace_id": ws_id, "label": item.get("label") or ("默认项目" if ws_id == "default" else ws_id), "path": "", "is_current": ws_id == current_id, "is_default": ws_id == "default", "running_task_count": int(running_by_workspace.get(ws_id, 0)), }) return projects def _build_host_workspaces_payload(catalog: dict, current_id: str) -> list[dict]: running_by_workspace = _active_task_counts("host") workspaces = [] for item in catalog.get("workspaces") or []: ws_id = item.get("workspace_id") workspaces.append({ "workspace_id": ws_id, "label": item.get("label"), "path": item.get("path"), "is_current": ws_id == current_id, "running_task_count": int(running_by_workspace.get(ws_id, 0)), }) return workspaces def _parse_android_version_from_gradle(project_root: Path) -> tuple[int, str]: """从 android-webview-app/app/build.gradle.kts 读取 versionCode/versionName。""" gradle_file = project_root / "android-webview-app" / "app" / "build.gradle.kts" if not gradle_file.exists(): return 1, "1.0.0" text = gradle_file.read_text(encoding="utf-8", errors="ignore") vc_match = re.search(r"versionCode\s*=\s*(\d+)", text) vn_match = re.search(r'versionName\s*=\s*"([^"]+)"', text) version_code = int(vc_match.group(1)) if vc_match else 1 version_name = vn_match.group(1) if vn_match else "1.0.0" return version_code, version_name def _resolve_android_apk_path(project_root: Path) -> Path: """优先使用 release APK，其次 fallback 到 debug APK。""" release_apk = project_root / "android-webview-app" / "app" / "release" / "app-release.apk" if release_apk.exists(): return release_apk return project_root / "android-webview-app" / "app" / "build" / "outputs" / "apk" / "debug" / "app-debug.apk" def _load_app_changelog(project_root: Path) -> str: """读取 App 更新说明（优先 APP_CHANGELOG.md）。""" candidates = [ project_root / "android-webview-app" / "APP_CHANGELOG.md", project_root / "android-webview-app" / "CHANGELOG.md", ] for path in candidates: if not path.exists(): continue text = path.read_text(encoding="utf-8", errors="ignore").strip() if text: return text[:4000] return "" def _is_host_mode_request() -> bool: return bool(session.get("host_mode")) and (TERMINAL_SANDBOX_MODE or "").lower() == "host" def _is_docker_project_request() -> bool: return bool(session.get("username")) and not bool(session.get("host_mode")) and not bool(session.get("is_api_user")) def _active_task_counts(username: str) -> dict: running_by_workspace = {} try: from .tasks import task_manager for rec in task_manager.list_tasks(username): if rec.status in {"pending", "running", "cancel_requested"}: ws_id = getattr(rec, "workspace_id", None) or "default" running_by_workspace[ws_id] = running_by_workspace.get(ws_id, 0) + 1 except Exception: running_by_workspace = {} return running_by_workspace def _run_project_git(project_path: Path, args: list[str]) -> tuple[bool, str]: git_bin = shutil.which("git") if not git_bin: return False, "" try: proc = subprocess.run( [git_bin, "-c", "core.quotePath=false", "-C", str(project_path), *args], cwd=str(project_path), text=True, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, timeout=2, check=False, ) except Exception: return False, "" return proc.returncode == 0, (proc.stdout or "").strip() def _run_project_git_raw(project_path: Path, args: list[str], timeout: int = 4) -> tuple[bool, str]: git_bin = shutil.which("git") if not git_bin: return False, "" try: proc = subprocess.run( [git_bin, "-c", "core.quotePath=false", "-C", str(project_path), *args], cwd=str(project_path), text=True, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, timeout=timeout, check=False, ) except Exception: return False, "" return proc.returncode == 0, (proc.stdout or "") def _sum_git_numstat(numstat_text: str) -> tuple[int, int]: additions = 0 deletions = 0 for raw_line in (numstat_text or "").splitlines(): parts = raw_line.split("\t", 2) if len(parts) < 2: continue if parts[0].isdigit(): additions += int(parts[0]) if parts[1].isdigit(): deletions += int(parts[1]) return additions, deletions def _paths_from_git_numstat(numstat_text: str) -> list[str]: paths: list[str] = [] for raw_line in (numstat_text or "").splitlines(): parts = raw_line.split("\t") if len(parts) < 3: continue path = parts[-1].strip() if path: paths.append(path) return paths def _count_text_lines(text: str) -> int: if text == "": return 0 return len(text.splitlines()) def _git_file_old_line_count(repo_root: Path, rel_path: str) -> int: ok, text = _run_project_git_raw(repo_root, ["show", f"HEAD:{rel_path}"], timeout=3) if ok: return _count_text_lines(text) target = (repo_root / rel_path).resolve() try: if target.exists() and repo_root in target.parents: return _count_text_lines(target.read_text(encoding="utf-8", errors="ignore")) except Exception: pass return 0 def _git_file_lines(repo_root: Path, rel_path: str, ref: str = "HEAD") -> list[str]: ok, text = _run_project_git_raw(repo_root, ["show", f"{ref}:{rel_path}"], timeout=3) if ok: return text.splitlines() return [] def _working_file_lines(repo_root: Path, rel_path: str) -> list[str]: target = (repo_root / rel_path).resolve() try: if target.exists() and repo_root in target.parents: return target.read_text(encoding="utf-8", errors="ignore").splitlines() except Exception: pass return _git_file_lines(repo_root, rel_path) def _parse_hunk_header(header: str) -> tuple[int, int, int, int]: match = re.match(r"@@ -(\d+)(?:,(\d+))? \+(\d+)(?:,(\d+))?", header) if not match: return 0, 0, 0, 0 old_start = int(match.group(1)) old_count = int(match.group(2) or "1") new_start = int(match.group(3)) new_count = int(match.group(4) or "1") return old_start, old_count, new_start, new_count def _parse_git_diff(repo_root: Path, diff_text: str, fold_context_map: dict[str, dict[str, int]] | None = None) -> list[dict]: files: list[dict] = [] current: dict | None = None current_hunk: dict | None = None old_line = 0 new_line = 0 previous_old_end = 0 previous_new_end = 0 def finish_hunk(): nonlocal current_hunk, current, previous_old_end, previous_new_end if not current_hunk or not current: current_hunk = None return path = current.get("path") or current.get("old_path") or "" old_start = current_hunk.get("old_start") or 0 old_count = current_hunk.get("old_count") or 0 new_start = current_hunk.get("new_start") or 0 new_count = current_hunk.get("new_count") or 0 current_hunk["before_hidden"] = max(0, int(old_start) - int(previous_old_end) - 1) before_fold_key = f"before:{old_start}" current_hunk["before_fold_key"] = before_fold_key requested_extra = int((fold_context_map or {}).get(path, {}).get(before_fold_key, 0) or 0) if requested_extra > 0 and current_hunk["before_hidden"] > 0: add_count = min(current_hunk["before_hidden"], requested_extra) old_lines = _git_file_lines(repo_root, current.get("old_path") or path) new_lines = _working_file_lines(repo_root, path) injected = [] for idx in range(add_count): old_no = int(old_start) - add_count + idx new_no = int(new_start) - add_count + idx text = "" if 0 < new_no <= len(new_lines): text = new_lines[new_no - 1] elif 0 < old_no <= len(old_lines): text = old_lines[old_no - 1] injected.append({"kind": "context", "old_line": old_no, "new_line": new_no, "text": text}) current_hunk["lines"] = injected + current_hunk.get("lines", []) current_hunk["before_hidden"] -= add_count previous_old_end = int(old_start) + int(old_count) - 1 previous_new_end = int(new_start) + int(new_count) - 1 current["hunks"].append(current_hunk) current_hunk = None def finish_file(): nonlocal current, previous_old_end, previous_new_end finish_hunk() if not current: return old_total = _git_file_old_line_count(repo_root, current.get("old_path") or current.get("path") or "") current["after_hidden"] = max(0, old_total - int(previous_old_end)) current["after_fold_key"] = "after" path = current.get("path") or current.get("old_path") or "" requested_extra = int((fold_context_map or {}).get(path, {}).get("after", 0) or 0) if requested_extra > 0 and current["after_hidden"] > 0 and current.get("hunks"): add_count = min(current["after_hidden"], requested_extra) old_lines = _git_file_lines(repo_root, current.get("old_path") or path) new_lines = _working_file_lines(repo_root, path) injected = [] for idx in range(add_count): old_no = int(previous_old_end) + 1 + idx new_no = int(previous_new_end) + 1 + idx text = "" if 0 < new_no <= len(new_lines): text = new_lines[new_no - 1] elif 0 < old_no <= len(old_lines): text = old_lines[old_no - 1] injected.append({"kind": "context", "old_line": old_no, "new_line": new_no, "text": text}) current["hunks"][-1]["lines"].extend(injected) current["after_hidden"] -= add_count files.append(current) current = None previous_old_end = 0 previous_new_end = 0 for raw in diff_text.splitlines(): if raw.startswith("diff --git "): finish_file() current = { "path": "", "old_path": "", "additions": 0, "deletions": 0, "hunks": [], "after_hidden": 0, } previous_old_end = 0 continue if current is None: continue if raw.startswith("--- "): old_path = raw[4:].strip() current["old_path"] = "" if old_path == "/dev/null" else re.sub(r"^a/", "", old_path) continue if raw.startswith("+++ "): new_path = raw[4:].strip() current["path"] = current["old_path"] if new_path == "/dev/null" else re.sub(r"^b/", "", new_path) continue if raw.startswith("@@ "): finish_hunk() old_start, old_count, new_start, new_count = _parse_hunk_header(raw) current_hunk = { "header": raw, "old_start": old_start, "old_count": old_count, "new_start": new_start, "new_count": new_count, "before_hidden": 0, "lines": [], } old_line = old_start new_line = new_start continue if current_hunk is None: continue marker = raw[:1] if marker not in {" ", "+", "-"}: continue text = raw[1:] line = {"kind": "context", "old_line": old_line, "new_line": new_line, "text": text} if marker == "+": line = {"kind": "add", "old_line": None, "new_line": new_line, "text": text} new_line += 1 current["additions"] += 1 elif marker == "-": line = {"kind": "delete", "old_line": old_line, "new_line": None, "text": text} old_line += 1 current["deletions"] += 1 else: old_line += 1 new_line += 1 current_hunk["lines"].append(line) finish_file() return [item for item in files if item.get("path") and item.get("hunks")] def _open_path_in_file_manager(path: Path) -> bool: target = path.expanduser().resolve() if not target.exists(): return False try: if sys.platform == "darwin": subprocess.Popen(["open", str(target)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) return True if os.name == "nt": os.startfile(str(target)) # type: ignore[attr-defined] return True opener = shutil.which("xdg-open") if opener: subprocess.Popen([opener, str(target)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) return True except Exception: return False return False def _resolve_project_git_file(workspace, rel_path: str) -> tuple[Path | None, str]: project_path = Path(getattr(workspace, "project_path", "") or "").expanduser().resolve() ok, root_text = _run_project_git(project_path, ["rev-parse", "--show-toplevel"]) repo_root = Path(root_text).expanduser().resolve() if ok and root_text else project_path try: target = (repo_root / rel_path).expanduser().resolve() target.relative_to(repo_root) except Exception: return None, "文件路径无效" if not target.exists() or not target.is_file(): return None, "文件不存在" return target, "" def _mac_open_app_candidates(file_path: Path | None = None) -> list[dict]: app_paths: list[Path] = [] default_app_path = "" if file_path is not None: try: from AppKit import NSWorkspace from Foundation import NSURL file_url = NSURL.fileURLWithPath_(str(file_path)) workspace = NSWorkspace.sharedWorkspace() urls = workspace.URLsForApplicationsToOpenURL_(file_url) or [] app_paths.extend(Path(str(url.path())) for url in urls if url and url.path()) default_url = workspace.URLForApplicationToOpenURL_(file_url) if default_url and default_url.path(): default_app_path = str(Path(str(default_url.path())).expanduser().absolute()) except Exception: app_paths = [] if not app_paths: # 只在 LaunchServices 不可用时做很小的兜底；正常路径必须使用系统针对该文件的结果。 for base in ["/System/Applications/TextEdit.app", "/Applications/Xcode.app", "/Applications/Cursor.app"]: path = Path(base).expanduser() if path.exists(): app_paths.append(path) seen = set() apps: list[dict] = [] for path in app_paths: try: visible_path = path.expanduser().absolute() except Exception: continue path_text = str(visible_path) if ( path_text in seen or "__pycache__" in path_text or "/Library/Application Support/" in path_text or "/System/Library/PrivateFrameworks/" in path_text or "/System/Library/CoreServices/" in path_text or "/System/Library/Services/" in path_text ): continue info = _read_mac_app_info(visible_path) label = info.get("label") or visible_path.stem bundle_id = str(info.get("bundle_id") or "") if bundle_id == "com.microsoft.VSCode" or label == "Code": label = visible_path.stem label_lower = label.lower() if label_lower.endswith("agent") or label_lower in {"finder", "system settings"}: continue seen.add(path_text) apps.append({ "id": path_text, "label": f"{label}（默认）" if default_app_path and path_text == default_app_path else label, "bundle_id": bundle_id, "icon_url": f"/api/project/app-icon?app_id={path_text}", "rank": 0 if default_app_path and path_text == default_app_path else 1, }) apps.sort(key=lambda item: (item.get("rank", 1000), item.get("label", "").lower())) return [{k: v for k, v in item.items() if k != "rank"} for item in apps[:40]] def _read_mac_app_info(app_path: Path) -> dict: info_path = app_path / "Contents" / "Info.plist" try: data = plistlib.loads(info_path.read_bytes()) except Exception: data = {} return { "label": data.get("CFBundleDisplayName") or data.get("CFBundleName") or app_path.stem, "bundle_id": data.get("CFBundleIdentifier") or "", "icon_file": data.get("CFBundleIconFile") or "", } def _mac_app_icon_path(app_path: Path) -> Path | None: info = _read_mac_app_info(app_path) icon_name = str(info.get("icon_file") or "").strip() resources = app_path / "Contents" / "Resources" candidates = [] if icon_name: candidates.append(resources / icon_name) if not icon_name.endswith(".icns"): candidates.append(resources / f"{icon_name}.icns") candidates.extend(resources.glob("*.icns")) for candidate in candidates: if candidate.exists(): return candidate return None def _windows_open_app_candidates(file_path: Path | None = None) -> list[dict]: if os.name != "nt" or file_path is None: return [] try: import ctypes import shlex import winreg except Exception: return [] ext = file_path.suffix.lower() if not ext: return [] def expand_env(value: str) -> str: try: return os.path.expandvars(value) except Exception: return value def parse_command_exe(command: str) -> str: command = expand_env((command or "").strip()) if not command: return "" try: parts = shlex.split(command, posix=False) if parts: return parts[0].strip('"') except Exception: pass if command.startswith('"'): end = command.find('"', 1) return command[1:end] if end > 1 else "" return command.split(" ", 1)[0].strip('"') def read_reg_value(root, subkey: str, value_name: str = "") -> str: try: with winreg.OpenKey(root, subkey) as key: value, _ = winreg.QueryValueEx(key, value_name) return str(value or "") except Exception: return "" def enum_subkey_values(root, subkey: str) -> list[str]: values: list[str] = [] try: with winreg.OpenKey(root, subkey) as key: idx = 0 while True: try: name, value, _ = winreg.EnumValue(key, idx) idx += 1 if name and name != "MRUList": values.append(str(value or name)) except OSError: break except Exception: pass return values def command_for_progid(progid: str) -> str: for root in (winreg.HKEY_CURRENT_USER, winreg.HKEY_CLASSES_ROOT, winreg.HKEY_LOCAL_MACHINE): for prefix in ("Software\\Classes\\", ""): command = read_reg_value(root, f"{prefix}{progid}\\shell\\open\\command") if command: return command return "" def app_label(exe: Path, default: bool = False) -> str: stem = exe.stem label_map = { "notepad": "记事本", "code": "Visual Studio Code", "cursor": "Cursor", "wordpad": "写字板", } label = label_map.get(stem.lower(), stem) return f"{label}（默认）" if default else label candidates: list[tuple[str, bool]] = [] # 默认打开程序：Windows Shell 关联查询，等价于系统“打开方式”的默认项。 try: ASSOCF_NONE = 0 ASSOCSTR_EXECUTABLE = 2 buffer_len = ctypes.c_ulong(0) ctypes.windll.Shlwapi.AssocQueryStringW( ASSOCF_NONE, ASSOCSTR_EXECUTABLE, ext, None, None, ctypes.byref(buffer_len) ) if buffer_len.value > 0: buffer = ctypes.create_unicode_buffer(buffer_len.value) result = ctypes.windll.Shlwapi.AssocQueryStringW( ASSOCF_NONE, ASSOCSTR_EXECUTABLE, ext, None, buffer, ctypes.byref(buffer_len) ) if result == 0 and buffer.value: candidates.append((buffer.value, True)) except Exception: pass progids = set() for root, subkey in ( (winreg.HKEY_CURRENT_USER, f"Software\\Microsoft\\Windows\\CurrentVersion\\Explorer\\FileExts\\{ext}\\OpenWithProgids"), (winreg.HKEY_CLASSES_ROOT, f"{ext}\\OpenWithProgids"), ): progids.update(enum_subkey_values(root, subkey)) default_progid = read_reg_value(winreg.HKEY_CLASSES_ROOT, ext) if default_progid: progids.add(default_progid) user_choice = read_reg_value( winreg.HKEY_CURRENT_USER, f"Software\\Microsoft\\Windows\\CurrentVersion\\Explorer\\FileExts\\{ext}\\UserChoice", "ProgId", ) if user_choice: progids.add(user_choice) for progid in progids: command = command_for_progid(progid) exe = parse_command_exe(command) if exe: candidates.append((exe, False)) app_names = enum_subkey_values( winreg.HKEY_CURRENT_USER, f"Software\\Microsoft\\Windows\\CurrentVersion\\Explorer\\FileExts\\{ext}\\OpenWithList", ) app_names += enum_subkey_values( winreg.HKEY_CLASSES_ROOT, f"{ext}\\OpenWithList", ) for app_name in app_names: app_key = f"Applications\\{app_name}\\shell\\open\\command" command = read_reg_value(winreg.HKEY_CLASSES_ROOT, app_key) exe = parse_command_exe(command) if exe: candidates.append((exe, False)) seen = set() apps = [] for exe, is_default in candidates: try: path = Path(expand_env(exe)).expanduser().resolve() except Exception: continue key = str(path).lower() if key in seen or not path.exists() or path.suffix.lower() != ".exe": continue seen.add(key) apps.append({"id": str(path), "label": app_label(path, is_default)}) return apps[:40] def _open_file_with_app(file_path: Path, app_id: str) -> bool: try: if sys.platform == "darwin": app_path = Path(app_id).expanduser().resolve() if not app_path.exists() or app_path.suffix != ".app": return False subprocess.Popen(["open", "-a", str(app_path), str(file_path)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) return True if os.name == "nt": exe_path = Path(app_id).expanduser().resolve() if not exe_path.exists() or exe_path.suffix.lower() != ".exe": return False subprocess.Popen([str(exe_path), str(file_path)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) return True except Exception: return False return False @status_bp.route('/api/health') @api_login_required def get_health(): """轻量健康检查：用于前端连接心跳，避免触发重型 status 计算。""" started_at = time.perf_counter() heartbeat_id = request.headers.get("X-Connection-Heartbeat", "").strip() username = session.get("username") or "-" payload = { "success": True, "status": "ok", "server_time_ms": int(time.time() * 1000), } elapsed_ms = (time.perf_counter() - started_at) * 1000 if request.args.get("diag") == "1" or elapsed_ms >= 800: log_conn_diag( f"health hb={heartbeat_id or '-'} user={username} " f"elapsed_ms={elapsed_ms:.1f} host_mode={bool(session.get('host_mode'))} " f"workspace_id={session.get('workspace_id') or '-'}" ) return jsonify(payload) @status_bp.route('/api/status') @api_login_required @with_terminal def get_status(terminal, workspace, username): """获取系统状态（包含对话、容器、版本等信息）""" started_at = time.perf_counter() heartbeat_id = request.headers.get("X-Connection-Heartbeat", "").strip() diag_requested = STATUS_DIAG_VERBOSE or request.args.get("diag") == "1" timings = { "terminal_ms": 0.0, "terminals_ms": 0.0, "conversation_meta_ms": 0.0, "container_ms": 0.0, "policy_ms": 0.0, } phase_started_at = time.perf_counter() try: status = terminal.get_status() except Exception as exc: timings["terminal_ms"] = (time.perf_counter() - phase_started_at) * 1000 total_ms = (time.perf_counter() - started_at) * 1000 log_conn_diag( f"status hb={heartbeat_id or '-'} user={username} phase=terminal.get_status " f"total_ms={total_ms:.1f} terminal_ms={timings['terminal_ms']:.1f} error={exc}" ) raise timings["terminal_ms"] = (time.perf_counter() - phase_started_at) * 1000 phase_started_at = time.perf_counter() if terminal.terminal_manager: status['terminals'] = terminal.terminal_manager.list_terminals() timings["terminals_ms"] = (time.perf_counter() - phase_started_at) * 1000 phase_started_at = time.perf_counter() try: current_conv = terminal.context_manager.current_conversation_id status.setdefault('conversation', {})['current_id'] = current_conv if current_conv and not current_conv.startswith('temp_'): current_conv_data = terminal.context_manager.conversation_manager.load_conversation(current_conv) if current_conv_data: status['conversation']['title'] = current_conv_data.get('title', '未知对话') status['conversation']['created_at'] = current_conv_data.get('created_at') status['conversation']['updated_at'] = current_conv_data.get('updated_at') meta = current_conv_data.get("metadata", {}) or {} status['conversation']['compression'] = { "in_progress": bool(meta.get("compression_in_progress", False)), "mode": meta.get("compression_mode"), "stage": meta.get("compression_stage"), "error": meta.get("compression_error"), "count": int(meta.get("compression_count", 0) or 0), "job_id": meta.get("compression_job_id"), } except Exception as exc: log_conn_diag(f"status conversation-meta-failed user={username} error={exc}") timings["conversation_meta_ms"] = (time.perf_counter() - phase_started_at) * 1000 status['project_path'] = str(workspace.project_path) phase_started_at = time.perf_counter() try: # 首屏状态只需要容器是否存在/运行等轻量信息；Docker stats 很慢， # 由 /api/container-status 在用量面板需要时单独拉取。 workspace_id = getattr(workspace, 'workspace_id', None) or session.get('workspace_id') or 'default' container_key = ( f"host::{workspace_id}" if bool(session.get("host_mode")) or username == "host" else f"{username}::{workspace_id}" ) status['container'] = container_manager.get_container_status(container_key, include_stats=False) except Exception as exc: status['container'] = {"success": False, "error": str(exc)} log_conn_diag(f"status container-status-failed user={username} error={exc}") timings["container_ms"] = (time.perf_counter() - phase_started_at) * 1000 status['version'] = AGENT_VERSION phase_started_at = time.perf_counter() try: policy = resolve_admin_policy(user_manager.get_user(username)) status['admin_policy'] = { "ui_blocks": policy.get("ui_blocks") or {}, "disabled_models": policy.get("disabled_models") or [], "forced_category_states": policy.get("forced_category_states") or {}, "version": policy.get("updated_at"), } except Exception: pass timings["policy_ms"] = (time.perf_counter() - phase_started_at) * 1000 total_ms = (time.perf_counter() - started_at) * 1000 is_slow = total_ms >= STATUS_DIAG_SLOW_MS if diag_requested or is_slow: conversation = status.get("conversation") or {} log_conn_diag( f"status hb={heartbeat_id or '-'} user={username} host_mode={bool(session.get('host_mode'))} " f"workspace_id={session.get('workspace_id') or '-'} " f"conversation_id={conversation.get('current_id') or '-'} " f"total_ms={total_ms:.1f} slow={is_slow} " f"terminal_ms={timings['terminal_ms']:.1f} terminals_ms={timings['terminals_ms']:.1f} " f"conversation_meta_ms={timings['conversation_meta_ms']:.1f} " f"container_ms={timings['container_ms']:.1f} policy_ms={timings['policy_ms']:.1f}" ) return jsonify(status) @status_bp.route('/api/project/git-summary') @api_login_required @with_terminal def get_project_git_summary(terminal, workspace, username): """返回当前项目自身 git 仓库概览（不使用隐藏版本管理 git）。""" project_path = Path(getattr(workspace, "project_path", "") or "").expanduser().resolve() if not project_path.exists(): return jsonify({"success": True, "data": {"has_git": False}}) ok, root_text = _run_project_git(project_path, ["rev-parse", "--show-toplevel"]) if not ok or not root_text: return jsonify({"success": True, "data": {"has_git": False}}) repo_root = Path(root_text).expanduser().resolve() ok, branch = _run_project_git(repo_root, ["branch", "--show-current"]) if not ok or not branch: ok, branch = _run_project_git(repo_root, ["rev-parse", "--short", "HEAD"]) branch = f"HEAD {branch}" if ok and branch else "HEAD" _, upstream = _run_project_git(repo_root, ["rev-parse", "--abbrev-ref", "--symbolic-full-name", "@{u}"]) _, numstat_text = _run_project_git(repo_root, ["diff", "--numstat", "HEAD", "--"]) additions, deletions = _sum_git_numstat(numstat_text) return jsonify({ "success": True, "data": { "has_git": True, "project_name": repo_root.name, "project_path": str(repo_root), "branch": branch, "upstream": upstream, "additions": additions, "deletions": deletions, } }) @status_bp.route('/api/project/git-diff') @api_login_required @with_terminal def get_project_git_diff(terminal, workspace, username): """返回当前项目自身 git 工作区 diff，用于前端侧边栏展示。""" project_path = Path(getattr(workspace, "project_path", "") or "").expanduser().resolve() if not project_path.exists(): return jsonify({"success": True, "data": {"has_git": False, "files": []}}) ok, root_text = _run_project_git(project_path, ["rev-parse", "--show-toplevel"]) if not ok or not root_text: return jsonify({"success": True, "data": {"has_git": False, "files": []}}) repo_root = Path(root_text).expanduser().resolve() try: context_lines = int(request.args.get("context", "3") or 3) except Exception: context_lines = 3 context_lines = max(0, min(200, context_lines)) context_map: dict[str, int] = {} raw_context_map = request.args.get("context_map", "").strip() if raw_context_map: try: parsed = json.loads(raw_context_map) if isinstance(parsed, dict): for key, value in parsed.items(): if not isinstance(key, str) or not key: continue try: context_map[key] = max(0, min(200, int(value))) except Exception: continue except Exception: context_map = {} fold_context_map: dict[str, dict[str, int]] = {} raw_fold_context_map = request.args.get("fold_context_map", "").strip() if raw_fold_context_map: try: parsed = json.loads(raw_fold_context_map) if isinstance(parsed, dict): for path_key, folds in parsed.items(): if not isinstance(path_key, str) or not isinstance(folds, dict): continue safe_folds: dict[str, int] = {} for fold_key, value in folds.items(): if not isinstance(fold_key, str) or not fold_key: continue try: safe_folds[fold_key] = max(0, min(200, int(value))) except Exception: continue if safe_folds: fold_context_map[path_key] = safe_folds except Exception: fold_context_map = {} ok, branch = _run_project_git(repo_root, ["branch", "--show-current"]) if not ok or not branch: ok, branch = _run_project_git(repo_root, ["rev-parse", "--short", "HEAD"]) branch = f"HEAD {branch}" if ok and branch else "HEAD" _, upstream = _run_project_git(repo_root, ["rev-parse", "--abbrev-ref", "--symbolic-full-name", "@{u}"]) _, numstat_text = _run_project_git(repo_root, ["diff", "--numstat", "HEAD", "--"]) additions, deletions = _sum_git_numstat(numstat_text) diff_text = "" if context_map: parts: list[str] = [] for rel_path in _paths_from_git_numstat(numstat_text): per_file_context = context_map.get(rel_path, context_lines) ok, file_diff = _run_project_git_raw( repo_root, ["--no-pager", "diff", f"--unified={per_file_context}", "HEAD", "--", rel_path], timeout=8, ) if ok and file_diff: parts.append(file_diff) diff_text = "\n".join(parts) else: ok, diff_text = _run_project_git_raw(repo_root, [f"--no-pager", "diff", f"--unified={context_lines}", "HEAD", "--"], timeout=8) if not ok: diff_text = "" return jsonify({ "success": True, "data": { "has_git": True, "project_name": repo_root.name, "project_path": str(repo_root), "branch": branch, "upstream": upstream, "additions": additions, "deletions": deletions, "context": context_lines, "files": _parse_git_diff(repo_root, diff_text, fold_context_map=fold_context_map), } }) @status_bp.route('/api/project/open-in-file-manager', methods=['POST']) @api_login_required @with_terminal def open_project_in_file_manager(terminal, workspace, username): project_path = Path(getattr(workspace, "project_path", "") or "").expanduser().resolve() ok, root_text = _run_project_git(project_path, ["rev-parse", "--show-toplevel"]) target = Path(root_text).expanduser().resolve() if ok and root_text else project_path if _open_path_in_file_manager(target): return jsonify({"success": True}) return jsonify({"success": False, "error": "无法打开文件管理器"}), 500 @status_bp.route('/api/project/file-open-apps') @api_login_required @with_terminal def list_project_file_open_apps(terminal, workspace, username): if not _is_host_mode_request(): return jsonify({"success": False, "error": "仅宿主机模式可用"}), 403 rel_path = request.args.get("path", "").strip() target, error = _resolve_project_git_file(workspace, rel_path) if not target: return jsonify({"success": False, "error": error or "文件路径无效"}), 400 if sys.platform == "darwin": apps = _mac_open_app_candidates(target) elif os.name == "nt": apps = _windows_open_app_candidates(target) else: apps = [] return jsonify({"success": True, "data": {"apps": apps}}) @status_bp.route('/api/project/app-icon') @api_login_required def get_project_open_app_icon(): if not _is_host_mode_request(): return jsonify({"success": False, "error": "仅宿主机模式可用"}), 403 app_id = request.args.get("app_id", "").strip() if sys.platform != "darwin" or not app_id: return jsonify({"success": False, "error": "应用图标不可用"}), 404 try: app_path = Path(app_id).expanduser().resolve() if not app_path.exists() or app_path.suffix != ".app": return jsonify({"success": False, "error": "应用不存在"}), 404 icon_path = _mac_app_icon_path(app_path) if not icon_path: return jsonify({"success": False, "error": "应用图标不存在"}), 404 cache_dir = Path(tempfile.gettempdir()) / "agents_app_icons" cache_dir.mkdir(parents=True, exist_ok=True) out_path = cache_dir / f"{abs(hash(str(app_path)))}.png" if not out_path.exists() or out_path.stat().st_mtime < icon_path.stat().st_mtime: sips = shutil.which("sips") if not sips: return jsonify({"success": False, "error": "图标转换工具不可用"}), 404 subprocess.run( [sips, "-s", "format", "png", str(icon_path), "--out", str(out_path)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=5, check=False, ) if out_path.exists(): return send_file(str(out_path), mimetype="image/png") except Exception: pass return jsonify({"success": False, "error": "应用图标不可用"}), 404 @status_bp.route('/api/project/open-file-with-app', methods=['POST']) @api_login_required @with_terminal def open_project_file_with_app(terminal, workspace, username): if not _is_host_mode_request(): return jsonify({"success": False, "error": "仅宿主机模式可用"}), 403 payload = request.get_json(silent=True) or {} rel_path = str(payload.get("path") or "").strip() app_id = str(payload.get("app_id") or "").strip() if not app_id: return jsonify({"success": False, "error": "未选择应用"}), 400 target, error = _resolve_project_git_file(workspace, rel_path) if not target: return jsonify({"success": False, "error": error or "文件路径无效"}), 400 available = _mac_open_app_candidates(target) if sys.platform == "darwin" else _windows_open_app_candidates(target) if os.name == "nt" else [] if not any(item.get("id") == app_id for item in available): return jsonify({"success": False, "error": "应用不可用"}), 400 if _open_file_with_app(target, app_id): return jsonify({"success": True}) return jsonify({"success": False, "error": "打开文件失败"}), 500 @status_bp.route('/api/host/workspaces') @api_login_required def list_host_workspaces(): if not _is_host_mode_request(): return jsonify({"success": False, "error": "仅宿主机模式可用"}), 403 catalog, current = resolve_host_workspace(session.get("host_workspace_id")) current_id = current.get("workspace_id") session['host_workspace_id'] = current_id session['workspace_id'] = current_id workspaces = _build_host_workspaces_payload(catalog, current_id) return jsonify({ "success": True, "data": { "source_path": catalog.get("source_path"), "default_workspace_id": catalog.get("default_workspace_id"), "current_workspace_id": current_id, "workspaces": workspaces, } }) @status_bp.route('/api/projects') @api_login_required def list_docker_projects(): if not _is_docker_project_request(): return jsonify({"success": False, "error": "仅 Docker Web 模式可用"}), 403 username = session.get("username") current_id = (session.get("workspace_id") or "default").strip() or "default" # 默认项目用于向后兼容旧 Docker Web 文件/对话。 user_manager.ensure_user_workspace(username, current_id) projects = _build_docker_projects_payload(username, current_id) return jsonify({ "success": True, "data": { "default_workspace_id": "default", "current_workspace_id": current_id, "workspaces": projects, } }) @status_bp.route('/api/projects/select', methods=['GET', 'POST']) @api_login_required def select_docker_project(): if not _is_docker_project_request(): return jsonify({"success": False, "error": "仅 Docker Web 模式可用"}), 403 payload = request.get_json(silent=True) if request.method != "GET" else None workspace_id = ( request.args.get("workspace_id") or (payload or {}).get("workspace_id") or "" ).strip() if not workspace_id: return jsonify({"success": False, "error": "缺少项目 ID"}), 400 username = session.get("username") previous_workspace_id = session.get("workspace_id") or "default" known_projects = user_manager.list_user_workspaces(username) if workspace_id not in known_projects: return jsonify({"success": False, "error": "项目不存在"}), 404 try: workspace = user_manager.ensure_user_workspace(username, workspace_id) except ValueError as exc: return jsonify({"success": False, "error": str(exc)}), 400 session["workspace_id"] = getattr(workspace, "workspace_id", workspace_id) try: container_manager.ensure_container( username, str(workspace.project_path), container_key=f"{username}::{session['workspace_id']}", preferred_mode="docker", ) except RuntimeError as exc: session["workspace_id"] = previous_workspace_id return jsonify({"success": False, "error": str(exc)}), 503 terminal = state.user_terminals.get(f"{username}::{session['workspace_id']}") if terminal: try: attach_user_broadcast(terminal, username) except Exception: pass return jsonify({ "success": True, "data": { "current_workspace_id": session["workspace_id"], "project_path": "", "default_workspace_id": "default", "reloaded": previous_workspace_id != session["workspace_id"], } }) @status_bp.route('/api/projects/create', methods=['POST']) @api_login_required def create_docker_project(): if not _is_docker_project_request(): return jsonify({"success": False, "error": "仅 Docker Web 模式可用"}), 403 payload = request.get_json(silent=True) or {} label = (payload.get("label") or payload.get("name") or "").strip() if not label: return jsonify({"success": False, "error": "项目名称不能为空"}), 400 username = session.get("username") try: workspace = user_manager.create_user_workspace(username, "", label=label) except ValueError as exc: return jsonify({"success": False, "error": str(exc)}), 400 projects = [] current_id = session.get("workspace_id") or "default" for ws_id, item in user_manager.list_user_workspaces(username).items(): projects.append({ "workspace_id": ws_id, "label": item.get("label") or ws_id, "path": "", "is_current": ws_id == current_id, "is_default": ws_id == "default", }) return jsonify({ "success": True, "data": { "created": True, "workspace": { "workspace_id": getattr(workspace, "workspace_id", ""), "label": label, "path": "", }, "default_workspace_id": "default", "current_workspace_id": current_id, "workspaces": projects, } }) @status_bp.route('/api/projects/rename', methods=['POST']) @api_login_required def rename_docker_project(): if not _is_docker_project_request(): return jsonify({"success": False, "error": "仅 Docker Web 模式可用"}), 403 payload = request.get_json(silent=True) or {} workspace_id = (payload.get("workspace_id") or "").strip() label = (payload.get("label") or payload.get("name") or "").strip() if not workspace_id: return jsonify({"success": False, "error": "缺少项目 ID"}), 400 if not label: return jsonify({"success": False, "error": "项目名称不能为空"}), 400 username = session.get("username") try: workspace = user_manager.rename_user_workspace(username, workspace_id, label) except ValueError as exc: return jsonify({"success": False, "error": str(exc)}), 400 current_id = session.get("workspace_id") or "default" return jsonify({ "success": True, "data": { "workspace": workspace, "default_workspace_id": "default", "current_workspace_id": current_id, "workspaces": _build_docker_projects_payload(username, current_id), } }) @status_bp.route('/api/projects/delete', methods=['POST']) @api_login_required def delete_docker_project(): if not _is_docker_project_request(): return jsonify({"success": False, "error": "仅 Docker Web 模式可用"}), 403 payload = request.get_json(silent=True) or {} workspace_id = (payload.get("workspace_id") or "").strip() if not workspace_id: return jsonify({"success": False, "error": "缺少项目 ID"}), 400 if workspace_id == "default": return jsonify({"success": False, "error": "默认项目不能删除"}), 400 username = session.get("username") if _active_task_counts(username).get(workspace_id): return jsonify({"success": False, "error": "该项目有运行中的任务，暂不能删除"}), 409 known_projects = user_manager.list_user_workspaces(username) if workspace_id not in known_projects: return jsonify({"success": False, "error": "项目不存在"}), 404 term_key = f"{username}::{workspace_id}" try: _close_terminal_for_key(term_key) container_manager.release_container(term_key, reason="delete_project") user_manager.delete_user_workspace(username, workspace_id) except ValueError as exc: return jsonify({"success": False, "error": str(exc)}), 400 current_id = session.get("workspace_id") or "default" if current_id == workspace_id: current_id = "default" session["workspace_id"] = current_id try: workspace = user_manager.ensure_user_workspace(username, current_id) container_manager.ensure_container( username, str(workspace.project_path), container_key=f"{username}::{current_id}", preferred_mode="docker", ) except RuntimeError as exc: return jsonify({"success": False, "error": str(exc)}), 503 return jsonify({ "success": True, "data": { "deleted_workspace_id": workspace_id, "default_workspace_id": "default", "current_workspace_id": current_id, "workspaces": _build_docker_projects_payload(username, current_id), } }) @status_bp.route('/api/host/workspaces/select', methods=['GET', 'POST']) @api_login_required def select_host_workspace(): if not _is_host_mode_request(): return jsonify({"success": False, "error": "仅宿主机模式可用"}), 403 payload = request.get_json(silent=True) if request.method != "GET" else None workspace_id = ( request.args.get("workspace_id") or (payload or {}).get("workspace_id") or "" ).strip() if not workspace_id: return jsonify({"success": False, "error": "缺少 workspace_id"}), 400 write_host_workspace_debug( "status.select_host_workspace.request", workspace_id=workspace_id, current_session_workspace_id=session.get("host_workspace_id"), ) catalog = load_host_workspace_catalog() current_id = (session.get("host_workspace_id") or "").strip() target = next( (item for item in (catalog.get("workspaces") or []) if item.get("workspace_id") == workspace_id), None, ) if not target: return jsonify({"success": False, "error": "workspace_id 不存在"}), 404 target_path = str(Path(target.get("path") or "").expanduser().resolve()) write_host_workspace_debug( "status.select_host_workspace.target", workspace_id=workspace_id, target_path=target_path, current_workspace_id=current_id, ) previous_workspace_id = session.get("workspace_id") with state.HOST_ACTIVE_WORKSPACE_LOCK: state.HOST_ACTIVE_WORKSPACE_ID = workspace_id state.HOST_ACTIVE_WORKSPACE_PATH = target_path state.HOST_ACTIVE_WORKSPACE_VERSION += 1 session['host_workspace_id'] = workspace_id session['workspace_id'] = workspace_id try: container_handle = state.container_manager.ensure_container( "host", target_path, container_key=f"host::{workspace_id}", preferred_mode="host", ) except RuntimeError as exc: if current_id: session['host_workspace_id'] = current_id if previous_workspace_id is not None: session['workspace_id'] = previous_workspace_id elif current_id: session['workspace_id'] = current_id write_host_workspace_debug( "status.select_host_workspace.ensure_container_failed", workspace_id=workspace_id, target_path=target_path, error=str(exc), ) return jsonify({"success": False, "error": str(exc)}), 503 host_terminal = state.user_terminals.get(f"host::{workspace_id}") if host_terminal: try: host_terminal.update_container_session(container_handle) attach_user_broadcast(host_terminal, "host") host_terminal.username = "host" host_terminal.user_role = "admin" write_host_workspace_debug( "status.select_host_workspace.updated_terminal", terminal_id=id(host_terminal), workspace_id=workspace_id, project_path=str(getattr(host_terminal, "project_path", "")), context_project_path=str(getattr(getattr(host_terminal, "context_manager", None), "project_path", "")), current_conversation_id=getattr(getattr(host_terminal, "context_manager", None), "current_conversation_id", None), ) except Exception: try: if getattr(host_terminal, "terminal_manager", None): host_terminal.terminal_manager.close_all() except Exception: pass state.user_terminals.pop(f"host::{workspace_id}", None) write_host_workspace_debug( "status.select_host_workspace.drop_terminal_after_error", workspace_id=workspace_id, ) write_host_workspace_debug( "status.select_host_workspace.success", workspace_id=workspace_id, target_path=target_path, ) return jsonify({ "success": True, "data": { "current_workspace_id": workspace_id, "project_path": target_path, "default_workspace_id": catalog.get("default_workspace_id"), "reloaded": current_id != workspace_id, } }) @status_bp.route('/api/host/workspaces/create', methods=['GET', 'POST']) @api_login_required def create_host_workspace_api(): if not _is_host_mode_request(): return jsonify({"success": False, "error": "仅宿主机模式可用"}), 403 payload = request.get_json(silent=True) if request.method != "GET" else None workspace_path = ( request.args.get("path") or (payload or {}).get("path") or "" ).strip() label = ( request.args.get("label") or (payload or {}).get("label") or "" ).strip() set_default_raw = ( request.args.get("set_default") or (payload or {}).get("set_default") or "" ) set_default = str(set_default_raw).lower() in {"1", "true", "yes", "on"} if not workspace_path: return jsonify({"success": False, "error": "缺少 path"}), 400 try: result = create_host_workspace(workspace_path, label=label, set_default=set_default) except Exception as exc: return jsonify({"success": False, "error": str(exc)}), 500 catalog = result.get("catalog") or {} workspaces = [] current_id = (session.get("host_workspace_id") or "").strip() for item in catalog.get("workspaces") or []: workspaces.append({ "workspace_id": item.get("workspace_id"), "label": item.get("label"), "path": item.get("path"), "is_current": item.get("workspace_id") == current_id, }) return jsonify({ "success": True, "data": { "created": bool(result.get("created")), "workspace": result.get("workspace") or {}, "source_path": catalog.get("source_path"), "default_workspace_id": catalog.get("default_workspace_id"), "current_workspace_id": current_id, "workspaces": workspaces, } }) @status_bp.route('/api/host/workspaces/rename', methods=['POST']) @api_login_required def rename_host_workspace_api(): if not _is_host_mode_request(): return jsonify({"success": False, "error": "仅宿主机模式可用"}), 403 payload = request.get_json(silent=True) or {} workspace_id = (payload.get("workspace_id") or "").strip() label = (payload.get("label") or payload.get("name") or "").strip() if not workspace_id: return jsonify({"success": False, "error": "缺少 workspace_id"}), 400 if not label: return jsonify({"success": False, "error": "工作区名称不能为空"}), 400 try: result = rename_host_workspace(workspace_id, label) except ValueError as exc: return jsonify({"success": False, "error": str(exc)}), 400 except Exception as exc: return jsonify({"success": False, "error": str(exc)}), 500 catalog = result.get("catalog") or load_host_workspace_catalog() current_id = (session.get("host_workspace_id") or "").strip() return jsonify({ "success": True, "data": { "workspace": result.get("workspace") or {}, "source_path": catalog.get("source_path"), "default_workspace_id": catalog.get("default_workspace_id"), "current_workspace_id": current_id, "workspaces": _build_host_workspaces_payload(catalog, current_id), } }) @status_bp.route('/api/host/workspaces/delete', methods=['POST']) @api_login_required def delete_host_workspace_api(): if not _is_host_mode_request(): return jsonify({"success": False, "error": "仅宿主机模式可用"}), 403 payload = request.get_json(silent=True) or {} workspace_id = (payload.get("workspace_id") or "").strip() if not workspace_id: return jsonify({"success": False, "error": "缺少 workspace_id"}), 400 if _active_task_counts("host").get(workspace_id): return jsonify({"success": False, "error": "该工作区有运行中的任务，暂不能删除"}), 409 try: _close_terminal_for_key(f"host::{workspace_id}") container_manager.release_container(f"host::{workspace_id}", reason="delete_host_workspace") result = delete_host_workspace(workspace_id) except ValueError as exc: return jsonify({"success": False, "error": str(exc)}), 400 except Exception as exc: return jsonify({"success": False, "error": str(exc)}), 500 catalog = result.get("catalog") or load_host_workspace_catalog() current_id = (session.get("host_workspace_id") or "").strip() if current_id == workspace_id or not any( item.get("workspace_id") == current_id for item in (catalog.get("workspaces") or []) ): _, current = resolve_host_workspace(catalog.get("default_workspace_id")) current_id = current.get("workspace_id") or catalog.get("default_workspace_id") or "default" current_path = str(Path(current.get("path") or "").expanduser().resolve()) session["host_workspace_id"] = current_id session["workspace_id"] = current_id with state.HOST_ACTIVE_WORKSPACE_LOCK: state.HOST_ACTIVE_WORKSPACE_ID = current_id state.HOST_ACTIVE_WORKSPACE_PATH = current_path state.HOST_ACTIVE_WORKSPACE_VERSION += 1 return jsonify({ "success": True, "data": { "deleted_workspace_id": workspace_id, "source_path": catalog.get("source_path"), "default_workspace_id": catalog.get("default_workspace_id"), "current_workspace_id": current_id, "workspaces": _build_host_workspaces_payload(catalog, current_id), } }) @status_bp.route('/api/container-status') @api_login_required @with_terminal def get_container_status_api(terminal, workspace, username): try: container_key = ( f"host::{getattr(workspace, 'workspace_id', None) or session.get('workspace_id') or 'default'}" if bool(session.get("host_mode")) or username == "host" else f"{username}::{getattr(workspace, 'workspace_id', None) or session.get('workspace_id') or 'default'}" ) status = container_manager.get_container_status(container_key) return jsonify({"success": True, "data": status}) except Exception as exc: return jsonify({"success": False, "error": str(exc)}), 500 @status_bp.route('/api/project-storage') @api_login_required @with_terminal def get_project_storage(terminal, workspace, username): now = time.time() workspace_id = getattr(workspace, "workspace_id", None) or session.get("workspace_id") or "default" cache_key = f"{username}::{workspace_id}" cache_entry = PROJECT_STORAGE_CACHE.get(cache_key) if cache_entry and (now - cache_entry.get("ts", 0)) < PROJECT_STORAGE_CACHE_TTL_SECONDS: return jsonify({"success": True, "data": cache_entry["data"]}) try: file_manager = getattr(terminal, 'file_manager', None) if not file_manager: return jsonify({"success": False, "error": "文件管理器未初始化"}), 500 used_bytes = file_manager._get_project_size() limit_bytes = PROJECT_MAX_STORAGE_MB * 1024 * 1024 if PROJECT_MAX_STORAGE_MB else None usage_percent = (used_bytes / limit_bytes * 100) if limit_bytes else None data = { "used_bytes": used_bytes, "limit_bytes": limit_bytes, "limit_label": f"{PROJECT_MAX_STORAGE_MB}MB" if PROJECT_MAX_STORAGE_MB else "未限制", "usage_percent": usage_percent } PROJECT_STORAGE_CACHE[cache_key] = {"ts": now, "data": data} return jsonify({"success": True, "data": data}) except Exception as exc: stale = PROJECT_STORAGE_CACHE.get(cache_key) if stale: return jsonify({"success": True, "data": stale.get("data"), "stale": True}), 200 return jsonify({"success": False, "error": str(exc)}), 500 @status_bp.route('/api/app/version') @api_login_required def get_app_version_info(): project_root = Path(__file__).resolve().parent.parent version_code, version_name = _parse_android_version_from_gradle(project_root) apk_path = _resolve_android_apk_path(project_root) apk_exists = apk_path.exists() file_size = apk_path.stat().st_size if apk_exists else 0 published_at = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(apk_path.stat().st_mtime)) if apk_exists else None changelog = _load_app_changelog(project_root) apk_url = request.url_root.rstrip("/") + "/api/app/apk/latest" current_code_raw = request.args.get("currentVersionCode", "").strip() current_code = int(current_code_raw) if current_code_raw.isdigit() else None has_update = (version_code > current_code) if current_code is not None else None return jsonify({ "success": True, "data": { "latestVersionCode": version_code, "latestVersionName": version_name, "apkUrl": apk_url, "apkSha256": "", "fileSizeBytes": file_size, "changelog": changelog, "forceUpdate": False, "publishedAt": published_at, "minSupportedVersionCode": 1, "apkExists": apk_exists, "hasUpdate": has_update, } }) @status_bp.route('/api/app/apk/latest') def download_latest_apk(): project_root = Path(__file__).resolve().parent.parent apk_path = _resolve_android_apk_path(project_root) if not apk_path.exists(): return jsonify({"success": False, "error": "APK 不存在，请先构建 release 包"}), 404 return send_file( apk_path, as_attachment=True, download_name="cyj-agent-latest.apk", mimetype="application/vnd.android.package-archive", conditional=False, etag=False, max_age=0, )