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

import os
import stat
from typing import Any, Dict, Optional

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

activity_logger: Optional[JobActivityLogger] = None

def _set_logger(cbs_job_id, is_cli: bool = False) -> None:
    global activity_logger
    activity_logger = JobActivityLogger(cbs_job_id, echo_to_stdout=is_cli)

def generate_ssh_key(request_body: Dict[str, Any], is_cli: bool = False) -> Dict[str, Any]:
    cbs_job_id = request_body.get("cbs_job_id")
    if cbs_job_id is None:
        raise BartError(ErrorCode.VALIDATION_FAILED, "cbs_job_id is required")
    _set_logger(cbs_job_id, is_cli)
    assert activity_logger is not None

key_req = request_body.get("generate_ssh_key_requests") or {}
    if not isinstance(key_req, dict) or not key_req:
        raise BartError(ErrorCode.VALIDATION_FAILED, "generate_ssh_key_requests must be a non-empty object")

job_folder = os.path.join(bart_jobs_folder, f"job_{int(cbs_job_id)}")
    os.makedirs(job_folder, exist_ok=True)

action_data: Dict[str, Any] = {}
    for req_id, spec in key_req.items():
        try:
            if not isinstance(spec, dict):
                raise BartError(ErrorCode.VALIDATION_FAILED, f"Invalid entry for request id {req_id!r}")
            ssh_key = spec.get("ssh_key")
            if ssh_key is None or not isinstance(ssh_key, str):
                raise BartError(ErrorCode.VALIDATION_FAILED, f"ssh_key is required for request id {req_id!r}")

user_dir = os.path.join(job_folder, str(req_id))
            os.makedirs(user_dir, exist_ok=True)
            key_path = os.path.join(user_dir, "ssh_key")

try:
                with open(key_path, "w", encoding="utf-8") as f:
                    f.write(ssh_key)
                os.chmod(key_path, stat.S_IRUSR | stat.S_IWUSR)
            except OSError as e:
                raise BartError(ErrorCode.PERMISSION_DENIED, f"Cannot write {key_path}: {e}") from e

activity_logger.success(f"Wrote ssh key file {key_path}")
            action_data[str(req_id)] = {"ssh_key_file_path": key_path, "ssh_key_file_status": "success"}
        except BartError:
            activity_logger.success(f"Error writing ssh key file for request {req_id}")
            action_data[str(req_id)] = {"ssh_key_file_status": "failed"}

return action_data

Register