feat: Initial implemention

Author: SilverRainZ

src/sphinxnotes/data/__init__.py

@@ -0,0 +1,251 @@
+"""
+sphinxnotes.dataview
+~~~~~~~~~~~~~~~~~~~~
+
+:copyright: Copyright 2025 by the Shengyu Zhang.
+:license: BSD, see LICENSE for details.
+"""
+
+from __future__ import annotations
+from typing import Any
+
+from docutils import nodes
+from docutils.parsers.rst import directives
+
+from sphinx.util import logging
+from sphinx.util.docutils import SphinxDirective, SphinxRole
+from sphinx.application import Sphinx
+from sphinx.transforms import SphinxTransform
+
+from . import meta
+from .freestyle import FreeStyleDirective, FreeStyleOptionSpec
+from .data import Data, Field, Schema, Raw
+from .render import Template, Phase, render, JinjaEnv
+from .utils import find_current_document, find_current_section, TempData
+from .context_proxy import proxy
+
+logger = logging.getLogger(__name__)
+
+
+class Node(nodes.Element): ...
+
+
+class pending_node(Node, nodes.meta): ...
+
+
+class rendered_node(Node, nodes.container): ...
+
+
+class TemplateStore(TempData[Template]): ...
+
+
+class RawDataStore(TempData[Raw]): ...
+
+
+class SchemaStore(TempData[Schema]): ...
+
+
+class TemplateDirective(SphinxDirective):
+    option_spec = {
+        'on': Phase.option_spec,
+        'debug': directives.flag,
+    }
+    has_content = True
+
+    def run(self) -> list[nodes.Node]:
+        self.assert_has_content()
+
+        tmpl = Template(
+            text='\n'.join(self.content),
+            phase=self.options.get('on', Phase.default()),
+            debug='debug' in self.options,
+        )
+        TemplateStore.set(self.state.document, tmpl)
+
+        return []
+
+
+class DataSchemaDirective(FreeStyleDirective):
+    optional_arguments = 1
+    option_spec = FreeStyleOptionSpec()
+    has_content = True
+
+    def run(self) -> list[nodes.Node]:
+        if self.arguments:
+            name = Field.from_str(self.arguments[0])
+        else:
+            name = None
+
+        attrs = {}
+        for k, v in self.options.items():
+            attrs[k] = Field.from_str(v)
+
+        if self.content:
+            content = Field.from_str(self.content[0])
+        else:
+            content = None
+
+        schema = Schema(name, attrs, content)
+        SchemaStore.set(self.state.document, schema)
+
+        return []
+
+
+def markup_context(v: SphinxDirective | SphinxRole | nodes.Element) -> dict[str, Any]:
+    ctx = {}
+
+    if isinstance(v, nodes.Element):
+        ctx['_markup'] = {}
+    else:
+        isdir = isinstance(v, SphinxDirective)
+        ctx['_markup'] = {
+            'type': 'directive' if isdir else 'role',
+            'name': v.name,
+            'lineno': v.lineno,
+            'rawtext': v.block_text if isdir else v.rawtext,
+        }
+    return ctx
+
+
+def doctree_context(v: SphinxDirective | SphinxRole | nodes.Element) -> dict[str, Any]:
+    ctx = {}
+    if isinstance(v, nodes.Element):
+        ctx['_doc'] = proxy(find_current_document(v))
+        ctx['_section'] = proxy(find_current_section(v))
+    else:
+        isdir = isinstance(v, SphinxDirective)
+        state = v.state if isdir else v.inliner
+        ctx['_doc'] = proxy(state.document)
+        ctx['_section'] = proxy(find_current_section(state.parent))
+    return ctx
+
+
+def sphinx_context(v: SphinxDirective | SphinxRole | SphinxTransform) -> dict[str, Any]:
+    ctx = {}
+    ctx['_env'] = proxy(v.env)
+    ctx['_config'] = proxy(v.config)
+    return ctx
+
+
+class DataDefineDirective(FreeStyleDirective):
+    optional_arguments = 1
+    has_content = True
+
+    def run(self) -> list[nodes.Node]:
+        tmpl = TemplateStore.get(self.state.document)
+        if tmpl is None:
+            return []
+
+        schema = SchemaStore.get(self.state.document)
+        rawdata = self._raw_data()
+
+        if tmpl.phase != Phase.Parsing:
+            n = pending_node()
+            RawDataStore.set(n, rawdata)
+            TemplateStore.set(n, tmpl)
+            if schema:
+                SchemaStore.set(n, schema)
+            return [n]
+
+        if schema:
+            try:
+                data = schema.parse(rawdata)
+            except ValueError as e:
+                raise self.error(str(e))
+        else:
+            data = Data.from_raw(rawdata)
+
+        extra_ctx = {
+            **sphinx_context(self),
+            **doctree_context(self),
+            **markup_context(self),
+        }
+        n = render(self.parse_text_to_nodes, tmpl, data, extra_ctx)
+
+        return [n]
+
+    def _raw_data(self) -> Raw:
+        return Raw(
+            self.arguments[0] if self.arguments else None,
+            self.options.copy(),
+            '\n'.join(self.content),
+        )
+
+
+class ParsedHookDirective(SphinxDirective):
+    def run(self) -> list[nodes.Node]:
+        for pending in self.state.document.findall(pending_node):
+            tmpl = TemplateStore.get(pending)
+            schema = SchemaStore.get(pending)
+            rawdata = RawDataStore.get(pending)
+
+            assert tmpl
+            assert rawdata
+
+            if schema:
+                try:
+                    data = schema.parse(rawdata)
+                except ValueError as e:
+                    raise self.error(str(e))
+            else:
+                data = Data.from_raw(rawdata)
+
+            extra_ctx = {
+                **sphinx_context(self),
+                **doctree_context(pending),
+                **markup_context(pending),
+            }
+            n = render(self.parse_text_to_nodes, tmpl, data, extra_ctx)
+            pending.replace_self(n)
+
+        return []  # nothing to return
+
+
+def on_source_read(app, docname, content):
+    # NOTE: content is a single element list, representing the content of the
+    # source file.
+    #
+    # .. seealso:: https://www.sphinx-doc.org/en/master/extdev/event_callbacks.html#event-source-read
+    content[-1] = content[-1] + '\n\n.. data:parsed-hook::'
+
+
+class ResolvingHookTransform(SphinxTransform):
+    default_priority = 210  # 在主要处理阶段运行
+
+    def apply(self, **kwargs):
+        for pending in self.document.findall(pending_node):
+            tmpl = TemplateStore.get(pending)
+            schema = SchemaStore.get(pending)
+            rawdata = RawDataStore.get(pending)
+
+            assert tmpl
+            assert rawdata
+
+            if schema:
+                try:
+                    data = schema.parse(rawdata)
+                except ValueError:
+                    continue  # FIXME
+            else:
+                data = Data.from_raw(rawdata)
+
+            n = render(self.parse_text_to_nodes, tmpl, data, {})
+            pending.replace_self(n)
+
+
+def setup(app: Sphinx):
+    meta.pre_setup(app)
+
+    app.add_directive('data:tmpl', TemplateDirective, False)
+    app.add_directive('data:schema', DataSchemaDirective, False)
+    # app.add_directive('data:use-tmpl', DataTemplateDirective, False)
+    app.add_directive('data:def', DataDefineDirective, False)
+    app.add_directive('data:parsed-hook', ParsedHookDirective, False)
+
+    app.connect('source-read', on_source_read)
+
+    app.add_transform(ResolvingHookTransform)
+
+    JinjaEnv.setup(app)
+
+    return meta.post_setup(app)

src/sphinxnotes/data/context_proxy.py

@@ -0,0 +1,147 @@
+from dataclasses import dataclass
+from functools import wraps
+from typing import Any, Callable
+from types import MappingProxyType
+
+from docutils import nodes
+
+from sphinx.util import logging
+from sphinx.config import Config as SphinxConfig
+
+logger = logging.getLogger(__name__)
+
+
+def proxy_property(func: Callable[[Any], Any]) -> property:
+    @wraps(func)
+    def wrapped(self: Proxy) -> Any:
+        return self._normalize(func(self))
+
+    return property(wrapped)
+
+
+@dataclass(frozen=True)
+class Proxy:
+    _obj: Any
+
+    def __getattr__(self, name: str) -> Any:
+        # Internal attr is not accessable.
+        if name.startswith('_'):
+            raise AttributeError(name)
+
+        v = getattr(self._obj, name)
+        if callable(v):
+            # Deny callable attr for safety.
+            raise AttributeError(name)
+
+        return self._wrap(v)
+
+    @staticmethod
+    def _wrap(v: Any) -> Any:
+        cls = SPECIFIC_TYPE_REGISTRY.get(type(v))
+        if cls:
+            return cls(v)
+        for types, cls in TYPE_REGISTRY.items():
+            if isinstance(v, types):
+                return cls(v)
+        return v
+
+    @staticmethod
+    def _normalize(val: Any) -> Any:
+        """
+        对显式 property 的返回值做模板安全化处理：
+        - 标量原样返回；
+        - 已是 ContextProxy 原样返回；
+        - 注册类型自动 wrap；
+        - 容器转换为不可变快照并递归处理；
+        - 其他对象默认转 str。
+        """
+        # 标量
+        if val is None or isinstance(val, (str, int, float, bool)):
+            return val
+
+        # 已包装过
+        if isinstance(val, Proxy):
+            return val
+
+        # 注册类型自动 wrap
+        wrapped_val = Proxy._wrap(val)
+        if wrapped_val is not val:
+            return wrapped_val
+
+        # 容器
+        if isinstance(val, (set, frozenset)):
+            return frozenset(Proxy._normalize(x) for x in val)
+        if isinstance(val, (list, tuple)):
+            return tuple(Proxy._normalize(x) for x in val)
+        if isinstance(val, dict):
+            copied = {k: Proxy._normalize(v) for k, v in val.items()}
+            return MappingProxyType(copied)
+
+        # 其他对象：安全起见转成字符串
+        return str(val)
+
+
+@dataclass(frozen=True)
+class Node(Proxy):
+    _obj: nodes.Element
+
+    @proxy_property
+    def attrs(self) -> dict[str, str]:
+        """Shortcut to :attr:`nodes.Element.attributes`."""
+        return self._obj.attributes
+
+    def __str__(self) -> str:
+        return self._obj.astext()
+
+
+@dataclass(frozen=True)
+class NodeWithTitle(Node):
+    @proxy_property
+    def title(self) -> Node | None:
+        return self._obj.next_node(nodes.Titular)
+
+
+@dataclass(frozen=True)
+class Section(NodeWithTitle):
+    _obj: nodes.section
+
+    @proxy_property
+    def sections(self) -> tuple['Section', ...]:
+        sect_nodes = self._obj[0].findall(
+            nodes.section, descend=False, ascend=False, siblings=True
+        )
+        return list(sect_nodes)
+
+
+@dataclass(frozen=True)
+class Document(NodeWithTitle):
+    _obj: nodes.document
+
+    def _top_section(self) -> Section:
+        section = self._obj.next_node(nodes.section)
+        assert section
+        return Section(section)
+
+    @proxy_property
+    def sections(self) -> tuple[Section, ...]:
+        return self._top_section().sections
+
+
+@dataclass(frozen=True)
+class ConfigProxy(Proxy):
+    _obj: SphinxConfig
+
+
+TYPE_REGISTRY: dict[type | tuple[type, ...], type[Proxy]] = {
+    nodes.Node: Node,
+}
+
+SPECIFIC_TYPE_REGISTRY: dict[type, type[Proxy]] = {
+    nodes.document: Document,
+    nodes.section: Section,
+    SphinxConfig: ConfigProxy,
+}
+
+
+def proxy(v: Any) -> Proxy | None:
+    return Proxy._wrap(v) if v is not None else None