<s> import csv <EOL> import gzip <EOL> import json <EOL> import logging <EOL> import re <EOL> from django . core . urlresolvers import reverse <EOL> from django . http import QueryDict <EOL> from django . utils . translation import ugettext as _ <EOL> from desktop . context_processors import get_app_name <EOL> from desktop . lib import django_mako , i18n <EOL> from desktop . lib . django_util import render <EOL> from desktop . lib . exceptions_renderable import PopupException <EOL> from desktop . lib . django_forms import MultiForm <EOL> from hadoop . fs import hadoopfs <EOL> from hadoop . fs . fsutils import remove_header <EOL> from beeswax . common import TERMINATORS <EOL> from beeswax . design import hql_query <EOL> from beeswax . forms import CreateTableForm , ColumnTypeFormSet , PartitionTypeFormSet , CreateByImportFileForm , CreateByImportDelimForm , TERMINATOR_CHOICES <EOL> from beeswax . server import dbms <EOL> from beeswax . server . dbms import QueryServerException <EOL> from beeswax . views import execute_directly <EOL> LOG = logging . getLogger ( __name__ ) <EOL> def create_table ( request , database = '<STR_LIT:default>' ) : <EOL> """<STR_LIT>""" <EOL> db = dbms . get ( request . user ) <EOL> dbs = db . get_databases ( ) <EOL> databases = [ { '<STR_LIT:name>' : db , '<STR_LIT:url>' : reverse ( '<STR_LIT>' , kwargs = { '<STR_LIT>' : db } ) } for db in dbs ] <EOL> form = MultiForm ( <EOL> table = CreateTableForm , <EOL> columns = ColumnTypeFormSet , <EOL> partitions = PartitionTypeFormSet <EOL> ) <EOL> if request . method == "<STR_LIT:POST>" : <EOL> form . bind ( request . POST ) <EOL> form . table . db = db <EOL> form . table . database = database <EOL> if request . POST . get ( '<STR_LIT>' ) : <EOL> if form . is_valid ( ) : <EOL> columns = [ f . cleaned_data for f in form . columns . forms ] <EOL> partition_columns = [ f . cleaned_data for f in form . partitions . forms ] <EOL> proposed_query = django_mako . render_to_string ( "<STR_LIT>" , { <EOL> '<STR_LIT>' : databases , <EOL> '<STR_LIT>' : database , <EOL> '<STR_LIT>' : form . table . cleaned_data , <EOL> '<STR_LIT>' : columns , <EOL> '<STR_LIT>' : partition_columns <EOL> } <EOL> ) <EOL> proposed_query = proposed_query . decode ( '<STR_LIT:utf-8>' ) <EOL> table_name = form . table . cleaned_data [ '<STR_LIT:name>' ] <EOL> return _submit_create_and_load ( request , proposed_query , table_name , None , False , database = database ) <EOL> else : <EOL> form . bind ( ) <EOL> return render ( "<STR_LIT>" , request , { <EOL> '<STR_LIT:action>' : "<STR_LIT:#>" , <EOL> '<STR_LIT>' : databases , <EOL> '<STR_LIT>' : form . table , <EOL> '<STR_LIT>' : form . columns , <EOL> '<STR_LIT>' : form . partitions , <EOL> '<STR_LIT>' : len ( dbms . get ( request . user ) . get_tables ( ) ) > <NUM_LIT:0> , <EOL> '<STR_LIT>' : database , <EOL> } ) <EOL> IMPORT_PEEK_SIZE = <NUM_LIT:5> * <NUM_LIT> ** <NUM_LIT:2> <EOL> IMPORT_PEEK_NLINES = <NUM_LIT:10> <EOL> DELIMITERS = [ hive_val for hive_val , desc , ascii in TERMINATORS ] <EOL> DELIMITER_READABLE = { '<STR_LIT>' : _ ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : _ ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : _ ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : _ ( '<STR_LIT>' ) , <EOL> '<STR_LIT:U+002C>' : _ ( '<STR_LIT>' ) , <EOL> '<STR_LIT:U+0020>' : _ ( '<STR_LIT>' ) } <EOL> FILE_READERS = [ ] <EOL> def import_wizard ( request , database = '<STR_LIT:default>' ) : <EOL> """<STR_LIT>""" <EOL> encoding = i18n . get_site_encoding ( ) <EOL> app_name = get_app_name ( request ) <EOL> db = dbms . get ( request . user ) <EOL> dbs = db . get_databases ( ) <EOL> databases = [ { '<STR_LIT:name>' : db , '<STR_LIT:url>' : reverse ( '<STR_LIT>' , kwargs = { '<STR_LIT>' : db } ) } for db in dbs ] <EOL> if request . method == '<STR_LIT:POST>' : <EOL> delim_is_auto = False <EOL> fields_list , n_cols = [ [ ] ] , <NUM_LIT:0> <EOL> s3_col_formset = None <EOL> s1_file_form = CreateByImportFileForm ( request . POST , db = db ) <EOL> if s1_file_form . is_valid ( ) : <EOL> do_s2_auto_delim = request . POST . get ( '<STR_LIT>' ) <EOL> do_s2_user_delim = request . POST . get ( '<STR_LIT>' ) <EOL> do_s3_column_def = request . POST . get ( '<STR_LIT>' ) <EOL> do_hive_create = request . POST . get ( '<STR_LIT>' ) <EOL> cancel_s2_user_delim = request . POST . get ( '<STR_LIT>' ) <EOL> cancel_s3_column_def = request . POST . get ( '<STR_LIT>' ) <EOL> if len ( filter ( None , ( do_s2_auto_delim , do_s2_user_delim , do_s3_column_def , do_hive_create , cancel_s2_user_delim , cancel_s3_column_def ) ) ) != <NUM_LIT:1> : <EOL> raise PopupException ( _ ( '<STR_LIT>' ) ) <EOL> if not do_s2_auto_delim : <EOL> s2_delim_form = CreateByImportDelimForm ( request . POST ) <EOL> if not s2_delim_form . is_valid ( ) : <EOL> do_s2_user_delim , do_s3_column_def , do_hive_create = True , False , False <EOL> if do_hive_create : <EOL> s3_col_formset = ColumnTypeFormSet ( prefix = '<STR_LIT>' , data = request . POST ) <EOL> if not s3_col_formset . is_valid ( ) : <EOL> do_s3_column_def , do_hive_create = True , False <EOL> if do_s2_auto_delim : <EOL> delim_is_auto = True <EOL> fields_list , n_cols , s2_delim_form = _delim_preview ( request . fs , s1_file_form , encoding , [ reader . TYPE for reader in FILE_READERS ] , DELIMITERS ) <EOL> if ( do_s2_user_delim or do_s3_column_def or cancel_s3_column_def ) and s2_delim_form . is_valid ( ) : <EOL> fields_list , n_cols , s2_delim_form = _delim_preview ( request . fs , s1_file_form , encoding , ( s2_delim_form . cleaned_data [ '<STR_LIT>' ] , ) , <EOL> ( s2_delim_form . cleaned_data [ '<STR_LIT>' ] , ) ) <EOL> if do_s2_auto_delim or do_s2_user_delim or cancel_s3_column_def : <EOL> return render ( '<STR_LIT>' , request , { <EOL> '<STR_LIT:action>' : reverse ( app_name + '<STR_LIT>' , kwargs = { '<STR_LIT>' : database } ) , <EOL> '<STR_LIT>' : DELIMITER_READABLE . get ( s2_delim_form [ '<STR_LIT>' ] . data [ <NUM_LIT:0> ] , s2_delim_form [ '<STR_LIT>' ] . data [ <NUM_LIT:1> ] ) , <EOL> '<STR_LIT>' : delim_is_auto , <EOL> '<STR_LIT>' : s1_file_form , <EOL> '<STR_LIT>' : s2_delim_form , <EOL> '<STR_LIT>' : fields_list , <EOL> '<STR_LIT>' : TERMINATOR_CHOICES , <EOL> '<STR_LIT>' : n_cols , <EOL> '<STR_LIT>' : database , <EOL> '<STR_LIT>' : databases <EOL> } ) <EOL> if do_s3_column_def : <EOL> if s3_col_formset is None : <EOL> columns = [ ] <EOL> for i in range ( n_cols ) : <EOL> columns . append ( { <EOL> '<STR_LIT>' : '<STR_LIT>' % ( i , ) , <EOL> '<STR_LIT>' : '<STR_LIT:string>' , <EOL> } ) <EOL> s3_col_formset = ColumnTypeFormSet ( prefix = '<STR_LIT>' , initial = columns ) <EOL> try : <EOL> fields_list_for_json = list ( fields_list ) <EOL> if fields_list_for_json : <EOL> fields_list_for_json [ <NUM_LIT:0> ] = map ( lambda a : re . sub ( '<STR_LIT>' , '<STR_LIT>' , a ) , fields_list_for_json [ <NUM_LIT:0> ] ) <EOL> return render ( '<STR_LIT>' , request , { <EOL> '<STR_LIT:action>' : reverse ( app_name + '<STR_LIT>' , kwargs = { '<STR_LIT>' : database } ) , <EOL> '<STR_LIT>' : s1_file_form , <EOL> '<STR_LIT>' : s2_delim_form , <EOL> '<STR_LIT>' : s3_col_formset , <EOL> '<STR_LIT>' : fields_list , <EOL> '<STR_LIT>' : json . dumps ( fields_list_for_json ) , <EOL> '<STR_LIT>' : n_cols , <EOL> '<STR_LIT>' : database , <EOL> '<STR_LIT>' : databases <EOL> } ) <EOL> except Exception , e : <EOL> raise PopupException ( _ ( "<STR_LIT>" ) , detail = e ) <EOL> if do_hive_create : <EOL> delim = s2_delim_form . cleaned_data [ '<STR_LIT>' ] <EOL> table_name = s1_file_form . cleaned_data [ '<STR_LIT:name>' ] <EOL> proposed_query = django_mako . render_to_string ( "<STR_LIT>" , { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:name>' : table_name , <EOL> '<STR_LIT>' : s1_file_form . cleaned_data [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : delim , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } , <EOL> '<STR_LIT>' : [ f . cleaned_data for f in s3_col_formset . forms ] , <EOL> '<STR_LIT>' : [ ] , <EOL> '<STR_LIT>' : database , <EOL> '<STR_LIT>' : databases <EOL> } <EOL> ) <EOL> do_load_data = s1_file_form . cleaned_data . get ( '<STR_LIT>' ) <EOL> path = s1_file_form . cleaned_data [ '<STR_LIT:path>' ] <EOL> try : <EOL> return _submit_create_and_load ( request , proposed_query , table_name , path , do_load_data , database = database ) <EOL> except QueryServerException , e : <EOL> raise PopupException ( _ ( '<STR_LIT>' ) , detail = e . message ) <EOL> else : <EOL> s1_file_form = CreateByImportFileForm ( ) <EOL> return render ( '<STR_LIT>' , request , { <EOL> '<STR_LIT:action>' : reverse ( app_name + '<STR_LIT>' , kwargs = { '<STR_LIT>' : database } ) , <EOL> '<STR_LIT>' : s1_file_form , <EOL> '<STR_LIT>' : database , <EOL> '<STR_LIT>' : databases <EOL> } ) <EOL> def _submit_create_and_load ( request , create_hql , table_name , path , do_load , database ) : <EOL> """<STR_LIT>""" <EOL> on_success_params = QueryDict ( '<STR_LIT>' , mutable = True ) <EOL> app_name = get_app_name ( request ) <EOL> if do_load : <EOL> on_success_params [ '<STR_LIT>' ] = table_name <EOL> on_success_params [ '<STR_LIT:path>' ] = path <EOL> on_success_params [ '<STR_LIT>' ] = request . POST . get ( '<STR_LIT>' ) <EOL> on_success_url = reverse ( app_name + '<STR_LIT>' , kwargs = { '<STR_LIT>' : database } ) + '<STR_LIT:?>' + on_success_params . urlencode ( ) <EOL> else : <EOL> on_success_url = reverse ( '<STR_LIT>' , kwargs = { '<STR_LIT>' : database , '<STR_LIT>' : table_name } ) <EOL> query = hql_query ( create_hql , database = database ) <EOL> return execute_directly ( request , query , <EOL> on_success_url = on_success_url , <EOL> on_success_params = on_success_params ) <EOL> def _delim_preview ( fs , file_form , encoding , file_types , delimiters ) : <EOL> """<STR_LIT>""" <EOL> assert file_form . is_valid ( ) <EOL> path = file_form . cleaned_data [ '<STR_LIT:path>' ] <EOL> try : <EOL> file_obj = fs . open ( path ) <EOL> delim , file_type , fields_list = _parse_fields ( path , file_obj , encoding , file_types , delimiters ) <EOL> file_obj . close ( ) <EOL> except IOError , ex : <EOL> msg = "<STR_LIT>" % ( path , ex ) <EOL> LOG . exception ( msg ) <EOL> raise PopupException ( msg ) <EOL> n_cols = max ( [ len ( row ) for row in fields_list ] ) <EOL> delimiter_0 = delim <EOL> delimiter_1 = '<STR_LIT>' <EOL> if not filter ( lambda val : val [ <NUM_LIT:0> ] == delim , TERMINATOR_CHOICES ) : <EOL> delimiter_0 = '<STR_LIT>' <EOL> delimiter_1 = delim <EOL> delim_form = CreateByImportDelimForm ( dict ( delimiter_0 = delimiter_0 , <EOL> delimiter_1 = delimiter_1 , <EOL> file_type = file_type , <EOL> n_cols = n_cols ) ) <EOL> if not delim_form . is_valid ( ) : <EOL> assert False , _ ( '<STR_LIT>' ) % { '<STR_LIT:error>' : delim_form . errors } <EOL> return fields_list , n_cols , delim_form <EOL> def _parse_fields ( path , file_obj , encoding , filetypes , delimiters ) : <EOL> """<STR_LIT>""" <EOL> file_readers = [ reader for reader in FILE_READERS if reader . TYPE in filetypes ] <EOL> for reader in file_readers : <EOL> LOG . debug ( "<STR_LIT>" % ( reader . TYPE , path ) ) <EOL> file_obj . seek ( <NUM_LIT:0> , hadoopfs . SEEK_SET ) <EOL> lines = reader . readlines ( file_obj , encoding ) <EOL> if lines is not None : <EOL> delim , fields_list = _readfields ( lines , delimiters ) <EOL> return delim , reader . TYPE , fields_list <EOL> else : <EOL> msg = _ ( "<STR_LIT>" ) % { '<STR_LIT:path>' : path , '<STR_LIT>' : encoding } <EOL> LOG . error ( msg ) <EOL> raise PopupException ( msg ) <EOL> def _readfields ( lines , delimiters ) : <EOL> """<STR_LIT>""" <EOL> def score_delim ( fields_list ) : <EOL> """<STR_LIT>""" <EOL> n_lines = len ( fields_list ) <EOL> len_list = [ len ( fields ) for fields in fields_list ] <EOL> if not len_list : <EOL> raise PopupException ( _ ( "<STR_LIT>" ) ) <EOL> if min ( len_list ) == <NUM_LIT:1> : <EOL> return <NUM_LIT:0> <EOL> avg_n_fields = sum ( len_list ) / n_lines <EOL> sq_of_exp = avg_n_fields * avg_n_fields <EOL> len_list_sq = [ l * l for l in len_list ] <EOL> exp_of_sq = sum ( len_list_sq ) / n_lines <EOL> var = exp_of_sq - sq_of_exp <EOL> return ( <NUM_LIT> / ( var + <NUM_LIT:1> ) ) + avg_n_fields <EOL> max_score = - <NUM_LIT:1> <EOL> res = ( None , None ) <EOL> for delim in delimiters : <EOL> delimiter = delim . decode ( '<STR_LIT>' ) <EOL> try : <EOL> fields_list = _get_rows ( lines , delimiter ) <EOL> except : <EOL> LOG . exception ( '<STR_LIT>' ) <EOL> fields_list = [ line . split ( delimiter ) for line in lines if line ] <EOL> score = score_delim ( fields_list ) <EOL> LOG . debug ( "<STR_LIT>" % ( delim , score ) ) <EOL> if score > max_score : <EOL> max_score = score <EOL> res = ( delim , fields_list ) <EOL> return res <EOL> def _get_rows ( lines , delimiter ) : <EOL> column_reader = csv . reader ( lines , delimiter = delimiter ) <EOL> return [ row for row in column_reader if row ] <EOL> def _peek_file ( fs , file_form ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> path = file_form . cleaned_data [ '<STR_LIT:path>' ] <EOL> file_obj = fs . open ( path ) <EOL> file_head = file_obj . read ( IMPORT_PEEK_SIZE ) <EOL> file_obj . close ( ) <EOL> return ( path , file_head ) <EOL> except IOError , ex : <EOL> msg = _ ( "<STR_LIT>" ) % { '<STR_LIT:path>' : path , '<STR_LIT:error>' : ex } <EOL> LOG . exception ( msg ) <EOL> raise PopupException ( msg ) <EOL> class GzipFileReader ( object ) : <EOL> """<STR_LIT>""" <EOL> TYPE = '<STR_LIT>' <EOL> @ staticmethod <EOL> def readlines ( fileobj , encoding ) : <EOL> """<STR_LIT>""" <EOL> gz = gzip . GzipFile ( fileobj = fileobj , mode = '<STR_LIT:rb>' ) <EOL> try : <EOL> data = gz . read ( IMPORT_PEEK_SIZE ) <EOL> except IOError : <EOL> return None <EOL> try : <EOL> return unicode ( data , encoding , errors = '<STR_LIT:replace>' ) . splitlines ( ) [ : IMPORT_PEEK_NLINES ] <EOL> except UnicodeError : <EOL> return None <EOL> FILE_READERS . append ( GzipFileReader ) <EOL> class TextFileReader ( object ) : <EOL> """<STR_LIT>""" <EOL> TYPE = '<STR_LIT:text>' <EOL> @ staticmethod <EOL> def readlines ( fileobj , encoding ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> data = fileobj . read ( IMPORT_PEEK_SIZE ) <EOL> return unicode ( data , encoding , errors = '<STR_LIT:replace>' ) . splitlines ( ) [ : IMPORT_PEEK_NLINES ] <EOL> except UnicodeError : <EOL> return None <EOL> FILE_READERS . append ( TextFileReader ) <EOL> def load_after_create ( request , database ) : <EOL> """<STR_LIT>""" <EOL> tablename = request . REQUEST . get ( '<STR_LIT>' ) <EOL> path = request . REQUEST . get ( '<STR_LIT:path>' ) <EOL> is_remove_header = request . REQUEST . get ( '<STR_LIT>' ) . lower ( ) == '<STR_LIT>' and not path . endswith ( '<STR_LIT>' ) <EOL> if not tablename or not path : <EOL> msg = _ ( '<STR_LIT>' ) <EOL> LOG . error ( msg ) <EOL> raise PopupException ( msg ) <EOL> if is_remove_header : <EOL> try : <EOL> remove_header ( request . fs , path ) <EOL> except Exception , e : <EOL> raise PopupException ( _ ( "<STR_LIT>" ) , detail = e ) <EOL> LOG . debug ( "<STR_LIT>" % ( path , tablename ) ) <EOL> hql = "<STR_LIT>" % ( path , database , tablename ) <EOL> query = hql_query ( hql ) <EOL> on_success_url = reverse ( '<STR_LIT>' , kwargs = { '<STR_LIT>' : database , '<STR_LIT>' : tablename } ) <EOL> try : <EOL> return execute_directly ( request , query , on_success_url = on_success_url ) <EOL> except Exception , e : <EOL> message = '<STR_LIT>' <EOL> LOG . exception ( message ) <EOL> detail = e . message if hasattr ( e , '<STR_LIT:message>' ) and e . message else None <EOL> raise PopupException ( _ ( message ) , detail = detail ) </s>
