#!/usr/bin/env python # -*- coding: utf-8 -*- # Copyright (C) 2013 Étienne Loks # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License as # published by the Free Software Foundation, either version 3 of the # License, or (at your option) any later version. # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . # See the file COPYING for details. """ Utils: import archaelogical operation from a DBF file """ import datetime import dbf from django.contrib.auth.models import User from django.db import transaction from archaeological_operations.import_from_csv import parse_operationtype, \ parse_multivalue, parse_person as _parse_person, parse_date from archaeological_operations.models import Operation, OperationType, Period, \ AdministrativeAct, ActType def parse_person(surname, name, owner): return _parse_person(surname, name, None, owner) def parse_ha(value): try: val = float(value) except ValueError: val = 0 return val * 10000 ope_types = { 'AET':None, 'APP':None, 'DOC':(u'documents_study', u'Étude documentaire'), 'EV':(u'evaluation', u'Évaluation'), 'FOU':(u'prev_excavation', u"Fouille archéologique préventive"), 'FP':(u'prog_excavation', u"Fouille archéologique programmée"), 'MH':(u'building_study', u"Étude de bâti (préventif)"), 'OPD':None, 'PCR':None, 'PMS':None, 'PRD':None, 'PRT':None, 'PRM':None, 'RAR':None, 'SD':(u'sampling_research', u"Sondage (préventif)"), 'SP':(u'prev_excavation', u"Fouille archéologique préventive"), 'SU':(u'emergency_excavation', u"Sauvetage urgent"), } def parse_patriarche_operationtype(value): if value not in ope_types.keys(): print value return None if not ope_types[value]: return None return OperationType.objects.get(txt_idx=ope_types[value][0]) PATRIARCHE_DBF_OPE_COLS = [ (('operation_type',), 'parse_patriarche_operationtype', []), (('common_name',), unicode, []), [], (('in_charge',), 'parse_person', [2]), [], #'etat', [], #'adresse', [], #'origine C(3)', [], # 'chronologi C(12)', [], #'programme C(254)', [], # 'rattach_pc C(254)', [], # 'code_dossi N(8,0)', (('administrative_act', 'ref_sra'), unicode, []), (('administrative_act', 'signature_date'), parse_date, []), (('start_date',), parse_date, []), (('end_date',), parse_date, []), (('year',), int, []), [], # 'identifica C(254)', (('code_patriarche',), int, []), [], # 'x_degre N(16,6)', [], # 'y_degre N(16,6)', [], # 'x_saisi C(12)', [], # 'y_saisi C(12)', [], # 'georeferen C(3)', [], # 'geometrie C(3)', (('surface',), parse_ha, []) ] DBF_OPE_COLS = PATRIARCHE_DBF_OPE_COLS def import_from_dbf(filename, update=False, col_defs=DBF_OPE_COLS, person=None, stdout=None): """ Import from a DBF file. Return number of operation treated and errors. """ try: table = dbf.Table(filename) except (dbf.DbfError, TypeError): return 0, [u"Incorrect DBF file."] new_ops, errors = import_operations_dbf(table, col_defs=col_defs, update=update, person=person, stdout=stdout) return new_ops, errors ERROR_LBLS = {'missing_ope_type':'* Missing operation type: ', 'missing_patriarche_code':'* Missing patriarche code: '} @transaction.commit_manually def import_operations_dbf(values, col_defs=DBF_OPE_COLS, update=False, person=None, stdout=None): default_person = person or User.objects.order_by('pk').all()[0] # key : (class, default, reverse) key_classes = { 'administrative_act':(AdministrativeAct, {'history_modifier':default_person, 'act_type':ActType.objects.get( txt_idx='excavation_order')}, 'operation'), } ope_default = {'history_modifier':default_person} current_import = [] new_ops = 0 errors_nb = {} for error in ERROR_LBLS.keys(): errors_nb[error] = 0 error_ope, error_reversed, error_multis = [], [], [] for line_idx, vals in enumerate(values): if stdout: stdout.write("\r* line %d" % (line_idx)) if not line_idx: continue # remove header args = {} for col_idx, val in enumerate(vals): if len(col_defs) <= col_idx or not col_defs[col_idx]: continue attrs, typ, extra_cols = col_defs[col_idx] if not callable(typ): typ = globals()[typ] c_args = args for attr in attrs: if attr not in c_args: c_args[attr] = {} c_args = c_args[attr] if not extra_cols: try: v = typ(val) except TypeError: v = None else: arguments = [vals[col_number] for col_number in extra_cols] if not [arg for arg in arguments if arg]: continue arguments += [default_person] v = typ(val, *arguments) if len(attrs) == 1: args[attrs[0]] = v elif len(attrs) == 2: args[attrs[0]][attrs[1]] = v elif len(attrs) == 3: args[attrs[0]][attrs[1]][attrs[2]] = v # manage exploded dates for k in args.keys(): if '__year' in k: key = k[:-len('__year')] try: v = datetime.datetime(args[k], args[key+'__month'], args[key+'__day']) args[key] = v except: pass args.pop(k) args.pop(key+'__month') args.pop(key+'__day') reversed_items, multis = [], [] for k in args.keys(): if k in key_classes: cls, default, reverse = key_classes[k] default.update(args[k]) if reverse: reversed_items.append((cls, default, reverse)) args.pop(k) continue try: obj = cls.objects.get(**default) except: obj = cls.objects.create(**default) obj.save() transaction.commit() args[k] = obj elif type(args[k]) == list: multis.append((k, args[k])) args.pop(k) op = None if not update and not args.get('operation_type'): errors_nb['missing_ope_type'] += 1 continue try: op = Operation.objects.get(code_patriarche=args['code_patriarche']) if not update and op.pk not in current_import: errors_nb['already_available_patriarche_code'] += 1 continue except: pass # check if not args.get('year') and args.get('start_date'): args['year'] = args['start_date'].year # creation if not op: args.update(ope_default) op = Operation.objects.create(**args) new_ops += 1 transaction.commit() current_import.append(op.pk) else: # mise à jour try: for k in args: if getattr(op, k): continue setattr(op, k, args[k]) op.save() except: transaction.rollback() continue transaction.commit() try: for cls, default, reverse in reversed_items: default[reverse] = op it = cls(**default).save() except: transaction.rollback() error_reversed.append((line_idx, reversed_items)) continue transaction.commit() try: for k, vals in multis: for v in vals: getattr(op, k).add(v) op.save() except: transaction.rollback() error_multis.append((line_idx, multis)) continue transaction.commit() errors = [] for error_key in errors_nb: nb_error = errors_nb[error_key] if nb_error: errors.append(ERROR_LBLS[error_key] + str(nb_error)) if error_ope: error = "Error while recording theses operations:\n" for line_idx, args in error_ope: error += "line: " + str(line_idx) + " args: " + str(args) + '\n' errors.append(error) if error_multis: error = "Error while recording theses multiples items attached to "\ "operation:" for line_idx, args in error_multis: error += "line: " + str(line_idx) + " args: " + str(args) + '\n' errors.append(error) if error_reversed: error = "Error while recording theses items that depend to operation:" for line_idx, args in error_reversed: error += "line: " + str(line_idx) + " args: " + str(args) + '\n' errors.append(error) return new_ops, errors