Viên ngọc phương Đông

ruạṛ

import json
import os
import secrets
import subprocess
import sys
import time
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional, Union

from app.config import bart_jobs_folder
from app.errors.error_codes import ErrorCode
from app.errors.exceptions import BartError
from app.utils.job_activity_logger import JobActivityLogger
from app.utils.utility_functions import read_json_file

DELAY_SECONDS = 15
SINGLE_USERS_INPUT_NAME_PREFIX = "single_users"
SINGLE_USERS_OUTPUT_NAME_PREFIX = "single_users_output"
# Combined dummy output (same row schema as ``bulk_users_output.json`` — JSON array)
SINGLE_USERS_AGGREGATE_OUTPUT_NAME = "single_users_aggregate_output.json"
START_SINGLE_USER_BACKUP_SCRIPT = "/usr/local/bin/start_single_user_backup.py"

# Long-running single backup; override with BART_SINGLE_USER_BACKUP_TIMEOUT_SEC if needed.
_DEFAULT_TIMEOUT_SEC = int(os.getenv("BART_SINGLE_USER_BACKUP_TIMEOUT_SEC", "86400"))

activity_logger: Optional[JobActivityLogger] = None

def single_user_backup(request_body: Dict[str, Any], is_cli: bool = False) -> Dict[str, Any]:
    cbs_job_id = request_body.get("cbs_job_id")
    set_job_logger(cbs_job_id, is_cli)
    activity_logger.info(f"<======== Starting Single-User Backups ========>")
    job_folder = create_job_folder(cbs_job_id)
    save_input_json_files(job_folder, request_body)

for request_id in request_body.get("backup_requests").keys():
        call_single_user_backup_script(cbs_job_id, job_folder, request_id)
        activity_logger.info(f"==> Single user backup for request_id {request_id} is completed")

aggregate_path = write_single_users_aggregate_from_opt_bart_jobs(job_folder, request_body)
    return generate_callback_data(aggregate_path)

def set_job_logger(job_id, is_cli):
    global activity_logger
    activity_logger = JobActivityLogger(job_id, echo_to_stdout=is_cli)

def create_job_folder(cbs_id: int) -> str:
    job_folder = f"{bart_jobs_folder}/job_{cbs_id}"
    os.makedirs(job_folder, exist_ok=True)
    activity_logger.success(f"Created job folder: {job_folder}")
    return job_folder

def save_input_json_files(job_folder: str, request_body: Dict[str, Any]) -> None:
    for request_id in request_body.get("backup_requests").keys():
        input_json_file = f"{SINGLE_USERS_INPUT_NAME_PREFIX}_{request_id}.json"
        input_json_path = os.path.join(job_folder, input_json_file)
        payload = request_body.get("backup_requests", {}).get(request_id)
        if not isinstance(payload, dict):
            activity_logger.error(f"backup_requests[{request_id!r}] must be a JSON object, got {type(payload).__name__}")
            raise BartError(ErrorCode.VALIDATION_FAILED, f"backup_requests entry for {request_id!r} must be an object")
        to_write = {**payload, "preference_id": str(request_id)}
        try:
            with open(input_json_path, "w", encoding="utf-8") as f:
                json.dump(to_write, f, indent=2)
        except OSError as e:
            activity_logger.error(f"Failed to write input JSON: {e}")
            raise BartError(ErrorCode.PERMISSION_DENIED, f"Cannot write {input_json_path}: {e}") from e
        activity_logger.note(f"Wrote request for {request_id} payload to {input_json_path}")
    activity_logger.note(f"Wrote all input json files")

def write_single_users_aggregate_from_opt_bart_jobs(job_folder: str, request_body: Dict[str, Any]) -> str:
    assert activity_logger is not None
    output_rows: List[Dict[str, Any]] = []
    for request_id, payload in request_body["backup_requests"].items():
        job_id = str(payload.get("job_id", "")).strip()
        if not job_id:
            activity_logger.error(f"Missing job_id in backup_requests[{request_id!r}]")
            raise BartError(
                ErrorCode.VALIDATION_FAILED,
                f"backup_requests[{request_id!r}] must include a non-empty job_id",
            )
        output_json_path = os.path.join(job_folder, f"{job_id}.json")
        if not os.path.isfile(output_json_path):
            activity_logger.error(f"Output JSON not found: {output_json_path}")
            raise BartError(
                ErrorCode.FILE_NOT_FOUND,
                f"Expected output file for job_id {job_id!r}: {output_json_path}",
            )
        try:
            row = _row_from_opt_bart_status_json(output_json_path)
        except ValueError as e:
            activity_logger.error(f"Invalid JSON in {output_json_path}: {e}")
            raise BartError(ErrorCode.VALIDATION_FAILED, str(e)) from e
        pref_s = str(request_id)
        row = {**row, "preference_id": pref_s}
        output_rows.append(row)
        activity_logger.note(f"Included status row from {output_json_path} (preference_id={pref_s})")

aggregate_path = os.path.join(job_folder, SINGLE_USERS_AGGREGATE_OUTPUT_NAME)
    try:
        with open(aggregate_path, "w", encoding="utf-8") as f:
            json.dump(output_rows, f, indent=2)
    except OSError as e:
        activity_logger.error(f"Failed to write aggregate output JSON: {e}")
        raise BartError(ErrorCode.PERMISSION_DENIED, f"Cannot write {aggregate_path}: {e}") from e
    activity_logger.success(
        f"Wrote single-user aggregate output ({len(output_rows)} row(s)) to {aggregate_path}"
    )
    return aggregate_path

def _row_from_opt_bart_status_json(path: str) -> Dict[str, Any]:

raw = read_json_file(path)
    if isinstance(raw, list):
        if len(raw) == 1 and isinstance(raw[0], dict):
            return dict(raw[0])
        raise BartError( ErrorCode.VALIDATION_FAILED,
                         f"{path} must be a JSON object or a one-element array of an object")
    if isinstance(raw, dict):
        return dict(raw)
    raise BartError(ErrorCode.VALIDATION_FAILED,
                    f"{path} must contain a JSON object or array, got {type(raw).__name__}")

def call_single_user_backup_script(cbs_job_id: int, job_folder: str, request_id: Union[str, int]):
    input_json_file = f"{SINGLE_USERS_INPUT_NAME_PREFIX}_{request_id}.json"
    input_json_path = os.path.join(job_folder, input_json_file)
    if not os.path.isfile(input_json_path):
        activity_logger.error(f"Input file not found: {input_json_path}")
        raise BartError(ErrorCode.FILE_NOT_FOUND, f"Missing single input file: {input_json_path}")

if not os.path.isfile(START_SINGLE_USER_BACKUP_SCRIPT):
        activity_logger.error(f"Backup script not found: {START_SINGLE_USER_BACKUP_SCRIPT}")
        raise BartError(
            ErrorCode.FILE_NOT_FOUND,
            f"Single-user backup script is not installed: {START_SINGLE_USER_BACKUP_SCRIPT}",
        )

cmd = ["sudo", "-n", sys.executable, START_SINGLE_USER_BACKUP_SCRIPT,
           "--single-user-json", input_json_path,
           "--parent-job-id", cbs_job_id]
    activity_logger.note(f"Running single user backup: {' '.join(cmd)}")
    activity_logger.info(f"Script timeout set to {_DEFAULT_TIMEOUT_SEC} seconds")

started = time.monotonic()
    try:
        result = subprocess.run(
            cmd,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            timeout=_DEFAULT_TIMEOUT_SEC,
            check=False,
        )
    except subprocess.TimeoutExpired as e:
        elapsed = time.monotonic() - started
        activity_logger.error(f"Single user backup script timed out after {elapsed:.1f}s")
        raise BartError(
            ErrorCode.TIMEOUT,
            f"start_single_user_backup.py exceeded {_DEFAULT_TIMEOUT_SEC} seconds",
        ) from e
    except OSError as e:
        activity_logger.error(f"Failed to execute single user backup script: {e}")
        raise BartError(ErrorCode.COMMAND_FAILED, f"Could not run backup script: {e}") from e

elapsed = time.monotonic() - started
    activity_logger.info(f"Single user backup script finished in {elapsed:.3f} seconds (exit {result.returncode})")

if result.stdout and result.stdout.strip():
        activity_logger.debug(f"Script stdout (last 2k): {result.stdout.strip()[-2000:]!r}")
    if result.stderr and result.stderr.strip():
        activity_logger.note(f"Script stderr (last 2k): {result.stderr.strip()[-2000:]!r}")

if result.returncode != 0:
        err_tail = result.stderr or result.stdout or "no output"
        activity_logger.error(f"Single user backup script failed with exit code {result.returncode}: {err_tail}")
        raise BartError(ErrorCode.COMMAND_FAILED, f"start_single_user_backup.py failed (rc={result.returncode}): {err_tail}")

def _docroot_paths_from_job(job: Dict[str, Any]) -> list:
    paths = []
    for d in job.get("docroots") or []:
        if isinstance(d, dict) and d.get("path"):
            paths.append(str(d["path"]))
    return paths

def _job_prefix_from_folder(job_folder: str) -> str:
    base = os.path.basename(os.path.normpath(job_folder))
    if base.startswith("job_"):
        return f"{bart_jobs_folder}/{base}"
    return job_folder

def _list_single_user_input_files(job_folder: str) -> list:
    """Paths to ``single_users_<id>.json`` inputs (excludes ``single_users_output_*``)."""
    found = []
    in_prefix = f"{SINGLE_USERS_INPUT_NAME_PREFIX}_"
    out_prefix = f"{SINGLE_USERS_OUTPUT_NAME_PREFIX}_"
    for fn in sorted(os.listdir(job_folder)):
        if not fn.endswith(".json"):
            continue
        if fn.startswith(out_prefix):
            continue
        if fn.startswith(in_prefix):
            found.append(os.path.join(job_folder, fn))
    return found

def _reference_id_from_single_user_input_filename(filename: str) -> str:
    # single_users_7.json -> 7
    base = os.path.basename(filename)
    if not base.endswith(".json") or not base.startswith(f"{SINGLE_USERS_INPUT_NAME_PREFIX}_"):
        return ""
    return base[len(f"{SINGLE_USERS_INPUT_NAME_PREFIX}_") : -len(".json")]

def _write_dummy_single_user_artifact_files(
    preference_id: str,
    file_list_path: str,
    changed_files_list_path: str,
) -> None:
    """Create placeholder file-list and backup-diff files for dummy single-user backup."""
    assert activity_logger is not None
    ref_s = str(preference_id)
    nonce = secrets.token_hex(8)
    try:
        os.makedirs(os.path.dirname(file_list_path), exist_ok=True)
        with open(file_list_path, "w", encoding="utf-8") as f:
            f.write(
                f"# BART dummy single-user file list (preference_id={ref_s} nonce={nonce})\n"
                f"public_html/index.php\n"
                f"homedir/tmp/.dummy-{nonce[:12]}.dat\n"
            )
        with open(changed_files_list_path, "w", encoding="utf-8") as f:
            f.write(
                f"# BART dummy single-user changed files (preference_id={ref_s} nonce={nonce})\n"
                f"M\tpublic_html/wp-config.php\n"
                f"A\tmail/.Trash_{nonce[:6]}\n"
            )
    except OSError as e:
        activity_logger.error(f"Failed to write dummy single-user artifacts under {ref_s}: {e}")
        raise BartError(
            ErrorCode.PERMISSION_DENIED,
            f"Cannot write dummy single-user files for reference {ref_s}: {e}",
        ) from e
    activity_logger.note(f"Wrote dummy artifacts {file_list_path} and {changed_files_list_path}")

def call_single_user_backup_script_dummy(job_folder: str) -> str:
    """
    For each ``single_users_<id>.json`` input, write ``single_users_output_<id>.json`` (one object)
    and write ``single_users_aggregate_output.json`` as a JSON array of rows (same shape as bulk output).
    """
    assert activity_logger is not None

input_paths = _list_single_user_input_files(job_folder)
    if not input_paths:
        activity_logger.error(f"No single-user input JSON files in {job_folder}")
        raise BartError(ErrorCode.FILE_NOT_FOUND, f"No {SINGLE_USERS_INPUT_NAME_PREFIX}_*.json files in {job_folder}")

job_prefix = _job_prefix_from_folder(job_folder)
    output_rows = []

for input_json_path in input_paths:
        preference_id = _reference_id_from_single_user_input_filename(input_json_path)
        if not preference_id:
            activity_logger.warn(f"Skipping file with unexpected name: {input_json_path}")
            continue

try:
            job = read_json_file(input_json_path)
        except ValueError as e:
            activity_logger.error(f"Invalid JSON in {input_json_path}: {e}")
            raise BartError(ErrorCode.VALIDATION_FAILED, str(e)) from e

if not isinstance(job, dict):
            raise BartError(ErrorCode.VALIDATION_FAILED, f"{input_json_path} must contain a JSON object")

pref_s = str(preference_id)
        user_subdir = os.path.normpath(os.path.join(job_prefix, pref_s))
        file_list_path = os.path.join(user_subdir, "file-list.txt")
        changed_files_list_path = os.path.join(user_subdir, "backup-diff.txt")

_write_dummy_single_user_artifact_files(pref_s, file_list_path, changed_files_list_path)

succeeded_docroots = _docroot_paths_from_job(job)
        succeeded_dbs = [str(x) for x in (job.get("databases") or []) if x is not None]
        row = {
            "job_id": str(job.get("job_id", preference_id)),
            "preference_id": preference_id,
            "status": "success",
            "size_bytes": 1024,
            "succeeded_docroots": succeeded_docroots,
            "succeeded_dbs": succeeded_dbs,
            "failed_docroots": [],
            "failed_dbs": [],
            "file_list_path": file_list_path,
            "changed_files_list_path": changed_files_list_path,
        }
        output_rows.append(row)

per_output_path = os.path.join(
            job_folder, f"{SINGLE_USERS_OUTPUT_NAME_PREFIX}_{preference_id}.json"
        )
        try:
            with open(per_output_path, "w", encoding="utf-8") as f:
                json.dump(row, f, indent=2)
        except OSError as e:
            activity_logger.error(f"Failed to write output JSON: {e}")
            raise BartError(ErrorCode.PERMISSION_DENIED, f"Cannot write {per_output_path}: {e}") from e

activity_logger.success(
        f"Wrote dummy single-user output ({len(output_rows)} row(s)) to {aggregate_path}"
    )
    return aggregate_path

def generate_callback_data_dummy(request_body, output_json_path):
    time.sleep(DELAY_SECONDS)
    backup_requests = request_body.get("backup_requests") or {}
    today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    action_data = {}
    for idx, req_id in enumerate(backup_requests.keys()):
        n = idx + 1
        action_data[req_id] = {
            "backup_status": "success",
            "file_list_path": f"customer_scheduled/custbacktest{n}/{today}/file_list.json",
            "changed_file_list_path": f"customer_scheduled/custbacktest{n}/{today}/changed_file_list.json",
            "backup_size_bytes": 1024,
        }
    return action_data

def generate_callback_data(output_json_path: str) -> Dict[str, Any]:
    """
    Read ``single_users_aggregate_output.json`` (JSON array of per-job results) and build CBS
    ``action_data`` keyed by ``preference_id``, same shape as :func:`generate_callback_data_dummy`.
    """
    assert activity_logger is not None

if not output_json_path or not os.path.isfile(output_json_path):
        activity_logger.error(f"Output file not found: {output_json_path}")
        raise BartError(ErrorCode.FILE_NOT_FOUND, f"Missing single-user aggregate output file: {output_json_path}")

try:
        data = read_json_file(output_json_path)
    except ValueError as e:
        activity_logger.error(f"Invalid JSON in {output_json_path}: {e}")
        raise BartError(ErrorCode.VALIDATION_FAILED, str(e)) from e

if not isinstance(data, list):
        activity_logger.error(f"Expected JSON array in {output_json_path}, got {type(data).__name__}")
        raise BartError(ErrorCode.VALIDATION_FAILED, f"{output_json_path} must contain a JSON array")

action_data: Dict[str, Any] = {}
    for row in data:
        if not isinstance(row, dict):
            continue
        ref = row.get("preference_id")
        if ref is None:
            ref = row.get("job_id")
        if ref is None:
            activity_logger.warn(f"Skipping output row without preference_id/job_id: {row!r}")
            continue
        ref_s = str(ref)
        action_data[ref_s] = {
            "backup_status": row.get("status", "unknown"),
            "file_list_path": row.get("file_list_path", ""),
            "changed_file_list_path": row.get("changed_files_list_path", ""),
            "backup_size_bytes": row.get("size_bytes", 0),
            "succeeded_docroots": row.get("succeeded_docroots", []),
            "succeeded_dbs": row.get("succeeded_dbs", []),
            "failed_docroots": row.get("failed_docroots", []),
            "failed_dbs": row.get("failed_dbs", []),
        }

activity_logger.note(f"Built action_data for {len(action_data)} reference(s) from {output_json_path}")
    return action_data

Register