from __future__ import annotations import time import re from pathlib import Path from flask import Blueprint, jsonify, request, send_file from .auth_helpers import api_login_required, resolve_admin_policy from .context import with_terminal from .state import ( PROJECT_STORAGE_CACHE, PROJECT_STORAGE_CACHE_TTL_SECONDS, PROJECT_MAX_STORAGE_MB, container_manager, user_manager, ) from config import AGENT_VERSION 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 "" @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/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, )