intersphinx: Add :intersphinx:***: role

Author: tk0miya

sphinx/ext/intersphinx.py

@@ -29,24 +29,28 @@
 import sys
 import time
 from os import path
-from typing import IO, Any, Dict, List, Tuple
+from types import ModuleType
+from typing import IO, Any, Dict, List, Optional, Tuple, cast
 from urllib.parse import urlsplit, urlunsplit
 
 from docutils import nodes
-from docutils.nodes import TextElement
-from docutils.utils import relative_path
+from docutils.nodes import Node, TextElement, system_message
+from docutils.utils import Reporter, relative_path
 
 import sphinx
 from sphinx.addnodes import pending_xref
 from sphinx.application import Sphinx
 from sphinx.builders.html import INVENTORY_FILENAME
 from sphinx.config import Config
 from sphinx.environment import BuildEnvironment
+from sphinx.errors import ExtensionError
 from sphinx.locale import _, __
+from sphinx.transforms.post_transforms import ReferencesResolver
 from sphinx.util import logging, requests
+from sphinx.util.docutils import CustomReSTDispatcher, SphinxRole
 from sphinx.util.inventory import InventoryFile
 from sphinx.util.nodes import find_pending_xref_condition
-from sphinx.util.typing import Inventory
+from sphinx.util.typing import Inventory, RoleFunction
 
 logger = logging.getLogger(__name__)
 
@@ -351,6 +355,115 @@ def missing_reference(app: Sphinx, env: BuildEnvironment, node: pending_xref,
     return None
 
 
+class IntersphinxDispatcher(CustomReSTDispatcher):
+    """Custom dispatcher for intersphinx role.
+
+    This enables :intersphinx:***: roles on parsing reST document.
+    """
+
+    def __init__(self) -> None:
+        super().__init__()
+
+    def role(self, role_name: str, language_module: ModuleType, lineno: int, reporter: Reporter
+             ) -> Tuple[RoleFunction, List[system_message]]:
+        if role_name.split(':')[0] == 'intersphinx':
+            return IntersphinxRole(), []
+        else:
+            return super().role(role_name, language_module, lineno, reporter)
+
+
+class IntersphinxRole(SphinxRole):
+    def run(self) -> Tuple[List[Node], List[system_message]]:
+        role_name = self.get_role_name(self.name)
+        if role_name is None:
+            logger.warning(__('role not found: %s'), self.name,
+                           location=(self.env.docname, self.lineno))
+            return [], []
+
+        result, messages = self.invoke_role(role_name)
+        for node in result:
+            if isinstance(node, pending_xref):
+                node['intersphinx'] = True
+
+        return result, messages
+
+    def get_role_name(self, name: str) -> Optional[Tuple[str, str]]:
+        names = name.split(':')
+        if len(names) == 2:
+            # :intersphinx:role:
+            domain = self.env.temp_data.get('default_domain')
+            role = names[1]
+        elif len(names) == 3:
+            # :intersphinx:domain:role:
+            domain = names[1]
+            role = names[2]
+        else:
+            return None
+
+        if domain and self.is_existent_role(domain, role):
+            return (domain, role)
+        elif self.is_existent_role('std', role):
+            return ('std', role)
+        else:
+            return None
+
+    def is_existent_role(self, domain_name: str, role_name: str) -> bool:
+        try:
+            domain = self.env.get_domain(domain_name)
+            if role_name in domain.roles:
+                return True
+            else:
+                return False
+        except ExtensionError:
+            return False
+
+    def invoke_role(self, role: Tuple[str, str]) -> Tuple[List[Node], List[system_message]]:
+        domain = self.env.get_domain(role[0])
+        if domain:
+            role_func = domain.role(role[1])
+
+            return role_func(':'.join(role), self.rawtext, self.text, self.lineno,
+                             self.inliner, self.options, self.content)
+        else:
+            return [], []
+
+
+class IntersphinxRoleResolver(ReferencesResolver):
+    """pending_xref node resolver for intersphinx role.
+
+    This resolves pending_xref nodes generated by :intersphinx:***: role.
+    """
+
+    default_priority = ReferencesResolver.default_priority - 1
+
+    def run(self, **kwargs: Any) -> None:
+        for node in self.document.traverse(pending_xref):
+            if 'intersphinx' in node:
+                contnode = cast(nodes.TextElement, node[0].deepcopy())
+                refdoc = node.get('refdoc', self.env.docname)
+                try:
+                    domain = self.env.get_domain(node['refdomain'])
+                except Exception:
+                    domain = None
+
+                newnode = missing_reference(self.app, self.env, node, contnode)
+                if newnode is None:
+                    self.warn_missing_reference(refdoc, node['reftype'], node['reftarget'],
+                                                node, domain)
+                else:
+                    node.replace_self(newnode)
+
+
+def install_dispatcher(app: Sphinx, docname: str, source: List[str]) -> None:
+    """Enable IntersphinxDispatcher.
+
+    .. note:: The installed dispatcher will uninstalled on disabling sphinx_domain
+              automatically.
+    """
+    dispatcher = IntersphinxDispatcher()
+    dispatcher.enable()
+
+
 def normalize_intersphinx_mapping(app: Sphinx, config: Config) -> None:
     for key, value in config.intersphinx_mapping.copy().items():
         try:
@@ -381,6 +494,7 @@ def setup(app: Sphinx) -> Dict[str, Any]:
     app.add_config_value('intersphinx_timeout', None, False)
     app.connect('config-inited', normalize_intersphinx_mapping, priority=800)
     app.connect('builder-inited', load_mappings)
+    app.connect('source-read', install_dispatcher)
     app.connect('missing-reference', missing_reference)
     return {
         'version': sphinx.__display_version__,