185 lines
7.4 KiB
Python
185 lines
7.4 KiB
Python
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')
|
||
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,
|
||
)
|