@ -1,6 +1,6 @@
import subprocess , json , psutil , time , os , win32security , sys , base64 , logging , ctypes , copy #Get input argument
import pickle
import inspect
from partd import Server
from . import Server
@ -29,6 +29,7 @@ import uuid # Generate uuid
import datetime # datetime
import math
import glob # search the files
import urllib
#Единый глобальный словарь (З а
gSettingsDict = None
@ -874,6 +875,15 @@ def WebRequestParseBodyJSON(inRequest):
"""
return json . loads ( WebRequestParseBodyStr ( inRequest = inRequest ) )
def WebRequestParsePath ( inRequest ) :
"""
Parse the request - extract the url . Example : / pyOpenRPA / Debugging / DefHelper / . . .
: param inRequest :
: return : Str , Example : / pyOpenRPA / Debugging / DefHelper / . . .
"""
return urllib . parse . unquote ( inRequest . path )
def WebRequestParseFile ( inRequest ) :
"""
Parse the request - extract the file ( name , body in bytes )
@ -900,6 +910,13 @@ def WebRequestParseFile(inRequest):
return lResultTurple
def WebRequestResponseSend ( inRequest , inResponeStr ) :
"""
Send response for the request
: return :
"""
inRequest . OpenRPAResponseDict [ " Body " ] = bytes ( inResponeStr , " utf8 " )
def WebUserInfoGet ( inRequest ) :
"""
Return User info about request
@ -1109,6 +1126,7 @@ def ProcessorAliasDefCreate(inDef, inAliasStr=None, inGSettings = None):
"""
Create alias for def ( can be used in ActivityItem in field Def )
! WHEN DEF ALIAS IS REQUIRED ! - Def alias is required when you try to call Python def from the Orchestrator WEB side ( because you can ' t transmit Python def object out of the Python environment)
Deprecated . See ActivityItemDefAliasCreate
. . code - block : : python
@ -1128,21 +1146,13 @@ def ProcessorAliasDefCreate(inDef, inAliasStr=None, inGSettings = None):
: param inAliasStr : String alias for associated def
: return : str Alias string ( Alias can be regenerated if previous alias was occupied )
"""
#TODO Pay attention - New alias can be used too - need to create more complex algorythm to create new alias!
inGSettings = GSettingsGet ( inGSettings = inGSettings ) # Set the global settings
lL = inGSettings [ " Logger " ]
if inAliasStr is None : inAliasStr = str ( inDef )
# Check if key is not exists
if inAliasStr in inGSettings [ " ProcessorDict " ] [ " AliasDefDict " ] :
inAliasStr = str ( inDef )
if lL : lL . warning ( f " Orchestrator.ProcessorAliasDefCreate: Alias { inAliasStr } already exists in alias dictionary. Another alias will be generated and returned " )
inGSettings [ " ProcessorDict " ] [ " AliasDefDict " ] [ inAliasStr ] = inDef
return inAliasStr
return ActivityItemDefAliasCreate ( inDef = inDef , inAliasStr = inAliasStr , inGSettings = inGSettings )
def ProcessorAliasDefUpdate ( inDef , inAliasStr , inGSettings = None ) :
"""
Update alias for def ( can be used in ActivityItem in field Def ) .
! WHEN DEF ALIAS IS REQUIRED ! - Def alias is required when you try to call Python def from the Orchestrator WEB side ( because you can ' t transmit Python def object out of the Python environment)
Deprecated . See ActivityItemDefAliasUpdate
. . code - block : : python
@ -1162,12 +1172,58 @@ def ProcessorAliasDefUpdate(inDef, inAliasStr, inGSettings = None):
: param inAliasStr : String alias for associated def
: return : str Alias string
"""
inGSettings = GSettingsGet ( inGSettings = inGSettings ) # Set the global settings
if callable ( inDef ) : inGSettings [ " ProcessorDict " ] [ " AliasDefDict " ] [ inAliasStr ] = inDef
else : raise Exception ( f " pyOpenRPA Exception: You can ' t use Orchestrator.ProcessorAliasDefUpdate with arg ' inDef ' string value. inDef is ' { inDef } ' , inAliasStr is ' { inAliasStr } ' " )
return inAliasStr
return ActivityItemDefAliasUpdate ( inDef = inDef , inAliasStr = inAliasStr , inGSettings = inGSettings )
def ProcessorActivityItemCreate ( inDef , inArgList = None , inArgDict = None , inArgGSettingsStr = None , inArgLoggerStr = None , inGUIDStr = None , inThreadBool = False ) :
# ActivityItem defs
def ActivityItemHelperDefList ( inDefQueryStr = None ) :
"""
Create list of the available Def names in activity item . You can use query def filter via arg inDefQueryStr
: param inDefStr :
: return : [ " ActivityItemDefAliasUpdate " , " ActivityItemDefAliasCreate " , etc . . . ]
"""
lResultList = [ ]
if inDefQueryStr is not None : # do search alg
for lKeyStr in GSettingsGet ( ) [ " ProcessorDict " ] [ " AliasDefDict " ] :
if inDefQueryStr in lKeyStr :
lResultList . append ( lKeyStr )
else :
for lKeyStr in GSettingsGet ( ) [ " ProcessorDict " ] [ " AliasDefDict " ] :
lResultList . append ( lKeyStr )
return lResultList
def ActivityItemHelperDefAutofill ( inDef ) :
"""
Detect def by the name and prepare the activity item dict with values .
: param inDef :
: return :
"""
lResultDict = {
" Def " : None ,
" ArgList " : [ ] ,
" ArgDict " : { } ,
" ArgGSettingsStr " : None ,
" ArgLoggerStr " : None
}
lResultDict [ " Def " ] = inDef
lGS = GSettingsGet ( )
lDefSignature = inspect . signature ( lGS [ " ProcessorDict " ] [ " AliasDefDict " ] [ inDef ] )
for lItemKeyStr in lDefSignature . parameters :
lItemValue = lDefSignature . parameters [ lItemKeyStr ]
# Check if arg name contains "GSetting" or "Logger"
if " GSETTING " in lItemKeyStr . upper ( ) :
lResultDict [ " ArgGSettingsStr " ] = lItemKeyStr
elif " LOGGER " in lItemKeyStr . upper ( ) :
lResultDict [ " ArgLoggerStr " ] = lItemKeyStr
else :
if lItemValue . default is inspect . _empty :
lResultDict [ " ArgDict " ] [ lItemKeyStr ] = None
else :
lResultDict [ " ArgDict " ] [ lItemKeyStr ] = lItemValue . default
return lResultDict
def ActivityItemCreate ( inDef , inArgList = None , inArgDict = None , inArgGSettingsStr = None , inArgLoggerStr = None , inGUIDStr = None , inThreadBool = False ) :
"""
Create activity item . Activity item can be used as list item in ProcessorActivityItemAppend or in Processor . ActivityListExecute .
@ -1179,7 +1235,7 @@ def ProcessorActivityItemCreate(inDef, inArgList=None, inArgDict=None, inArgGSet
# EXAMPLE 1
def TestDef ( inArg1Str , inGSettings , inLogger ) :
pass
lActivityItem = Orchestrator . Processor ActivityItemCreate(
lActivityItem = Orchestrator . (
inDef = TestDef ,
inArgList = [ ] ,
inArgDict = { " inArg1Str " : " ArgValueStr " } ,
@ -1197,11 +1253,11 @@ def ProcessorActivityItemCreate(inDef, inArgList=None, inArgDict=None, inArgGSet
# EXAMPLE 2
def TestDef ( inArg1Str ) :
pass
Orchestrator . ProcessorAliasDef Update(
Orchestrator . ActivityItemDefAlias Update(
inGSettings = gSettings ,
inDef = TestDef ,
inAliasStr = " TestDefAlias " )
lActivityItem = Orchestrator . Processor ActivityItemCreate(
lActivityItem = Orchestrator . (
inDef = " TestDefAlias " ,
inArgList = [ ] ,
inArgDict = { " inArg1Str " : " ArgValueStr " } ,
@ -1241,6 +1297,132 @@ def ProcessorActivityItemCreate(inDef, inArgList=None, inArgDict=None, inArgGSet
}
return lActivityItemDict
def ActivityItemDefAliasCreate ( inDef , inAliasStr = None , inGSettings = None ) :
"""
Create alias for def ( can be used in ActivityItem in field Def )
! WHEN DEF ALIAS IS REQUIRED ! - Def alias is required when you try to call Python def from the Orchestrator WEB side ( because you can ' t transmit Python def object out of the Python environment)
. . code - block : : python
# USAGE
from pyOpenRPA import Orchestrator
def TestDef ( ) :
pass
lAliasStr = Orchestrator . ActivityItemDefAliasCreate (
inGSettings = gSettings ,
inDef = TestDef ,
inAliasStr = " TestDefAlias " )
# Now you can call TestDef by the alias from var lAliasStr with help of ActivityItem (key Def = lAliasStr)
: param inGSettings : Global settings dict ( singleton )
: param inDef : Def
: param inAliasStr : String alias for associated def
: return : str Alias string ( Alias can be regenerated if previous alias was occupied )
"""
#TODO Pay attention - New alias can be used too - need to create more complex algorythm to create new alias!
inGSettings = GSettingsGet ( inGSettings = inGSettings ) # Set the global settings
lL = inGSettings [ " Logger " ]
if inAliasStr is None : inAliasStr = str ( inDef )
# Check if key is not exists
if inAliasStr in inGSettings [ " ProcessorDict " ] [ " AliasDefDict " ] :
inAliasStr = str ( inDef )
if lL : lL . warning ( f " Orchestrator.ProcessorAliasDefCreate: Alias { inAliasStr } already exists in alias dictionary. Another alias will be generated and returned " )
inGSettings [ " ProcessorDict " ] [ " AliasDefDict " ] [ inAliasStr ] = inDef
return inAliasStr
def ActivityItemDefAliasUpdate ( inDef , inAliasStr , inGSettings = None ) :
"""
Update alias for def ( can be used in ActivityItem in field Def ) .
! WHEN DEF ALIAS IS REQUIRED ! - Def alias is required when you try to call Python def from the Orchestrator WEB side ( because you can ' t transmit Python def object out of the Python environment)
. . code - block : : python
# USAGE
from pyOpenRPA import Orchestrator
def TestDef ( ) :
pass
Orchestrator . ActivityItemDefAliasUpdate (
inGSettings = gSettings ,
inDef = TestDef ,
inAliasStr = " TestDefAlias " )
# Now you can call TestDef by the alias "TestDefAlias" with help of ActivityItem (key Def = "TestDefAlias")
: param inGSettings : Global settings dict ( singleton )
: param inDef : Def
: param inAliasStr : String alias for associated def
: return : str Alias string
"""
inGSettings = GSettingsGet ( inGSettings = inGSettings ) # Set the global settings
if callable ( inDef ) : inGSettings [ " ProcessorDict " ] [ " AliasDefDict " ] [ inAliasStr ] = inDef
else : raise Exception ( f " pyOpenRPA Exception: You can ' t use Orchestrator.ActivityItemDefAliasUpdate with arg ' inDef ' string value. inDef is ' { inDef } ' , inAliasStr is ' { inAliasStr } ' " )
return inAliasStr
def ProcessorActivityItemCreate ( inDef , inArgList = None , inArgDict = None , inArgGSettingsStr = None , inArgLoggerStr = None , inGUIDStr = None , inThreadBool = False ) :
"""
Create activity item . Activity item can be used as list item in ProcessorActivityItemAppend or in Processor . ActivityListExecute .
Deprecated . See ActivityItemCreate
. . code - block : : python
# USAGE
from pyOpenRPA import Orchestrator
# EXAMPLE 1
def TestDef ( inArg1Str , inGSettings , inLogger ) :
pass
lActivityItem = Orchestrator . ProcessorActivityItemCreate (
inDef = TestDef ,
inArgList = [ ] ,
inArgDict = { " inArg1Str " : " ArgValueStr " } ,
inArgGSettingsStr = " inGSettings " ,
inArgLoggerStr = " inLogger " )
# lActivityItem:
# {
# "Def":TestDef,
# "ArgList":inArgList,
# "ArgDict":inArgDict,
# "ArgGSettings": "inArgGSettings",
# "ArgLogger": "inLogger"
# }
# EXAMPLE 2
def TestDef ( inArg1Str ) :
pass
Orchestrator . ProcessorAliasDefUpdate (
inGSettings = gSettings ,
inDef = TestDef ,
inAliasStr = " TestDefAlias " )
lActivityItem = Orchestrator . ProcessorActivityItemCreate (
inDef = " TestDefAlias " ,
inArgList = [ ] ,
inArgDict = { " inArg1Str " : " ArgValueStr " } ,
inArgGSettingsStr = None ,
inArgLoggerStr = None )
# lActivityItem:
# {
# "Def":"TestDefAlias",
# "ArgList":inArgList,
# "ArgDict":inArgDict,
# "ArgGSettings": None,
# "ArgLogger": None
# }
: param inDef : def link or def alias ( look gSettings [ " Processor " ] [ " AliasDefDict " ] )
: param inArgList : Args list for the Def
: param inArgDict : Args dict for the def
: param inArgGSettingsStr : Name of def argument of the GSettings dict
: param inArgLoggerStr : Name of def argument of the logging object
: param inGUIDStr : GUID which you can specify . If None the GUID will be generated
: param inThreadBool : True - execute ActivityItem in new thread ; False - in processor thread
: return : { }
"""
return ActivityItemCreate ( inDef = inDef , inArgList = inArgList , inArgDict = inArgDict , inArgGSettingsStr = inArgGSettingsStr , inArgLoggerStr = inArgLoggerStr ,
inGUIDStr = inGUIDStr , inThreadBool = inThreadBool )
def ProcessorActivityItemAppend ( inGSettings = None , inDef = None , inArgList = None , inArgDict = None , inArgGSettingsStr = None , inArgLoggerStr = None , inActivityItemDict = None ) :
"""
Create and add activity item in processor queue .
Ivan Maslov
3 years ago