-
Notifications
You must be signed in to change notification settings - Fork 0
/
FileHandle.py
166 lines (150 loc) · 6.55 KB
/
FileHandle.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
# MIT License
#
# Copyright (C) 2021 <FacuFalcone - CaidevOficial>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import json
import os
import shutil
def SingleMessage(message: str) -> None:
"""[summary]
Prints a formatted message, receiving a string as a parameter.
Args:
message (str): [Message to show in the console.]
"""
print("######################################################")
print(f'### {message}')
print("######################################################\n")
def DoubleMessage(message1: str, message2: str) -> None:
"""[summary]
Prints a formatted message, receiving two strings as a parameter.
Args:
message1 (str): [First Message to show in the console.]
message2 (str): [Second Message to show in the console.]
"""
print("######################################################")
print(f'### {message1}')
print(f'### {message2}')
print("######################################################\n")
def OpenFile(abspath: str) -> dict:
"""[summary]
Opens the file specified in 'abspath'.
Args:
abspath (str): The path with the name of the file to open.
Returns:
[dict]: [The json parsed with the configuration of the tables inside.]
"""
try:
SingleMessage(f'Trying to open {abspath}')
with open(abspath, 'r+') as schemafile:
TABLE_FIELDS = json.load(schemafile)
print("Success: File Opened!")
except Exception as e:
TABLE_FIELDS = {}
DoubleMessage(f'Error opening the file {abspath}','An empty dictionary will be returned.')
SingleMessage(f"Exception: {e}")
return TABLE_FIELDS
def WriteJSON(myDict: dict, datasetName: str, jsonFileNameToSave: str) -> None:
"""
Writes and creates a file with format json.
Args:
myDict (dict): A dictionary with the data of the json to create.
fileName (str): The name of the final File.
"""
formatJSON = f"{datasetName}.{jsonFileNameToSave}"
try:
SingleMessage(f'Trying write the file {formatJSON}')
with open(formatJSON, 'w+') as jsonFile:
jsonFile.write(json.dumps(myDict, sort_keys=True, indent=4, separators=(',', ':')))
jsonFile.close()
SingleMessage('Success: Json file created!')
except Exception as e:
SingleMessage('Error: The json could not be written')
SingleMessage(f"Exception: {e}")
def WriteCSV(myList: list, datasetName: str, tableName: str) -> None:
"""
Writes and creates a file with format csv.
Args:
myList (list): A list with all the registers inside.
tableName (str): Name of the table with all the registers to make the file.
"""
formatCSV = f"{datasetName}.{tableName}.csv"
try:
SingleMessage(f'Trying write the file {formatCSV}')
with open(formatCSV, 'w+') as csvFile:
csvFile.write(str(myList))
csvFile.close()
SingleMessage('Success: CSV file created!')
except Exception as e:
SingleMessage('Error: The CSV could not be written')
SingleMessage(f"Exception: {e}")
def WriteSQL(myListOfQuerys: list, datasetName: str, tableName: str) -> None:
"""
Writes and creates a file with format sql.
Args:
myList (list): A list with all the registers inside.
tableName (str): Name of the table with all the registers to make the file.
"""
formatSQL = f"{datasetName}.{tableName}.sql"
try:
SingleMessage(f'Trying write the file {formatSQL}')
with open(formatSQL, 'w+') as sqlFile:
sqlFile.write(str(myListOfQuerys))
sqlFile.close()
SingleMessage('Success: SQL file created!')
except Exception as e:
SingleMessage('Error: The SQL could not be written')
SingleMessage(f"Exception: {e}")
def SortSingleFile(filename:str, formatFile:str, datasetName: str,directoryToSave:str, moved:bool) -> bool:
"""[summary]
Moves a single file to a directory.
Args:
filename (str): [Name of the file to be moved.]
formatFile (str): [Format extension of the file to be moved.]
datasetName (str): [Name of the dataset to be used in the directory to be created.]
directoryToSave (str): [Name of the directory to save the file.]
moved (bool): [Boolean state that indicates if the file has been moved or not.]
Returns:
bool: [True if the file was moved, False otherwise.]
"""
if filename.endswith(f'.{formatFile}'):
directory = f'{datasetName}.{directoryToSave}'
if not os.path.exists(directory):
os.makedirs(directory)
shutil.move(filename, directory)
SingleMessage(f'Success: {filename} moved to {directory}.')
moved = True
return moved
def SortFiles(exceptedFile: str, datasetName: str,directoryOfCSV:str, directoryOfJson:str, directoryOfSQL:str, currentDir:str) -> bool:
"""
Moves the files with format json and csv (except the configuration's json) to two directories, one for all the json and the other for the csv.
Args:
exceptedFile (str): File of tables's configuration (json)
currentDir ([type]): Current directory with the files.
Returns:
bool: [True if the specified files was moved, False otherwise.]
"""
moveFiles = False
try:
SingleMessage(f'Trying read files in {currentDir}.')
for filename in os.listdir(currentDir):
if (not filename.endswith("Configurations.json") and (not filename.endswith(f"{exceptedFile}"))):
moveFiles = SortSingleFile(filename, 'csv', datasetName, directoryOfCSV, moveFiles)
moveFiles = SortSingleFile(filename, 'json', datasetName, directoryOfJson, moveFiles)
moveFiles = SortSingleFile(filename, 'sql', datasetName, directoryOfSQL, moveFiles)
except Exception as e:
SingleMessage(f'Error: Reading of {currentDir} has failed.')
SingleMessage(f"Exception: {e}")
finally:
return moveFiles