c16a4f23b4
Phase 1a of a multi-session refactor toward a clean blueprint
structure. Pulls the helper code that lived alongside the routes in
the 1621-line app.py into focused modules. app.py is now 625 lines
of mostly routes plus a small Flask wiring header. Behaviour is
unchanged: smoke-tested against the 8 main GET routes (200 OK).
New modules:
- config.py env vars + IMAGE_TYPES + FRIENDLY_NAMES +
SHARED_DEPLOY_* taxonomy + unattend XML
namespaces.
- services/audit.py audit log file handler + audit() helper.
- services/csrf.py session CSRF token + before_request validator
wired via init_csrf(app).
- services/fs.py image_root / deploy_path / unattend_path /
control_path / tools_path + load_json /
save_json + resolve_destination.
- services/system.py service_status / find_usb_mounts /
find_upload_sources.
- services/images.py image_status + load_image_config.
- services/deploy.py import_deploy + _merge_tree +
_replace_with_symlink + allowed_import_source.
- services/unattend.py parse_unattend / build_unattend_xml /
extract_form_data and the qn / qwcm / settings
pass helpers.
- services/wim.py extract_startnet / update_startnet / list_files
wrapping wimextract / wimupdate / wimdir.
Endpoint names kept stable (dashboard, clonezilla_backups, etc.) so
existing url_for(...) calls in templates are unchanged. Phase 1b
(Flask blueprints with ".endpoint" naming) deferred to a future
session because it requires updating ~30 url_for sites in templates
and is mostly cosmetic.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
56 lines
1.7 KiB
Python
56 lines
1.7 KiB
Python
"""Host-system probes: systemd service status, USB mounts, upload sources."""
|
|
|
|
import os
|
|
import subprocess
|
|
|
|
import config
|
|
|
|
|
|
def service_status(service_name):
|
|
"""Check whether a systemd service is active."""
|
|
try:
|
|
result = subprocess.run(
|
|
["systemctl", "is-active", service_name],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=5,
|
|
)
|
|
state = result.stdout.strip()
|
|
return {"name": service_name, "active": state == "active", "state": state}
|
|
except Exception as exc:
|
|
return {"name": service_name, "active": False, "state": str(exc)}
|
|
|
|
|
|
def find_usb_mounts():
|
|
"""Return a list of mount-point paths that look like removable media."""
|
|
mounts = []
|
|
try:
|
|
with open("/proc/mounts", "r") as fh:
|
|
for line in fh:
|
|
parts = line.split()
|
|
if len(parts) >= 2:
|
|
mount_point = parts[1]
|
|
if mount_point.startswith(("/mnt/", "/media/")):
|
|
if os.path.isdir(mount_point):
|
|
mounts.append(mount_point)
|
|
except OSError:
|
|
pass
|
|
return sorted(set(mounts))
|
|
|
|
|
|
def find_upload_sources():
|
|
"""Return sub-directories inside UPLOAD_DIR that look like image content."""
|
|
sources = []
|
|
if os.path.isdir(config.UPLOAD_DIR):
|
|
try:
|
|
entries = os.listdir(config.UPLOAD_DIR)
|
|
if entries:
|
|
sources.append(config.UPLOAD_DIR)
|
|
for entry in entries:
|
|
full = os.path.join(config.UPLOAD_DIR, entry)
|
|
if os.path.isdir(full):
|
|
sources.append(full)
|
|
except OSError:
|
|
pass
|
|
return sources
|