-
Notifications
You must be signed in to change notification settings - Fork 0
/
parse_and_convert_to_json_and_csv.py
executable file
·85 lines (60 loc) · 2.97 KB
/
parse_and_convert_to_json_and_csv.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
#!/usr/bin/env python3
import xml.etree.ElementTree as ET
import json
import subprocess
import argparse
import sys
import globals
def parse_health_export() -> tuple[list[dict], list[str]]:
records: list[dict] = []
record_types: list[str] = []
for _, elem in ET.iterparse(globals.data_file_path):
if elem.tag == 'Record':
records.append(elem.attrib)
if elem.attrib['type'] not in record_types:
record_types.append(elem.attrib['type'])
return records, record_types
def print_all_record_types(record_types: list[str]):
print(json.dumps(record_types, indent=4))
def write_all_records_json_file(records: list[dict]):
with open(globals.get_output_file_path('all_records', 'json', 'jq'), 'w') as json_file:
json_file.write(json.dumps(records, indent=4) + '\n')
def write_all_records_txt_file(records: list[dict]):
with open(globals.get_output_file_path('all_records', 'txt', 'jq'), 'w') as txt_file:
for record in records:
txt_file.write(json.dumps(record) + '\n')
def write_all_records_csv_file_with_jq():
cat_command = f'cat {globals.get_output_file_path('all_records', 'txt', 'jq')}'
jq_command = 'jq -r "[.type, .creationDate, .startDate, .endDate, .value, .unit, .device, .sourceName, .sourceVersion] | @csv"'
output_file_path = globals.get_output_file_path('all_records', 'csv', 'jq')
subprocess.run(f'{cat_command} | {jq_command} > {output_file_path}', shell=True)
def main() -> int:
parser = argparse.ArgumentParser(description=globals.get_argparse_description('JSON and CSV files'), formatter_class=argparse.RawDescriptionHelpFormatter)
parser.add_argument('-p', '--print-types', help='whether all record types should be printed', action='store_true')
parser.add_argument('-j', '--write-json', help='whether all records JSON file should be written', action='store_true')
parser.add_argument('-c', '--write-csv', help='whether all records CSV file should be written with jq', action='store_true')
args = parser.parse_args()
if not any(vars(args).values()):
parser.print_help()
return 0
cli_options = {
'print_all_record_types': args.print_types,
'write_all_records_json_file': args.write_json,
'write_all_records_csv_file': args.write_csv
}
records, record_types = parse_health_export()
if cli_options['print_all_record_types']:
print('All record types:\n')
print_all_record_types(record_types)
return 0
if cli_options['write_all_records_json_file']:
print('All records JSON file is being written …')
write_all_records_json_file(records)
if cli_options['write_all_records_csv_file']:
print('All records text file is being written …')
write_all_records_txt_file(records)
print('All records CSV file is being written with jq …')
write_all_records_csv_file_with_jq()
return 0
if __name__ == '__main__':
sys.exit(main())