<s> import json <EOL> from couchdbkit import ResourceNotFound <EOL> from collections import OrderedDict <EOL> from django . contrib import messages <EOL> from django . core . urlresolvers import reverse <EOL> from django . http import HttpResponseBadRequest , HttpResponseRedirect , Http404 , HttpResponse <EOL> from django . http . response import HttpResponseServerError <EOL> from django . shortcuts import render <EOL> from django . utils . decorators import method_decorator <EOL> from django . utils . translation import ugettext as _ , ugettext_noop <EOL> from django . views . decorators . csrf import csrf_exempt <EOL> from django . views . decorators . http import require_POST <EOL> from django . views . generic . base import TemplateView <EOL> from corehq . apps . domain . decorators import login_or_digest , login_and_domain_required <EOL> from corehq . apps . domain . views import BaseDomainView <EOL> from corehq . apps . fixtures . tasks import fixture_upload_async , fixture_download_async <EOL> from corehq . apps . fixtures . dispatcher import require_can_edit_fixtures <EOL> from corehq . apps . fixtures . download import prepare_fixture_download , prepare_fixture_html <EOL> from corehq . apps . fixtures . exceptions import ( <EOL> FixtureDownloadError , <EOL> ExcelMalformatException , <EOL> FixtureAPIException , <EOL> DuplicateFixtureTagException , <EOL> FixtureUploadError <EOL> ) <EOL> from corehq . apps . fixtures . models import FixtureDataType , FixtureDataItem , FieldList , FixtureTypeField <EOL> from corehq . apps . fixtures . upload import run_upload , validate_file_format , get_workbook <EOL> from corehq . apps . fixtures . fixturegenerators import item_lists_by_domain <EOL> from corehq . apps . fixtures . utils import is_identifier_invalid <EOL> from corehq . apps . reports . datatables import DataTablesHeader , DataTablesColumn <EOL> from corehq . apps . reports . util import format_datatables_data <EOL> from corehq . apps . users . models import Permissions <EOL> from corehq . util . files import file_extention_from_filename <EOL> from corehq . util . spreadsheets . excel import JSONReaderError , HeaderValueError , WorksheetNotFound <EOL> from dimagi . utils . couch . bulk import CouchTransaction <EOL> from dimagi . utils . logging import notify_exception <EOL> from dimagi . utils . web import json_response <EOL> from dimagi . utils . decorators . view import get_file <EOL> from copy import deepcopy <EOL> from soil import CachedDownload , DownloadBase <EOL> from soil . exceptions import TaskFailedError <EOL> from soil . util import expose_cached_download , get_download_context <EOL> def strip_json ( obj , disallow_basic = None , disallow = None ) : <EOL> disallow = disallow or [ ] <EOL> if disallow_basic is None : <EOL> disallow_basic = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> disallow += disallow_basic <EOL> ret = { } <EOL> try : <EOL> obj = obj . to_json ( ) <EOL> except Exception : <EOL> pass <EOL> for key in obj : <EOL> if key not in disallow : <EOL> ret [ key ] = obj [ key ] <EOL> return ret <EOL> def _to_kwargs ( req ) : <EOL> return dict ( ( str ( k ) , v ) for k , v in json . load ( req , object_pairs_hook = OrderedDict ) . items ( ) ) <EOL> @ require_can_edit_fixtures <EOL> def tables ( request , domain ) : <EOL> if request . method == '<STR_LIT:GET>' : <EOL> return json_response ( <EOL> [ strip_json ( x ) for x in <EOL> sorted ( FixtureDataType . by_domain ( domain ) , key = lambda data_type : data_type . tag ) ] <EOL> ) <EOL> @ require_can_edit_fixtures <EOL> def update_tables ( request , domain , data_type_id , test_patch = None ) : <EOL> """<STR_LIT>""" <EOL> if test_patch is None : <EOL> test_patch = { } <EOL> if data_type_id : <EOL> try : <EOL> data_type = FixtureDataType . get ( data_type_id ) <EOL> except ResourceNotFound : <EOL> raise Http404 ( ) <EOL> assert ( data_type . doc_type == FixtureDataType . _doc_type ) <EOL> assert ( data_type . domain == domain ) <EOL> if request . method == '<STR_LIT:GET>' : <EOL> return json_response ( strip_json ( data_type ) ) <EOL> elif request . method == '<STR_LIT>' : <EOL> with CouchTransaction ( ) as transaction : <EOL> data_type . recursive_delete ( transaction ) <EOL> return json_response ( { } ) <EOL> elif not request . method == '<STR_LIT>' : <EOL> return HttpResponseBadRequest ( ) <EOL> if request . method == '<STR_LIT:POST>' or request . method == "<STR_LIT>" : <EOL> fields_update = test_patch or _to_kwargs ( request ) <EOL> fields_patches = fields_update [ "<STR_LIT>" ] <EOL> data_tag = fields_update [ "<STR_LIT>" ] <EOL> is_global = fields_update [ "<STR_LIT>" ] <EOL> validation_errors = [ ] <EOL> if is_identifier_invalid ( data_tag ) : <EOL> validation_errors . append ( data_tag ) <EOL> for field_name , options in fields_update [ '<STR_LIT>' ] . items ( ) : <EOL> method = options . keys ( ) <EOL> if '<STR_LIT>' in method : <EOL> field_name = options [ '<STR_LIT>' ] <EOL> if is_identifier_invalid ( field_name ) and '<STR_LIT>' not in method : <EOL> validation_errors . append ( field_name ) <EOL> validation_errors = map ( lambda e : _ ( "<STR_LIT>" <EOL> "<STR_LIT>" ) % e , validation_errors ) <EOL> if validation_errors : <EOL> return json_response ( { <EOL> '<STR_LIT>' : validation_errors , <EOL> '<STR_LIT>' : _ ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) , <EOL> } ) <EOL> with CouchTransaction ( ) as transaction : <EOL> if data_type_id : <EOL> data_type = update_types ( fields_patches , domain , data_type_id , data_tag , is_global , transaction ) <EOL> update_items ( fields_patches , domain , data_type_id , transaction ) <EOL> else : <EOL> if FixtureDataType . fixture_tag_exists ( domain , data_tag ) : <EOL> return HttpResponseBadRequest ( "<STR_LIT>" ) <EOL> else : <EOL> data_type = create_types ( fields_patches , domain , data_tag , is_global , transaction ) <EOL> return json_response ( strip_json ( data_type ) ) <EOL> def update_types ( patches , domain , data_type_id , data_tag , is_global , transaction ) : <EOL> data_type = FixtureDataType . get ( data_type_id ) <EOL> fields_patches = deepcopy ( patches ) <EOL> assert ( data_type . doc_type == FixtureDataType . _doc_type ) <EOL> assert ( data_type . domain == domain ) <EOL> old_fields = data_type . fields <EOL> new_fixture_fields = [ ] <EOL> setattr ( data_type , "<STR_LIT>" , data_tag ) <EOL> setattr ( data_type , "<STR_LIT>" , is_global ) <EOL> for old_field in old_fields : <EOL> patch = fields_patches . pop ( old_field . field_name , { } ) <EOL> if not any ( patch ) : <EOL> new_fixture_fields . append ( old_field ) <EOL> if "<STR_LIT>" in patch : <EOL> setattr ( old_field , "<STR_LIT>" , patch [ "<STR_LIT>" ] ) <EOL> new_fixture_fields . append ( old_field ) <EOL> if "<STR_LIT>" in patch : <EOL> continue <EOL> new_fields = fields_patches . keys ( ) <EOL> for new_field_name in new_fields : <EOL> patch = fields_patches . pop ( new_field_name ) <EOL> if "<STR_LIT>" in patch : <EOL> new_fixture_fields . append ( FixtureTypeField ( <EOL> field_name = new_field_name , <EOL> properties = [ ] <EOL> ) ) <EOL> setattr ( data_type , "<STR_LIT>" , new_fixture_fields ) <EOL> transaction . save ( data_type ) <EOL> return data_type <EOL> def update_items ( fields_patches , domain , data_type_id , transaction ) : <EOL> data_items = FixtureDataItem . by_data_type ( domain , data_type_id ) <EOL> for item in data_items : <EOL> fields = item . fields <EOL> updated_fields = { } <EOL> patches = deepcopy ( fields_patches ) <EOL> for old_field in fields . keys ( ) : <EOL> patch = patches . pop ( old_field , { } ) <EOL> if not any ( patch ) : <EOL> updated_fields [ old_field ] = fields . pop ( old_field ) <EOL> if "<STR_LIT>" in patch : <EOL> new_field_name = patch [ "<STR_LIT>" ] <EOL> updated_fields [ new_field_name ] = fields . pop ( old_field ) <EOL> if "<STR_LIT>" in patch : <EOL> continue <EOL> for new_field_name in patches . keys ( ) : <EOL> patch = patches . pop ( new_field_name , { } ) <EOL> if "<STR_LIT>" in patch : <EOL> updated_fields [ new_field_name ] = FieldList ( <EOL> field_list = [ ] <EOL> ) <EOL> setattr ( item , "<STR_LIT>" , updated_fields ) <EOL> transaction . save ( item ) <EOL> data_items = FixtureDataItem . by_data_type ( domain , data_type_id ) <EOL> def create_types ( fields_patches , domain , data_tag , is_global , transaction ) : <EOL> data_type = FixtureDataType ( <EOL> domain = domain , <EOL> tag = data_tag , <EOL> is_global = is_global , <EOL> fields = [ FixtureTypeField ( field_name = field , properties = [ ] ) for field in fields_patches ] , <EOL> item_attributes = [ ] , <EOL> ) <EOL> transaction . save ( data_type ) <EOL> return data_type <EOL> @ require_can_edit_fixtures <EOL> def data_table ( request , domain ) : <EOL> table_ids = request . GET . getlist ( "<STR_LIT>" ) <EOL> try : <EOL> sheets = prepare_fixture_html ( table_ids , domain ) <EOL> except FixtureDownloadError as e : <EOL> messages . info ( request , unicode ( e ) ) <EOL> raise Http404 ( ) <EOL> sheets . pop ( "<STR_LIT>" ) <EOL> if not sheets : <EOL> return { <EOL> "<STR_LIT>" : DataTablesHeader ( DataTablesColumn ( "<STR_LIT>" ) ) , <EOL> "<STR_LIT>" : [ ] <EOL> } <EOL> selected_sheet = sheets . values ( ) [ <NUM_LIT:0> ] <EOL> selected_sheet_tag = sheets . keys ( ) [ <NUM_LIT:0> ] <EOL> data_table = { <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : selected_sheet_tag <EOL> } <EOL> headers = [ DataTablesColumn ( header ) for header in selected_sheet [ "<STR_LIT>" ] ] <EOL> data_table [ "<STR_LIT>" ] = DataTablesHeader ( * headers ) <EOL> if selected_sheet [ "<STR_LIT>" ] and selected_sheet [ "<STR_LIT>" ] : <EOL> data_table [ "<STR_LIT>" ] = [ [ format_datatables_data ( x or "<STR_LIT>" , "<STR_LIT:a>" ) for x in row ] for row in selected_sheet [ "<STR_LIT>" ] ] <EOL> else : <EOL> messages . info ( request , _ ( "<STR_LIT>" ) ) <EOL> data_table [ "<STR_LIT>" ] = [ [ "<STR_LIT>" for x in range ( <NUM_LIT:0> , len ( headers ) ) ] ] <EOL> return data_table <EOL> @ require_can_edit_fixtures <EOL> def download_item_lists ( request , domain ) : <EOL> """<STR_LIT>""" <EOL> download = DownloadBase ( ) <EOL> download . set_task ( fixture_download_async . delay ( <EOL> prepare_fixture_download , <EOL> table_ids = request . POST . getlist ( "<STR_LIT>" , [ ] ) , <EOL> domain = domain , <EOL> download_id = download . download_id , <EOL> ) ) <EOL> return download . get_start_response ( ) <EOL> @ require_can_edit_fixtures <EOL> def download_file ( request , domain ) : <EOL> download_id = request . GET . get ( "<STR_LIT>" ) <EOL> try : <EOL> dw = CachedDownload . get ( download_id ) <EOL> if dw : <EOL> return dw . toHttpResponse ( ) <EOL> else : <EOL> raise IOError <EOL> except IOError : <EOL> notify_exception ( request ) <EOL> messages . error ( request , _ ( "<STR_LIT>" ) ) <EOL> return HttpResponseRedirect ( reverse ( "<STR_LIT>" , args = [ ] , kwargs = { '<STR_LIT>' : domain , '<STR_LIT>' : '<STR_LIT>' } ) ) <EOL> def fixtures_home ( domain ) : <EOL> return reverse ( "<STR_LIT>" , args = [ ] , <EOL> kwargs = { '<STR_LIT>' : domain , '<STR_LIT>' : '<STR_LIT>' } ) <EOL> class FixtureViewMixIn ( object ) : <EOL> section_name = ugettext_noop ( "<STR_LIT>" ) <EOL> @ property <EOL> def section_url ( self ) : <EOL> return fixtures_home ( self . domain ) <EOL> class UploadItemLists ( TemplateView ) : <EOL> def get_context_data ( self , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> return { <EOL> '<STR_LIT>' : self . domain <EOL> } <EOL> def get ( self , request ) : <EOL> return HttpResponseRedirect ( fixtures_home ( self . domain ) ) <EOL> @ method_decorator ( get_file ) <EOL> def post ( self , request ) : <EOL> replace = '<STR_LIT:replace>' in request . POST <EOL> file_ref = expose_cached_download ( <EOL> request . file . read ( ) , <EOL> file_extension = file_extention_from_filename ( request . file . name ) , <EOL> expiry = <NUM_LIT:1> * <NUM_LIT> * <NUM_LIT> , <EOL> ) <EOL> try : <EOL> validate_file_format ( file_ref . get_filename ( ) ) <EOL> except ( FixtureUploadError , JSONReaderError , HeaderValueError ) as e : <EOL> messages . error ( request , _ ( u'<STR_LIT>' ) % e ) <EOL> return HttpResponseRedirect ( fixtures_home ( self . domain ) ) <EOL> task = fixture_upload_async . delay ( <EOL> self . domain , <EOL> file_ref . download_id , <EOL> replace , <EOL> ) <EOL> file_ref . set_task ( task ) <EOL> return HttpResponseRedirect ( <EOL> reverse ( <EOL> FixtureUploadStatusView . urlname , <EOL> args = [ self . domain , file_ref . download_id ] <EOL> ) <EOL> ) <EOL> @ method_decorator ( require_can_edit_fixtures ) <EOL> def dispatch ( self , request , domain , * args , ** kwargs ) : <EOL> self . domain = domain <EOL> return super ( UploadItemLists , self ) . dispatch ( request , * args , ** kwargs ) <EOL> class FixtureUploadStatusView ( FixtureViewMixIn , BaseDomainView ) : <EOL> urlname = '<STR_LIT>' <EOL> page_title = ugettext_noop ( '<STR_LIT>' ) <EOL> def get ( self , request , * args , ** kwargs ) : <EOL> context = super ( FixtureUploadStatusView , self ) . main_context <EOL> context . update ( { <EOL> '<STR_LIT>' : self . domain , <EOL> '<STR_LIT>' : kwargs [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : reverse ( '<STR_LIT>' , args = [ self . domain , kwargs [ '<STR_LIT>' ] ] ) , <EOL> '<STR_LIT:title>' : _ ( self . page_title ) , <EOL> '<STR_LIT>' : _ ( "<STR_LIT>" ) , <EOL> '<STR_LIT>' : _ ( "<STR_LIT>" ) , <EOL> '<STR_LIT>' : reverse ( '<STR_LIT>' , args = [ self . domain ] ) , <EOL> '<STR_LIT>' : _ ( "<STR_LIT>" ) , <EOL> } ) <EOL> return render ( request , '<STR_LIT>' , context ) <EOL> def page_url ( self ) : <EOL> return reverse ( self . urlname , args = self . args , kwargs = self . kwargs ) <EOL> @ require_can_edit_fixtures <EOL> def fixture_upload_job_poll ( request , domain , download_id , template = "<STR_LIT>" ) : <EOL> try : <EOL> context = get_download_context ( download_id , check_state = True ) <EOL> except TaskFailedError : <EOL> return HttpResponseServerError ( ) <EOL> context . update ( { <EOL> '<STR_LIT>' : _ ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : _ ( '<STR_LIT>' ) , <EOL> } ) <EOL> return render ( request , template , context ) <EOL> @ csrf_exempt <EOL> @ require_POST <EOL> @ login_or_digest <EOL> @ require_can_edit_fixtures <EOL> def upload_fixture_api ( request , domain , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> response_codes = { "<STR_LIT>" : <NUM_LIT> , "<STR_LIT>" : <NUM_LIT> , "<STR_LIT:success>" : <NUM_LIT:200> } <EOL> error_messages = { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } <EOL> def _return_response ( code , message ) : <EOL> resp_json = { } <EOL> resp_json [ "<STR_LIT:code>" ] = code <EOL> resp_json [ "<STR_LIT:message>" ] = message <EOL> return HttpResponse ( json . dumps ( resp_json ) , content_type = "<STR_LIT:application/json>" ) <EOL> try : <EOL> upload_file = request . FILES [ "<STR_LIT>" ] <EOL> replace = request . POST [ "<STR_LIT:replace>" ] <EOL> if replace . lower ( ) == "<STR_LIT:true>" : <EOL> replace = True <EOL> elif replace . lower ( ) == "<STR_LIT:false>" : <EOL> replace = False <EOL> except Exception : <EOL> return _return_response ( response_codes [ "<STR_LIT>" ] , error_messages [ "<STR_LIT>" ] ) <EOL> if not request . couch_user . has_permission ( domain , Permissions . edit_data . name ) : <EOL> error_message = error_messages [ "<STR_LIT>" ] . format ( attr = request . couch_user . username ) <EOL> return _return_response ( response_codes [ "<STR_LIT>" ] , error_message ) <EOL> try : <EOL> workbook = get_workbook ( upload_file ) <EOL> except Exception : <EOL> return _return_response ( response_codes [ "<STR_LIT>" ] , error_messages [ "<STR_LIT>" ] ) <EOL> try : <EOL> upload_resp = run_upload ( domain , workbook , replace = replace ) <EOL> except WorksheetNotFound as e : <EOL> error_message = error_messages [ "<STR_LIT>" ] . format ( attr = e . title ) <EOL> return _return_response ( response_codes [ "<STR_LIT>" ] , error_message ) <EOL> except ExcelMalformatException as e : <EOL> return _return_response ( response_codes [ "<STR_LIT>" ] , str ( e ) ) <EOL> except DuplicateFixtureTagException as e : <EOL> return _return_response ( response_codes [ "<STR_LIT>" ] , str ( e ) ) <EOL> except FixtureAPIException as e : <EOL> return _return_response ( response_codes [ "<STR_LIT>" ] , str ( e ) ) <EOL> except Exception as e : <EOL> error_message = error_messages [ "<STR_LIT>" ] . format ( attr = e ) <EOL> return _return_response ( response_codes [ "<STR_LIT>" ] , error_message ) <EOL> num_unknown_groups = len ( upload_resp . unknown_groups ) <EOL> num_unknown_users = len ( upload_resp . unknown_users ) <EOL> resp_json = { } <EOL> if not num_unknown_users and not num_unknown_groups : <EOL> num_uploads = upload_resp . number_of_fixtures <EOL> success_message = "<STR_LIT>" % ( num_uploads , '<STR_LIT:s>' if num_uploads > <NUM_LIT:1> else '<STR_LIT>' ) <EOL> return _return_response ( response_codes [ "<STR_LIT:success>" ] , success_message ) <EOL> else : <EOL> resp_json [ "<STR_LIT:code>" ] = response_codes [ "<STR_LIT>" ] <EOL> warn_groups = "<STR_LIT>" % ( num_unknown_groups , '<STR_LIT>' if num_unknown_groups > <NUM_LIT:1> else '<STR_LIT>' ) <EOL> warn_users = "<STR_LIT>" % ( num_unknown_users , '<STR_LIT>' if num_unknown_users > <NUM_LIT:1> else '<STR_LIT>' ) <EOL> resp_json [ "<STR_LIT:message>" ] = "<STR_LIT>" <EOL> if num_unknown_groups : <EOL> resp_json [ "<STR_LIT:message>" ] += "<STR_LIT>" % ( warn_groups , upload_resp . unknown_groups ) <EOL> if num_unknown_users : <EOL> resp_json [ "<STR_LIT:message>" ] += "<STR_LIT>" % ( ( "<STR_LIT>" if num_unknown_groups else "<STR_LIT>" ) , warn_users , upload_resp . unknown_users ) <EOL> return HttpResponse ( json . dumps ( resp_json ) , content_type = "<STR_LIT:application/json>" ) <EOL> @ login_and_domain_required <EOL> def fixture_metadata ( request , domain ) : <EOL> """<STR_LIT>""" <EOL> return json_response ( item_lists_by_domain ( domain ) ) </s>
