sphinxext-ip/jandd/sphinxext/ip/__init__.py

# -*- coding: utf-8 -*-
"""
    jandd.sphinxext.ip
    ~~~~~~~~~~~~~~~~~~

    The IP domain.

    :copyright: Copyright (c) 2016 Jan Dittberner
    :license: GPLv3+, see COPYING file for details.
"""
__version__ = '0.1.0'

import re

from ipcalc import Network

from docutils import nodes
from docutils.parsers.rst import Directive

from sphinx import addnodes
from sphinx.domains import Domain, ObjType
from sphinx.environment import NoUri
from sphinx.locale import l_
from sphinx.roles import XRefRole
from sphinx.util.nodes import make_refnode


def ip_object_anchor(typ, path):
    path = re.sub(r'[.:/]', '-', path)
    return typ.lower() + '-' + path


class ip_node(nodes.Inline, nodes.TextElement): pass

class ip_range(nodes.General, nodes.Element): pass


class IPXRefRole(XRefRole):
    """
    Cross referencing role for the IP domain.
    """
    def __init__(self, method, **kwargs):
        self.method = method
        innernodeclass=None
        if method in ('v4', 'v6'):
            innernodeclass = ip_node
        super(IPXRefRole, self).__init__(
            innernodeclass=innernodeclass, **kwargs)

    def process_link(self, env, refnode, has_explicit_title, title, target):
        domaindata = env.domaindata['ip']
        domaindata[self.method][target] = (target, refnode)
        return title, target

    def result_nodes(self, document, env, node, is_ref):
        try:
            node['typ'] = self.method
            indexnode = addnodes.index()
            targetid = 'index-%s' % env.new_serialno('index')
            targetnode = nodes.target('', '', ids=[targetid])
            idxtext = "%s; %s" % (node.astext(), env.docname)
            indexnode['entries'] = [('single', idxtext, targetid, '', None)]
            return [indexnode, targetnode, node], []
        except KeyError as e:
            return [node], [e.args[0]]


class IPRange(Directive):
    has_content = True
    required_arguments = 1
    optional_arguments = 0
    final_argument_whitespace = False
    option_spec = {}

    def handle_rangespec(self, node):
        titlenode = nodes.title()
        node.append(titlenode)
        titlenode.append(nodes.inline('', self.get_prefix_title()))
        titlenode.append(nodes.literal('', self.rangespec))
        ids = ip_object_anchor(self.typ, self.rangespec)
        node['ids'].append(ids)
        self.env.domaindata[self.domain][self.typ][ids] = (
            self.env.docname,
            self.options.get('synopsis', ''))
        return ids

    def run(self):
        if ':' in self.name:
            self.domain, self.objtype = self.name.split(':', 1)
        else:
            self.domain, self.objtype = '', self.name
        self.env = self.state.document.settings.env
        self.rangespec = self.arguments[0]
        node = nodes.section()
        name = self.handle_rangespec(node)
        self.indexnode = addnodes.index(entries=[
            ('single', self.get_index_text(), name, '')])

        if self.content:
            contentnode = nodes.paragraph('')
            node.append(contentnode)
            self.state.nested_parse(
                self.content, self.content_offset, contentnode)

        iprange = ip_range()
        node.append(iprange)
        iprange['rangespec'] = self.rangespec
        return [self.indexnode, node]


class IPv4Range(IPRange):
    typ = 'v4range'

    def get_prefix_title(self):
        return l_('IPv4 address range ')

    def get_index_text(self):
        return "%s; %s" % (l_('IPv4 range'), self.rangespec)


class IPv6Range(IPRange):
    typ = 'v6range'

    def get_prefix_title(self):
        return l_('IPv6 address range ')

    def get_index_text(self):
        return "%s; %s" % (l_('IPv6 range'), self.rangespec)


class IPDomain(Domain):
    """
    IP address and range domain.
    """
    name = 'ip'
    label = 'IP addresses and ranges.'

    object_types = {
        'v4':      ObjType(l_('v4'), 'v4', 'obj'),
        'v6':      ObjType(l_('v6'), 'v6', 'obj'),
        'v4range': ObjType(l_('v4range'), 'v4range', 'obj'),
        'v6range': ObjType(l_('v6range'), 'v6range', 'obj'),
    }

    directives = {
        'v4range': IPv4Range,
        'v6range': IPv6Range,
    }

    roles = {
        'v4':      IPXRefRole('v4'),
        'v6':      IPXRefRole('v6'),
        'v4range': IPXRefRole('v4range'),
        'v6range': IPXRefRole('v6range'),
    }

    initial_data = {
        'v4': {},
        'v6': {},
        'v4range': {},
        'v6range': {},
        'ips': [],
    }

    def clear_doc(self, docname):
        pass

    def resolve_xref(self, env, fromdocname, builder, typ, target, node,
                     contnode):
        key = ip_object_anchor(typ, target)
        try:
            info = self.data[typ][key]
        except KeyError:
            text = contnode.rawsource
            role = self.roles.get(typ)
            if role is None:
                return None
            resnode = role.result_nodes(env.get_doctree(fromdocname),
                                        env, node, True)[0][0]
            if isinstance(resnode, addnodes.pending_xref):
                text = node[0][0]
                reporter = env.get_doctree(fromdocname).reporter
                reporter.warning('Cannot resolve reference to %r' % text,
                                 line=node.line)
                return None
            return resnode
        else:
            title = typ.upper() + ' ' + target
            anchor = ip_object_anchor(typ, target)
            return make_refnode(builder, fromdocname, info[0], anchor,
                                contnode, title)

    @property
    def items(self):
        return dict((key, self.data[key]) for key in self.object_types)

    def get_objects(self):
        for typ, items in self.items.items():
            for path, info in items.items():
                anchor = ip_object_anchor(typ, path)
                yield (path, path, typ, info[0], anchor, 1)


def process_ips(app, doctree):
    env = app.builder.env
    domaindata = env.domaindata[IPDomain.name]

    for node in doctree.traverse(ip_node):
        ip = node.astext()
        domaindata['ips'].append({
            'docname': env.docname,
            'source': node.parent.source or env.doc2path(env.docname),
            'lineno': node.parent.line,
            'ip': ip,
            'typ': node.parent['typ'],
        })
        node.replace_self(nodes.literal('', ip))


def sort_ip_info(item):
    return item['ip']


def process_ip_nodes(app, doctree, fromdocname):
    env = app.builder.env
    domaindata = env.domaindata[IPDomain.name]

    for node in doctree.traverse(ip_range):
        content = []
        net = Network(node['rangespec'])
        for ip_info in sorted(domaindata['ips'], key=sort_ip_info):
            if ip_info['ip'] in net:
                para = nodes.paragraph()
                para += nodes.literal('', ip_info['ip'])
                ids = ip_object_anchor(ip_info['typ'], ip_info['ip'])
                para['ids'].append(ids)
                domaindata[ip_info['typ']][ids] = (fromdocname, '')
                newnode = nodes.reference('', '', internal=True)
                try:
                    newnode['refuri'] = app.builder.get_relative_uri(
                        fromdocname, ip_info['docname'])
                except NoUri:
                    pass
                title = env.titles[ip_info['docname']]
                innernode = nodes.Text(title.astext())
                newnode.append(innernode)
                para += nodes.Text(" in ")
                para += newnode
                content.append(para)

                #print(ip_info, 'in', node['rangespec'])
        node.replace_self(content)


def setup(app):
    app.add_domain(IPDomain)
    app.connect('doctree-read', process_ips)
    app.connect('doctree-resolved', process_ip_nodes)
    return {'version': __version__}