Skip to content

Migrate concept determinations

hypha.apply.determinations.management.commands.migrate_concept_determinations

Command

Bases: BaseCommand

help class-attribute instance-attribute 

help = 'Determination migration script. Requires a source JSON file.'

data class-attribute instance-attribute 

data = []

STREAMFIELD_MAP class-attribute instance-attribute 

STREAMFIELD_MAP = {'field_cnr_technical': {'id': 'technical', 'type': 'value', 'key': 'safe_value'}, 'field_cnr_principles': {'id': 'principles', 'type': 'value', 'key': 'safe_value'}, 'field_cnr_sustainable': {'id': 'sustainable', 'type': 'value', 'key': 'safe_value'}, 'field_cnr_comments': {'id': 'comments', 'type': 'value', 'key': 'safe_value'}, 'field_cnsr_determination_message': {'id': 'message', 'type': 'value', 'key': 'safe_value'}}

add_arguments 

add_arguments(parser)
Source code in hypha/apply/determinations/management/commands/migrate_concept_determinations.py 
46
47
48
49
def add_arguments(self, parser):
    parser.add_argument(
        "source", type=argparse.FileType("r"), help="Migration source JSON file"
    )

handle 

handle(*args, **options)
Source code in hypha/apply/determinations/management/commands/migrate_concept_determinations.py 
51
52
53
54
55
56
57
58
59
60
@transaction.atomic
def handle(self, *args, **options):
    with options["source"] as json_data:
        self.data = json.load(json_data)

        blacklist = {"8104"}

        for id in self.data:
            if id not in blacklist:
                self.process(id)

process 

process(id)
Source code in hypha/apply/determinations/management/commands/migrate_concept_determinations.py 
 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
105
106
107
108
109
110
def process(self, id):
    node = self.data[id]
    form_data = {}

    try:
        determination = Determination.objects.get(drupal_id=node["nid"])
    except Determination.DoesNotExist:
        determination = Determination(drupal_id=node["nid"])

    # Disable auto_* on date fields so imported dates are used.
    for field in determination._meta.local_fields:
        if field.name == "created_at":
            field.auto_now_add = False
        elif field.name == "updated_at":
            field.auto_now = False

    # TODO timezone?
    determination.created_at = datetime.fromtimestamp(
        int(node["created"]), timezone.utc
    )
    determination.updated_at = datetime.fromtimestamp(
        int(node["changed"]), timezone.utc
    )
    determination.author = self.get_user(node["uid"])
    determination.submission = self.get_submission(node)
    determination.outcome = self.get_determination(node)
    determination.message = self.get_field_value(
        "field_cnsr_determination_message", node
    )

    for field in node:
        if field in self.STREAMFIELD_MAP:
            try:
                id = self.STREAMFIELD_MAP[field]["id"]
                if id != "message":
                    form_data[id] = self.get_field_value(field, node)
            except TypeError:
                pass

    determination.data = form_data

    try:
        determination.save()
        self.stdout.write(f"Processed \"{node['title']}\" ({node['nid']})")
    except IntegrityError:
        self.stdout.write(
            f"Skipped \"{node['title']}\" ({node['nid']}) due to IntegrityError"
        )
        pass

get_field_value 

get_field_value(field, node)

Handles the following formats: field: {(safe_)value: VALUE} field: {target_id: ID} -- Drupal ForeignKey. Reference to other node or user entities. field: {tid: ID} -- or term ID. fk to Categories field: [] field: [{value|target_id|tid: VALUE},]

Source code in hypha/apply/determinations/management/commands/migrate_concept_determinations.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
def get_field_value(self, field, node):
    """
    Handles the following formats:
    field: {(safe_)value: VALUE}
    field: {target_id: ID} -- Drupal ForeignKey. Reference to other node or user entities.
    field: {tid: ID} -- or term ID. fk to Categories
    field: []
    field: [{value|target_id|tid: VALUE},]
    """
    mapping = self.STREAMFIELD_MAP[field]
    mapping_type = mapping["type"]
    key = mapping.get("key", "value")
    source_value = node[field]
    value = None

    if mapping_type == "direct":
        value = source_value
    elif mapping_type == "value":
        if key in source_value:
            value = self.nl2br(source_value[key]) if source_value else ""
        else:
            value = self.nl2br(source_value["value"]) if source_value else ""
    elif mapping_type == "map" and "map" in "mapping":
        value = mapping["map"].get(source_value[key])
    elif mapping_type == "address" and "map" in mapping:
        try:
            value_map = mapping["map"]
            value = {}
            for item in value_map:
                value[value_map[item]] = source_value[item]
            value = json.dumps(value)
        except TypeError:
            value = {}
    elif mapping_type == "boolean":
        value = source_value[key] == "1" if source_value else False
    elif mapping_type == "category":
        if not source_value:
            value = []
        else:
            if isinstance(source_value, dict):
                option = self.get_referenced_term(source_value[key])
                value = [option] if option else []
            else:
                value = []
                for item in source_value:
                    option = self.get_referenced_term(item[key])
                    if option:
                        value.append(option)
    elif mapping_type == "file":
        # TODO finish mapping. Requires access to the files.
        value = {}

    return value

get_user 

get_user(uid)
Source code in hypha/apply/determinations/management/commands/migrate_concept_determinations.py 
166
167
168
169
170
171
def get_user(self, uid):
    try:
        User = get_user_model()
        return User.objects.get(drupal_id=uid)
    except User.DoesNotExist:
        return None

get_submission 

get_submission(node)
Source code in hypha/apply/determinations/management/commands/migrate_concept_determinations.py 
173
174
175
176
177
178
def get_submission(self, node):
    try:
        nid = node["field_submission_concept_note"]["target_id"]
        return ApplicationSubmission.objects.get(drupal_id=nid)
    except ApplicationSubmission.DoesNotExist:
        return "None"

get_determination 

get_determination(node)
Source code in hypha/apply/determinations/management/commands/migrate_concept_determinations.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
def get_determination(self, node):
    choices = {
        "invited": 2,
        "approved": 2,
        "dropped": 0,
        "unapproved": 0,
        "undetermined": 1,
    }

    try:
        determination = choices.get(node["field_cnsr_determination"]["value"], 1)
    except TypeError:
        determination = 1

    return determination

nl2br 

nl2br(value)
Source code in hypha/apply/determinations/management/commands/migrate_concept_determinations.py 
196
197
def nl2br(self, value):
    return value.replace("\r\n", "<br>\n")