#1840

Committed 26 Jun 2024 09:52AM UTC coverage: 84.867% (-0.01%) from 84.879%

Build # #1840

Build Type

push

coveralls-python

Committed by

web-flow

Commit Message

Merge pull request #1528 from akvo/feature/1524-hh-consent-question

Feature/1524 hh consent question

Run Details

3068 of 3747 branches covered (81.88%)

Branch coverage included in aggregate %.

6858 of 7949 relevant lines covered (86.28%)

0.86 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

80.25

backend/api/v1/v1_forms/management/commands/download_forms_from_afs.py

import requests as r
import json
import pandas as pd
import re

from django.core.management import BaseCommand

base_url = 'https://form-service.akvotest.org/api/form/'
forms = [
    {
        "id": 1701757876668,
        "name": "RTMIS School WASH Form",
    },
]
question_types = {'input': 'text', 'image': 'photo'}


def to_snake_case(s):
    if not s:
        return s
    # Remove special characters (except underscore and space)
    s = re.sub(r'[^\w\s]', '', s)
    # Replace whitespace and hyphens with underscores
    s = re.sub(r'[\s-]+', '_', s)
    # Convert camelCase to snake_case
    s = re.sub(r'(?<!^)(?=[A-Z])', '_', s).lower()
    # Ensure no double underscores
    s = re.sub(r'_+', '_', s)
    return s


def option_mapping(row):
    if row[0] != row[0]:
        return row[0]
    options = [
        {
            'value': r.get(
                'value',
                r.get('code', to_snake_case(r.get('name')))
            ),
            'label': r.get('label', r.get('name')),
            **{
                k: v for k, v in r.items()
                if k not in ['value', 'label', 'code', 'name']
            }
        }
        for r in row[0]
    ]
    return options


def question_mapping(row):
    qs = pd.DataFrame(row[0])\
        .rename(columns={'option': 'options'})

    if 'label' not in qs.columns:
        qs.insert(2, 'label', qs['name'])
        qs['name'] = qs['name'].apply(to_snake_case)

    if 'options' in qs.columns:
        qs['options'] = pd.DataFrame(qs['options']).apply(
            option_mapping, axis=1
        )

    qs = qs.where(pd.notnull(qs), None)
    qs = qs.to_dict('records')
    qs = list(map(
        lambda q: {
            **q,
            'type': question_types.get(str(q['type'])) if question_types.get(
                str(q['type'])
            ) else q['type']
        }, qs))
    return qs


def generate_form(form_id: int, data: list, test: bool = False):
    df = pd.json_normalize(data)
    df['type'] = 1
    df = df.rename(columns={
        'name': 'form',
        'question_group': 'question_groups'
    })
    df2 = pd.DataFrame(df['question_groups'][0])
    df2 = df2.rename(columns={
        'question': 'questions'
    })
    if 'label' not in df2.columns:
        df2.insert(2, 'label', df2['name'])
        df2['name'] = df2['name'].apply(to_snake_case)

    df2['questions'] = pd.DataFrame(df2['questions'])\
        .apply(question_mapping, axis=1)
    qg = df2.to_dict('records')
    df['question_groups'] = df['question_groups'].apply(lambda x: qg)
    j = df.to_dict('records')[0]
    if not test:
        with open(f"./source/forms/{form_id}.prod.json", 'w') as f:
            json.dump(j, f)


class Command(BaseCommand):
    def add_arguments(self, parser):
        parser.add_argument("-t",
                            "--test",
                            nargs="?",
                            const=1,
                            default=False,
                            type=int)

    def handle(self, *args, **options):
        TEST = options.get("test")
        for form in forms:
            form_id = form['id']
            form_name = form['name']
            req = r.get(f"{base_url}{form_id}")
            generate_form(form_id=form_id, data=req.json(), test=TEST)
            self.stdout.write(
                f"Form Downloaded | {form_id} {form_name}"
            )

1	import requests as r	1✔
2	import json	1✔
3	import pandas as pd	1✔
4	import re	1✔
5
6	from django.core.management import BaseCommand	1✔
7
8	base_url = 'https://form-service.akvotest.org/api/form/'	1✔
9	forms = [	1✔
10	{
11	"id": 1701757876668,
12	"name": "RTMIS School WASH Form",
13	},
14	]
15	question_types = {'input': 'text', 'image': 'photo'}	1✔
16
17
18	def to_snake_case(s):	1✔
19	if not s:	1!
20	return s	1✔
21	# Remove special characters (except underscore and space)
22	s = re.sub(r'[^\w\s]', '', s)	×
23	# Replace whitespace and hyphens with underscores
24	s = re.sub(r'[\s-]+', '_', s)	×
25	# Convert camelCase to snake_case
26	s = re.sub(r'(?<!^)(?=[A-Z])', '_', s).lower()	×
27	# Ensure no double underscores
28	s = re.sub(r'_+', '_', s)	×
29	return s	×
30
31
32	def option_mapping(row):	1✔
33	if row[0] != row[0]:	1✔
34	return row[0]	1✔
35	options = [	1✔
36	{
37	'value': r.get(
38	'value',
39	r.get('code', to_snake_case(r.get('name')))
40	),
41	'label': r.get('label', r.get('name')),
42	**{
43	k: v for k, v in r.items()
44	if k not in ['value', 'label', 'code', 'name']
45	}
46	}
47	for r in row[0]
48	]
49	return options	1✔
50
51
52	def question_mapping(row):	1✔
53	qs = pd.DataFrame(row[0])\	1✔
54	.rename(columns={'option': 'options'})
55
56	if 'label' not in qs.columns:	1!
57	qs.insert(2, 'label', qs['name'])	×
58	qs['name'] = qs['name'].apply(to_snake_case)	×
59
60	if 'options' in qs.columns:	1!
61	qs['options'] = pd.DataFrame(qs['options']).apply(	1✔
62	option_mapping, axis=1
63	)
64
65	qs = qs.where(pd.notnull(qs), None)	1✔
66	qs = qs.to_dict('records')	1✔
67	qs = list(map(	1✔
68	lambda q: {
69	**q,
70	'type': question_types.get(str(q['type'])) if question_types.get(
71	str(q['type'])
72	) else q['type']
73	}, qs))
74	return qs	1✔
75
76
77	def generate_form(form_id: int, data: list, test: bool = False):	1✔
78	df = pd.json_normalize(data)	1✔
79	df['type'] = 1	1✔
80	df = df.rename(columns={	1✔
81	'name': 'form',
82	'question_group': 'question_groups'
83	})
84	df2 = pd.DataFrame(df['question_groups'][0])	1✔
85	df2 = df2.rename(columns={	1✔
86	'question': 'questions'
87	})
88	if 'label' not in df2.columns:	1!
89	df2.insert(2, 'label', df2['name'])	×
90	df2['name'] = df2['name'].apply(to_snake_case)	×
91
92	df2['questions'] = pd.DataFrame(df2['questions'])\	1✔
93	.apply(question_mapping, axis=1)
94	qg = df2.to_dict('records')	1✔
95	df['question_groups'] = df['question_groups'].apply(lambda x: qg)	1✔
96	j = df.to_dict('records')[0]	1✔
97	if not test:	1!
98	with open(f"./source/forms/{form_id}.prod.json", 'w') as f:	×
99	json.dump(j, f)	×
100
101
102	class Command(BaseCommand):	1✔
103	def add_arguments(self, parser):	1✔
104	parser.add_argument("-t",	1✔
105	"--test",
106	nargs="?",
107	const=1,
108	default=False,
109	type=int)
110
111	def handle(self, args, *options):	1✔
112	TEST = options.get("test")	1✔
113	for form in forms:	1✔
114	form_id = form['id']	1✔
115	form_name = form['name']	1✔
116	req = r.get(f"{base_url}{form_id}")	1✔
117	generate_form(form_id=form_id, data=req.json(), test=TEST)	1✔
118	self.stdout.write(	1✔
119	f"Form Downloaded \| {form_id} {form_name}"
120	)

akvo / rtmis / #1840

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous