IP' not in dashboard_html:
- dashboard_html = dashboard_html.replace(header_old, header_new).replace(name_cell, name_ip_cell).replace('colspan="4" class="muted">Виртуальных машин пока нет', 'colspan="5" class="muted">Виртуальных машин пока нет')
- changed.append('Dashboard VM IP column was added')
-elif 'vm.ip' not in dashboard_html:
- dashboard_html = dashboard_html.replace(name_cell, name_ip_cell)
- changed.append('Dashboard VM IP cell was added')
-else:
- changed.append('Dashboard VM IP column was already present')
-dashboard_template_path.write_text(dashboard_html, encoding='utf-8')
-
-if network_template_path.exists():
- network_html = network_template_path.read_text(encoding='utf-8')
- manual_card = '''
-
-
-
Ручные IP VM
- override
-
-
Если VM в bridge/static-сети и IP не виден через DHCP/ARP, укажи адрес вручную. Этот IP будет первым источником для таблицы и проброса портов.
-
-
VM
Текущий IP
Ручной IP
Действие
-
- {% for vm in vms %}
-
-
{{ vm.name }}
-
{{ vm.ip|default("—") }}
-
-
-
-
- {% else %}
-
VM пока нет
- {% endfor %}
-
-
-
-'''
- if 'Ручные IP VM' not in network_html:
- marker = ' \n
\n
Диагностика публичного доступа
'
- if marker in network_html:
- network_html = network_html.replace(marker, manual_card + '\n' + marker, 1)
- changed.append('manual VM IP card was added to network page')
- else:
- warnings.append('network diagnostics card marker not found, manual IP card skipped')
- else:
- changed.append('manual VM IP card was already present')
- network_template_path.write_text(network_html, encoding='utf-8')
-else:
- warnings.append('network.html not found, manual IP card skipped')
-
-print('DHCP leases empty-state patch completed:')
-for item in changed:
- print(f'- {item}')
-for item in warnings:
- print(f'WARN: {item}')
-
-raise SystemExit(0)
diff --git a/scripts/patch_disk_archives.py b/scripts/patch_disk_archives.py
deleted file mode 100644
index 93938a7..0000000
--- a/scripts/patch_disk_archives.py
+++ /dev/null
@@ -1,316 +0,0 @@
-#!/usr/bin/env python3
-from pathlib import Path
-import sys
-
-app_path = Path(sys.argv[1]) if len(sys.argv) > 1 else Path('/opt/virtuality/web/app.py')
-if not app_path.exists():
- raise SystemExit(f'app.py not found: {app_path}')
-
-text = app_path.read_text()
-changed = []
-
-if 'import zipfile' not in text:
- text = text.replace('import uuid\n', 'import uuid\nimport lzma\nimport tarfile\nimport zipfile\n', 1)
- changed.append('archive imports added')
-elif 'import lzma' not in text:
- text = text.replace('import tarfile\n', 'import lzma\nimport tarfile\n', 1) if 'import tarfile\n' in text else text.replace('import zipfile\n', 'import lzma\nimport zipfile\n', 1)
- changed.append('lzma import added')
-else:
- changed.append('archive imports already present')
-
-helpers = r'''
-
-def safe_disk_upload_filename(filename: str) -> str | None:
- original = Path(filename or "").name.strip()
- lower = original.lower()
- if lower.endswith(".tar.gz"):
- suffix = ".tar.gz"
- stem = original[:-7]
- elif lower.endswith(".img.xz"):
- suffix = ".img.xz"
- stem = original[:-7]
- elif lower.endswith(".tgz"):
- suffix = ".tgz"
- stem = original[:-4]
- else:
- suffix = Path(original).suffix.lower()
- stem = Path(original).stem
- if suffix not in (".img", ".raw", ".qcow2", ".img.xz", ".zip", ".tar.gz", ".tgz"):
- return None
- stem = re.sub(r"\s+", "-", stem.strip())
- stem = re.sub(r"[^a-zA-Z0-9_.-]", "_", stem)
- stem = stem.strip("._-")[:120]
- if not stem:
- stem = "virtuality-disk"
- return f"{stem}{suffix}"
-
-
-def disk_upload_is_archive(name: str) -> bool:
- lower = str(name or "").lower()
- return lower.endswith((".zip", ".tar.gz", ".tgz"))
-
-
-def disk_upload_is_xz_image(name: str) -> bool:
- return str(name or "").lower().endswith(".img.xz")
-
-
-def disk_upload_is_image(name: str) -> bool:
- lower = str(name or "").lower()
- return lower.endswith((".img", ".raw", ".qcow2", ".img.xz"))
-
-
-def archive_member_is_safe(name: str) -> bool:
- if not name:
- return False
- p = Path(name)
- if p.is_absolute():
- return False
- return ".." not in p.parts
-
-
-def archive_member_basename(name: str) -> str | None:
- return safe_disk_image_filename(Path(name).name)
-
-
-def unique_disk_image_path(name: str) -> Path:
- DISK_IMAGES_DIR.mkdir(parents=True, exist_ok=True)
- safe = safe_disk_image_filename(name)
- if not safe:
- raise ValueError("Некорректное имя образа диска")
- target = DISK_IMAGES_DIR / safe
- if not target.exists():
- return target
- stem = target.stem
- suffix = target.suffix
- for index in range(1, 1000):
- candidate = DISK_IMAGES_DIR / f"{stem}-{index}{suffix}"
- if not candidate.exists():
- return candidate
- raise ValueError("Не удалось подобрать свободное имя файла")
-
-
-def find_archive_disk_members(archive_path: Path) -> list[dict[str, Any]]:
- members: list[dict[str, Any]] = []
- lower = archive_path.name.lower()
- if lower.endswith(".zip"):
- with zipfile.ZipFile(archive_path) as archive:
- for info in archive.infolist():
- if info.is_dir() or not archive_member_is_safe(info.filename):
- continue
- safe_name = archive_member_basename(info.filename)
- if safe_name:
- members.append({"kind": "zip", "name": info.filename, "safe_name": safe_name, "size": int(info.file_size)})
- elif lower.endswith((".tar.gz", ".tgz")):
- with tarfile.open(archive_path, "r:gz") as archive:
- for info in archive.getmembers():
- if not info.isfile() or not archive_member_is_safe(info.name):
- continue
- safe_name = archive_member_basename(info.name)
- if safe_name:
- members.append({"kind": "tar", "name": info.name, "safe_name": safe_name, "size": int(info.size)})
- else:
- raise ValueError("Поддерживаются только .zip, .tar.gz и .tgz")
- return sorted(members, key=lambda item: item.get("size", 0), reverse=True)
-
-
-def extract_disk_archive(archive_path: Path) -> list[Path]:
- members = find_archive_disk_members(archive_path)
- if not members:
- raise ValueError("В архиве не найдено .img, .raw или .qcow2 файлов")
- extracted: list[Path] = []
- lower = archive_path.name.lower()
- if lower.endswith(".zip"):
- with zipfile.ZipFile(archive_path) as archive:
- for member in members:
- target = unique_disk_image_path(member["safe_name"])
- with archive.open(member["name"], "r") as src, target.open("wb") as dst:
- shutil.copyfileobj(src, dst, length=1024 * 1024)
- extracted.append(target)
- else:
- with tarfile.open(archive_path, "r:gz") as archive:
- for member in members:
- file_obj = archive.extractfile(member["name"])
- if file_obj is None:
- continue
- target = unique_disk_image_path(member["safe_name"])
- with file_obj as src, target.open("wb") as dst:
- shutil.copyfileobj(src, dst, length=1024 * 1024)
- extracted.append(target)
- return extracted
-
-
-def extract_xz_disk_image(compressed_path: Path, safe_name: str | None = None) -> Path:
- source_name = safe_name or compressed_path.name
- if not source_name.lower().endswith(".img.xz"):
- raise ValueError("Поддерживаются только сжатые образы .img.xz")
- raw_name = source_name[:-3]
- target = unique_disk_image_path(raw_name)
- try:
- with lzma.open(compressed_path, "rb") as src, target.open("wb") as dst:
- shutil.copyfileobj(src, dst, length=1024 * 1024)
- except Exception:
- target.unlink(missing_ok=True)
- raise
- return target
-
-
-def disk_convert_target_path(source_path: Path) -> Path:
- if source_path.suffix.lower() == ".qcow2":
- return source_path
- target = source_path.with_suffix(".qcow2")
- if not target.exists():
- return target
- for index in range(1, 1000):
- candidate = source_path.with_name(f"{source_path.stem}-{index}.qcow2")
- if not candidate.exists():
- return candidate
- raise ValueError("Не удалось подобрать имя qcow2 для конвертации")
-
-
-def disk_convert_progress(line: str, current: int) -> int:
- match = re.search(r"\((\d+(?:\.\d+)?)/100%\)", line or "")
- if match:
- return max(current, int(float(match.group(1))))
- match = re.search(r"(\d+(?:\.\d+)?)%", line or "")
- if match:
- return max(current, int(float(match.group(1))))
- return current
-
-
-def run_disk_convert_worker(operation_id: str, source_path: str, target_path: str) -> None:
- operation = read_operation(operation_id)
- if not operation:
- return
- source = Path(source_path)
- target = Path(target_path)
- update_operation(operation, status="running", progress=1, message=f"Конвертация {source.name} в qcow2", started_at=utc_now())
- cmd = ["qemu-img", "convert", "-p", "-f", disk_image_format(source), "-O", "qcow2", str(source), str(target)]
- append_operation_log(operation_id, "Запуск конвертации:")
- append_operation_log(operation_id, " ".join(cmd))
- try:
- process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1)
- progress = 1
- if process.stdout:
- for line in process.stdout:
- append_operation_log(operation_id, line)
- progress = disk_convert_progress(line, progress)
- fresh = read_operation(operation_id) or operation
- update_operation(fresh, progress=progress, message=f"Конвертация {source.name}: {progress}%")
- exit_code = process.wait()
- fresh = read_operation(operation_id) or operation
- if exit_code == 0:
- append_operation_log(operation_id, f"Конвертация завершена: {target}")
- update_operation(fresh, status="success", progress=100, exit_code=exit_code, message=f"Готово: {target.name}", finished_at=utc_now(), target_path=str(target))
- else:
- append_operation_log(operation_id, f"qemu-img завершился с ошибкой. Exit code: {exit_code}")
- target.unlink(missing_ok=True)
- update_operation(fresh, status="error", progress=100, exit_code=exit_code, message=f"qemu-img завершился с ошибкой: {exit_code}", finished_at=utc_now())
- except Exception as exc:
- fresh = read_operation(operation_id) or operation
- append_operation_log(operation_id, f"Ошибка конвертации: {exc}")
- target.unlink(missing_ok=True)
- update_operation(fresh, status="error", progress=100, exit_code=-1, message=str(exc), finished_at=utc_now())
-
-
-def start_disk_convert_operation(source_path: Path) -> dict[str, Any] | None:
- if source_path.suffix.lower() == ".qcow2":
- return None
- target_path = disk_convert_target_path(source_path)
- operation_id = str(uuid.uuid4())
- operation = {
- "id": operation_id,
- "type": "disk_convert",
- "title": f"Конвертация {source_path.name}",
- "status": "queued",
- "progress": 0,
- "message": "Конвертация поставлена в очередь",
- "created_at": utc_now(),
- "updated_at": utc_now(),
- "created_by": AUTH_USER,
- "source_path": str(source_path),
- "target_path": str(target_path),
- }
- write_operation(operation)
- append_operation_log(operation_id, "Операция поставлена в очередь.")
- threading.Thread(target=run_disk_convert_worker, args=(operation_id, str(source_path), str(target_path)), daemon=True).start()
- return operation
-
-
-def disk_upload_response(request: Request, payload: dict[str, Any]):
- if request.headers.get("x-requested-with") == "XMLHttpRequest" or "application/json" in request.headers.get("accept", ""):
- return JSONResponse(payload)
- return RedirectResponse(url="/disk-images", status_code=303)
-'''
-
-if 'def safe_disk_upload_filename(' not in text:
- marker = '\n\ndef valid_vm_name(name: str) -> bool:'
- if marker not in text:
- marker = '\n\ndef bridge_exists(name: str) -> bool:'
- if marker not in text:
- raise SystemExit('helper insert marker not found')
- text = text.replace(marker, helpers + marker, 1)
- changed.append('disk archive helpers added')
-else:
- pattern = r"\n\ndef safe_disk_upload_filename\(filename: str\).*?\n\ndef disk_upload_response\(request: Request, payload: dict\[str, Any\]\):.*?\n return RedirectResponse\(url=\"/disk-images\", status_code=303\)\n"
- text, count = re.subn(pattern, helpers, text, count=1, flags=re.S)
- changed.append('disk archive helpers replaced with img.xz support' if count else 'disk archive helpers already present')
-
-old_route_start = text.find('@app.post("/disk-images/upload"')
-if old_route_start == -1:
- raise SystemExit('disk image upload route not found')
-next_route = text.find('\n\n@app.post("/disk-images/{name}/delete")', old_route_start)
-if next_route == -1:
- raise SystemExit('disk image delete route marker not found')
-new_route = r'''@app.post("/disk-images/upload", response_class=HTMLResponse)
-def disk_image_upload(request: Request, image_file: UploadFile = File(...)):
- auth_redirect = require_auth(request)
- if auth_redirect:
- return auth_redirect
- safe_name = safe_disk_upload_filename(image_file.filename or "")
- if not safe_name:
- return templates.TemplateResponse("disk_images.html", {"request": request, "app_name": APP_NAME, "user": AUTH_USER, "images": list_disk_image_files(), "error": "Можно загружать только .img, .raw, .qcow2, .img.xz, .zip, .tar.gz или .tgz файлы."}, status_code=400)
- DISK_IMAGES_DIR.mkdir(parents=True, exist_ok=True)
- tmp_target = DISK_IMAGES_DIR / f".{safe_name}.uploading"
- saved_paths: list[Path] = []
- try:
- with tmp_target.open("wb") as out:
- shutil.copyfileobj(image_file.file, out, length=1024 * 1024)
- if disk_upload_is_archive(safe_name):
- saved_paths = extract_disk_archive(tmp_target)
- tmp_target.unlink(missing_ok=True)
- elif disk_upload_is_xz_image(safe_name):
- saved_paths = [extract_xz_disk_image(tmp_target, safe_name)]
- tmp_target.unlink(missing_ok=True)
- else:
- target = unique_disk_image_path(safe_name)
- tmp_target.rename(target)
- saved_paths = [target]
- except Exception as exc:
- tmp_target.unlink(missing_ok=True)
- for path in saved_paths:
- path.unlink(missing_ok=True)
- return templates.TemplateResponse("disk_images.html", {"request": request, "app_name": APP_NAME, "user": AUTH_USER, "images": list_disk_image_files(), "error": f"Ошибка загрузки/распаковки образа: {exc}"}, status_code=500)
-
- operations = []
- for path in saved_paths:
- operation = start_disk_convert_operation(path)
- if operation:
- operations.append(operation)
-
- payload = {
- "ok": True,
- "mode": "converting" if operations else "ready",
- "operation_id": operations[0]["id"] if operations else None,
- "operation_ids": [op["id"] for op in operations],
- "files": [path.name for path in saved_paths],
- "message": f"Загружено файлов: {len(saved_paths)}. Конвертаций запущено: {len(operations)}.",
- }
- return disk_upload_response(request, payload)
-'''
-text = text[:old_route_start] + new_route + text[next_route:]
-changed.append('disk image upload route supports img.xz and archives')
-
-app_path.write_text(text)
-print('disk archive import patch applied:')
-for item in changed:
- print(f'- {item}')
diff --git a/scripts/patch_disk_convert_progress.py b/scripts/patch_disk_convert_progress.py
deleted file mode 100644
index 441810a..0000000
--- a/scripts/patch_disk_convert_progress.py
+++ /dev/null
@@ -1,171 +0,0 @@
-#!/usr/bin/env python3
-from pathlib import Path
-import sys
-
-app_path = Path(sys.argv[1]) if len(sys.argv) > 1 else Path('/opt/virtuality/web/app.py')
-if not app_path.exists():
- print(f'WARN: app.py not found: {app_path}')
- raise SystemExit(0)
-
-text = app_path.read_text()
-changed = []
-warnings = []
-
-old_progress = '''def progress_from_line(current: int, line: str) -> int:
- low = line.lower()
- if "allocating" in low or "creating storage" in low:
- return max(current, 30)
- if "starting install" in low or "installing" in low:
- return max(current, 50)
- if "creating domain" in low:
- return max(current, 75)
- if "domain creation completed" in low or "installation continues" in low:
- return max(current, 90)
- return current
-'''
-new_progress = '''def progress_from_line(current: int, line: str) -> int:
- low = line.lower()
- convert_match = re.search(r"\\((\\d+(?:\\.\\d+)?)\\s*/\\s*100%\\)", line)
- if convert_match:
- return max(1, min(99, int(float(convert_match.group(1)))))
- percent_match = re.search(r"(\\d+(?:\\.\\d+)?)%", line)
- if "qemu-img" in low and percent_match:
- return max(1, min(99, int(float(percent_match.group(1)))))
- if "allocating" in low or "creating storage" in low:
- return max(current, 30)
- if "starting install" in low or "installing" in low:
- return max(current, 50)
- if "creating domain" in low:
- return max(current, 75)
- if "domain creation completed" in low or "installation continues" in low:
- return max(current, 90)
- return current
-'''
-if old_progress in text:
- text = text.replace(old_progress, new_progress, 1)
- changed.append('qemu-img percent parser added')
-elif 'convert_match = re.search' in text:
- changed.append('qemu-img percent parser already present')
-else:
- warnings.append('progress_from_line marker not found, progress parser skipped')
-
-helpers = r'''
-
-def qemu_img_format_for_path(path: Path) -> str:
- suffix = path.suffix.lower()
- if suffix == ".qcow2":
- return "qcow2"
- return "raw"
-
-
-def qcow2_name_for_upload(safe_name: str) -> str:
- base = Path(safe_name).stem
- return f"{base}.qcow2"
-
-
-def start_disk_image_convert_operation(source_path: Path, target_path: Path, source_format: str) -> dict[str, Any]:
- operation_id = str(uuid.uuid4())
- cmd = ["qemu-img", "convert", "-p", "-f", source_format, "-O", "qcow2", str(source_path), str(target_path)]
- operation = {
- "id": operation_id,
- "type": "disk_image_convert",
- "title": f"Конвертация образа {source_path.name}",
- "status": "queued",
- "progress": 0,
- "message": "Ожидание конвертации образа",
- "created_at": utc_now(),
- "updated_at": utc_now(),
- "created_by": AUTH_USER,
- "source_path": str(source_path),
- "target_path": str(target_path),
- "source_format": source_format,
- "cmd": " ".join(cmd),
- }
- start_background_operation(operation, cmd)
- return operation
-'''
-if 'def start_disk_image_convert_operation(' not in text:
- marker = '\n\ndef valid_vm_name(name: str) -> bool:'
- if marker in text:
- text = text.replace(marker, helpers + marker, 1)
- changed.append('disk conversion operation helpers added')
- else:
- warnings.append('valid_vm_name marker not found, conversion helpers skipped')
-else:
- changed.append('disk conversion operation helpers already present')
-
-old_upload = r'''@app.post("/disk-images/upload", response_class=HTMLResponse)
-def disk_image_upload(request: Request, image_file: UploadFile = File(...)):
- auth_redirect = require_auth(request)
- if auth_redirect:
- return auth_redirect
- safe_name = safe_disk_image_filename(image_file.filename or "")
- if not safe_name:
- return templates.TemplateResponse("disk_images.html", {"request": request, "app_name": APP_NAME, "user": AUTH_USER, "images": list_disk_image_files(), "error": "Можно загружать только .img, .raw или .qcow2 файлы с безопасным именем."}, status_code=400)
- DISK_IMAGES_DIR.mkdir(parents=True, exist_ok=True)
- target = DISK_IMAGES_DIR / safe_name
- if target.exists():
- return templates.TemplateResponse("disk_images.html", {"request": request, "app_name": APP_NAME, "user": AUTH_USER, "images": list_disk_image_files(), "error": f"Образ уже существует: {safe_name}"}, status_code=400)
- tmp_target = DISK_IMAGES_DIR / f".{safe_name}.uploading"
- try:
- with tmp_target.open("wb") as out:
- shutil.copyfileobj(image_file.file, out)
- tmp_target.rename(target)
- except Exception as exc:
- tmp_target.unlink(missing_ok=True)
- return templates.TemplateResponse("disk_images.html", {"request": request, "app_name": APP_NAME, "user": AUTH_USER, "images": list_disk_image_files(), "error": f"Ошибка загрузки образа: {exc}"}, status_code=500)
- return RedirectResponse(url="/disk-images", status_code=303)
-'''
-new_upload = r'''@app.post("/disk-images/upload")
-def disk_image_upload(request: Request, image_file: UploadFile = File(...)):
- auth_redirect = require_auth(request)
- if auth_redirect:
- return auth_redirect
- safe_name = safe_disk_image_filename(image_file.filename or "")
- wants_json = "application/json" in request.headers.get("accept", "") or request.headers.get("x-requested-with") == "XMLHttpRequest"
-
- def disk_error(message: str, status_code: int = 400):
- if wants_json:
- return JSONResponse({"ok": False, "error": message}, status_code=status_code)
- return templates.TemplateResponse("disk_images.html", {"request": request, "app_name": APP_NAME, "user": AUTH_USER, "images": list_disk_image_files(), "error": message}, status_code=status_code)
-
- if not safe_name:
- return disk_error("Можно загружать только .img, .raw или .qcow2 файлы с безопасным именем.")
- DISK_IMAGES_DIR.mkdir(parents=True, exist_ok=True)
- target = DISK_IMAGES_DIR / safe_name
- final_qcow2 = DISK_IMAGES_DIR / qcow2_name_for_upload(safe_name)
- if target.exists() or final_qcow2.exists():
- return disk_error(f"Образ уже существует: {safe_name}")
- tmp_target = DISK_IMAGES_DIR / f".{safe_name}.uploading"
- try:
- with tmp_target.open("wb") as out:
- shutil.copyfileobj(image_file.file, out)
- tmp_target.rename(target)
- if target.suffix.lower() == ".qcow2":
- payload = {"ok": True, "mode": "ready", "name": target.name, "redirect": "/disk-images"}
- else:
- operation = start_disk_image_convert_operation(target, final_qcow2, qemu_img_format_for_path(target))
- payload = {"ok": True, "mode": "converting", "name": target.name, "target": final_qcow2.name, "operation_id": operation["id"], "redirect": "/disk-images"}
- except Exception as exc:
- tmp_target.unlink(missing_ok=True)
- return disk_error(f"Ошибка загрузки образа: {exc}", status_code=500)
-
- if wants_json:
- return JSONResponse(payload)
- return RedirectResponse(url="/disk-images", status_code=303)
-'''
-if old_upload in text:
- text = text.replace(old_upload, new_upload, 1)
- changed.append('disk upload returns JSON and starts conversion operation')
-elif 'mode": "converting"' in text and ('start_disk_image_convert_operation' in text or 'start_disk_convert_operation' in text):
- changed.append('disk upload conversion already present')
-else:
- warnings.append('disk_image_upload marker not found, upload conversion patch skipped')
-
-app_path.write_text(text)
-print('disk convert progress patch completed:')
-for item in changed:
- print(f'- {item}')
-for item in warnings:
- print(f'WARN: {item}')
-raise SystemExit(0)
diff --git a/scripts/patch_disk_images.py b/scripts/patch_disk_images.py
deleted file mode 100644
index 03688f3..0000000
--- a/scripts/patch_disk_images.py
+++ /dev/null
@@ -1,298 +0,0 @@
-#!/usr/bin/env python3
-from pathlib import Path
-import re
-import sys
-
-app_path = Path(sys.argv[1]) if len(sys.argv) > 1 else Path('/opt/virtuality/web/app.py')
-if not app_path.exists():
- raise SystemExit(f'app.py not found: {app_path}')
-
-text = app_path.read_text()
-changed = []
-warnings = []
-
-if 'DISK_IMAGES_DIR = Path("/var/lib/virtuality/disk-images")' not in text:
- text = text.replace('IMAGES_DIR = Path("/var/lib/virtuality/images")\n', 'IMAGES_DIR = Path("/var/lib/virtuality/images")\nDISK_IMAGES_DIR = Path("/var/lib/virtuality/disk-images")\n', 1)
- changed.append('DISK_IMAGES_DIR added')
-else:
- changed.append('DISK_IMAGES_DIR already present')
-
-helpers = r'''
-
-def vm_arch_options() -> list[dict[str, str]]:
- return [
- {"value": "auto", "label": "Auto — по профилю хоста"},
- {"value": "x86_64", "label": "x86_64 / amd64"},
- {"value": "aarch64", "label": "ARM64 / aarch64"},
- {"value": "generic", "label": "Generic / no arch override"},
- ]
-
-
-def normalize_guest_arch(value: str, profile: dict[str, Any]) -> str:
- value = (value or "auto").strip()
- if value == "auto":
- return str(profile.get("recommended_guest_arch") or "x86_64")
- if value in ("x86_64", "amd64"):
- return "x86_64"
- if value in ("aarch64", "arm64"):
- return "aarch64"
- if value == "generic":
- return "generic"
- return str(profile.get("recommended_guest_arch") or "x86_64")
-
-
-def list_disk_image_files() -> list[dict[str, str]]:
- DISK_IMAGES_DIR.mkdir(parents=True, exist_ok=True)
- files = []
- for item in sorted(list(DISK_IMAGES_DIR.glob("*.img")) + list(DISK_IMAGES_DIR.glob("*.raw")) + list(DISK_IMAGES_DIR.glob("*.qcow2"))):
- try:
- stat = item.stat()
- size_gb = stat.st_size / 1024 / 1024 / 1024
- updated = datetime.fromtimestamp(stat.st_mtime).strftime("%Y-%m-%d %H:%M")
- except OSError:
- size_gb = 0
- updated = "unknown"
- files.append({"name": item.name, "path": str(item), "format": item.suffix.lower().lstrip('.'), "size": f"{size_gb:.2f} GB", "updated": updated})
- return files
-
-
-def safe_disk_image_filename(filename: str) -> str | None:
- name = Path(filename or "").name.strip().replace(" ", "-")
- name = re.sub(r"[^a-zA-Z0-9_.-]", "_", name)
- if not name.lower().endswith((".img", ".raw", ".qcow2")):
- return None
- if not re.fullmatch(r"[a-zA-Z0-9][a-zA-Z0-9_.-]{1,180}\.(img|raw|qcow2)", name, re.IGNORECASE):
- return None
- return name
-
-
-def disk_image_path_by_name(name: str) -> Path | None:
- safe_name = safe_disk_image_filename(name)
- if not safe_name:
- return None
- path = (DISK_IMAGES_DIR / safe_name).resolve()
- if DISK_IMAGES_DIR.resolve() not in path.parents:
- return None
- return path
-
-
-def disk_image_format(path: Path) -> str:
- suffix = path.suffix.lower().lstrip('.')
- if suffix == 'qcow2':
- return 'qcow2'
- return 'raw'
-
-
-def bridge_exists(name: str) -> bool:
- if not name or not re.fullmatch(r"[a-zA-Z0-9_.:-]+", name):
- return False
- return run_cmd(["ip", "link", "show", name], timeout=5)["ok"]
-'''
-
-if 'def list_disk_image_files() -> list[dict[str, str]]:' not in text:
- marker = '\n\ndef valid_vm_name(name: str) -> bool:'
- if marker not in text:
- raise SystemExit('valid_vm_name marker not found')
- text = text.replace(marker, helpers + marker, 1)
- changed.append('disk image and architecture helpers added')
-else:
- if 'def vm_arch_options()' not in text:
- insert = helpers.split('\n\ndef list_disk_image_files()', 1)[0]
- marker = '\n\ndef list_disk_image_files() -> list[dict[str, str]]:'
- text = text.replace(marker, insert + marker, 1)
- changed.append('architecture helpers added')
- else:
- changed.append('architecture helpers already present')
- changed.append('disk image helpers already present')
-
-text = text.replace('\n\ndef bridge_exists(name: str) -> bool:\n if not name or not re.fullmatch(r"[a-zA-Z0-9_.:-]+", name):\n return False\n return run_cmd(["ip", "link", "show", name], timeout=5)["ok"]\n\n\ndef bridge_exists(name: str) -> bool:\n if not name or not re.fullmatch(r"[a-zA-Z0-9_.:-]+", name):\n return False\n return run_cmd(["ip", "link", "show", name], timeout=5)["ok"]\n', '\n\ndef bridge_exists(name: str) -> bool:\n if not name or not re.fullmatch(r"[a-zA-Z0-9_.:-]+", name):\n return False\n return run_cmd(["ip", "link", "show", name], timeout=5)["ok"]\n')
-
-old_context = '"isos": list_iso_files(), "error": error, "profile": profile, "form": form or {"memory": 2048, "vcpus": 2, "disk_size": 20, "network_mode": default_mode, "bridge": DEFAULT_BRIDGE}}'
-new_context = '"isos": list_iso_files(), "disk_images": list_disk_image_files(), "arch_options": vm_arch_options(), "error": error, "profile": profile, "form": form or {"memory": 2048, "vcpus": 2, "disk_size": 20, "source_type": "iso", "guest_arch": "auto", "network_mode": default_mode, "bridge": DEFAULT_BRIDGE}}'
-if old_context in text:
- text = text.replace(old_context, new_context, 1)
- changed.append('vm form context gets disk images and arch options')
-elif '"disk_images": list_disk_image_files()' in text and '"arch_options": vm_arch_options()' not in text:
- text = text.replace('"disk_images": list_disk_image_files(),', '"disk_images": list_disk_image_files(), "arch_options": vm_arch_options(),')
- text = text.replace('"source_type": "iso", "network_mode"', '"source_type": "iso", "guest_arch": "auto", "network_mode"')
- changed.append('vm form context upgraded with arch options')
-elif '"arch_options": vm_arch_options()' in text:
- changed.append('vm form context already has arch options')
-else:
- warnings.append('vm_form_context marker not found')
-
-routes = r'''
-
-@app.get("/disk-images", response_class=HTMLResponse)
-def disk_images_page(request: Request, error: str | None = None):
- auth_redirect = require_auth(request)
- if auth_redirect:
- return auth_redirect
- return templates.TemplateResponse("disk_images.html", {"request": request, "app_name": APP_NAME, "user": AUTH_USER, "images": list_disk_image_files(), "error": error})
-
-
-@app.post("/disk-images/upload", response_class=HTMLResponse)
-def disk_image_upload(request: Request, image_file: UploadFile = File(...)):
- auth_redirect = require_auth(request)
- if auth_redirect:
- return auth_redirect
- safe_name = safe_disk_image_filename(image_file.filename or "")
- if not safe_name:
- return templates.TemplateResponse("disk_images.html", {"request": request, "app_name": APP_NAME, "user": AUTH_USER, "images": list_disk_image_files(), "error": "Можно загружать только .img, .raw или .qcow2 файлы с безопасным именем."}, status_code=400)
- DISK_IMAGES_DIR.mkdir(parents=True, exist_ok=True)
- target = DISK_IMAGES_DIR / safe_name
- if target.exists():
- return templates.TemplateResponse("disk_images.html", {"request": request, "app_name": APP_NAME, "user": AUTH_USER, "images": list_disk_image_files(), "error": f"Образ уже существует: {safe_name}"}, status_code=400)
- tmp_target = DISK_IMAGES_DIR / f".{safe_name}.uploading"
- try:
- with tmp_target.open("wb") as out:
- shutil.copyfileobj(image_file.file, out)
- tmp_target.rename(target)
- except Exception as exc:
- tmp_target.unlink(missing_ok=True)
- return templates.TemplateResponse("disk_images.html", {"request": request, "app_name": APP_NAME, "user": AUTH_USER, "images": list_disk_image_files(), "error": f"Ошибка загрузки образа: {exc}"}, status_code=500)
- return RedirectResponse(url="/disk-images", status_code=303)
-
-
-@app.post("/disk-images/{name}/delete")
-def disk_image_delete(request: Request, name: str):
- auth_redirect = require_auth(request)
- if auth_redirect:
- return auth_redirect
- path = disk_image_path_by_name(name)
- if path and path.exists() and path.is_file():
- path.unlink()
- return RedirectResponse(url="/disk-images", status_code=303)
-'''
-
-if '@app.get("/disk-images"' not in text:
- marker = '\n\n@app.get("/network", response_class=HTMLResponse)'
- if marker not in text:
- raise SystemExit('network route marker not found')
- text = text.replace(marker, routes + marker, 1)
- changed.append('disk image routes added')
-else:
- changed.append('disk image routes already present')
-
-old_sig = 'def vm_create_submit(request: Request, name: str = Form(...), memory: int = Form(...), vcpus: int = Form(...), disk_size: int = Form(...), iso_path: str = Form(...), network_mode: str = Form("nat"), bridge: str = Form(DEFAULT_BRIDGE)):'
-new_sig = 'def vm_create_submit(request: Request, name: str = Form(...), memory: int = Form(...), vcpus: int = Form(...), disk_size: int = Form(...), iso_path: str = Form(""), disk_image_path: str = Form(""), source_type: str = Form("iso"), guest_arch: str = Form("auto"), network_mode: str = Form("nat"), bridge: str = Form(DEFAULT_BRIDGE)):'
-if old_sig in text:
- text = text.replace(old_sig, new_sig, 1)
- changed.append('vm create signature supports disk images and guest arch')
-elif 'source_type: str = Form("iso")' in text and 'guest_arch: str = Form("auto")' not in text:
- text = text.replace('source_type: str = Form("iso"), network_mode:', 'source_type: str = Form("iso"), guest_arch: str = Form("auto"), network_mode:', 1)
- changed.append('vm create signature upgraded with guest arch')
-elif 'guest_arch: str = Form("auto")' in text:
- changed.append('vm create signature already supports guest arch')
-else:
- warnings.append('vm_create_submit signature marker not found')
-
-old_body = 'form = {"name": name, "memory": memory, "vcpus": vcpus, "disk_size": disk_size, "iso_path": iso_path, "network_mode": network_mode, "bridge": bridge}'
-new_body = 'form = {"name": name, "memory": memory, "vcpus": vcpus, "disk_size": disk_size, "iso_path": iso_path, "disk_image_path": disk_image_path, "source_type": source_type, "guest_arch": guest_arch, "network_mode": network_mode, "bridge": bridge}'
-if old_body in text:
- text = text.replace(old_body, new_body, 1)
- changed.append('vm create form state supports source type and guest arch')
-elif '"source_type": source_type' in text and '"guest_arch": guest_arch' not in text:
- text = text.replace('"source_type": source_type, "network_mode"', '"source_type": source_type, "guest_arch": guest_arch, "network_mode"')
- changed.append('vm create form state upgraded with guest arch')
-
-old_iso_validation = ''' else:
- iso = Path(iso_path).resolve()
- if ISO_DIR.resolve() not in iso.parents or iso.suffix.lower() != ".iso" or not iso.exists():
- error = "ISO должен быть существующим .iso файлом из /var/lib/virtuality/iso."
-'''
-new_iso_validation = ''' elif source_type not in ("iso", "disk_image"):
- error = "Некорректный источник VM."
- elif guest_arch not in ("auto", "x86_64", "aarch64", "generic"):
- error = "Некорректная архитектура VM."
- elif network_mode == "bridge" and not bridge_exists(bridge):
- error = f"Bridge {bridge} не найден на сервере. Для VPS выбери режим NAT Router — virtuality-nat, либо сначала создай bridge {bridge}."
- else:
- if source_type == "iso":
- iso = Path(iso_path).resolve()
- if ISO_DIR.resolve() not in iso.parents or iso.suffix.lower() != ".iso" or not iso.exists():
- error = "ISO должен быть существующим .iso файлом из /var/lib/virtuality/iso."
- else:
- disk_image = Path(disk_image_path).resolve()
- if DISK_IMAGES_DIR.resolve() not in disk_image.parents or disk_image.suffix.lower() not in (".img", ".raw", ".qcow2") or not disk_image.exists():
- error = "Образ диска должен быть существующим .img, .raw или .qcow2 файлом из /var/lib/virtuality/disk-images."
-'''
-if old_iso_validation in text:
- text = text.replace(old_iso_validation, new_iso_validation, 1)
- changed.append('vm create validation supports disk image source and guest arch')
-elif 'guest_arch not in' not in text and 'source_type == "iso"' in text:
- text = text.replace('elif network_mode not in ("nat", "bridge"):\n error = "Некорректный режим сети."', 'elif network_mode not in ("nat", "bridge"):\n error = "Некорректный режим сети."\n elif guest_arch not in ("auto", "x86_64", "aarch64", "generic"):\n error = "Некорректная архитектура VM."', 1)
- changed.append('vm create validation upgraded with guest arch')
-elif 'guest_arch not in' in text:
- changed.append('vm create validation already supports guest arch')
-
-old_cmd = ''' IMAGES_DIR.mkdir(parents=True, exist_ok=True)
- disk_path = IMAGES_DIR / f"{name}.qcow2"
- if disk_path.exists():
- return vm_form_context(request, error=f"Диск уже существует: {disk_path}", form=form, status_code=400)
-
- profile = host_profile.load_host_profile()
- is_arm = profile.get("recommended_guest_arch") == "aarch64"
- network_arg = f"network={network_core.NETWORK_NAME},model=virtio" if network_mode == "nat" else f"bridge={bridge},model=virtio"
- cmd = ["virt-install", "--name", name, "--memory", str(memory), "--vcpus", str(vcpus)]
- if is_arm:
- cmd += ["--arch", "aarch64", "--machine", "virt", "--cpu", "host", "--virt-type", "kvm", "--boot", "uefi"]
- cmd += ["--disk", f"path={disk_path},size={disk_size},format=qcow2,bus=virtio", "--cdrom", iso_path, "--os-variant", "generic", "--network", network_arg, "--graphics", "vnc,listen=0.0.0.0", "--noautoconsole"]
-'''
-new_cmd = ''' IMAGES_DIR.mkdir(parents=True, exist_ok=True)
- disk_path = IMAGES_DIR / f"{name}.qcow2"
- if disk_path.exists():
- return vm_form_context(request, error=f"Диск уже существует: {disk_path}", form=form, status_code=400)
-
- profile = host_profile.load_host_profile()
- selected_arch = normalize_guest_arch(guest_arch, profile)
- host_arch = str(profile.get("arch") or platform.machine() or "")
- is_arm = selected_arch == "aarch64"
- # ARM64 guest on x86 host cannot use KVM. It must use QEMU emulation.
- virt_type = "qemu" if (is_arm and host_arch not in ("aarch64", "arm64")) else ("kvm" if profile.get("kvm_device") else "qemu")
- network_arg = f"network={network_core.NETWORK_NAME},model=virtio" if network_mode == "nat" else f"bridge={bridge},model=virtio"
- cmd = ["virt-install", "--name", name, "--memory", str(memory), "--vcpus", str(vcpus), "--virt-type", virt_type]
- if selected_arch == "x86_64":
- cmd += ["--arch", "x86_64"]
- elif is_arm:
- cmd += ["--arch", "aarch64", "--machine", "virt", "--cpu", "host" if virt_type == "kvm" else "cortex-a57", "--boot", "uefi"]
-
- if source_type == "disk_image":
- source_disk = Path(disk_image_path).resolve()
- source_format = disk_image_format(source_disk)
- convert_cmd = f"qemu-img convert -p -f {source_format} -O qcow2 {source_disk} {disk_path}"
- virt_cmd = " ".join(cmd + ["--import", "--disk", f"path={disk_path},format=qcow2,bus=virtio", "--os-variant", "generic", "--network", network_arg, "--graphics", "vnc,listen=0.0.0.0", "--noautoconsole"])
- cmd = ["bash", "-lc", f"set -euo pipefail; {convert_cmd}; {virt_cmd}"]
- else:
- cmd += ["--disk", f"path={disk_path},size={disk_size},format=qcow2,bus=virtio", "--cdrom", iso_path, "--os-variant", "generic", "--network", network_arg, "--graphics", "vnc,listen=0.0.0.0", "--noautoconsole"]
-'''
-if old_cmd in text:
- text = text.replace(old_cmd, new_cmd, 1)
- changed.append('vm create command supports ARM64 qemu fallback, guest arch and disk image import')
-elif 'selected_arch = normalize_guest_arch' in text:
- # Upgrade existing selected_arch block if it still uses kvm for aarch64 on x86.
- text = re.sub(
- r'virt_type = "kvm" if profile\.get\("kvm_device"\) else "qemu"',
- 'host_arch = str(profile.get("arch") or platform.machine() or "")\n # ARM64 guest on x86 host cannot use KVM. It must use QEMU emulation.\n virt_type = "qemu" if (is_arm and host_arch not in ("aarch64", "arm64")) else ("kvm" if profile.get("kvm_device") else "qemu")',
- text,
- count=1,
- )
- changed.append('existing vm create command was upgraded with ARM64 qemu fallback')
-elif 'virt_type = "kvm" if profile.get("kvm_device") else "qemu"' in text:
- text = text.replace('is_arm = profile.get("recommended_guest_arch") == "aarch64"\n virt_type =', 'selected_arch = normalize_guest_arch(guest_arch, profile)\n host_arch = str(profile.get("arch") or platform.machine() or "")\n is_arm = selected_arch == "aarch64"\n virt_type = "qemu" if (is_arm and host_arch not in ("aarch64", "arm64")) else', 1)
- text = text.replace('if is_arm:\n cmd += ["--arch", "aarch64", "--machine", "virt", "--cpu", "host" if virt_type == "kvm" else "cortex-a57", "--boot", "uefi"]', 'if selected_arch == "x86_64":\n cmd += ["--arch", "x86_64"]\n elif is_arm:\n cmd += ["--arch", "aarch64", "--machine", "virt", "--cpu", "host" if virt_type == "kvm" else "cortex-a57", "--boot", "uefi"]', 1)
- changed.append('existing vm create command was upgraded with guest arch and ARM64 qemu fallback')
-else:
- warnings.append('vm command marker not found')
-
-if '"guest_arch": selected_arch' not in text and '"guest_arch": profile.get("recommended_guest_arch")' in text:
- text = text.replace('"guest_arch": profile.get("recommended_guest_arch")', '"guest_arch": selected_arch')
-if '"disk_image_path": disk_image_path' not in text:
- text = text.replace('"iso_path": iso_path, "host_profile"', '"iso_path": iso_path, "disk_image_path": disk_image_path, "source_type": source_type, "guest_arch": selected_arch, "host_profile"')
-
-app_path.write_text(text)
-print('disk images patch applied:')
-for item in changed:
- print(f'- {item}')
-for item in warnings:
- print(f'WARN: {item}')
diff --git a/scripts/patch_existing_vm_boot_order.py b/scripts/patch_existing_vm_boot_order.py
deleted file mode 100644
index e9aae63..0000000
--- a/scripts/patch_existing_vm_boot_order.py
+++ /dev/null
@@ -1,224 +0,0 @@
-#!/usr/bin/env python3
-from pathlib import Path
-import sys
-
-app_path = Path(sys.argv[1]) if len(sys.argv) > 1 else Path('/opt/virtuality/web/app.py')
-if not app_path.exists():
- raise SystemExit(f'app.py not found: {app_path}')
-
-app_dir = app_path.resolve().parent
-template_path = app_dir / 'templates' / 'vm_detail.html'
-changed = []
-warnings = []
-
-text = app_path.read_text()
-
-if 'import tempfile' not in text:
- text = text.replace('import subprocess\n', 'import subprocess\nimport tempfile\n', 1)
- changed.append('tempfile import added')
-else:
- changed.append('tempfile import already present')
-
-if 'import xml.etree.ElementTree as ET' not in text:
- text = text.replace('from typing import Any\n', 'from typing import Any\nimport xml.etree.ElementTree as ET\n', 1)
- changed.append('ElementTree import added')
-else:
- changed.append('ElementTree import already present')
-
-helpers = r'''
-
-def vm_boot_order_label(value: str) -> str:
- labels = {item["value"]: item["label"] for item in vm_boot_order_options()}
- return labels.get(value or "auto", "Auto — по источнику VM")
-
-
-def boot_order_to_devs(value: str) -> list[str]:
- value = normalize_boot_order(value, "disk_image")
- mapping = {
- "disk": ["hd"],
- "cdrom_disk": ["cdrom", "hd"],
- "disk_cdrom": ["hd", "cdrom"],
- "network_disk": ["network", "hd"],
- }
- return mapping.get(value, ["hd"])
-
-
-def boot_devs_to_order(devs: list[str]) -> str:
- clean = [item for item in devs if item in ("hd", "cdrom", "network")]
- if clean[:2] == ["cdrom", "hd"]:
- return "cdrom_disk"
- if clean[:2] == ["hd", "cdrom"]:
- return "disk_cdrom"
- if clean[:2] == ["network", "hd"]:
- return "network_disk"
- if clean[:1] == ["hd"]:
- return "disk"
- return "auto"
-
-
-def current_vm_boot_order(name: str) -> str:
- result = run_cmd(["virsh", "dumpxml", name], timeout=12)
- if not result.get("ok"):
- return "auto"
- try:
- root = ET.fromstring(result.get("stdout") or "")
- except Exception:
- return "auto"
- os_node = root.find("os")
- if os_node is None:
- return "auto"
- devs = []
- for boot in os_node.findall("boot"):
- dev = boot.attrib.get("dev", "").strip()
- if dev:
- devs.append(dev)
- return boot_devs_to_order(devs)
-
-
-def apply_vm_boot_order(name: str, boot_order: str) -> tuple[bool, str]:
- if not valid_vm_name(name) or not vm_exists(name):
- return False, "VM не найдена."
- if boot_order not in ("auto", "disk", "cdrom_disk", "disk_cdrom", "network_disk"):
- return False, "Некорректный порядок загрузки VM."
-
- selected = normalize_boot_order(boot_order, "disk_image")
- result = run_cmd(["virsh", "dumpxml", name], timeout=15)
- if not result.get("ok"):
- return False, result.get("stderr") or "Не удалось получить XML VM."
-
- try:
- root = ET.fromstring(result.get("stdout") or "")
- except Exception as exc:
- return False, f"Не удалось разобрать XML VM: {exc}"
-
- os_node = root.find("os")
- if os_node is None:
- os_node = ET.SubElement(root, "os")
-
- for boot in list(os_node.findall("boot")):
- os_node.remove(boot)
-
- insert_at = 0
- for idx, child in enumerate(list(os_node)):
- if child.tag in ("type", "loader", "nvram", "firmware", "smbios", "bootmenu"):
- insert_at = idx + 1
-
- for dev in reversed(boot_order_to_devs(selected)):
- boot_node = ET.Element("boot", {"dev": dev})
- os_node.insert(insert_at, boot_node)
-
- xml_text = ET.tostring(root, encoding="unicode")
- with tempfile.NamedTemporaryFile("w", encoding="utf-8", suffix=".xml", delete=False) as handle:
- handle.write(xml_text)
- tmp_name = handle.name
- try:
- define = run_cmd(["virsh", "define", tmp_name], timeout=30)
- finally:
- Path(tmp_name).unlink(missing_ok=True)
- if not define.get("ok"):
- return False, define.get("stderr") or "virsh define завершился ошибкой."
- return True, f"Порядок загрузки применён: {vm_boot_order_label(selected)}. Если VM запущена, изменение сработает после перезапуска."
-'''
-
-if 'def apply_vm_boot_order(name: str, boot_order: str)' not in text:
- marker = '\n\ndef vm_details(name: str) -> dict[str, Any]:'
- if marker in text:
- text = text.replace(marker, helpers + marker, 1)
- changed.append('existing VM boot order helpers added')
- else:
- warnings.append('vm_details marker not found, helpers skipped')
-else:
- changed.append('existing VM boot order helpers already present')
-
-old = '"host_ip": system_summary()["ip"]})'
-new = '"host_ip": system_summary()["ip"], "boot_options": vm_boot_order_options(), "current_boot_order": current_vm_boot_order(name), "boot_message": request.query_params.get("boot_message", ""), "boot_error": request.query_params.get("boot_error", "")})'
-if '"current_boot_order": current_vm_boot_order(name)' not in text:
- if old in text:
- text = text.replace(old, new, 1)
- changed.append('vm detail context gets boot order')
- else:
- warnings.append('vm_detail context marker not found, skipped')
-else:
- changed.append('vm detail context already has boot order')
-
-route = r'''
-
-@app.post("/vm/{name}/boot-order")
-def vm_boot_order_apply(request: Request, name: str, boot_order: str = Form("auto")):
- auth_redirect = require_auth(request)
- if auth_redirect:
- return auth_redirect
- ok, message = apply_vm_boot_order(name, boot_order)
- if ok:
- return RedirectResponse(url=f"/vm/{name}?boot_message={message}", status_code=303)
- return RedirectResponse(url=f"/vm/{name}?boot_error={message}", status_code=303)
-'''
-
-if '@app.post("/vm/{name}/boot-order")' not in text:
- marker = '\n\n@app.post("/vm/{name}/{action}")'
- if marker in text:
- text = text.replace(marker, route + marker, 1)
- changed.append('existing VM boot order route added')
- else:
- warnings.append('generic vm action marker not found, route skipped')
-else:
- changed.append('existing VM boot order route already present')
-
-app_path.write_text(text)
-
-if template_path.exists():
- tpl = template_path.read_text()
- original = tpl
- if 'action="/vm/{{ vm.name }}/boot-order"' not in tpl:
- block = r'''
-
- {% if boot_message %}
-
{{ boot_message }}
- {% endif %}
- {% if boot_error %}
-
{{ boot_error }}
- {% endif %}
-
-
-
-
Порядок загрузки
- boot order
-
-
-
Настройка меняет XML-конфигурацию VM через virsh define. Если машина сейчас запущена, новый порядок загрузки сработает после перезапуска.
-
-'''
- markers = ['\n\n ', '\n\n ', '\n\n ']
- inserted = False
- for marker in markers:
- if marker in tpl:
- tpl = tpl.replace(marker, block + marker, 1)
- inserted = True
- changed.append('boot order card added to vm_detail.html')
- break
- if not inserted:
- warnings.append('vm detail insertion marker not found, boot order card skipped')
- else:
- changed.append('boot order card already present in vm_detail.html')
- if tpl != original:
- template_path.write_text(tpl)
-else:
- warnings.append(f'vm_detail.html not found: {template_path}')
-
-print('existing VM boot order patch applied:')
-for item in changed:
- print(f'- {item}')
-if warnings:
- print('Warnings:')
- for item in warnings:
- print(f'- {item}')
diff --git a/scripts/patch_existing_vm_iso_mount.py b/scripts/patch_existing_vm_iso_mount.py
deleted file mode 100644
index b8eef27..0000000
--- a/scripts/patch_existing_vm_iso_mount.py
+++ /dev/null
@@ -1,280 +0,0 @@
-#!/usr/bin/env python3
-from pathlib import Path
-import sys
-
-app_path = Path(sys.argv[1]) if len(sys.argv) > 1 else Path('/opt/virtuality/web/app.py')
-if not app_path.exists():
- raise SystemExit(f'app.py not found: {app_path}')
-
-app_dir = app_path.resolve().parent
-template_path = app_dir / 'templates' / 'vm_detail.html'
-changed = []
-
-text = app_path.read_text()
-
-helpers = r'''
-
-def vm_state(name: str) -> str:
- result = run_cmd(["virsh", "domstate", name], timeout=8)
- return (result.get("stdout") or "unknown").strip().lower()
-
-
-def current_vm_iso(name: str) -> str:
- result = run_cmd(["virsh", "domblklist", name, "--details"], timeout=10)
- if not result.get("ok"):
- return ""
- for line in (result.get("stdout") or "").splitlines():
- if ".iso" not in line.lower():
- continue
- parts = line.split()
- if parts:
- return parts[-1]
- return ""
-
-
-def detach_vm_iso(name: str) -> tuple[bool, str]:
- if not valid_vm_name(name) or not vm_exists(name):
- return False, "VM не найдена."
- xml = run_cmd(["virsh", "dumpxml", name], timeout=12)
- cdrom_targets: list[str] = []
- if xml.get("ok"):
- try:
- root = ET.fromstring(xml.get("stdout") or "")
- devices = root.find("devices")
- if devices is not None:
- for disk in devices.findall("disk"):
- if disk.attrib.get("device") != "cdrom":
- continue
- target = disk.find("target")
- dev = target.attrib.get("dev") if target is not None else ""
- if dev:
- cdrom_targets.append(dev)
- except Exception:
- pass
- if not cdrom_targets:
- cdrom_targets = ["sda", "hda", "sdb", "hdc"]
-
- running = vm_state(name) == "running"
- errors = []
- changed_any = False
- for target in cdrom_targets:
- commands = []
- if running:
- commands.append(["virsh", "detach-disk", name, target, "--live"])
- commands.append(["virsh", "detach-disk", name, target, "--config"])
- for cmd in commands:
- result = run_cmd(cmd, timeout=30)
- if result.get("ok"):
- changed_any = True
- elif result.get("stderr"):
- errors.append(result.get("stderr"))
- if changed_any:
- return True, "ISO был отмонтирован."
- return False, errors[-1] if errors else "Подключенный ISO не найден."
-
-
-def mount_vm_iso(name: str, iso_path: str) -> tuple[bool, str]:
- if not valid_vm_name(name) or not vm_exists(name):
- return False, "VM не найдена."
- iso = Path(iso_path or "").resolve()
- try:
- iso_root = ISO_DIR.resolve()
- except Exception:
- iso_root = Path("/var/lib/virtuality/iso")
- if iso_root not in iso.parents or iso.suffix.lower() != ".iso" or not iso.exists():
- return False, "ISO должен быть существующим .iso файлом из /var/lib/virtuality/iso."
-
- detach_vm_iso(name)
- running = vm_state(name) == "running"
- base = ["virsh", "attach-disk", name, str(iso), "sda", "--type", "cdrom", "--mode", "readonly"]
- if running:
- live = run_cmd(base + ["--live"], timeout=30)
- if not live.get("ok"):
- return False, live.get("stderr") or "Не удалось подключить ISO к запущенной VM."
- config = run_cmd(base + ["--config"], timeout=30)
- if not config.get("ok"):
- return False, config.get("stderr") or "Не удалось сохранить ISO в конфигурации VM."
- return True, "ISO был смонтирован в VM. Если гостевая ОС его не увидела сразу, перезагрузи VM или обнови устройства внутри гостевой ОС."
-'''
-
-if 'def mount_vm_iso(name: str, iso_path: str)' not in text:
- marker = '\n\ndef vm_details(name: str) -> dict[str, Any]:'
- if marker in text:
- text = text.replace(marker, helpers + marker, 1)
- changed.append('existing VM ISO mount helpers added')
- else:
- changed.append('ISO helpers skipped: vm_details marker not found')
-else:
- changed.append('existing VM ISO mount helpers already present')
-
-if '"current_iso": current_vm_iso(name)' not in text:
- replacements = [
- (
- '"resource_message": request.query_params.get("resource_message", ""), "resource_error": request.query_params.get("resource_error", "")})',
- '"resource_message": request.query_params.get("resource_message", ""), "resource_error": request.query_params.get("resource_error", ""), "isos": list_iso_files(), "current_iso": current_vm_iso(name), "iso_message": request.query_params.get("iso_message", ""), "iso_error": request.query_params.get("iso_error", "")})',
- ),
- (
- '"boot_message": request.query_params.get("boot_message", ""), "boot_error": request.query_params.get("boot_error", "")})',
- '"boot_message": request.query_params.get("boot_message", ""), "boot_error": request.query_params.get("boot_error", ""), "isos": list_iso_files(), "current_iso": current_vm_iso(name), "iso_message": request.query_params.get("iso_message", ""), "iso_error": request.query_params.get("iso_error", "")})',
- ),
- (
- '"host_ip": system_summary()["ip"]})',
- '"host_ip": system_summary()["ip"], "isos": list_iso_files(), "current_iso": current_vm_iso(name), "iso_message": request.query_params.get("iso_message", ""), "iso_error": request.query_params.get("iso_error", "")})',
- ),
- ]
- for old, new in replacements:
- if old in text:
- text = text.replace(old, new, 1)
- changed.append('vm detail context gets ISO mount data')
- break
- else:
- changed.append('ISO context skipped: vm detail context marker not found')
-else:
- changed.append('vm detail context already has ISO mount data')
-
-routes = r'''
-
-@app.post("/vm/{name}/iso/mount")
-def vm_iso_mount_apply(request: Request, name: str, iso_path: str = Form(...)):
- auth_redirect = require_auth(request)
- if auth_redirect:
- return auth_redirect
- ok, message = mount_vm_iso(name, iso_path)
- if ok:
- return RedirectResponse(url=f"/vm/{name}?iso_message={message}", status_code=303)
- return RedirectResponse(url=f"/vm/{name}?iso_error={message}", status_code=303)
-
-
-@app.post("/vm/{name}/iso/unmount")
-def vm_iso_unmount_apply(request: Request, name: str):
- auth_redirect = require_auth(request)
- if auth_redirect:
- return auth_redirect
- ok, message = detach_vm_iso(name)
- if ok:
- return RedirectResponse(url=f"/vm/{name}?iso_message={message}", status_code=303)
- return RedirectResponse(url=f"/vm/{name}?iso_error={message}", status_code=303)
-'''
-
-if '@app.post("/vm/{name}/iso/mount")' not in text:
- markers = ['\n\n@app.post("/vm/{name}/{action}")', '\n\n@app.get("/vm/create"', '\n\n@app.get("/operations"']
- for marker in markers:
- if marker in text:
- text = text.replace(marker, routes + marker, 1)
- changed.append('existing VM ISO mount routes added')
- break
- else:
- changed.append('ISO routes skipped: safe route marker not found')
-else:
- changed.append('existing VM ISO mount routes already present')
-
-app_path.write_text(text)
-
-iso_card = r'''
-
-
-
ISO-привод
- cdrom
-
- {% if current_iso %}
-
Сейчас подключен ISO: {{ current_iso }}
- {% else %}
-
ISO сейчас не подключён.
- {% endif %}
-
-
- {% if not isos %}
-
ISO-образов пока нет. Сначала загрузи .iso в разделе ISO.
- {% endif %}
-
Для запущенной VM ISO подключается live и сохраняется в конфигурации. Для выключенной VM ISO будет доступен при следующем старте.
-
-'''
-
-boot_card = r'''
-
-
-
Порядок загрузки
- drag boot
-
-
-
Перетащи нужный источник выше. Для загрузки с ISO поставь ISO / CD-ROM первым. Изменение сработает после перезапуска VM.
-
-
-'''
-
-if template_path.exists():
- tpl = template_path.read_text()
- original = tpl
- settings_grid = '\n\n \n' + iso_card + boot_card + ' \n'
-
- if 'action="/vm/{{ vm.name }}/iso/mount"' not in tpl and 'vm-boot-iso-grid' not in tpl:
- markers = ['\n\n ', '\n\n ', '\n\n ']
- for marker in markers:
- if marker in tpl:
- if 'vm-resource-boot-iso-grid' in marker:
- insert_pos = tpl.find(marker) + len(marker)
- tpl = tpl[:insert_pos] + '\n' + iso_card + boot_card + tpl[insert_pos:]
- changed.append('ISO and boot cards inserted into three-column grid')
- else:
- tpl = tpl.replace(marker, settings_grid + marker, 1)
- changed.append('ISO and boot cards inserted by fallback')
- break
- else:
- changed.append('ISO layout skipped: safe template marker not found')
- elif 'vm-boot-iso-grid' in tpl and 'boot-order-list' not in tpl:
- changed.append('ISO grid exists but draggable boot marker absent; left unchanged safely')
- else:
- changed.append('ISO mount template already present')
-
- if tpl != original:
- template_path.write_text(tpl)
-else:
- changed.append('ISO template skipped: vm_detail.html not found')
-
-print('existing VM ISO mount patch applied:')
-for item in changed:
- print(f'- {item}')
diff --git a/scripts/patch_existing_vm_resources.py b/scripts/patch_existing_vm_resources.py
deleted file mode 100644
index 8f3ee26..0000000
--- a/scripts/patch_existing_vm_resources.py
+++ /dev/null
@@ -1,205 +0,0 @@
-#!/usr/bin/env python3
-from pathlib import Path
-import sys
-
-app_path = Path(sys.argv[1]) if len(sys.argv) > 1 else Path('/opt/virtuality/web/app.py')
-if not app_path.exists():
- raise SystemExit(f'app.py not found: {app_path}')
-
-changed = []
-warnings = []
-text = app_path.read_text()
-
-if 'import tempfile' not in text:
- text = text.replace('import subprocess\n', 'import subprocess\nimport tempfile\n', 1)
- changed.append('tempfile import added')
-else:
- changed.append('tempfile import already present')
-
-if 'import xml.etree.ElementTree as ET' not in text:
- text = text.replace('from typing import Any\n', 'from typing import Any\nimport xml.etree.ElementTree as ET\n', 1)
- changed.append('ElementTree import added')
-else:
- changed.append('ElementTree import already present')
-
-helpers = r'''
-
-def parse_dominfo_value(dominfo: str, label: str) -> str:
- for line in (dominfo or "").splitlines():
- if line.strip().lower().startswith(label.lower()):
- return line.split(":", 1)[1].strip()
- return ""
-
-
-def kib_text_to_mb(value: str) -> int:
- match = re.search(r"(\d+)", value or "")
- if not match:
- return 0
- return max(0, int(int(match.group(1)) / 1024))
-
-
-def current_vm_arch(name: str) -> str:
- result = run_cmd(["virsh", "dumpxml", name], timeout=12)
- if not result.get("ok"):
- return "unknown"
- try:
- root = ET.fromstring(result.get("stdout") or "")
- except Exception:
- return "unknown"
- os_type = root.find("os/type")
- return (os_type.attrib.get("arch") if os_type is not None else "") or "unknown"
-
-
-def vm_runtime_state(name: str) -> str:
- result = run_cmd(["virsh", "domstate", name], timeout=8)
- return (result.get("stdout") or "unknown").strip().lower()
-
-
-def vm_resource_settings(name: str) -> dict[str, Any]:
- dominfo = run_cmd(["virsh", "dominfo", name], timeout=10).get("stdout") or ""
- state = (parse_dominfo_value(dominfo, "State") or vm_runtime_state(name)).lower()
- vcpus_raw = parse_dominfo_value(dominfo, "CPU(s)")
- used_memory_raw = parse_dominfo_value(dominfo, "Used memory")
- max_memory_raw = parse_dominfo_value(dominfo, "Max memory")
- memory_mb = kib_text_to_mb(used_memory_raw) or kib_text_to_mb(max_memory_raw) or 1024
- try:
- vcpus = int(re.search(r"\d+", vcpus_raw or "1").group(0))
- except Exception:
- vcpus = 1
- return {
- "state": state,
- "is_shutoff": state in ("shut off", "shutoff", "shut-off"),
- "memory_mb": memory_mb,
- "vcpus": vcpus,
- "arch": current_vm_arch(name),
- }
-
-
-def apply_vm_resources(name: str, memory_mb: int, vcpus: int, guest_arch: str) -> tuple[bool, str]:
- if not valid_vm_name(name) or not vm_exists(name):
- return False, "VM не найдена."
- resources = vm_resource_settings(name)
- if not resources.get("is_shutoff"):
- return False, "CPU/RAM/архитектуру можно менять только когда VM выключена. Сначала выключи VM."
- if memory_mb < 512 or memory_mb > 262144:
- return False, "RAM должна быть от 512 MB до 262144 MB."
- if vcpus < 1 or vcpus > 128:
- return False, "CPU должен быть от 1 до 128 vCPU."
- if guest_arch not in ("keep", "x86_64", "aarch64"):
- return False, "Некорректная архитектура VM."
-
- result = run_cmd(["virsh", "dumpxml", name], timeout=15)
- if not result.get("ok"):
- return False, result.get("stderr") or "Не удалось получить XML VM."
- try:
- root = ET.fromstring(result.get("stdout") or "")
- except Exception as exc:
- return False, f"Не удалось разобрать XML VM: {exc}"
-
- memory_kib = str(int(memory_mb) * 1024)
- for tag in ("memory", "currentMemory"):
- node = root.find(tag)
- if node is None:
- node = ET.SubElement(root, tag)
- node.text = memory_kib
- node.set("unit", "KiB")
-
- vcpu_node = root.find("vcpu")
- if vcpu_node is None:
- vcpu_node = ET.SubElement(root, "vcpu")
- vcpu_node.text = str(int(vcpus))
- vcpu_node.set("placement", "static")
-
- arch_changed = False
- if guest_arch != "keep":
- os_type = root.find("os/type")
- if os_type is None:
- os_node = root.find("os")
- if os_node is None:
- os_node = ET.SubElement(root, "os")
- os_type = ET.SubElement(os_node, "type")
- os_type.text = "hvm"
- old_arch = os_type.attrib.get("arch", "")
- if old_arch != guest_arch:
- os_type.set("arch", guest_arch)
- if guest_arch == "aarch64":
- os_type.set("machine", "virt")
- arch_changed = True
-
- xml_text = ET.tostring(root, encoding="unicode")
- with tempfile.NamedTemporaryFile("w", encoding="utf-8", suffix=".xml", delete=False) as handle:
- handle.write(xml_text)
- tmp_name = handle.name
- try:
- define = run_cmd(["virsh", "define", tmp_name], timeout=30)
- finally:
- Path(tmp_name).unlink(missing_ok=True)
- if not define.get("ok"):
- return False, define.get("stderr") or "virsh define завершился ошибкой."
-
- message = f"Ресурсы VM применены: CPU {vcpus}, RAM {memory_mb} MB"
- if arch_changed:
- message += f", архитектура {guest_arch}. Важно: смена архитектуры может потребовать совместимый диск/загрузчик."
- return True, message
-'''
-
-if 'def apply_vm_resources(name: str, memory_mb: int, vcpus: int, guest_arch: str)' not in text:
- marker = '\n\ndef vm_details(name: str) -> dict[str, Any]:'
- if marker in text:
- text = text.replace(marker, helpers + marker, 1)
- changed.append('existing VM resources helpers added')
- else:
- warnings.append('vm_details marker not found, helpers skipped')
-else:
- changed.append('existing VM resources helpers already present')
-
-if '"resource_settings": vm_resource_settings(name)' not in text:
- replacements = [
- ('"iso_message": request.query_params.get("iso_message", ""), "iso_error": request.query_params.get("iso_error", "")})', '"iso_message": request.query_params.get("iso_message", ""), "iso_error": request.query_params.get("iso_error", ""), "resource_settings": vm_resource_settings(name), "resource_message": request.query_params.get("resource_message", ""), "resource_error": request.query_params.get("resource_error", "")})'),
- ('"boot_message": request.query_params.get("boot_message", ""), "boot_error": request.query_params.get("boot_error", "")})', '"boot_message": request.query_params.get("boot_message", ""), "boot_error": request.query_params.get("boot_error", ""), "resource_settings": vm_resource_settings(name), "resource_message": request.query_params.get("resource_message", ""), "resource_error": request.query_params.get("resource_error", "")})'),
- ('"host_ip": system_summary()["ip"]})', '"host_ip": system_summary()["ip"], "resource_settings": vm_resource_settings(name), "resource_message": request.query_params.get("resource_message", ""), "resource_error": request.query_params.get("resource_error", "")})'),
- ]
- for old, new in replacements:
- if old in text:
- text = text.replace(old, new, 1)
- changed.append('vm detail context gets resource settings')
- break
- else:
- warnings.append('vm detail context marker not found, resource context skipped')
-else:
- changed.append('vm detail context already has resource settings')
-
-route = r'''
-
-@app.post("/vm/{name}/resources")
-def vm_resources_apply(request: Request, name: str, memory_mb: int = Form(...), vcpus: int = Form(...), guest_arch: str = Form("keep")):
- auth_redirect = require_auth(request)
- if auth_redirect:
- return auth_redirect
- ok, message = apply_vm_resources(name, memory_mb, vcpus, guest_arch)
- if ok:
- return RedirectResponse(url=f"/vm/{name}?resource_message={message}", status_code=303)
- return RedirectResponse(url=f"/vm/{name}?resource_error={message}", status_code=303)
-'''
-
-if '@app.post("/vm/{name}/resources")' not in text:
- markers = ['\n\n@app.post("/vm/{name}/boot-order")', '\n\n@app.post("/vm/{name}/iso/mount")', '\n\n@app.post("/vm/{name}/{action}")']
- for marker in markers:
- if marker in text:
- text = text.replace(marker, route + marker, 1)
- changed.append('existing VM resources route added')
- break
- else:
- warnings.append('vm action marker not found, resources route skipped')
-else:
- changed.append('existing VM resources route already present')
-
-app_path.write_text(text)
-
-print('existing VM resources patch applied:')
-for item in changed:
- print(f'- {item}')
-if warnings:
- print('Warnings:')
- for item in warnings:
- print(f'- {item}')
diff --git a/scripts/patch_live_status.py b/scripts/patch_live_status.py
deleted file mode 100644
index bc99ce9..0000000
--- a/scripts/patch_live_status.py
+++ /dev/null
@@ -1,154 +0,0 @@
-#!/usr/bin/env python3
-from pathlib import Path
-import sys
-
-app_path = Path(sys.argv[1]) if len(sys.argv) > 1 else Path('/opt/virtuality/web/app.py')
-if not app_path.exists():
- raise SystemExit(f'app.py not found: {app_path}')
-
-app_dir = app_path.parent
-static_dir = app_dir / 'static'
-static_dir.mkdir(exist_ok=True)
-panel_js_path = static_dir / 'panel.js'
-app_css_path = static_dir / 'app.css'
-changed = []
-
-text = app_path.read_text()
-live_routes = r'''
-
-@app.get("/live/status")
-def live_status(request: Request):
- if not get_current_user(request):
- return JSONResponse({"ok": False, "error": "Unauthorized"}, status_code=401)
- vms = []
- for vm in parse_virsh_list():
- name = vm.get("name", "")
- state = vm.get("state", "unknown")
- try:
- ip = vm_ip(name) if name else "—"
- except Exception:
- ip = "—"
- css = "ok" if "running" in state else "err" if "shut" in state else "warn"
- vms.append({"id": vm.get("id", "-"), "name": name, "state": state, "state_css": css, "ip": ip if ip and ip != "not available" else "—"})
- return JSONResponse({"ok": True, "generated_at": utc_now(), "vms": vms, "services": {"libvirtd": service_state("libvirtd.service"), "virtlogd": service_state("virtlogd.service"), "cockpit": service_state("cockpit.socket"), "web": service_state("virtuality-web.service")}, "operations": list_operations(5)})
-
-
-@app.get("/live/operations")
-def live_operations(request: Request):
- if not get_current_user(request):
- return JSONResponse({"ok": False, "error": "Unauthorized"}, status_code=401)
- return JSONResponse({"ok": True, "generated_at": utc_now(), "operations": list_operations(25)})
-'''
-if '@app.get("/live/status"' not in text:
- marker = '\n\n@app.get("/api/health")'
- text = text.replace(marker, live_routes + marker, 1) if marker in text else text.rstrip() + live_routes + '\n'
- changed.append('live routes added')
-else:
- changed.append('live routes already present')
-app_path.write_text(text)
-
-panel_js = panel_js_path.read_text() if panel_js_path.exists() else ''
-live_js = r'''
-
-/* Virtuality live status + toast layer */
-(() => {
- if (window.VirtualityLiveStatus) return;
- window.VirtualityLiveStatus = true;
- function qs(s, r = document) { return r.querySelector(s); }
- function qsa(s, r = document) { return Array.from(r.querySelectorAll(s)); }
- function toast(message, type = 'ok') {
- if (!message) return;
- let wrap = qs('#v-toast-wrap');
- if (!wrap) { wrap = document.createElement('div'); wrap.id = 'v-toast-wrap'; document.body.appendChild(wrap); }
- const item = document.createElement('div');
- item.className = 'v-toast ' + type;
- item.textContent = message;
- wrap.appendChild(item);
- setTimeout(() => item.classList.add('show'), 20);
- setTimeout(() => { item.classList.remove('show'); setTimeout(() => item.remove(), 260); }, 3600);
- }
- function vmNameFromRow(row) {
- const link = qs('a[href^="/vm/"]', row);
- if (!link) return '';
- try { const p = new URL(link.href, location.href).pathname.split('/').filter(Boolean); return p[0] === 'vm' ? decodeURIComponent(p[1] || '') : ''; } catch (_) { return ''; }
- }
- function stateClass(state) { const v = String(state || '').toLowerCase(); return v.includes('running') ? 'ok' : v.includes('shut') ? 'err' : 'warn'; }
- function updateDashboardRows(vms) {
- if (!Array.isArray(vms)) return;
- const map = new Map(vms.map((vm) => [vm.name, vm]));
- qsa('table tbody tr').forEach((row) => {
- const name = vmNameFromRow(row);
- if (!name || !map.has(name)) return;
- const vm = map.get(name);
- const cells = qsa('td', row);
- if (cells[0]) cells[0].textContent = vm.id || '-';
- if (cells[2]) cells[2].textContent = vm.ip || '—';
- if (cells[3]) {
- let badge = qs('.status', cells[3]);
- if (!badge) { badge = document.createElement('span'); cells[3].textContent = ''; cells[3].appendChild(badge); }
- badge.className = 'status ' + (vm.state_css || stateClass(vm.state));
- badge.textContent = vm.state || 'unknown';
- }
- });
- }
- function updateDetailHeader(vms) {
- const title = qs('.brand');
- if (!title || !Array.isArray(vms)) return;
- const vm = vms.find((item) => item.name === title.textContent.trim());
- if (!vm) return;
- let badge = qs('#vm-live-badge');
- if (!badge) { badge = document.createElement('span'); badge.id = 'vm-live-badge'; title.insertAdjacentElement('afterend', badge); }
- badge.className = 'status live-badge ' + (vm.state_css || stateClass(vm.state));
- badge.textContent = vm.state || 'unknown';
- }
- async function refreshLiveStatus() {
- if (document.hidden || (!qs('.v-main') && !qs('.shell'))) return;
- try {
- const r = await fetch('/live/status', { cache: 'no-store', headers: { 'Accept': 'application/json' }});
- if (!r.ok) return;
- const p = await r.json();
- if (!p.ok) return;
- updateDashboardRows(p.vms);
- updateDetailHeader(p.vms);
- } catch (_) {}
- }
- document.addEventListener('submit', (event) => {
- const form = event.target;
- if (!form || !form.action) return;
- const parts = new URL(form.action, location.href).pathname.split('/').filter(Boolean);
- if (parts[0] !== 'vm' || parts.length < 3) return;
- const labels = { start: 'Команда запуска VM отправлена', shutdown: 'Команда мягкого выключения VM отправлена', reboot: 'Команда перезагрузки VM отправлена', destroy: 'Команда принудительного выключения VM отправлена', autostart: 'Автозапуск VM включается', 'autostart-disable': 'Автозапуск VM отключается', delete: 'Удаление VM запущено' };
- const action = parts[2];
- if (labels[action]) { const type = action === 'destroy' || action === 'delete' ? 'warn' : 'ok'; sessionStorage.setItem('virtualityToast', labels[action]); toast(labels[action], type); setTimeout(refreshLiveStatus, 900); setTimeout(refreshLiveStatus, 2500); }
- }, true);
- function boot() { const m = sessionStorage.getItem('virtualityToast'); if (m) { sessionStorage.removeItem('virtualityToast'); toast(m, 'ok'); } refreshLiveStatus(); setInterval(refreshLiveStatus, 5000); }
- if (document.readyState === 'loading') document.addEventListener('DOMContentLoaded', boot); else boot();
-})();
-'''
-if 'Virtuality live status + toast layer' not in panel_js:
- panel_js_path.write_text(panel_js.rstrip() + live_js + '\n')
- changed.append('live javascript added')
-else:
- changed.append('live javascript already present')
-
-app_css = app_css_path.read_text() if app_css_path.exists() else ''
-live_css = r'''
-
-/* Virtuality live status + toast layer */
-#v-toast-wrap { position: fixed; top: 14px; right: 14px; z-index: 10000; display: grid; gap: 8px; width: min(360px, calc(100vw - 28px)); pointer-events: none; }
-.v-toast { transform: translateY(-8px); opacity: 0; padding: 10px 12px; border-radius: 8px; border: 1px solid var(--line); background: #fff; color: var(--text); box-shadow: 0 12px 32px rgba(15,23,42,.18); font-size: 13px; font-weight: 800; transition: .22s ease; }
-.v-toast.show { transform: translateY(0); opacity: 1; }
-.v-toast.ok { border-color: rgba(22,163,74,.28); box-shadow: 0 12px 32px rgba(22,163,74,.16); }
-.v-toast.warn { border-color: rgba(217,119,6,.32); box-shadow: 0 12px 32px rgba(217,119,6,.16); }
-.v-toast.err { border-color: rgba(220,38,38,.32); box-shadow: 0 12px 32px rgba(220,38,38,.16); }
-.live-badge { margin-left: 10px; vertical-align: middle; }
-'''
-if 'Virtuality live status + toast layer' not in app_css:
- app_css_path.write_text(app_css.rstrip() + live_css + '\n')
- changed.append('live css added')
-else:
- changed.append('live css already present')
-
-print('live status patch applied:')
-for item in changed:
- print(f'- {item}')
diff --git a/scripts/patch_logs_center.py b/scripts/patch_logs_center.py
deleted file mode 100644
index 2494186..0000000
--- a/scripts/patch_logs_center.py
+++ /dev/null
@@ -1,166 +0,0 @@
-#!/usr/bin/env python3
-from pathlib import Path
-import sys
-
-app_path = Path(sys.argv[1]) if len(sys.argv) > 1 else Path('/opt/virtuality/web/app.py')
-if not app_path.exists():
- raise SystemExit(f'app.py not found: {app_path}')
-
-templates_dir = app_path.parent / 'templates'
-static_dir = app_path.parent / 'static'
-text = app_path.read_text()
-changed = []
-
-helpers = r'''
-
-LOG_SOURCES = {
- "web": {"title": "Virtuality Web", "kind": "journal", "unit": "virtuality-web.service"},
- "update": {"title": "Update Center", "kind": "file", "path": "/var/log/virtuality/update.log"},
- "install": {"title": "Последняя установка", "kind": "glob", "pattern": "/var/log/virtuality/install_web_panel_*.log"},
- "operations": {"title": "Операции", "kind": "operations"},
- "auto-update": {"title": "Auto Update", "kind": "journal", "unit": "virtuality-auto-update.service"},
- "libvirtd": {"title": "libvirtd", "kind": "journal", "unit": "libvirtd.service"},
- "virtlogd": {"title": "virtlogd", "kind": "journal", "unit": "virtlogd.service"},
-}
-
-
-def read_log_source(source: str, lines: int = 220) -> dict[str, Any]:
- key = source if source in LOG_SOURCES else "web"
- cfg = LOG_SOURCES[key]
- lines = max(20, min(int(lines or 220), 2000))
- content = ""
- path = ""
- cmd = ""
- if cfg["kind"] == "journal":
- unit = cfg["unit"]
- cmd = f"journalctl -u {unit} -n {lines} --no-pager"
- result = run_cmd(["journalctl", "-u", unit, "-n", str(lines), "--no-pager"], timeout=15)
- content = result["stdout"] or result["stderr"] or "Лог пуст или journalctl недоступен"
- elif cfg["kind"] == "file":
- path = cfg["path"]
- cmd = f"tail -n {lines} {path}"
- content = tail_text(Path(path), max_lines=lines) or "Файл лога пока пуст или не найден"
- elif cfg["kind"] == "glob":
- pattern = cfg["pattern"]
- files = sorted(Path('/').glob(pattern.lstrip('/')), key=lambda item: item.stat().st_mtime if item.exists() else 0, reverse=True)
- if files:
- path = str(files[0])
- cmd = f"tail -n {lines} {path}"
- content = tail_text(files[0], max_lines=lines) or "Файл лога пуст"
- else:
- cmd = f"ls {pattern}"
- content = "Логи установки ещё не найдены"
- elif cfg["kind"] == "operations":
- cmd = f"tail -n {lines} /var/log/virtuality/operations/*.log"
- parts = []
- ensure_operations_dir()
- for item in sorted(OPERATIONS_DIR.glob('*.log'), key=lambda p: p.stat().st_mtime, reverse=True)[:10]:
- parts.append(f"===== {item.name} =====\n" + tail_text(item, max_lines=max(20, lines // 5)))
- content = "\n\n".join(parts) or "Журналы операций пока пусты"
- return {"key": key, "title": cfg["title"], "content": content, "path": path, "cmd": cmd, "lines": lines}
-'''
-
-if 'LOG_SOURCES = {' not in text:
- markers = [
- '\n\ndef parse_virsh_list() -> list[dict[str, str]]:',
- '\n\ndef list_recent_operations(limit: int = 20) -> list[dict[str, Any]]:',
- '\n\ndef safe_iso_filename(filename: str) -> str | None:',
- ]
- for marker in markers:
- if marker in text:
- text = text.replace(marker, helpers + marker, 1)
- changed.append('log center helpers added')
- break
- else:
- print('WARN: log helper marker not found, skip helper injection')
- changed.append('log center helpers skipped')
-else:
- start = text.find('LOG_SOURCES = {')
- end = text.find('\n}\n\n\ndef read_log_source', start)
- if start != -1 and end != -1:
- replacement = helpers.strip().split('\n\n\ndef read_log_source', 1)[0]
- text = text[:start] + replacement + text[end + 3:]
- changed.append('log sources refreshed without telegram')
- else:
- text = text.replace(' "telegram": {"title": "Telegram notifier", "kind": "file", "path": "/var/log/virtuality/telegram_version_bot.log"},\n', '')
- changed.append('telegram log source removed')
-
-route = r'''
-
-@app.get("/logs", response_class=HTMLResponse)
-def logs_page(request: Request, source: str = "web", lines: int = 220):
- auth_redirect = require_auth(request)
- if auth_redirect:
- return auth_redirect
- selected = read_log_source(source, lines)
- return templates.TemplateResponse("logs.html", {
- "request": request,
- "app_name": APP_NAME,
- "user": AUTH_USER,
- "sources": LOG_SOURCES,
- "selected": selected,
- })
-
-
-@app.get("/api/logs")
-def api_logs(request: Request, source: str = "web", lines: int = 220):
- if not get_current_user(request):
- return JSONResponse({"ok": False, "error": "Unauthorized"}, status_code=401)
- return {"ok": True, "log": read_log_source(source, lines)}
-'''
-
-if '@app.get("/logs"' not in text:
- markers = ['\n\n@app.get("/vm/create", response_class=HTMLResponse)', '\n\n@app.get("/api/operations")', '\n\n@app.get("/iso", response_class=HTMLResponse)']
- for marker in markers:
- if marker in text:
- text = text.replace(marker, route + marker, 1)
- changed.append('logs routes added')
- break
- else:
- print('WARN: logs route insert marker not found, skip route injection')
- changed.append('logs routes skipped')
-else:
- changed.append('logs routes already present')
-
-app_path.write_text(text)
-
-sidebar_html = '''{% set path = request.url.path %}
-
-'''
-
-if templates_dir.exists():
- sidebar_path = templates_dir / '_sidebar.html'
- sidebar_path.write_text(sidebar_html)
- changed.append('_sidebar.html ensured')
-
- for path in sorted(templates_dir.glob('*.html')):
- if path.name in {'login.html', 'console.html', '_sidebar.html'}:
- continue
- html = path.read_text()
- if '' not in html:
- html = html.replace('