tokenizers

1
from collections import defaultdict, abc
2
from typing import cast
3

4
from docutils import nodes
5
from docutils.parsers.rst import Directive
6

7
import sphinx
8
from sphinx.locale import _
9
from sphinx.util.docutils import SphinxDirective
10
from sphinx.errors import ExtensionError
11

12
from conf import languages as LANGUAGES
13

14
logger = sphinx.util.logging.getLogger(__name__)
15

16
GLOBALNAME = "$GLOBAL$"
17

18

19
def update(d, u):
20
    for k, v in u.items():
21
        if isinstance(v, abc.Mapping):
22
            d[k] = update(d.get(k, {}), v)
23
        else:
24
            d[k] = v
25
    return d
26

27

28
class EntityNode(nodes.General, nodes.Element):
29
    pass
30

31

32
class EntitiesNode(nodes.General, nodes.Element):
33
    pass
34

35

36
class AllEntities:
37
    def __init__(self):
38
        self.entities = defaultdict(dict)
39

40
    @classmethod
41
    def install(cls, env):
42
        if not hasattr(env, "entity_all_entities"):
43
            entities = cls()
44
            env.entity_all_entities = entities
45
        return env.entity_all_entities
46

47
    def merge(self, other):
48
        self.entities.update(other.entities)
49

50
    def purge(self, docname):
51
        for env_docname in [GLOBALNAME, docname]:
52
            self.entities[env_docname] = dict(
53
                [
54
                    (name, entity)
55
                    for name, entity in self.entities[env_docname].items()
56
                    if entity["docname"] != docname
57
                ]
58
            )
59

60
    def _extract_entities(self, nodes):
61
        pass
62

63
    def _extract_options(self, nodes):
64
        pass
65

66
    def _add_entities(self, entities, language, is_global, docname):
67
        scope = GLOBALNAME if is_global else docname
68
        for entity in entities:
69
            name = f'{language}-{entity["name"]}'
70
            content = entity["content"]
71

72
            if name in self.entities[scope]:
73
                logger.warning(
74
                    f'Entity "{name}" has already been defined{" globally" if is_global else ""}',
75
                    location=docname,
76
                )
77

78
            self.entities[scope][name] = {"docname": docname, "content": content}
79

80
    def _extract_global(self, nodes):
81
        for node in nodes:
82
            if node.tagname != "field":
83
                raise Exception(f"Expected a field, found {node.tagname}")
84

85
            name, _ = node.children
86
            if name.tagname != "field_name":
87
                raise Exception(f"Expected a field name here, found {name_node.tagname}")
88

89
            if str(name.children[0]) == "global":
90
                return True
91

92
    def _extract_entities(self, nodes):
93
        entities = []
94
        for node in nodes:
95
            if node.tagname != "definition_list_item":
96
                raise Exception(f"Expected a list item here, found {node.tagname}")
97

98
            name_node, content_node = node.children
99
            if name_node.tagname != "term":
100
                raise Exception(f"Expected a term here, found {name_node.tagname}")
101
            if content_node.tagname != "definition":
102
                raise Exception(f"Expected a definition here, found {content_node.tagname}")
103

104
            name = str(name_node.children[0])
105
            if len(content_node.children) == 1 and content_node.children[0].tagname == "paragraph":
106
                content = content_node.children[0].children[0]
107
            else:
108
                content = content_node
109

110
            entities.append({"name": name, "content": content})
111
        return entities
112

113
    def extract(self, node, docname):
114
        is_global = False
115
        entities = []
116

117
        language = None
118
        for node in node.children:
119
            if language is None and node.tagname != "paragraph":
120
                raise Exception(f"Expected language name:\n.. entities:: <LANGUAGE>")
121
            elif language is None and node.tagname == "paragraph":
122
                language = str(node.children[0])
123
                if language not in LANGUAGES:
124
                    raise Exception(
125
                        f'Unknown language "{language}. Might be missing a newline after language"'
126
                    )
127
            elif node.tagname == "field_list":
128
                is_global = self._extract_global(node.children)
129
            elif node.tagname == "definition_list":
130
                entities.extend(self._extract_entities(node.children))
131
            else:
132
                raise Exception(f"Expected a list of terms/options, found {node.tagname}")
133

134
        self._add_entities(entities, language, is_global, docname)
135

136
    def resolve_pendings(self, app):
137
        env = app.builder.env
138

139
        updates = defaultdict(dict)
140
        for env_docname in self.entities.keys():
141
            for name, entity in self.entities[env_docname].items():
142
                docname = entity["docname"]
143
                node = entity["content"]
144

145
                for node in node.traverse(sphinx.addnodes.pending_xref):
146
                    contnode = cast(nodes.TextElement, node[0].deepcopy())
147
                    newnode = None
148

149
                    typ = node["reftype"]
150
                    target = node["reftarget"]
151
                    refdoc = node.get("refdoc", docname)
152
                    domain = None
153

154
                    try:
155
                        if "refdomain" in node and node["refdomain"]:
156
                            # let the domain try to resolve the reference
157
                            try:
158
                                domain = env.domains[node["refdomain"]]
159
                            except KeyError as exc:
160
                                raise NoUri(target, typ) from exc
161
                            newnode = domain.resolve_xref(
162
                                env, refdoc, app.builder, typ, target, node, contnode
163
                            )
164
                    except NoUri:
165
                        newnode = contnode
166

167
                    updates[env_docname][name] = {
168
                        "docname": docname,
169
                        "content": newnode or contnode,
170
                    }
171

172
        update(self.entities, updates)
173

174
    def get(self, language, name, docname):
175
        name = f"{language}-{name}"
176
        if name in self.entities[docname]:
177
            return self.entities[docname][name]
178
        elif name in self.entities[GLOBALNAME]:
179
            return self.entities[GLOBALNAME][name]
180
        else:
181
            return None
182

183

184
class EntitiesDirective(SphinxDirective):
185
    has_content = True
186

187
    def run(self):
188
        content = nodes.definition_list()
189
        self.state.nested_parse(self.content, self.content_offset, content)
190

191
        try:
192
            entities = AllEntities.install(self.env)
193
            entities.extract(content, self.env.docname)
194
        except Exception as err:
195
            raise self.error(f'Malformed directive "entities": {err}')
196

197
        return []
198

199

200
def entity_role(name, rawtext, text, lineno, inliner, options={}, content=[]):
201
    node = EntityNode()
202
    node.entity = text
203

204
    return [node], []
205

206

207
def process_entity_nodes(app, doctree, docname):
208
    """ Replace all the entities by their content """
209
    env = app.builder.env
210

211
    entities = AllEntities.install(env)
212
    entities.resolve_pendings(app)
213

214
    language = None
215
    try:
216
        language = next(l for l in LANGUAGES if l in app.tags)
217
    except Exception:
218
        logger.warning(f"No language tag specified, not resolving entities in {docname}")
219

220
    for node in doctree.traverse(EntityNode):
221
        if language is None:
222
            node.replace_self(nodes.Text(_(node.entity), _(node.entity)))
223
        else:
224
            entity = entities.get(language, node.entity, docname)
225
            if entity is None:
226
                node.replace_self(nodes.Text(_(node.entity), _(node.entity)))
227
                logger.warning(f'Entity "{node.entity}" has not been defined', location=node)
228
            else:
229
                node.replace_self(entity["content"])
230

231

232
def purge_entities(app, env, docname):
233
    """ Purge any entity that comes from the given docname """
234
    entities = AllEntities.install(env)
235
    entities.purge(docname)
236

237

238
def merge_entities(app, env, docnames, other):
239
    """ Merge multiple environment entities """
240
    entities = AllEntities.install(env)
241
    other_entities = AllEntities.install(other)
242
    entities.merge(other_entities)
243

244

245
def setup(app):
246
    app.add_node(EntityNode)
247
    app.add_node(EntitiesNode)
248
    app.add_directive("entities", EntitiesDirective)
249
    app.add_role("entity", entity_role)
250

251
    app.connect("doctree-resolved", process_entity_nodes)
252
    app.connect("env-merge-info", merge_entities)
253
    app.connect("env-purge-doc", purge_entities)
254

255
    return {
256
        "version": "0.1",
257
        "parallel_read_safe": True,
258
        "parallel_write_safe": True,
259
    }
260