You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
287 lines
8.0 KiB
287 lines
8.0 KiB
# Funktionen zum Dateizugriff mit Suchen, Lesen, Schreiben
|
|
# ------------------------------------------------------------
|
|
"""
|
|
|
|
"""
|
|
import codecs
|
|
import json
|
|
import os
|
|
import os.path
|
|
import re
|
|
import time
|
|
|
|
import xmltodict
|
|
import yaml
|
|
import platform
|
|
|
|
import basic.message
|
|
import basic.program
|
|
import utils.data_const as D
|
|
from pprint import pp
|
|
import utils.tdata_tool
|
|
import utils.date_tool
|
|
|
|
def getDump(obj):
|
|
result=""
|
|
print (str(type(obj)))
|
|
result = vars(obj)
|
|
return str(result)
|
|
# if type(obj) == "__dict__"
|
|
|
|
|
|
def getFiles(msg, job, path, pattern, conn):
|
|
"""
|
|
search filenames in the directory - if conn is set search remote
|
|
:param msg: -- msg-Objekt
|
|
:param path: -- path - String
|
|
:param pattern: -- filename or pattern
|
|
:param conn:
|
|
:return: Array with filenames
|
|
"""
|
|
if conn is not None:
|
|
return getRemoteFiles(msg, path, pattern, conn)
|
|
# job = basic.program.Job.getInstance()
|
|
verify = int(job.getDebugLevel("file_tool"))
|
|
out = []
|
|
msg.debug(verify, "getFiles " + path + " , " + pattern)
|
|
if not os.path.exists(path):
|
|
return out
|
|
for f in os.listdir(path):
|
|
msg.debug(verify, "getFiles " + f)
|
|
if re.search(pattern, f):
|
|
msg.debug(verify, "match " + f)
|
|
out.append(f)
|
|
return out
|
|
|
|
|
|
def removeFiles(msg, path, pattern, conn):
|
|
"""
|
|
search filenames in the directory and removes it
|
|
- if conn is set search remote
|
|
:param msg: -- msg-Objekt
|
|
:param path: -- path - String
|
|
:param pattern: -- filename as Pattern
|
|
:param conn:
|
|
:return: Array filenames
|
|
"""
|
|
pass
|
|
|
|
|
|
def copyFiles(job, fileList, source, target, comp):
|
|
"""
|
|
copies files from source to target
|
|
:param job:
|
|
:param fileList:
|
|
:param source:
|
|
:param target:
|
|
:param comp:
|
|
:return:
|
|
"""
|
|
pass
|
|
|
|
|
|
def getRemoteFiles(msg, path, pattern, conn):
|
|
"""
|
|
search filenames in the directory - if conn is set search remote
|
|
:param msg: -- msg-Objekt
|
|
:param path: -- path - String
|
|
:param pattern: -- filename as Pattern
|
|
:param conn:
|
|
:return: Array filenames
|
|
"""
|
|
pass
|
|
|
|
|
|
def getFilesRec(msg, job, path, pattern):
|
|
"""
|
|
Sucht Dateien im Verzeichnis rekursiv
|
|
:param msg: -- msg-Objekt
|
|
:param path: -- Pfad - String
|
|
:param pattern: -- Dateiname als Pattern
|
|
:return: Array mit gefundenen Dateien, absoluter Pfad
|
|
"""
|
|
#job = basic.program.Job.getInstance()
|
|
verify = int(job.getDebugLevel("file_tool"))
|
|
out = []
|
|
msg.debug(verify, "getFilesRec " + path + " , " + pattern)
|
|
for (r, dirs, files) in os.walk(path):
|
|
for f in files:
|
|
msg.debug(verify, "getFilesRec " + f)
|
|
if re.search(pattern, f):
|
|
msg.debug(verify, "match " + f)
|
|
out.append(os.path.join(r, f))
|
|
return out
|
|
|
|
|
|
def getTree(msg, job, pfad):
|
|
# job = basic.program.Job.getInstance()
|
|
verify = int(job.getDebugLevel("file_tool"))
|
|
msg.debug(verify, "getTree " + pfad )
|
|
tree = {}
|
|
files = []
|
|
for f in os.listdir(pfad):
|
|
if os.path.isDir(os.path.join(pfad, f)):
|
|
tree[f] = getTree(msg, job, os.path.join(pfad, f))
|
|
elif os.path.isFile(os.path.join(pfad, f)):
|
|
files.append(f)
|
|
tree["_files_"] = files
|
|
return tree
|
|
|
|
|
|
def mkPaths(job, path, msg):
|
|
# job = basic.program.Job.getInstance()
|
|
verify = int(job.getDebugLevel("file_tool"))
|
|
#modus = job.conf.confs["paths"]["mode"]
|
|
dirname = os.path.dirname(path)
|
|
if os.path.exists(dirname):
|
|
return
|
|
os.makedirs(dirname, exist_ok=True)
|
|
|
|
|
|
def getFileEncoding(msg, job, path):
|
|
print("--- getFileEncoding "+path)
|
|
encodings = ['utf-8', 'iso-8859-1'] # add more
|
|
for e in encodings:
|
|
print(e)
|
|
try:
|
|
fh = codecs.open(path, 'r', encoding=e)
|
|
fh.readlines()
|
|
fh.seek(0)
|
|
except UnicodeDecodeError:
|
|
print('got unicode error with %s , trying different encoding' % e)
|
|
except:
|
|
print("except")
|
|
else:
|
|
print('opening the file with encoding: %s ' % e)
|
|
return e
|
|
return detectFileEncode(job, path, msg)
|
|
|
|
|
|
def detectFileEncode(job, path, msg): # return ""
|
|
#job = basic.program.Job.getInstance()
|
|
verify = int(job.getDebugLevel("file_tool"))
|
|
print(path)
|
|
cntIso = 0
|
|
cntUtf = 0
|
|
j = 0
|
|
CHAR_ISO = [ 196, 228, 214, 246, 220, 252, 191 ]
|
|
with open(path, 'rb') as file:
|
|
byte = file.read(1)
|
|
while (byte):
|
|
i = int.from_bytes(byte, "little")
|
|
#byte = file.read(1)
|
|
if (i in CHAR_ISO):
|
|
cntIso += 1
|
|
if (i == 160):
|
|
pass
|
|
elif (i > 127):
|
|
cntUtf += 1
|
|
j += 1
|
|
l = i
|
|
byte = file.read(1)
|
|
file.close()
|
|
if (cntIso > cntUtf):
|
|
return 'iso-8859-1'
|
|
return 'utf-8'
|
|
|
|
|
|
def readFileLines(job, path, msg):
|
|
lines = readFileText(job, path, msg)
|
|
if isinstance(lines, (str)):
|
|
return lines.splitlines()
|
|
return []
|
|
|
|
|
|
def readFileText(job, path, msg):
|
|
#job = basic.program.Job.getInstance()
|
|
verify = int(job.getDebugLevel("file_tool"))
|
|
if not os.path.exists(path):
|
|
return ""
|
|
enc = detectFileEncode(job, path, msg)
|
|
with open(path, 'r', encoding=enc) as file:
|
|
text = file.read()
|
|
file.close()
|
|
return text
|
|
|
|
def getModTime(job, filepath):
|
|
out = ""
|
|
mtime = os.path.getmtime(filepath)
|
|
out = utils.date_tool.formatParsedDate(time.ctime(mtime), utils.date_tool.F_TIME_DEFAULT)
|
|
return out
|
|
|
|
def readFileDict(job, path, msg):
|
|
"""
|
|
reads and gets general a dict from any kind of filetyp
|
|
:param path: with extension of filetype
|
|
:param msg: optionally
|
|
:return:
|
|
"""
|
|
# 20220329 generalize
|
|
#job = basic.program.Job.getInstance()
|
|
verify = int(job.getDebugLevel("file_tool"))
|
|
doc = {}
|
|
if not os.path.exists(path):
|
|
return doc
|
|
enc = detectFileEncode(job, path, msg)
|
|
if D.DFILE_TYPE_YML in path[-4:]:
|
|
with open(path, 'r', encoding=enc) as file:
|
|
doc = yaml.full_load(file)
|
|
file.close()
|
|
elif D.DFILE_TYPE_JSON in path[-5:]:
|
|
with open(path, 'r', encoding=enc) as file:
|
|
doc = json.load(file)
|
|
file.close()
|
|
elif D.DFILE_TYPE_XML in path[-4:]:
|
|
with open(path, 'r', encoding=enc) as file:
|
|
res = xmltodict.parse(file.read())
|
|
# doc = dict(res)
|
|
doc = castOrderedDict(res)
|
|
file.close()
|
|
elif D.DFILE_TYPE_CSV in path[-5:]:
|
|
doc = utils.tdata_tool.getCsvSpec(msg, job, path, D.CSV_SPECTYPE_CONF)
|
|
return doc
|
|
|
|
def castOrderedDict(res, job=None, key=""):
|
|
if isinstance(res, dict):
|
|
doc = dict(res)
|
|
for x in doc:
|
|
doc[x] = castOrderedDict(doc[x], job, x)
|
|
elif isinstance(res, list):
|
|
sublist = []
|
|
for i in range(0, len(res)):
|
|
sublist.append(castOrderedDict(res[i], job, ""))
|
|
doc = sublist
|
|
else:
|
|
doc = res
|
|
return doc
|
|
|
|
|
|
def writeFileText(msg, job, path, text, enc="utf-8"):
|
|
#job = basic.program.Job.getInstance()
|
|
verify = int(job.getDebugLevel("file_tool"))
|
|
mkPaths(job, path, msg)
|
|
with open(path, 'w', encoding=enc) as file:
|
|
file.write(text)
|
|
file.close()
|
|
|
|
|
|
def writeFileDict(msg, job, path, dict, enc="utf-8"):
|
|
#job = basic.program.Job.getInstance()
|
|
mkPaths(job, path, msg)
|
|
if D.DFILE_TYPE_YML in path[-5:]:
|
|
with open(path, 'w', encoding=enc) as file:
|
|
yaml.dump(dict, file)
|
|
file.close()
|
|
elif D.DFILE_TYPE_JSON in path[-5:]:
|
|
with open(path, 'w', encoding=enc) as file:
|
|
doc = json.dumps(dict, indent=4)
|
|
file.write(doc)
|
|
file.close()
|
|
elif D.DFILE_TYPE_XML in path[-4:]:
|
|
with open(path, 'w', encoding=enc) as file:
|
|
text = xmltodict.unparse(dict, pretty=True)
|
|
if "<?xml version=" not in text:
|
|
text = "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n" + text
|
|
file.write(text)
|
|
file.close()
|
|
|
|
|