cic-internal-integration/apps/data-seeding/cic_ussd/import_ussd_data.py

68 lines
2.1 KiB
Python
Raw Normal View History

2021-05-15 09:40:34 +02:00
# standard imports
import argparse
import json
import logging
import os
# external imports
import celery
from confini import Config
# local imports
logging.basicConfig(level=logging.WARNING)
logg = logging.getLogger()
default_config_dir = '/usr/local/etc/cic'
arg_parser = argparse.ArgumentParser()
arg_parser.add_argument('-c', type=str, default=default_config_dir, help='config file')
arg_parser.add_argument('-q', type=str, default='cic-eth', help='Task queue')
arg_parser.add_argument('-v', action='store_true', help='Be verbose')
arg_parser.add_argument('-vv', action='store_true', help='Be more verbose')
arg_parser.add_argument('user_dir', type=str, help='path to users export dir tree')
args = arg_parser.parse_args()
if args.v:
logg.setLevel(logging.INFO)
elif args.vv:
logg.setLevel(logging.DEBUG)
config_dir = args.c
config = Config(config_dir, os.environ.get('CONFINI_ENV_PREFIX'))
config.process()
2021-06-23 06:29:38 +02:00
logg.debug('config loaded from {}:\n{}'.format(args.c, config))
2021-05-15 09:40:34 +02:00
2021-06-23 06:29:38 +02:00
ussd_data_dir = os.path.join(args.user_dir, 'ussd')
2021-05-15 09:40:34 +02:00
db_configs = {
'database': config.get('DATABASE_NAME'),
'host': config.get('DATABASE_HOST'),
'port': config.get('DATABASE_PORT'),
'user': config.get('DATABASE_USER'),
'password': config.get('DATABASE_PASSWORD')
}
celery_app = celery.Celery(broker=config.get('CELERY_BROKER_URL'), backend=config.get('CELERY_RESULT_URL'))
if __name__ == '__main__':
2021-06-23 06:29:38 +02:00
for x in os.walk(ussd_data_dir):
2021-05-15 09:40:34 +02:00
for y in x[2]:
2021-06-23 06:29:38 +02:00
if y[len(y) - 5:] == '.json':
2021-05-15 09:40:34 +02:00
filepath = os.path.join(x[0], y)
f = open(filepath, 'r')
try:
ussd_data = json.load(f)
2021-06-23 06:29:38 +02:00
logg.debug(f'LOADING USSD DATA: {ussd_data}')
2021-05-15 09:40:34 +02:00
except json.decoder.JSONDecodeError as e:
f.close()
logg.error('load error for {}: {}'.format(y, e))
continue
f.close()
s_set_ussd_data = celery.signature(
'import_task.set_ussd_data',
[db_configs, ussd_data]
)
s_set_ussd_data.apply_async(queue='cic-import-ussd')