<s> """<STR_LIT>""" <EOL> __docformat__ = '<STR_LIT>' <EOL> import sys <EOL> import os . path <EOL> import csv <EOL> from docutils import io , nodes , statemachine , utils <EOL> from docutils . error_reporting import SafeString <EOL> from docutils . utils import SystemMessagePropagation <EOL> from docutils . parsers . rst import Directive <EOL> from docutils . parsers . rst import directives <EOL> class Table ( Directive ) : <EOL> """<STR_LIT>""" <EOL> optional_arguments = <NUM_LIT:1> <EOL> final_argument_whitespace = True <EOL> option_spec = { '<STR_LIT:class>' : directives . class_option , <EOL> '<STR_LIT:name>' : directives . unchanged } <EOL> has_content = True <EOL> def make_title ( self ) : <EOL> if self . arguments : <EOL> title_text = self . arguments [ <NUM_LIT:0> ] <EOL> text_nodes , messages = self . state . inline_text ( title_text , <EOL> self . lineno ) <EOL> title = nodes . title ( title_text , '<STR_LIT>' , * text_nodes ) <EOL> else : <EOL> title = None <EOL> messages = [ ] <EOL> return title , messages <EOL> def process_header_option ( self ) : <EOL> source = self . state_machine . get_source ( self . lineno - <NUM_LIT:1> ) <EOL> table_head = [ ] <EOL> max_header_cols = <NUM_LIT:0> <EOL> if '<STR_LIT>' in self . options : <EOL> rows , max_header_cols = self . parse_csv_data_into_rows ( <EOL> self . options [ '<STR_LIT>' ] . split ( '<STR_LIT:\n>' ) , self . HeaderDialect ( ) , <EOL> source ) <EOL> table_head . extend ( rows ) <EOL> return table_head , max_header_cols <EOL> def check_table_dimensions ( self , rows , header_rows , stub_columns ) : <EOL> if len ( rows ) < header_rows : <EOL> error = self . state_machine . reporter . error ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> % ( header_rows , len ( rows ) , self . name ) , nodes . literal_block ( <EOL> self . block_text , self . block_text ) , line = self . lineno ) <EOL> raise SystemMessagePropagation ( error ) <EOL> if len ( rows ) == header_rows > <NUM_LIT:0> : <EOL> error = self . state_machine . reporter . error ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> % ( len ( rows ) , self . name ) , nodes . literal_block ( <EOL> self . block_text , self . block_text ) , line = self . lineno ) <EOL> raise SystemMessagePropagation ( error ) <EOL> for row in rows : <EOL> if len ( row ) < stub_columns : <EOL> error = self . state_machine . reporter . error ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' % <EOL> ( stub_columns , len ( row ) , self . name ) , nodes . literal_block ( <EOL> self . block_text , self . block_text ) , line = self . lineno ) <EOL> raise SystemMessagePropagation ( error ) <EOL> if len ( row ) == stub_columns > <NUM_LIT:0> : <EOL> error = self . state_machine . reporter . error ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> % ( len ( row ) , self . name ) , nodes . literal_block ( <EOL> self . block_text , self . block_text ) , line = self . lineno ) <EOL> raise SystemMessagePropagation ( error ) <EOL> def get_column_widths ( self , max_cols ) : <EOL> if '<STR_LIT>' in self . options : <EOL> col_widths = self . options [ '<STR_LIT>' ] <EOL> if len ( col_widths ) != max_cols : <EOL> error = self . state_machine . reporter . error ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' % ( self . name , max_cols ) , nodes . literal_block ( <EOL> self . block_text , self . block_text ) , line = self . lineno ) <EOL> raise SystemMessagePropagation ( error ) <EOL> elif max_cols : <EOL> col_widths = [ <NUM_LIT:100> // max_cols ] * max_cols <EOL> else : <EOL> error = self . state_machine . reporter . error ( <EOL> '<STR_LIT>' , nodes . literal_block ( <EOL> self . block_text , self . block_text ) , line = self . lineno ) <EOL> raise SystemMessagePropagation ( error ) <EOL> return col_widths <EOL> def extend_short_rows_with_empty_cells ( self , columns , parts ) : <EOL> for part in parts : <EOL> for row in part : <EOL> if len ( row ) < columns : <EOL> row . extend ( [ ( <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> , [ ] ) ] * ( columns - len ( row ) ) ) <EOL> class RSTTable ( Table ) : <EOL> def run ( self ) : <EOL> if not self . content : <EOL> warning = self . state_machine . reporter . warning ( <EOL> '<STR_LIT>' <EOL> % self . name , nodes . literal_block ( <EOL> self . block_text , self . block_text ) , line = self . lineno ) <EOL> return [ warning ] <EOL> title , messages = self . make_title ( ) <EOL> node = nodes . Element ( ) <EOL> self . state . nested_parse ( self . content , self . content_offset , node ) <EOL> if len ( node ) != <NUM_LIT:1> or not isinstance ( node [ <NUM_LIT:0> ] , nodes . table ) : <EOL> error = self . state_machine . reporter . error ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' % self . name , nodes . literal_block ( <EOL> self . block_text , self . block_text ) , line = self . lineno ) <EOL> return [ error ] <EOL> table_node = node [ <NUM_LIT:0> ] <EOL> table_node [ '<STR_LIT>' ] += self . options . get ( '<STR_LIT:class>' , [ ] ) <EOL> self . add_name ( table_node ) <EOL> if title : <EOL> table_node . insert ( <NUM_LIT:0> , title ) <EOL> return [ table_node ] + messages <EOL> class CSVTable ( Table ) : <EOL> option_spec = { '<STR_LIT>' : directives . nonnegative_int , <EOL> '<STR_LIT>' : directives . nonnegative_int , <EOL> '<STR_LIT>' : directives . unchanged , <EOL> '<STR_LIT>' : directives . positive_int_list , <EOL> '<STR_LIT:file>' : directives . path , <EOL> '<STR_LIT:url>' : directives . uri , <EOL> '<STR_LIT>' : directives . encoding , <EOL> '<STR_LIT:class>' : directives . class_option , <EOL> '<STR_LIT:name>' : directives . unchanged , <EOL> '<STR_LIT>' : directives . single_char_or_whitespace_or_unicode , <EOL> '<STR_LIT>' : directives . flag , <EOL> '<STR_LIT>' : directives . single_char_or_unicode , <EOL> '<STR_LIT>' : directives . single_char_or_unicode , } <EOL> class DocutilsDialect ( csv . Dialect ) : <EOL> """<STR_LIT>""" <EOL> delimiter = '<STR_LIT:U+002C>' <EOL> quotechar = '<STR_LIT:">' <EOL> doublequote = True <EOL> skipinitialspace = True <EOL> lineterminator = '<STR_LIT:\n>' <EOL> quoting = csv . QUOTE_MINIMAL <EOL> def __init__ ( self , options ) : <EOL> if '<STR_LIT>' in options : <EOL> self . delimiter = str ( options [ '<STR_LIT>' ] ) <EOL> if '<STR_LIT>' in options : <EOL> self . skipinitialspace = False <EOL> if '<STR_LIT>' in options : <EOL> self . quotechar = str ( options [ '<STR_LIT>' ] ) <EOL> if '<STR_LIT>' in options : <EOL> self . doublequote = False <EOL> self . escapechar = str ( options [ '<STR_LIT>' ] ) <EOL> csv . Dialect . __init__ ( self ) <EOL> class HeaderDialect ( csv . Dialect ) : <EOL> """<STR_LIT>""" <EOL> delimiter = '<STR_LIT:U+002C>' <EOL> quotechar = '<STR_LIT:">' <EOL> escapechar = '<STR_LIT:\\>' <EOL> doublequote = False <EOL> skipinitialspace = True <EOL> lineterminator = '<STR_LIT:\n>' <EOL> quoting = csv . QUOTE_MINIMAL <EOL> def check_requirements ( self ) : <EOL> pass <EOL> def run ( self ) : <EOL> try : <EOL> if ( not self . state . document . settings . file_insertion_enabled <EOL> and ( '<STR_LIT:file>' in self . options <EOL> or '<STR_LIT:url>' in self . options ) ) : <EOL> warning = self . state_machine . reporter . warning ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' % self . name , nodes . literal_block ( <EOL> self . block_text , self . block_text ) , line = self . lineno ) <EOL> return [ warning ] <EOL> self . check_requirements ( ) <EOL> title , messages = self . make_title ( ) <EOL> csv_data , source = self . get_csv_data ( ) <EOL> table_head , max_header_cols = self . process_header_option ( ) <EOL> rows , max_cols = self . parse_csv_data_into_rows ( <EOL> csv_data , self . DocutilsDialect ( self . options ) , source ) <EOL> max_cols = max ( max_cols , max_header_cols ) <EOL> header_rows = self . options . get ( '<STR_LIT>' , <NUM_LIT:0> ) <EOL> stub_columns = self . options . get ( '<STR_LIT>' , <NUM_LIT:0> ) <EOL> self . check_table_dimensions ( rows , header_rows , stub_columns ) <EOL> table_head . extend ( rows [ : header_rows ] ) <EOL> table_body = rows [ header_rows : ] <EOL> col_widths = self . get_column_widths ( max_cols ) <EOL> self . extend_short_rows_with_empty_cells ( max_cols , <EOL> ( table_head , table_body ) ) <EOL> except SystemMessagePropagation , detail : <EOL> return [ detail . args [ <NUM_LIT:0> ] ] <EOL> except csv . Error , detail : <EOL> error = self . state_machine . reporter . error ( <EOL> '<STR_LIT>' <EOL> % ( self . name , detail ) , nodes . literal_block ( <EOL> self . block_text , self . block_text ) , line = self . lineno ) <EOL> return [ error ] <EOL> table = ( col_widths , table_head , table_body ) <EOL> table_node = self . state . build_table ( table , self . content_offset , <EOL> stub_columns ) <EOL> table_node [ '<STR_LIT>' ] += self . options . get ( '<STR_LIT:class>' , [ ] ) <EOL> self . add_name ( table_node ) <EOL> if title : <EOL> table_node . insert ( <NUM_LIT:0> , title ) <EOL> return [ table_node ] + messages <EOL> def get_csv_data ( self ) : <EOL> """<STR_LIT>""" <EOL> encoding = self . options . get ( <EOL> '<STR_LIT>' , self . state . document . settings . input_encoding ) <EOL> if self . content : <EOL> if '<STR_LIT:file>' in self . options or '<STR_LIT:url>' in self . options : <EOL> error = self . state_machine . reporter . error ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' % self . name , nodes . literal_block ( <EOL> self . block_text , self . block_text ) , line = self . lineno ) <EOL> raise SystemMessagePropagation ( error ) <EOL> source = self . content . source ( <NUM_LIT:0> ) <EOL> csv_data = self . content <EOL> elif '<STR_LIT:file>' in self . options : <EOL> if '<STR_LIT:url>' in self . options : <EOL> error = self . state_machine . reporter . error ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' % self . name , <EOL> nodes . literal_block ( self . block_text , self . block_text ) , <EOL> line = self . lineno ) <EOL> raise SystemMessagePropagation ( error ) <EOL> source_dir = os . path . dirname ( <EOL> os . path . abspath ( self . state . document . current_source ) ) <EOL> source = os . path . normpath ( os . path . join ( source_dir , <EOL> self . options [ '<STR_LIT:file>' ] ) ) <EOL> source = utils . relative_path ( None , source ) <EOL> try : <EOL> self . state . document . settings . record_dependencies . add ( source ) <EOL> csv_file = io . FileInput ( <EOL> source_path = source , encoding = encoding , <EOL> error_handler = ( self . state . document . settings . input_encoding_error_handler ) , <EOL> handle_io_errors = None ) <EOL> csv_data = csv_file . read ( ) . splitlines ( ) <EOL> except IOError , error : <EOL> severe = self . state_machine . reporter . severe ( <EOL> u'<STR_LIT>' <EOL> % ( self . name , SafeString ( error ) ) , <EOL> nodes . literal_block ( self . block_text , self . block_text ) , <EOL> line = self . lineno ) <EOL> raise SystemMessagePropagation ( severe ) <EOL> elif '<STR_LIT:url>' in self . options : <EOL> import urllib2 <EOL> source = self . options [ '<STR_LIT:url>' ] <EOL> try : <EOL> csv_text = urllib2 . urlopen ( source ) . read ( ) <EOL> except ( urllib2 . URLError , IOError , OSError , ValueError ) , error : <EOL> severe = self . state_machine . reporter . severe ( <EOL> '<STR_LIT>' <EOL> % ( self . name , self . options [ '<STR_LIT:url>' ] , SafeString ( error ) ) , <EOL> nodes . literal_block ( self . block_text , self . block_text ) , <EOL> line = self . lineno ) <EOL> raise SystemMessagePropagation ( severe ) <EOL> csv_file = io . StringInput ( <EOL> source = csv_text , source_path = source , encoding = encoding , <EOL> error_handler = ( self . state . document . settings . input_encoding_error_handler ) ) <EOL> csv_data = csv_file . read ( ) . splitlines ( ) <EOL> else : <EOL> error = self . state_machine . reporter . warning ( <EOL> '<STR_LIT>' <EOL> % self . name , nodes . literal_block ( <EOL> self . block_text , self . block_text ) , line = self . lineno ) <EOL> raise SystemMessagePropagation ( error ) <EOL> return csv_data , source <EOL> if sys . version_info < ( <NUM_LIT:3> , ) : <EOL> def decode_from_csv ( s ) : <EOL> return s . decode ( '<STR_LIT:utf-8>' ) <EOL> def encode_for_csv ( s ) : <EOL> return s . encode ( '<STR_LIT:utf-8>' ) <EOL> else : <EOL> def decode_from_csv ( s ) : <EOL> return s <EOL> def encode_for_csv ( s ) : <EOL> return s <EOL> decode_from_csv = staticmethod ( decode_from_csv ) <EOL> encode_for_csv = staticmethod ( encode_for_csv ) <EOL> def parse_csv_data_into_rows ( self , csv_data , dialect , source ) : <EOL> csv_reader = csv . reader ( [ self . encode_for_csv ( line + '<STR_LIT:\n>' ) <EOL> for line in csv_data ] , <EOL> dialect = dialect ) <EOL> rows = [ ] <EOL> max_cols = <NUM_LIT:0> <EOL> for row in csv_reader : <EOL> row_data = [ ] <EOL> for cell in row : <EOL> cell_text = self . decode_from_csv ( cell ) <EOL> cell_data = ( <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> , statemachine . StringList ( <EOL> cell_text . splitlines ( ) , source = source ) ) <EOL> row_data . append ( cell_data ) <EOL> rows . append ( row_data ) <EOL> max_cols = max ( max_cols , len ( row ) ) <EOL> return rows , max_cols <EOL> class ListTable ( Table ) : <EOL> """<STR_LIT>""" <EOL> option_spec = { '<STR_LIT>' : directives . nonnegative_int , <EOL> '<STR_LIT>' : directives . nonnegative_int , <EOL> '<STR_LIT>' : directives . positive_int_list , <EOL> '<STR_LIT:class>' : directives . class_option , <EOL> '<STR_LIT:name>' : directives . unchanged } <EOL> def run ( self ) : <EOL> if not self . content : <EOL> error = self . state_machine . reporter . error ( <EOL> '<STR_LIT>' % self . name , <EOL> nodes . literal_block ( self . block_text , self . block_text ) , <EOL> line = self . lineno ) <EOL> return [ error ] <EOL> title , messages = self . make_title ( ) <EOL> node = nodes . Element ( ) <EOL> self . state . nested_parse ( self . content , self . content_offset , node ) <EOL> try : <EOL> num_cols , col_widths = self . check_list_content ( node ) <EOL> table_data = [ [ item . children for item in row_list [ <NUM_LIT:0> ] ] <EOL> for row_list in node [ <NUM_LIT:0> ] ] <EOL> header_rows = self . options . get ( '<STR_LIT>' , <NUM_LIT:0> ) <EOL> stub_columns = self . options . get ( '<STR_LIT>' , <NUM_LIT:0> ) <EOL> self . check_table_dimensions ( table_data , header_rows , stub_columns ) <EOL> except SystemMessagePropagation , detail : <EOL> return [ detail . args [ <NUM_LIT:0> ] ] <EOL> table_node = self . build_table_from_list ( table_data , col_widths , <EOL> header_rows , stub_columns ) <EOL> table_node [ '<STR_LIT>' ] += self . options . get ( '<STR_LIT:class>' , [ ] ) <EOL> self . add_name ( table_node ) <EOL> if title : <EOL> table_node . insert ( <NUM_LIT:0> , title ) <EOL> return [ table_node ] + messages <EOL> def check_list_content ( self , node ) : <EOL> if len ( node ) != <NUM_LIT:1> or not isinstance ( node [ <NUM_LIT:0> ] , nodes . bullet_list ) : <EOL> error = self . state_machine . reporter . error ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' % self . name , <EOL> nodes . literal_block ( self . block_text , self . block_text ) , <EOL> line = self . lineno ) <EOL> raise SystemMessagePropagation ( error ) <EOL> list_node = node [ <NUM_LIT:0> ] <EOL> for item_index in range ( len ( list_node ) ) : <EOL> item = list_node [ item_index ] <EOL> if len ( item ) != <NUM_LIT:1> or not isinstance ( item [ <NUM_LIT:0> ] , nodes . bullet_list ) : <EOL> error = self . state_machine . reporter . error ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> % ( self . name , item_index + <NUM_LIT:1> ) , nodes . literal_block ( <EOL> self . block_text , self . block_text ) , line = self . lineno ) <EOL> raise SystemMessagePropagation ( error ) <EOL> elif item_index : <EOL> if len ( item [ <NUM_LIT:0> ] ) != num_cols : <EOL> error = self . state_machine . reporter . error ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> % ( self . name , item_index + <NUM_LIT:1> , len ( item [ <NUM_LIT:0> ] ) , num_cols ) , <EOL> nodes . literal_block ( self . block_text , self . block_text ) , <EOL> line = self . lineno ) <EOL> raise SystemMessagePropagation ( error ) <EOL> else : <EOL> num_cols = len ( item [ <NUM_LIT:0> ] ) <EOL> col_widths = self . get_column_widths ( num_cols ) <EOL> return num_cols , col_widths <EOL> def build_table_from_list ( self , table_data , col_widths , header_rows , stub_columns ) : <EOL> table = nodes . table ( ) <EOL> tgroup = nodes . tgroup ( cols = len ( col_widths ) ) <EOL> table += tgroup <EOL> for col_width in col_widths : <EOL> colspec = nodes . colspec ( colwidth = col_width ) <EOL> if stub_columns : <EOL> colspec . attributes [ '<STR_LIT>' ] = <NUM_LIT:1> <EOL> stub_columns -= <NUM_LIT:1> <EOL> tgroup += colspec <EOL> rows = [ ] <EOL> for row in table_data : <EOL> row_node = nodes . row ( ) <EOL> for cell in row : <EOL> entry = nodes . entry ( ) <EOL> entry += cell <EOL> row_node += entry <EOL> rows . append ( row_node ) <EOL> if header_rows : <EOL> thead = nodes . thead ( ) <EOL> thead . extend ( rows [ : header_rows ] ) <EOL> tgroup += thead <EOL> tbody = nodes . tbody ( ) <EOL> tbody . extend ( rows [ header_rows : ] ) <EOL> tgroup += tbody <EOL> return table </s>
