from __future__ import annotations import time import re 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, load_host_workspace_catalog, resolve_host_workspace, ) from utils.host_workspace_debug import write_host_workspace_debug from . import state status_bp = Blueprint('status', __name__) 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" @status_bp.route('/api/status') @api_login_required @with_terminal def get_status(terminal, workspace, username): """获取系统状态（包含对话、容器、版本等信息）""" status = terminal.get_status() if terminal.terminal_manager: status['terminals'] = terminal.terminal_manager.list_terminals() 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: print(f"[Status] 获取当前对话信息失败: {exc}") status['project_path'] = str(workspace.project_path) try: status['container'] = container_manager.get_container_status(username) except Exception as exc: status['container'] = {"success": False, "error": str(exc)} status['version'] = AGENT_VERSION 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 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 = [] 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": { "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/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, ) if current_id == workspace_id: session['workspace_id'] = workspace_id try: container_handle = state.container_manager.ensure_container( "host", target_path, container_key="host", preferred_mode="host", ) host_terminal = state.user_terminals.get("host") if host_terminal: host_terminal.update_project_path(target_path) 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.same_id_reapply", 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", "")), ) except Exception: pass return jsonify({ "success": True, "data": { "current_workspace_id": workspace_id, "project_path": target_path, "default_workspace_id": catalog.get("default_workspace_id"), "reloaded": False, } }) try: from .tasks import task_manager running = [ rec for rec in task_manager.list_tasks("host") if rec.status in {"pending", "running", "cancel_requested"} ] if running: return jsonify({ "success": False, "error": "存在运行中的任务，请先停止后再切换工作区", }), 409 except Exception: pass previous_workspace_id = session.get("workspace_id") session['host_workspace_id'] = workspace_id session['workspace_id'] = workspace_id try: container_handle = state.container_manager.ensure_container( "host", target_path, container_key="host", 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("host") if host_terminal: try: host_terminal.update_project_path(target_path) 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("host", 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": True, } }) @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/container-status') @api_login_required @with_terminal def get_container_status_api(terminal, workspace, username): try: status = container_manager.get_container_status(username) 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() cache_entry = PROJECT_STORAGE_CACHE.get(username) 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[username] = {"ts": now, "data": data} return jsonify({"success": True, "data": data}) except Exception as exc: stale = PROJECT_STORAGE_CACHE.get(username) 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, )