swf-monitor/scripts/import-questionnaires.py

0001 #!/usr/bin/env python
0002 """Import ePIC production-request questionnaire CSV into PCS.
0003
0004 Standalone Django-bootstrap script, usable by hand or cron. It imports the
0005 Google Form responses-sheet CSV export through the same PCS service used by
0006 the web import button.
0007
0008 Usage:
0009     cd /data/wenauseic/github/swf-monitor
0010     source ../swf-testbed/.venv/bin/activate && source ~/.env
0011     scripts/import-questionnaires.py --url 'https://docs.google.com/.../export?format=csv'
0012     scripts/import-questionnaires.py --file responses.csv
0013 """
0014 import argparse
0015 import os
0016 import sys
0017 from urllib.request import urlopen
0018
0019
0020 THIS_DIR = os.path.dirname(os.path.abspath(__file__))
0021 sys.path.insert(0, os.path.join(THIS_DIR, '..', 'src'))
0022 os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'swf_monitor_project.settings')
0023
0024 import django  # noqa: E402
0025 django.setup()
0026
0027
0028 def _read_input(args):
0029     if args.url:
0030         with urlopen(args.url, timeout=args.timeout) as response:
0031             return response.read().decode('utf-8-sig'), args.url
0032     if args.file:
0033         with open(args.file, 'r', encoding='utf-8-sig') as f:
0034             return f.read(), args.file
0035     raise ValueError('provide --url or --file')
0036
0037
0038 def main(argv=None):
0039     parser = argparse.ArgumentParser()
0040     group = parser.add_mutually_exclusive_group(required=True)
0041     group.add_argument('--url', help='Link-readable Google Sheet CSV export URL')
0042     group.add_argument('--file', help='Local CSV file')
0043     parser.add_argument('--created-by', default='questionnaire_import')
0044     parser.add_argument('--timeout', type=int, default=30)
0045     args = parser.parse_args(argv)
0046
0047     from pcs.services import questionnaire_intake_csv, ServiceError
0048
0049     try:
0050         csv_text, source_url = _read_input(args)
0051         summary = questionnaire_intake_csv(
0052             csv_text, source_url=source_url, created_by=args.created_by)
0053     except (OSError, ValueError, ServiceError) as e:
0054         print(f'ERROR: {e}', file=sys.stderr)
0055         return 1
0056
0057     print(
0058         f"request questionnaire: {summary['created']} new, "
0059         f"{summary['updated']} updated, {summary['unchanged']} unchanged"
0060     )
0061     return 0
0062
0063
0064 if __name__ == '__main__':
0065     sys.exit(main())