Claus Lohmar
7808b16946
Archives like 64bit.7z that already contain a 64bit/ top-level dir were extracted into STAGING_IN/64bit/, creating 64bit/64bit/ nesting. Now extract into STAGING_IN root, then: - If the archive produced a single dir matching the stem, use it - Otherwise, collect scattered files into a stem-named subdirectory Removes the need for nested path: 64bit/Debian.vmdk instead of 64bit/64bit/Debian.vmdk
288 lines
9.3 KiB
Python
288 lines
9.3 KiB
Python
"""
|
|
Archive extraction, disk probing, and OS detection.
|
|
|
|
Runs on the Proxmox host (srv2) where qemu-img, 7z, unzip, tar are available.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
from pathlib import Path
|
|
from dataclasses import dataclass
|
|
from typing import Optional
|
|
|
|
STAGING_IN = Path("/mnt/converter/in")
|
|
STAGING_OUT = Path("/mnt/converter/out")
|
|
|
|
DISK_EXTENSIONS = {".vmdk", ".qcow2", ".qcow", ".img", ".raw", ".vhd", ".vhdx"}
|
|
|
|
ARCHIVE_EXTENSIONS = {".7z", ".zip", ".tar.gz", ".tgz", ".tar", ".gz", ".bz2", ".xz"}
|
|
|
|
MIN_DISK_BYTES = 1024
|
|
|
|
|
|
@dataclass
|
|
class DiskInfo:
|
|
path: Path
|
|
format: str # from qemu-img
|
|
size_bytes: int
|
|
size_gb: float
|
|
os_type: Optional[str] = None
|
|
bootable: bool = False
|
|
efi_detectable: bool = False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Extraction
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def extract_if_needed(filename: str) -> Path:
|
|
"""Extract an archive into STAGING_IN if it's compressed. Returns the path
|
|
to the extracted directory (or the original file if not compressed)."""
|
|
filepath = STAGING_IN / filename
|
|
if not filepath.exists():
|
|
raise FileNotFoundError(f"Source not found: {filepath}")
|
|
|
|
# Detect archive type
|
|
name = filepath.name.lower()
|
|
ext = _detect_archive_ext(name)
|
|
|
|
if ext is None:
|
|
return filepath # not an archive, return as-is
|
|
|
|
# Determine extraction tool
|
|
tool_map = {
|
|
".7z": ("7z", ["7z", "x", "-y", str(filepath)]),
|
|
".zip": ("unzip", ["unzip", "-o", str(filepath)]),
|
|
".tar.gz": ("tar", ["tar", "-xzf", str(filepath)]),
|
|
".tgz": ("tar", ["tar", "-xzf", str(filepath)]),
|
|
".tar": ("tar", ["tar", "-xf", str(filepath)]),
|
|
".gz": ("gunzip", ["gunzip", "-f", str(filepath)]),
|
|
".bz2": ("bunzip2", ["bunzip2", "-f", str(filepath)]),
|
|
".xz": ("xz", ["xz", "-d", str(filepath)]),
|
|
}
|
|
tool, cmd = tool_map[ext]
|
|
|
|
if not shutil.which(tool):
|
|
raise RuntimeError(f"Extraction tool '{tool}' not found on host")
|
|
|
|
# Directory stem: strip extension(s)
|
|
stem = name
|
|
for e in sorted(tool_map, key=len, reverse=True):
|
|
if stem.endswith(e):
|
|
stem = stem[:-len(e)]
|
|
break
|
|
|
|
# Snapshot STAGING_IN before extraction
|
|
before = set(STAGING_IN.iterdir()) if STAGING_IN.exists() else set()
|
|
|
|
# Extract into STAGING_IN root (not a subdir) to avoid double nesting
|
|
# when the archive already has a top-level directory
|
|
result = subprocess.run(cmd, cwd=str(STAGING_IN), capture_output=True, text=True, timeout=600)
|
|
if result.returncode != 0:
|
|
raise RuntimeError(f"{tool} failed: {result.stderr.strip()[:500]}")
|
|
|
|
# Determine what was created
|
|
after = set(STAGING_IN.iterdir())
|
|
new_items = after - before
|
|
|
|
# If the archive had a single top-level directory matching the stem, use it
|
|
expected_dir = STAGING_IN / stem
|
|
if expected_dir in new_items and expected_dir.is_dir():
|
|
return expected_dir
|
|
|
|
# Otherwise, collect scattered files into a subdirectory
|
|
out_dir = STAGING_IN / stem
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
for item in new_items:
|
|
dest = out_dir / item.name
|
|
if item.is_dir():
|
|
shutil.move(str(item), str(dest))
|
|
else:
|
|
shutil.move(str(item), str(dest))
|
|
return out_dir
|
|
|
|
|
|
def _detect_archive_ext(name: str) -> Optional[str]:
|
|
for ext in sorted(ARCHIVE_EXTENSIONS, key=len, reverse=True):
|
|
if name.endswith(ext):
|
|
return ext
|
|
return None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Disk discovery
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def discover_disk(source: Path) -> DiskInfo:
|
|
"""Walk the source directory and find the primary disk image."""
|
|
if source.is_file():
|
|
return _probe(source)
|
|
|
|
candidates = []
|
|
for dirpath, _, filenames in os.walk(source):
|
|
for fname in filenames:
|
|
fpath = Path(dirpath) / fname
|
|
if fpath.suffix.lower() in DISK_EXTENSIONS and fpath.stat().st_size >= MIN_DISK_BYTES:
|
|
candidates.append(fpath)
|
|
|
|
if not candidates:
|
|
raise FileNotFoundError(f"No disk image found in {source}")
|
|
|
|
# Pick largest as primary
|
|
candidates.sort(key=lambda p: p.stat().st_size, reverse=True)
|
|
return _probe(candidates[0])
|
|
|
|
|
|
def _probe(disk_path: Path) -> DiskInfo:
|
|
"""Run qemu-img info and extract format + virtual size."""
|
|
result = subprocess.run(
|
|
["qemu-img", "info", "--output=json", str(disk_path)],
|
|
capture_output=True, text=True, timeout=30,
|
|
)
|
|
|
|
fmt = disk_path.suffix.lstrip(".").lower()
|
|
virtual_size = disk_path.stat().st_size
|
|
|
|
if result.returncode == 0:
|
|
import json
|
|
data = json.loads(result.stdout)
|
|
fmt = data.get("format", fmt)
|
|
virtual_size = data.get("virtual-size", virtual_size)
|
|
|
|
size_gb = round(virtual_size / (1024 ** 3), 1)
|
|
|
|
return DiskInfo(
|
|
path=disk_path,
|
|
format=fmt,
|
|
size_bytes=virtual_size,
|
|
size_gb=size_gb,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# OS detection
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def detect_os(disk_path: Path, extract_dir: Path) -> Optional[str]:
|
|
"""Try multiple strategies to identify the guest OS.
|
|
|
|
1. Parse .vmx sidecar file for guestOS field
|
|
2. Read /etc/os-release from disk via libguestfs if available
|
|
3. Heuristic from filename / directory name
|
|
"""
|
|
# Strategy 1: .vmx sidecar
|
|
vmx = _find_sidecar(extract_dir, ".vmx")
|
|
if vmx:
|
|
os_type = _parse_vmx_guest_os(vmx)
|
|
if os_type:
|
|
return os_type
|
|
|
|
# Strategy 2: libguestfs (if available)
|
|
if shutil.which("virt-inspector"):
|
|
try:
|
|
result = subprocess.run(
|
|
["virt-inspector", str(disk_path)],
|
|
capture_output=True, text=True, timeout=60,
|
|
)
|
|
if result.returncode == 0:
|
|
import xml.etree.ElementTree as ET
|
|
root = ET.fromstring(result.stdout)
|
|
os_elem = root.find(".//operatingsystem")
|
|
if os_elem is not None and os_elem.text:
|
|
return f"{os_elem.text}"
|
|
except Exception:
|
|
pass
|
|
|
|
# Strategy 3: filename heuristics
|
|
name = disk_path.name.lower()
|
|
os_hints = {
|
|
"debian": "Debian",
|
|
"ubuntu": "Ubuntu",
|
|
"centos": "CentOS",
|
|
"rhel": "RHEL",
|
|
"fedora": "Fedora",
|
|
"windows": "Windows",
|
|
"win10": "Windows 10",
|
|
"win11": "Windows 11",
|
|
"alpine": "Alpine",
|
|
"arch": "Arch Linux",
|
|
"opensuse": "openSUSE",
|
|
}
|
|
for keyword, label in os_hints.items():
|
|
if keyword in name:
|
|
return label
|
|
|
|
# Also check parent directory name
|
|
parent = str(extract_dir).lower()
|
|
for keyword, label in os_hints.items():
|
|
if keyword in parent:
|
|
return label
|
|
|
|
return None
|
|
|
|
|
|
def _find_sidecar(base: Path, ext: str) -> Optional[Path]:
|
|
"""Find first file with given extension in the directory tree."""
|
|
if not base.is_dir():
|
|
return None
|
|
for dirpath, _, filenames in os.walk(base):
|
|
for fname in filenames:
|
|
if fname.lower().endswith(ext):
|
|
return Path(dirpath) / fname
|
|
return None
|
|
|
|
|
|
def _parse_vmx_guest_os(vmx_path: Path) -> Optional[str]:
|
|
"""Parse VMware .vmx file for guestOS field."""
|
|
try:
|
|
text = vmx_path.read_text(errors="ignore")
|
|
except OSError:
|
|
return None
|
|
m = re.search(r'guestOS\s*=\s*"([^"]+)"', text, re.IGNORECASE)
|
|
if not m:
|
|
return None
|
|
raw = m.group(1)
|
|
# Map common VMware guestOS values
|
|
mapping = {
|
|
"debian": "Debian", "ubuntu": "Ubuntu", "centos": "CentOS",
|
|
"rhel": "RHEL", "fedora": "Fedora", "windows": "Windows",
|
|
"other": "Other Linux", "other-64": "Other Linux (64-bit)",
|
|
"other26xlinux": "Linux 2.6.x", "other3xlinux": "Linux 3.x+",
|
|
"other4xlinux": "Linux 4.x+", "other5xlinux": "Linux 5.x+",
|
|
}
|
|
raw_lower = raw.lower().replace("_", "").replace("-", "").replace(" ", "")
|
|
for k, v in mapping.items():
|
|
if k in raw_lower:
|
|
return v
|
|
return raw.replace("-", " ").replace("_", " ").title()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# EFI detection
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def detect_efi(disk_path: Path) -> Optional[bool]:
|
|
"""Check whether the disk has an EFI System Partition using guestfish.
|
|
|
|
Returns True if a VFAT EFI partition is found, False if only non-EFI
|
|
filesystems detected, None if guestfish isn't available or fails.
|
|
"""
|
|
if not shutil.which("guestfish"):
|
|
return None
|
|
try:
|
|
result = subprocess.run(
|
|
["guestfish", "--ro", "-a", str(disk_path), "-i", "list-filesystems"],
|
|
capture_output=True, text=True, timeout=30,
|
|
)
|
|
if result.returncode != 0:
|
|
return None
|
|
# EFI System Partition is typically VFAT; look for it
|
|
if "vfat" in result.stdout.lower():
|
|
return True
|
|
return False
|
|
except Exception:
|
|
return None
|