tables.py

来自「Requirement ===========================」· Python 代码 · 共 445 行 · 第 1/2 页

    elif options.has_key('url'):        # CSV data is from a URL
        if not urllib2:
            severe = state_machine.reporter.severe(
                  'Problems with the "%s" directive and its "url" option: '
                  'unable to access the required functionality (from the '
                  '"urllib2" module).' % name,
                  nodes.literal_block(block_text, block_text), line=lineno)
            raise SystemMessagePropagation(severe)
        source = options['url']
        try:
            csv_text = urllib2.urlopen(source).read()
        except (urllib2.URLError, IOError, OSError, ValueError), error:
            severe = state_machine.reporter.severe(
                  'Problems with "%s" directive URL "%s":\n%s.'
                  % (name, options['url'], error),
                  nodes.literal_block(block_text, block_text), line=lineno)
            raise SystemMessagePropagation(severe)
        csv_file = io.StringInput(
            source=csv_text, source_path=source, encoding=encoding,
            error_handler=state.document.settings.input_encoding_error_handler)
        csv_data = csv_file.read().splitlines()
    else:
        error = state_machine.reporter.warning(
            'The "%s" directive requires content; none supplied.' % (name),
            nodes.literal_block(block_text, block_text), line=lineno)
        raise SystemMessagePropagation(error)
    return csv_data, source

def process_header_option(options, state_machine, lineno):
    source = state_machine.get_source(lineno - 1)
    table_head = []
    max_header_cols = 0
    if options.has_key('header'):       # separate table header in option
        rows, max_header_cols = parse_csv_data_into_rows(
            options['header'].split('\n'), HeaderDialect(), source, options)
        table_head.extend(rows)
    return table_head, max_header_cols

def parse_csv_data_into_rows(csv_data, dialect, source, options):
    # csv.py doesn't do Unicode; encode temporarily as UTF-8
    csv_reader = csv.reader([line.encode('utf-8') for line in csv_data],
                            dialect=dialect)
    rows = []
    max_cols = 0
    for row in csv_reader:
        row_data = []
        for cell in row:
            # decode UTF-8 back to Unicode
            cell_text = unicode(cell, 'utf-8')
            cell_data = (0, 0, 0, statemachine.StringList(
                cell_text.splitlines(), source=source))
            row_data.append(cell_data)
        rows.append(row_data)
        max_cols = max(max_cols, len(row))
    return rows, max_cols

def check_table_dimensions(rows, header_rows, stub_columns, name, lineno,
                           block_text, state_machine):
    if len(rows) < header_rows:
        error = state_machine.reporter.error(
            '%s header row(s) specified but only %s row(s) of data supplied '
            '("%s" directive).' % (header_rows, len(rows), name),
            nodes.literal_block(block_text, block_text), line=lineno)
        raise SystemMessagePropagation(error)
    if len(rows) == header_rows > 0:
        error = state_machine.reporter.error(
            'Insufficient data supplied (%s row(s)); no data remaining for '
            'table body, required by "%s" directive.' % (len(rows), name),
            nodes.literal_block(block_text, block_text), line=lineno)
        raise SystemMessagePropagation(error)
    for row in rows:
        if len(row) < stub_columns:
            error = state_machine.reporter.error(
                '%s stub column(s) specified but only %s columns(s) of data '
                'supplied ("%s" directive).' % (stub_columns, len(row), name),
                nodes.literal_block(block_text, block_text), line=lineno)
            raise SystemMessagePropagation(error)
        if len(row) == stub_columns > 0:
            error = state_machine.reporter.error(
                'Insufficient data supplied (%s columns(s)); no data remaining '
                'for table body, required by "%s" directive.'
                % (len(row), name),
                nodes.literal_block(block_text, block_text), line=lineno)
            raise SystemMessagePropagation(error)

def get_column_widths(max_cols, name, options, lineno, block_text,
                      state_machine):
    if options.has_key('widths'):
        col_widths = options['widths']
        if len(col_widths) != max_cols:
            error = state_machine.reporter.error(
              '"%s" widths do not match the number of columns in table (%s).'
              % (name, max_cols),
              nodes.literal_block(block_text, block_text), line=lineno)
            raise SystemMessagePropagation(error)
    elif max_cols:
        col_widths = [100 / max_cols] * max_cols
    else:
        error = state_machine.reporter.error(
            'No table data detected in CSV file.',
            nodes.literal_block(block_text, block_text), line=lineno)
        raise SystemMessagePropagation(error)
    return col_widths

def extend_short_rows_with_empty_cells(columns, parts):
    for part in parts:
        for row in part:
            if len(row) < columns:
                row.extend([(0, 0, 0, [])] * (columns - len(row)))

def list_table(name, arguments, options, content, lineno,
               content_offset, block_text, state, state_machine):
    """
    Implement tables whose data is encoded as a uniform two-level bullet list.
    For further ideas, see
    http://docutils.sf.net/docs/dev/rst/alternatives.html#list-driven-tables
    """ 
    if not content:
        error = state_machine.reporter.error(
            'The "%s" directive is empty; content required.' % name,
            nodes.literal_block(block_text, block_text), line=lineno)
        return [error]
    title, messages = make_title(arguments, state, lineno)
    node = nodes.Element()          # anonymous container for parsing
    state.nested_parse(content, content_offset, node)
    try:
        num_cols, col_widths = check_list_content(
            node, name, options, content, lineno, block_text, state_machine)
        table_data = [[item.children for item in row_list[0]]
                      for row_list in node[0]]
        header_rows = options.get('header-rows', 0) # default 0
        stub_columns = options.get('stub-columns', 0) # default 0
        check_table_dimensions(
            table_data, header_rows, stub_columns, name, lineno,
            block_text, state_machine)
    except SystemMessagePropagation, detail:
        return [detail.args[0]]
    table_node = build_table_from_list(table_data, col_widths,
                                       header_rows, stub_columns)
    table_node['classes'] += options.get('class', [])
    if title:
        table_node.insert(0, title)
    return [table_node] + messages

list_table.arguments = (0, 1, 1)
list_table.options = {'header-rows': directives.nonnegative_int,
                      'stub-columns': directives.nonnegative_int,
                      'widths': directives.positive_int_list,
                      'class': directives.class_option}
list_table.content = 1

def check_list_content(node, name, options, content, lineno, block_text,
                       state_machine):
    if len(node) != 1 or not isinstance(node[0], nodes.bullet_list):
        error = state_machine.reporter.error(
            'Error parsing content block for the "%s" directive: '
            'exactly one bullet list expected.' % name,
            nodes.literal_block(block_text, block_text), line=lineno)
        raise SystemMessagePropagation(error)
    list_node = node[0]
    # Check for a uniform two-level bullet list:
    for item_index in range(len(list_node)):
        item = list_node[item_index]
        if len(item) != 1 or not isinstance(item[0], nodes.bullet_list):
            error = state_machine.reporter.error(
                'Error parsing content block for the "%s" directive: '
                'two-level bullet list expected, but row %s does not contain '
                'a second-level bullet list.' % (name, item_index + 1),
                nodes.literal_block(block_text, block_text), line=lineno)
            raise SystemMessagePropagation(error)
        elif item_index:
            # ATTN pychecker users: num_cols is guaranteed to be set in the
            # "else" clause below for item_index==0, before this branch is
            # triggered.
            if len(item[0]) != num_cols:
                error = state_machine.reporter.error(
                    'Error parsing content block for the "%s" directive: '
                    'uniform two-level bullet list expected, but row %s does '
                    'not contain the same number of items as row 1 (%s vs %s).'
                    % (name, item_index + 1, len(item[0]), num_cols),
                    nodes.literal_block(block_text, block_text), line=lineno)
                raise SystemMessagePropagation(error)
        else:
            num_cols = len(item[0])
    col_widths = get_column_widths(
        num_cols, name, options, lineno, block_text, state_machine)
    if len(col_widths) != num_cols:
        error = state_machine.reporter.error(
            'Error parsing "widths" option of the "%s" directive: '
            'number of columns does not match the table data (%s vs %s).'
            % (name, len(col_widths), num_cols),
            nodes.literal_block(block_text, block_text), line=lineno)
        raise SystemMessagePropagation(error)
    return num_cols, col_widths

def build_table_from_list(table_data, col_widths, header_rows, stub_columns):
    table = nodes.table()
    tgroup = nodes.tgroup(cols=len(col_widths))
    table += tgroup
    for col_width in col_widths:
        colspec = nodes.colspec(colwidth=col_width)
        if stub_columns:
            colspec.attributes['stub'] = 1
            stub_columns -= 1
        tgroup += colspec
    rows = []
    for row in table_data:
        row_node = nodes.row()
        for cell in row:
            entry = nodes.entry()
            entry += cell
            row_node += entry
        rows.append(row_node)
    if header_rows:
        thead = nodes.thead()
        thead.extend(rows[:header_rows])
        tgroup += thead
    tbody = nodes.tbody()
    tbody.extend(rows[header_rows:])
    tgroup += tbody
    return table