nas-tools/dashboard/backend/routers/media_mgmt.py

import json
import logging
import os
import re
from datetime import datetime

import asyncssh
from fastapi import APIRouter, Depends, HTTPException, BackgroundTasks, Query
import config

router = APIRouter()
logger = logging.getLogger(__name__)

# Absolute paths to media-mgmt scripts on NAS host
HEALTH_CHECK_SCRIPT = "/volume1/homes/zjgump/media-mgmt/immich/scripts/health-check.sh"
STORAGE_REPORT_SCRIPT = "/volume1/homes/zjgump/media-mgmt/diagnostics/storage-report.sh"
SCAN_LIBRARY_SCRIPT = "/volume1/homes/zjgump/media-mgmt/immich/scripts/scan-library.sh"
THUMBNAIL_CLINIC_SCRIPT = "/volume1/homes/zjgump/media-mgmt/diagnostics/thumbnail-clinic.sh"
SHUTTLE_SCRIPT = "/volume1/homes/zjgump/media-mgmt/transfer/ssd-shuttle.sh"
PHONE_INGEST_SCRIPT = "/volume1/homes/zjgump/media-mgmt/transfer/phone-to-nas.sh"

LOG_DIR = "/volume1/homes/zjgump/media-mgmt/logs"

UNAVAILABLE_RESPONSE = {"status": "unavailable", "error": "NAS unreachable"}


async def run_host_command(cmd: str) -> tuple[int, str, str]:
    """Execute a command on the NAS host via SSH.

    Returns (exit_code, stdout, stderr). On SSH connection failure,
    exit_code=999 and stderr contains the error message.
    """
    try:
        full_cmd = f'PATH="/volume1/homes/zjgump/bin:$PATH" NAS_HOST="naslan" {cmd}'
        async with asyncssh.connect(
            config.SSH_HOST,
            username=config.SSH_USER,
            client_keys=[config.SSH_KEY_PATH],
            known_hosts=None,
        ) as conn:
            result = await conn.run(full_cmd)
            return result.exit_status, result.stdout, result.stderr
    except Exception as e:
        logger.error(f"SSH command failed ({cmd}): {e}")
        return 999, "", str(e)


@router.get("/health")
async def get_health():
    """Runs health-check.sh and returns structured status."""
    code, stdout, stderr = await run_host_command(f"bash {HEALTH_CHECK_SCRIPT}")
    if code == 999:
        return UNAVAILABLE_RESPONSE
    return {
        "status": "ok" if code == 0 else "warning" if code == 1 else "error",
        "exit_code": code,
        "stdout": stdout,
        "stderr": stderr,
    }


@router.get("/storage")
async def get_storage():
    """Runs storage-report.sh -j and returns the parsed JSON storage report."""
    code, stdout, stderr = await run_host_command(f"bash {STORAGE_REPORT_SCRIPT} -j")
    if code == 999:
        return UNAVAILABLE_RESPONSE
    if code != 0:
        return {"error": "Storage report failed", "stderr": stderr}
    try:
        return json.loads(stdout)
    except json.JSONDecodeError:
        return {"error": "Failed to parse JSON from storage report", "raw_output": stdout}


@router.get("/shuttle/status")
async def get_shuttle_status():
    """Shuttle pipeline status: queue depth, last transfer time, recent errors.

    Reads the shuttle log file to extract pipeline state.
    """
    log_path = f"{LOG_DIR}/shuttle.log"
    code, stdout, stderr = await run_host_command(f"cat {log_path} 2>/dev/null || echo 'NO_LOG'")
    if code == 999:
        return UNAVAILABLE_RESPONSE

    result = {
        "running": False,
        "queue_depth": 0,
        "last_transfer": None,
        "last_transfer_size": None,
        "recent_errors": [],
        "current_stage": None,
    }

    content = stdout or ""
    if content.strip() == "NO_LOG":
        return result

    # Check if shuttle is currently running via pgrep
    running_code, running_out, _ = await run_host_command(f"pgrep -f {SHUTTLE_SCRIPT}")
    result["running"] = running_code == 0
    if running_code == 0:
        result["current_stage"] = "active"
        result["pid"] = running_out.strip()

    # Parse log for last transfer, errors, current stage
    lines = content.strip().split("\n")
    error_lines = []
    transfer_lines = []

    for line in lines:
        if "❌" in line or "error" in line.lower() or "failed" in line.lower():
            error_lines.append(line.strip())
        if "copied" in line.lower() or "transferred" in line.lower() or "rsync" in line.lower():
            transfer_lines.append(line.strip())
        if "Stage" in line and ":" in line:
            result["current_stage"] = line.strip()

    if error_lines:
        result["recent_errors"] = error_lines[-10:]

    if transfer_lines:
        last = transfer_lines[-1]
        result["last_transfer"] = last
        # Try to extract size from rsync-like output
        size_match = re.search(r"(\d+\.?\d*)\s*(K|M|G)B", last)
        if size_match:
            result["last_transfer_size"] = size_match.group(0)

    # Count queued source directories
    queue_matches = [l for l in lines if "--source" in l or "--stage" in l]
    result["queue_depth"] = len(queue_matches)

    return result


@router.post("/maintenance/scan")
async def run_scan(background_tasks: BackgroundTasks, library: str = Query(None)):
    """Triggers scan-library.sh in the background."""
    cmd = f"bash {SCAN_LIBRARY_SCRIPT}"
    if library:
        cmd += f" {library}"

    log_file = f"{LOG_DIR}/scan.log"
    run_cmd = f"mkdir -p {LOG_DIR} && echo '=== Scan Started at $(date) ===' > {log_file} && nohup {cmd} >> {log_file} 2>&1 &"

    code, stdout, stderr = await run_host_command(run_cmd)
    if code == 999:
        return UNAVAILABLE_RESPONSE
    if code != 0:
        return {"error": "Failed to start scan", "stderr": stderr}

    return {"message": "Scan started in background", "log_file": log_file}


@router.post("/maintenance/requeue")
async def run_requeue(background_tasks: BackgroundTasks):
    """Triggers thumbnail-clinic.sh -r in the background."""
    cmd = f"bash {THUMBNAIL_CLINIC_SCRIPT} -r"
    log_file = f"{LOG_DIR}/requeue.log"
    run_cmd = f"mkdir -p {LOG_DIR} && echo '=== Requeue Started at $(date) ===' > {log_file} && nohup {cmd} >> {log_file} 2>&1 &"

    code, stdout, stderr = await run_host_command(run_cmd)
    if code == 999:
        return UNAVAILABLE_RESPONSE
    if code != 0:
        return {"error": "Failed to start thumbnail requeue", "stderr": stderr}

    return {"message": "Thumbnail requeue started in background", "log_file": log_file}


@router.get("/logs/{task_name}")
async def get_logs(task_name: str):
    """Reads logs for a task from the media-mgmt logs directory."""
    if task_name not in ["scan", "requeue", "shuttle", "phone-ingest"]:
        raise HTTPException(status_code=400, detail="Invalid task name")

    log_path = f"{LOG_DIR}/{task_name}.log"
    code, stdout, stderr = await run_host_command(f"cat {log_path} 2>/dev/null || echo 'NO_LOG'")
    if code == 999:
        return UNAVAILABLE_RESPONSE

    if stdout.strip() == "NO_LOG":
        return {"content": "No log file found."}
    lines = stdout.strip().split("\n")
    return {"content": "\n".join(lines[-200:])}


@router.get("/status")
async def get_active_status():
    """Checks if any ingestion/maintenance processes are currently running on the host."""
    processes = {
        "scan": SCAN_LIBRARY_SCRIPT,
        "requeue": THUMBNAIL_CLINIC_SCRIPT,
        "shuttle": SHUTTLE_SCRIPT,
        "phone-ingest": PHONE_INGEST_SCRIPT,
    }

    status = {}
    for name, script_path in processes.items():
        code, stdout, stderr = await run_host_command(f"pgrep -f {script_path}")
        if code == 999:
            return UNAVAILABLE_RESPONSE
        status[name] = {
            "running": code == 0,
            "pid": stdout.strip() if code == 0 else None,
        }
    return status