<s> import os <EOL> import re <EOL> from django . db import models <EOL> from django . db . models . signals import post_save <EOL> from guardian . shortcuts import assign_perm , get_perms_for_model <EOL> from pyxform import SurveyElementBuilder <EOL> from pyxform . builder import create_survey_from_xls <EOL> from pyxform . question import Question <EOL> from pyxform . section import RepeatingSection <EOL> from pyxform . xform2json import create_survey_element_from_xml <EOL> from xml . dom import Node <EOL> from onadata . apps . logger . models . xform import XForm <EOL> from onadata . apps . logger . xform_instance_parser import clean_and_parse_xml <EOL> from onadata . apps . viewer . models . parsed_instance import _encode_for_mongo <EOL> from onadata . libs . utils . common_tags import UUID , SUBMISSION_TIME , TAGS , NOTES <EOL> from onadata . libs . utils . export_tools import question_types_to_exclude , DictOrganizer <EOL> from onadata . libs . utils . model_tools import queryset_iterator , set_uuid <EOL> class ColumnRename ( models . Model ) : <EOL> xpath = models . CharField ( max_length = <NUM_LIT:255> , unique = True ) <EOL> column_name = models . CharField ( max_length = <NUM_LIT:32> ) <EOL> class Meta : <EOL> app_label = "<STR_LIT>" <EOL> @ classmethod <EOL> def get_dict ( cls ) : <EOL> return dict ( [ ( cr . xpath , cr . column_name ) for cr in cls . objects . all ( ) ] ) <EOL> def upload_to ( instance , filename , username = None ) : <EOL> if instance : <EOL> username = instance . xform . user . username <EOL> return os . path . join ( <EOL> username , <EOL> '<STR_LIT>' , <EOL> os . path . split ( filename ) [ <NUM_LIT:1> ] <EOL> ) <EOL> class DataDictionary ( XForm ) : <EOL> GEODATA_SUFFIXES = [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' <EOL> ] <EOL> PREFIX_NAME_REGEX = re . compile ( r'<STR_LIT>' ) <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> self . instances_for_export = lambda d : d . instances . all ( ) <EOL> super ( DataDictionary , self ) . __init__ ( * args , ** kwargs ) <EOL> def _set_uuid_in_xml ( self , file_name = None ) : <EOL> """<STR_LIT>""" <EOL> if not file_name : <EOL> file_name = self . file_name ( ) <EOL> file_name , file_ext = os . path . splitext ( file_name ) <EOL> doc = clean_and_parse_xml ( self . xml ) <EOL> model_nodes = doc . getElementsByTagName ( "<STR_LIT>" ) <EOL> if len ( model_nodes ) != <NUM_LIT:1> : <EOL> raise Exception ( u"<STR_LIT>" ) <EOL> model_node = model_nodes [ <NUM_LIT:0> ] <EOL> instance_nodes = [ node for node in model_node . childNodes if <EOL> node . nodeType == Node . ELEMENT_NODE and <EOL> node . tagName . lower ( ) == "<STR_LIT>" and <EOL> not node . hasAttribute ( "<STR_LIT:id>" ) ] <EOL> if len ( instance_nodes ) != <NUM_LIT:1> : <EOL> raise Exception ( u"<STR_LIT>" <EOL> u"<STR_LIT>" ) <EOL> instance_node = instance_nodes [ <NUM_LIT:0> ] <EOL> survey_nodes = [ node for node in instance_node . childNodes <EOL> if node . nodeType == Node . ELEMENT_NODE and <EOL> ( node . tagName == file_name or <EOL> node . attributes . get ( '<STR_LIT:id>' ) ) ] <EOL> if len ( survey_nodes ) != <NUM_LIT:1> : <EOL> raise Exception ( <EOL> u"<STR_LIT>" % self . id_string ) <EOL> survey_node = survey_nodes [ <NUM_LIT:0> ] <EOL> formhub_nodes = [ n for n in survey_node . childNodes <EOL> if n . nodeType == Node . ELEMENT_NODE and <EOL> n . tagName == "<STR_LIT>" ] <EOL> if len ( formhub_nodes ) > <NUM_LIT:1> : <EOL> raise Exception ( <EOL> u"<STR_LIT>" ) <EOL> elif len ( formhub_nodes ) == <NUM_LIT:1> : <EOL> formhub_node = formhub_nodes [ <NUM_LIT:0> ] <EOL> else : <EOL> formhub_node = survey_node . insertBefore ( <EOL> doc . createElement ( "<STR_LIT>" ) , survey_node . firstChild ) <EOL> uuid_nodes = [ node for node in formhub_node . childNodes if <EOL> node . nodeType == Node . ELEMENT_NODE and <EOL> node . tagName == "<STR_LIT>" ] <EOL> if len ( uuid_nodes ) == <NUM_LIT:0> : <EOL> formhub_node . appendChild ( doc . createElement ( "<STR_LIT>" ) ) <EOL> if len ( formhub_nodes ) == <NUM_LIT:0> : <EOL> calculate_node = doc . createElement ( "<STR_LIT>" ) <EOL> calculate_node . setAttribute ( <EOL> "<STR_LIT>" , "<STR_LIT>" % file_name ) <EOL> calculate_node . setAttribute ( "<STR_LIT:type>" , "<STR_LIT:string>" ) <EOL> calculate_node . setAttribute ( "<STR_LIT>" , "<STR_LIT>" % self . uuid ) <EOL> model_node . appendChild ( calculate_node ) <EOL> self . xml = doc . toprettyxml ( indent = "<STR_LIT:U+0020>" , encoding = '<STR_LIT:utf-8>' ) <EOL> text_re = re . compile ( '<STR_LIT>' , re . DOTALL ) <EOL> output_re = re . compile ( '<STR_LIT>' ) <EOL> prettyXml = text_re . sub ( '<STR_LIT>' , self . xml ) <EOL> inlineOutput = output_re . sub ( '<STR_LIT>' , prettyXml ) <EOL> inlineOutput = re . compile ( '<STR_LIT>' ) . sub ( <EOL> '<STR_LIT>' , inlineOutput ) <EOL> self . xml = inlineOutput <EOL> class Meta : <EOL> app_label = "<STR_LIT>" <EOL> proxy = True <EOL> def add_instances ( self ) : <EOL> if not hasattr ( self , "<STR_LIT>" ) : <EOL> _dict_organizer = DictOrganizer ( ) <EOL> obs = [ ] <EOL> for d in self . get_list_of_parsed_instances ( flat = False ) : <EOL> obs . append ( _dict_organizer . get_observation_from_dict ( d ) ) <EOL> return obs <EOL> def save ( self , * args , ** kwargs ) : <EOL> if self . xls : <EOL> survey = create_survey_from_xls ( self . xls ) <EOL> self . json = survey . to_json ( ) <EOL> self . xml = survey . to_xml ( ) <EOL> self . _mark_start_time_boolean ( ) <EOL> set_uuid ( self ) <EOL> self . _set_uuid_in_xml ( ) <EOL> super ( DataDictionary , self ) . save ( * args , ** kwargs ) <EOL> def file_name ( self ) : <EOL> return os . path . split ( self . xls . name ) [ - <NUM_LIT:1> ] <EOL> def get_survey ( self ) : <EOL> if not hasattr ( self , "<STR_LIT>" ) : <EOL> try : <EOL> builder = SurveyElementBuilder ( ) <EOL> self . _survey = builder . create_survey_element_from_json ( self . json ) <EOL> except ValueError : <EOL> xml = bytes ( bytearray ( self . xml , encoding = '<STR_LIT:utf-8>' ) ) <EOL> self . _survey = create_survey_element_from_xml ( xml ) <EOL> return self . _survey <EOL> survey = property ( get_survey ) <EOL> def get_survey_elements ( self ) : <EOL> return self . survey . iter_descendants ( ) <EOL> def get_survey_element ( self , name_or_xpath ) : <EOL> element = self . get_element ( name_or_xpath ) <EOL> name = ( element and element [ '<STR_LIT:name>' ] ) or name_or_xpath <EOL> for field in self . get_survey_elements ( ) : <EOL> if field . name == name : <EOL> return field <EOL> return None <EOL> def get_choice_label ( self , field , choice_value , lang = '<STR_LIT>' ) : <EOL> for choice in field . children : <EOL> if choice . name == choice_value : <EOL> label = choice . label <EOL> if isinstance ( label , dict ) : <EOL> label = label . get ( lang , choice . label . values ( ) [ <NUM_LIT:0> ] ) <EOL> return label <EOL> return choice_value <EOL> def get_mongo_field_names_dict ( self ) : <EOL> """<STR_LIT>""" <EOL> names = { } <EOL> for elem in self . get_survey_elements ( ) : <EOL> names [ _encode_for_mongo ( unicode ( elem . get_abbreviated_xpath ( ) ) ) ] = elem . get_abbreviated_xpath ( ) <EOL> return names <EOL> survey_elements = property ( get_survey_elements ) <EOL> def geopoint_xpaths ( self ) : <EOL> geo_xpaths = [ ] <EOL> for e in self . get_survey_elements ( ) : <EOL> if e . bind . get ( u'<STR_LIT:type>' ) == u'<STR_LIT>' : <EOL> geo_xpaths . append ( e . get_abbreviated_xpath ( ) ) <EOL> return geo_xpaths <EOL> def xpath_of_first_geopoint ( self ) : <EOL> geo_xpaths = self . geopoint_xpaths ( ) <EOL> return len ( geo_xpaths ) > <NUM_LIT:0> and geo_xpaths [ <NUM_LIT:0> ] <EOL> def has_instances_with_geopoints ( self ) : <EOL> return self . geocoded_submission_count ( ) > <NUM_LIT:0> <EOL> def xpaths ( self , prefix = '<STR_LIT>' , survey_element = None , result = None , <EOL> repeat_iterations = <NUM_LIT:4> ) : <EOL> """<STR_LIT>""" <EOL> if survey_element is None : <EOL> survey_element = self . survey <EOL> elif question_types_to_exclude ( survey_element . type ) : <EOL> return [ ] <EOL> if result is None : <EOL> result = [ ] <EOL> path = '<STR_LIT:/>' . join ( [ prefix , unicode ( survey_element . name ) ] ) <EOL> if survey_element . children is not None : <EOL> indices = [ '<STR_LIT>' ] if type ( survey_element ) != RepeatingSection else [ '<STR_LIT>' % ( i + <NUM_LIT:1> ) for i in range ( repeat_iterations ) ] <EOL> for i in indices : <EOL> for e in survey_element . children : <EOL> self . xpaths ( path + i , e , result , repeat_iterations ) <EOL> if isinstance ( survey_element , Question ) : <EOL> result . append ( path ) <EOL> if survey_element . bind . get ( u'<STR_LIT:type>' ) == u'<STR_LIT>' : <EOL> result . pop ( ) <EOL> for child in survey_element . children : <EOL> result . append ( '<STR_LIT:/>' . join ( [ path , child . name ] ) ) <EOL> elif survey_element . bind . get ( u'<STR_LIT:type>' ) == u'<STR_LIT>' : <EOL> result += self . get_additional_geopoint_xpaths ( path ) <EOL> return result <EOL> @ classmethod <EOL> def get_additional_geopoint_xpaths ( cls , xpath ) : <EOL> """<STR_LIT>""" <EOL> match = cls . PREFIX_NAME_REGEX . match ( xpath ) <EOL> prefix = '<STR_LIT>' <EOL> name = '<STR_LIT>' <EOL> if match : <EOL> prefix = match . groupdict ( ) [ '<STR_LIT>' ] <EOL> name = match . groupdict ( ) [ '<STR_LIT:name>' ] <EOL> else : <EOL> name = xpath <EOL> return [ prefix + '<STR_LIT:_>' + name + '<STR_LIT:_>' + suffix <EOL> for suffix in cls . GEODATA_SUFFIXES ] <EOL> def _additional_headers ( self ) : <EOL> return [ u'<STR_LIT>' , u'<STR_LIT>' , u'<STR_LIT>' , <EOL> u'<STR_LIT>' , u'<STR_LIT>' , u'<STR_LIT>' ] <EOL> def get_headers ( self , include_additional_headers = False ) : <EOL> """<STR_LIT>""" <EOL> def shorten ( xpath ) : <EOL> l = xpath . split ( '<STR_LIT:/>' ) <EOL> return '<STR_LIT:/>' . join ( l [ <NUM_LIT:2> : ] ) <EOL> header_list = [ shorten ( xpath ) for xpath in self . xpaths ( ) ] <EOL> header_list += [ UUID , SUBMISSION_TIME , TAGS , NOTES ] <EOL> if include_additional_headers : <EOL> header_list += self . _additional_headers ( ) <EOL> return header_list <EOL> def get_keys ( self ) : <EOL> def remove_first_index ( xpath ) : <EOL> return re . sub ( r'<STR_LIT>' , '<STR_LIT>' , xpath ) <EOL> return [ remove_first_index ( header ) for header in self . get_headers ( ) ] <EOL> def get_element ( self , abbreviated_xpath ) : <EOL> if not hasattr ( self , "<STR_LIT>" ) : <EOL> self . _survey_elements = { } <EOL> for e in self . get_survey_elements ( ) : <EOL> self . _survey_elements [ e . get_abbreviated_xpath ( ) ] = e <EOL> def remove_all_indices ( xpath ) : <EOL> return re . sub ( r"<STR_LIT>" , u"<STR_LIT>" , xpath ) <EOL> clean_xpath = remove_all_indices ( abbreviated_xpath ) <EOL> return self . _survey_elements . get ( clean_xpath ) <EOL> def get_label ( self , abbreviated_xpath ) : <EOL> e = self . get_element ( abbreviated_xpath ) <EOL> if e : <EOL> return e . label <EOL> def get_xpath_cmp ( self ) : <EOL> if not hasattr ( self , "<STR_LIT>" ) : <EOL> self . _xpaths = [ e . get_abbreviated_xpath ( ) <EOL> for e in self . survey_elements ] <EOL> def xpath_cmp ( x , y ) : <EOL> new_x = re . sub ( r"<STR_LIT>" , u"<STR_LIT>" , x ) <EOL> new_y = re . sub ( r"<STR_LIT>" , u"<STR_LIT>" , y ) <EOL> if new_x == new_y : <EOL> return cmp ( x , y ) <EOL> if new_x not in self . _xpaths and new_y not in self . _xpaths : <EOL> return <NUM_LIT:0> <EOL> elif new_x not in self . _xpaths : <EOL> return <NUM_LIT:1> <EOL> elif new_y not in self . _xpaths : <EOL> return - <NUM_LIT:1> <EOL> return cmp ( self . _xpaths . index ( new_x ) , self . _xpaths . index ( new_y ) ) <EOL> return xpath_cmp <EOL> def get_variable_name ( self , abbreviated_xpath ) : <EOL> """<STR_LIT>""" <EOL> if not hasattr ( self , "<STR_LIT>" ) : <EOL> self . _keys = self . get_keys ( ) <EOL> if not hasattr ( self , "<STR_LIT>" ) : <EOL> self . _headers = self . get_headers ( ) <EOL> assert abbreviated_xpath in self . _keys , abbreviated_xpath <EOL> i = self . _keys . index ( abbreviated_xpath ) <EOL> header = self . _headers [ i ] <EOL> if not hasattr ( self , "<STR_LIT>" ) : <EOL> self . _variable_names = ColumnRename . get_dict ( ) <EOL> assert type ( self . _variable_names ) == dict <EOL> if header in self . _variable_names and self . _variable_names [ header ] : <EOL> return self . _variable_names [ header ] <EOL> return header <EOL> def get_list_of_parsed_instances ( self , flat = True ) : <EOL> for i in queryset_iterator ( self . instances_for_export ( self ) ) : <EOL> yield i . get_dict ( flat = flat ) <EOL> def _rename_key ( self , d , old_key , new_key ) : <EOL> assert new_key not in d , d <EOL> d [ new_key ] = d [ old_key ] <EOL> del d [ old_key ] <EOL> def _expand_select_all_that_apply ( self , d , key , e ) : <EOL> if e and e . bind . get ( u"<STR_LIT:type>" ) == u"<STR_LIT>" : <EOL> options_selected = d [ key ] . split ( ) <EOL> for i , child in enumerate ( e . children ) : <EOL> new_key = child . get_abbreviated_xpath ( ) <EOL> if child . name in options_selected : <EOL> d [ new_key ] = True <EOL> else : <EOL> d [ new_key ] = False <EOL> del d [ key ] <EOL> def _expand_geocodes ( self , d , key , e ) : <EOL> if e and e . bind . get ( u"<STR_LIT:type>" ) == u"<STR_LIT>" : <EOL> geodata = d [ key ] . split ( ) <EOL> for i in range ( len ( geodata ) ) : <EOL> new_key = "<STR_LIT>" % ( key , self . geodata_suffixes [ i ] ) <EOL> d [ new_key ] = geodata [ i ] <EOL> def get_data_for_excel ( self ) : <EOL> for d in self . get_list_of_parsed_instances ( ) : <EOL> for key in d . keys ( ) : <EOL> e = self . get_element ( key ) <EOL> self . _expand_select_all_that_apply ( d , key , e ) <EOL> self . _expand_geocodes ( d , key , e ) <EOL> yield d <EOL> def _mark_start_time_boolean ( self ) : <EOL> starttime_substring = '<STR_LIT>' <EOL> if self . xml . find ( starttime_substring ) != - <NUM_LIT:1> : <EOL> self . has_start_time = True <EOL> else : <EOL> self . has_start_time = False <EOL> def get_survey_elements_of_type ( self , element_type ) : <EOL> return [ e for e in self . get_survey_elements ( ) <EOL> if e . type == element_type ] <EOL> def set_object_permissions ( sender , instance = None , created = False , ** kwargs ) : <EOL> if created : <EOL> for perm in get_perms_for_model ( XForm ) : <EOL> assign_perm ( perm . codename , instance . user , instance ) <EOL> post_save . connect ( set_object_permissions , sender = DataDictionary , <EOL> dispatch_uid = '<STR_LIT>' ) </s>
