-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathexport_mods_main.py
104 lines (82 loc) · 3.27 KB
/
export_mods_main.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
"""Script that creates all directories"""
import os
# Change to the project repository location
my_wd = os.getcwd()
my_repo = "research-and-development"
if not my_wd.endswith(my_repo):
os.chdir(my_repo)
from src.utils.singleton_boto import SingletonBoto # noqa
config = {
"s3": {
"ssl_file": "/etc/pki/tls/certs/ca-bundle.crt",
"s3_bucket": "onscdp-dev-data01-5320d6ca",
}
}
boto3_client = SingletonBoto.get_client(config)
import src.utils.s3_mods as mods # noqa
if __name__ == "__main__":
my_path = "/bat/res_dev/project_data/2023_surveys/BERD/01_staging/staging_qa/full_responses_qa/2023_staged_full_responses_24-10-02_v20.csv" # noqa
# to_delete_path = "/bat/res_dev/project_data/2023_surveys/BERD/01_staging/staging_qa/full_responses_qa/2023_staged_full_responses_test_to_delete.csv" # noqa
my_dir = "/bat/res_dev/project_data/2023_surveys/BERD/01_staging/staging_qa/full_responses_qa/" # noqa
# # Checking that file exists
my_size = mods.rd_file_size(my_path)
print(f"File size is {my_size}")
# Calculating md5sum
my_sum = mods.rd_md5sum(my_path)
expected_output = "ea94424aceecf11c8a70d289e51c34ea" # pragma: allowlist secret
print(type(my_sum))
if expected_output == my_sum:
print("Same md5sum")
# Calculating rd_isdir
mydir = "/bat"
response = mods.rd_isdir(mydir)
print("Got response")
print(response)
# Checking rd_isfile
response = mods.rd_isfile(my_path)
print(response)
# Checking that rd_stat_size works for files and directories
file_size = mods.rd_stat_size(my_path)
print(f"File {my_path} size is {file_size} bytes.")
dir_size = mods.rd_stat_size(my_dir)
print(f"Directory {my_dir} size is {dir_size} bytes.")
# Testing rd_read_header
response = mods.rd_read_header(my_path)
print(response)
# Testing rd_write_string_to_file
out_path = "/bat/res_dev/project_data/new_write_string_test_2.txt"
content = "New content"
mods.rd_write_string_to_file(content.encode(encoding="utf-8"), out_path)
print("all done")
# Testing rd_copy_file
src_path = "/bat/res_dev/project_data/new_write_string_test_2.txt"
dst_path = "/bat/res_dev/"
success = mods.rd_copy_file(src_path, dst_path)
if success:
print("File copied successfully")
else:
print("File not copied successfully")
# Testing rd_move_file
src_path = "/bat/res_dev/new_write_string_test_2.txt"
dst_path = "/bat/res_dev/project_data/"
success = mods.rd_move_file(src_path, dst_path)
if success:
print("File moved successfully")
else:
print("File not moved successfully")
# Testing rd_search_file
bat_path = "bat/res_dev/project_data/"
dir_path = ( # pragma: allowlist secret noqa
"2023_surveys/BERD/01_staging/staging_qa/full_responses_qa/"
)
ending = "24-10-02_v20.csv"
found_file = mods.rd_search_file(bat_path + dir_path, ending)
print(f"Found file: {found_file}")
# Deleting a file
# status = mods.rd_delete_file(my_path)
# if status:
# print(f"File {to_delete_path} successfully deleted")
# Testing read_excel
# my_path = "bat/res_dev/project_data/test_excel_gz.xlsx"
# df = mods.read_excel(my_path)
# print(df.head())