from __future__ import annotations import time import re import os 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 @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/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, )