KnowHow

#!/usr/bin/python3

import subprocess
from subprocess import PIPE
import logging
import time
import datetime
import signal
import os
import sys
import gc

dir_path = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.append(dir_path)

from config import settings

"""
created by me.
submit shell command
version 2024.09.23.1
"""

#logging.basicConfig(
#        level=logging.DEBUG)

handler = logging.StreamHandler(sys.stdout)
formatter = logging.Formatter('%(asctime)s:%(name)s:%(levelname)s:%(message)s')
handler.setFormatter(formatter)

logger = logging.getLogger(__name__)
#logger.setLevel(logging.DEBUG)
logger.setLevel(logging.INFO)
logger.addHandler(handler)


logger.debug({'add path': dir_path})


class ShellCommand(object):
    def __init__(self, dt_now, timeout: int, command: str):
        self.stdout = False
        self.stderr = False
        self.returncode = False
        self.command = False
        self._timeout = timeout
        self._command = command
        self._dt_now = dt_now
        self._command_result = False
        self._errlog = False

    def submit_command(self, command):
        self.command = command
        result = subprocess.run(
                self.command,
                shell=True,
                stdout=PIPE,
                stderr=PIPE,
                timeout=self._timeout)
        self.stdout = result.stdout.decode('utf-8')
        self.stderr = result.stderr.decode('utf-8')
        self.returncode = result.returncode

        if result.returncode != 0:
            raise Exception(self.stderr)

    def execute_command(self):
        try:
            self.submit_command(self._command)
            self._command_result = self.stdout
            logger.debug({'shellcommand':'success'})
        except Exception as e:
            self._command_result = self.stderr
            self._errlog = str(e)
            logger.error({
                'time': self._dt_now,
                'status': 'failed',
                'action':'ExceuteShellComand',
                'error': self._errlog,
                'command': self._command})


class ShellCommandSignal(ShellCommand):

    def _signal_handler(self, signum, frame):
        logger.error({
            'statud': "Time out!",
            'action': 'ExecuteShellCommand_signal'})
        raise Exception("catch error timeout")

    def submit_command(self, command):
        self.command = command
        result = subprocess.run(
                self.command,
                shell=True,
                stdout=PIPE,
                stderr=PIPE)
        self.stdout = result.stdout.decode('utf-8')
        self.stderr = result.stderr.decode('utf-8')
        self.returncode = result.returncode

    def execute_command(self):
        signal.signal(signal.SIGALRM, self._signal_handler)
        signal.alarm(self._timeout)

        try:
            self.submit_command(self._command)
            self._command_result = self.stdout
        except Exception as e:
            self._command_result = self.stderr
            self._errlog = str(e)
            logger.error({
                'time': self._dt_now,
                'status': 'failed',
                'action':'ExceuteShellComand_signal',
                'error': self._errlog,
                'command': self._command})


def test_main(_dt_now, _timeout):
    command = 'sleep 3; ls -l'
    shell_command = ShellCommand(_dt_now, _timeout, command)
    shell_command.execute_command()
    logger.info({'test result': shell_command._command_result})


def test_main_signal(_dt_now, _timeout):
    command = 'sleep 3; ls -l'
    shell_command = ShellCommandSignal(_dt_now, _timeout, command)
    shell_command.execute_command()
    logger.info({'test signal result': shell_command._command_result})


if __name__ == '__main__':
    dt_now = datetime.datetime.now().strftime('%Y/%m/%d %H:%M:%S')
    timeout = 10

    test_main(dt_now, timeout)

    test_main_signal(dt_now, timeout)