Execute PowerShell from Python

import subprocess
import datetime
import time
import os
import ast
import json

# ======================================================= 
#                                               ~@.*.@~
# 66½                                              y 
# =======================================================

'''
Windows service running as Flask to execute
PowerShell from web call.
'''
class PowerShellProc():
    """
    WEB REQUEST <- JSON {requestor: "auth_user_id", return_code: 200, request_package: {}, timestamp: nnn}
    OUT -> JSON (return code, {key: val})
    """

    def __init__(self):
        self.pwrshell_path = r'C:\Windows\system32\WindowsPowerShell\v1.0\powershell.exe'
        self.station = r'\\mswg9\groups\integration\repo\ps'
        self.marshal = 'PSCop.ps1' # Auth: MSA and exec actual PS scripts based on request.

    def exec_pwrshell(self, pwr_script=None, script_path=None, args=[], ps_tuple_to_py=True, gm_flag=False, date2str=False):
        '''Out.exec_pwrshell('psscript.ps1' ,script_path=None, ps_tuple_to_py=True)
        script_path: overrides envrionment path.exists
        ps_tuple_to_py: Powershell tuples convert to dictionaries, set True to return a Python tuple.
        '''
        data = None
        exec_path = self.pwrshell_path
        
        if not os.path.exists(exec_path):
            raise IOError('Powershell not found at %s.' % exec_path)

        if not script_path:
            script_path = self.station

        if not os.path.exists(script_path):
            raise IOError('Script path not found at %s.' % script_path)

        if pwr_script:
            script_path = os.path.join(script_path, pwr_script)
        else:
            script_path = os.path.join(script_path, self.marshal)

        cmd = [exec_path, script_path]

        if args:
            if not isinstance(args, list):
                raise TypeError("Arguments must be passed as a list not %s." % str(type(args)))
            cmd.extend(args)

        proc = subprocess.Popen(cmd,
                    shell=True,
                    stdin=subprocess.PIPE,
                    stdout=subprocess.PIPE,
                    )

        rtrn,perr = proc.communicate()
        rcode = proc.returncode

        #if rtrn:
            # Line directly below converts PowerShell Json to Python data structs.
            #data = self.ps_output_converstion(rtrn, tup=ps_tuple_to_py, gm=gm_flag, date2str=date2str)

        if perr:
            # Error string...
            data = json.dumps(perr.decode('utf-8','ignore'))
        else:
            #data = rtrn.decode('utf-8','ignore')
            data = self.ps_output_converstion(rtrn, tup=ps_tuple_to_py, gm=gm_flag, date2str=date2str)

        return rcode,data

    def ps_output_converstion(self, res, tup=True, gm=False, date2str=True):
        '''Converts Powershell string output to Python type'''

        res = res.decode('utf-8','ignore')

        if '\n' in res:
            res = res.replace('\n','')

        if '\r' in res:
            res = res.split('\r')[0]

        res = ast.literal_eval(res)

        # Data conversion if necessary.
        if type(res) == str:
            if 'date' in res.lower() or 'time' in res.lower():
                epoch_elem = elem[elem.find("(")+1:elem.rfind(")")]
                dt = self.ps_epoch_to_datetime(float(elem), gm=gm, date2str=date2str)
                res = dt

        if type(res) == list:
            for index, elem in enumerate(res):
                if 'date' in str(elem).lower() or 'date' in str(elem).lower():
                    epoch_elem = elem[elem.find("(")+1:elem.rfind(")")]
                    dt = self.ps_epoch_to_datetime(float(elem), gm=gm, date2str=date2str)
                    res.remove(elem)
                    res.insert(index,dt)

        if type(res) == dict:
            for field, value in res.items():
                if 'date' in str(value).lower():
                    epoch_elem = value[value.find("(")+1:value.rfind(")")]
                    dt = self.ps_epoch_to_datetime(float(epoch_elem), gm=gm, date2str=date2str)
                    res[field] = dt

            if tup:
                if 'Item1' in res:
                    l = [v for v in res.values()]
                    res = tuple(l)

        return res

    def ps_epoch_to_datetime(self, epoch_time, gm=False, date2str=False):
        '''Converts Powershell Date to Python datetime.'''
        
        if gm:
            struct = time.gmtime(epoch_time/1000.)
        else:
            struct = time.localtime(epoch_time/1000.)

        dt = datetime.datetime.fromtimestamp(time.mktime(struct))

        if date2str: dt = dt.ctime()
        return dt

    def py2json(self, data):
        return json.dumps(data) # This will convert datetime to string.

    def json2py(self, data):
        return json.loads(data)

$Data = "Hello"
#$Data = 1
#$Data = 2.5
#$Data = [DateTime]"January 1, 2018 6:00 AM"
#$Data = [System.Tuple]::Create("Theresa Wilson", [DateTime]"January 1, 2018 6:00 AM", 1)
#$Data = @( "a","b","c","d","e",1,2,3 )
#$Data = [System.Collections.ArrayList]$listArray = @( "a","b","c","d","e",1,2,3)
$Data = @{"Washington" = "Olympia"; "Oregon" = "Salem"; California = "Sacramento"}

#$Data = $null # Do not return NULL, Python doesn't understand it, return zero instead.

$Data | ConvertTo-Json -Compress

Write-Host $Data

#Write-Output $Data

import lib_powershell

ps = lib_powershell.PowerShellProc()

ps.exec_pwrshell(pwr_script='test.ps1',script_path=r'C:\Path',ps_tuple_to_py=False) # Tuple to dict.
(0, {'Item3': 1, 'Length': 3, 'Item1': 'Theresa Wilson', 'Item2': datetime.datetime(2018, 1, 1, 6, 0)})
 
ps.exec_pwrshell(pwr_script='test.ps1',script_path=r'C:\Path')
(0, (1, 3, 'Theresa Wilson', datetime.datetime(2018, 1, 1, 6, 0)))
 
ps.exec_pwrshell(pwr_script='test.ps1',script_path=r'C:\Path', gm_flag=True) # Mixed Tuple
(0, (1, 3, 'Theresa Wilson', datetime.datetime(2018, 1, 1, 11, 0)))
 
ps.exec_pwrshell(pwr_script='test.ps1',script_path=r'C:\Path')
(0, {'Oregon': 'Salem', 'Washington': 'Olympia', 'California': 'Sacramento'})
 
ps.exec_pwrshell(pwr_script='test.ps1',script_path=r'C:\Path') # Mixed array/arraylist
(0, ['a', 'b', 'c', 'd', 'e', 1, 2, 3])
 
ps.exec_pwrshell(pwr_script='test.ps1',script_path=r'C:\Path') # Int
(0, 1)
 
ps.exec_pwrshell(pwr_script='test.ps1',script_path=r'C:\Path') # String
(0, 'Hello')
 
# NOTE: Tuple returned is (process_return_code, powershell_return_data)