✨ Render full RST need directives from @rst blocks

src/sphinx_codelinks/analyse/analyse.py

@@ -76,9 +76,7 @@ def __init__(
         self.git_commit_rev: str | None = (
             utils.get_current_rev(self.git_root) if self.git_root else None
         )
-        self.project_path: Path = (
-            self.git_root if self.git_root else self.analyse_config.src_dir
-        )
+        self.project_path: Path = self.git_root or self.analyse_config.src_dir
         self.oneline_warnings: list[AnalyseWarning] = []
 
     def get_src_strings(self) -> Generator[tuple[Path, bytes], Any, None]:  # type: ignore[explicit-any]

src/sphinx_codelinks/analyse/utils.py

@@ -2,6 +2,7 @@
 import configparser
 import logging
 from pathlib import Path
+import re
 from typing import TypedDict
 from urllib.request import pathname2url
 
@@ -355,6 +356,170 @@ class ExtractedRstType(TypedDict):
     end_idx: int
 
 
+class ParsedDirective(TypedDict):
+    """A single parsed RST directive."""
+
+    name: str
+    argument: str
+    options: dict[str, str]
+    content: str
+    has_extra_content: bool
+    directive_line_offset: int
+    """0-based line index of the ``.. name::`` line within the input text."""
+    content_line_offset: int | None
+    """0-based line index where the directive content starts within the input text.
+
+    ``None`` if the directive has no content body.
+    """
+
+
+_RE_DIRECTIVE = re.compile(r"^(\s*)\.\.\s+([\w:.+-]+)\s*::\s*(.*)")
+_RE_OPTION = re.compile(r"^\s+:([^:]+):\s*(.*)")
+
+
+def _parse_options(body_lines: list[str]) -> tuple[dict[str, str], int]:
+    """Parse field-list options from the start of directive body lines.
+
+    Supports multi-line option values: continuation lines must be indented
+    and are joined with a single space.
+
+    :return: Tuple of (options dict, content_start index into body_lines).
+    """
+    options: dict[str, str] = {}
+    content_start = 0
+    current_key: str | None = None
+    for j, line in enumerate(body_lines):
+        if not line.strip():
+            # Blank line ends the option block.
+            content_start = j + 1
+            current_key = None
+            break
+        opt_match = _RE_OPTION.match(line)
+        if opt_match:
+            current_key = opt_match.group(1).strip()
+            options[current_key] = opt_match.group(2).strip()
+            content_start = j + 1
+        elif current_key is not None and line[:1] == " ":
+            # Continuation line for the previous option value.
+            # NOTE: In standard RST (docutils),
+            # continuation indent is measured relative to the field body
+            # start.  Here any leading space is accepted, which is looser
+            # but correct within a directive body where all lines are
+            # already indented past the directive marker.
+            prev = options[current_key]
+            continuation = line.strip()
+            options[current_key] = f"{prev} {continuation}" if prev else continuation
+            content_start = j + 1
+        else:
+            content_start = j
+            break
+    else:
+        content_start = len(body_lines)
+    return options, content_start
+
+
+def _extract_content(
+    body_lines: list[str], content_start: int
+) -> tuple[list[str], int]:
+    """Extract and dedent the content portion of a directive body.
+
+    :return: Tuple of (dedented content lines, number of leading blank lines removed).
+    """
+    content_lines = body_lines[content_start:]
+    content_blanks_removed = 0
+    while content_lines and not content_lines[0].strip():
+        content_lines.pop(0)
+        content_blanks_removed += 1
+    while content_lines and not content_lines[-1].strip():
+        content_lines.pop()
+    if content_lines:
+        min_indent = min(
+            len(cl) - len(cl.lstrip()) for cl in content_lines if cl.strip()
+        )
+        content_lines = [cl[min_indent:] if cl.strip() else "" for cl in content_lines]
+    return content_lines, content_blanks_removed
+
+
+def parse_single_directive(rst_text: str) -> ParsedDirective | None:
+    """Parse a single RST directive from text.
+
+    Expects text whose first non-blank line is a directive, e.g.::
+
+        .. need-type:: argument
+           :option: value
+
+           Content body here.
+
+    :param rst_text: The RST text to parse.
+    :return: Parsed directive, or ``None`` if the first non-blank line
+        is not a directive.
+    """
+    lines = rst_text.splitlines()
+
+    # Find directive on the first non-blank line
+    dir_idx: int | None = None
+    dir_match: re.Match[str] | None = None
+    for i, line in enumerate(lines):
+        if line.strip():
+            dir_match = _RE_DIRECTIVE.match(line)
+            if dir_match:
+                dir_idx = i
+            break
+
+    if dir_idx is None or dir_match is None:
+        return None
+
+    dir_indent = len(dir_match.group(1))
+    name = dir_match.group(2)
+    # NOTE: In standard RST (docutils), directive
+    # arguments may span multiple lines before the first field-list
+    # marker.  Here only the ``.. name::`` line is captured; this is
+    # sufficient for NeedDirective where the argument is a single-line
+    # title.
+    argument = dir_match.group(3).strip()
+
+    # Collect body: indented (or blank) lines after the directive.
+    # body_end tracks the last non-blank indented line so trailing
+    # blank lines between the directive and outside content are excluded.
+    body_end = dir_idx
+    for i in range(dir_idx + 1, len(lines)):
+        line = lines[i]
+        if not line.strip():
+            continue
+        if len(line) - len(line.lstrip()) > dir_indent:
+            body_end = i
+        else:
+            break
+
+    body_lines = lines[dir_idx + 1 : body_end + 1]
+
+    options, content_start = _parse_options(body_lines)
+    content_lines, content_blanks_removed = _extract_content(body_lines, content_start)
+    content = "\n".join(content_lines)
+
+    # Extra content = any non-blank line outside the directive body.
+    has_extra = any(lines[i].strip() for i in range(body_end + 1, len(lines)))
+
+    # Line offsets relative to the start of rst_text (0-based).
+    directive_line_offset = dir_idx
+    if content_lines:
+        content_line_offset: int | None = (
+            dir_idx + 1 + content_start + content_blanks_removed
+        )
+    else:
+        content_line_offset = None
+
+    return ParsedDirective(
+        name=name,
+        argument=argument,
+        options=options,
+        content=content,
+        has_extra_content=has_extra,
+        directive_line_offset=directive_line_offset,
+        content_line_offset=content_line_offset,
+    )
+
+
 # @Extract reStructuredText blocks embedded in comments, IMPL_RST_1, impl, [FE_RST_EXTRACTION]
 def extract_rst(
     text: str, start_marker: str, end_marker: str

src/sphinx_codelinks/sphinx_extension/directives/src_trace.py

@@ -5,14 +5,17 @@
 
 from docutils import nodes
 from docutils.parsers.rst import directives
+from docutils.statemachine import StringList
 from packaging.version import Version
 import sphinx
 from sphinx.util.docutils import SphinxDirective
 from sphinx_needs.api import add_need  # type: ignore[import-untyped]
+from sphinx_needs.directives.need import NeedDirective  # type: ignore[import-untyped]
 from sphinx_needs.utils import add_doc  # type: ignore[import-untyped]
 
 from sphinx_codelinks.analyse.analyse import SourceAnalyse
-from sphinx_codelinks.analyse.models import OneLineNeed
+from sphinx_codelinks.analyse.models import Metadata
+from sphinx_codelinks.analyse.utils import parse_single_directive
 from sphinx_codelinks.config import (
     CodeLinksConfig,
     CodeLinksProjectConfigType,
@@ -43,7 +46,7 @@ def get_rel_path(doc_path: Path, code_path: Path, base_dir: Path) -> tuple[Path,
 
 
 def generate_str_link_name(
-    oneline_need: OneLineNeed,
+    oneline_need: Metadata,
     target_filepath: Path,
     dirs: dict[str, Path],
     local: bool = False,
@@ -180,6 +183,16 @@ def run(self) -> list[nodes.Node]:
             dirs,
         )
 
+        # render needs from marked RST blocks
+        rendered_needs.extend(
+            self.render_marked_rst_needs(
+                src_analyse,
+                local_url_field,
+                remote_url_field,
+                dirs,
+            )
+        )
+
         # for post-processing of need links
         # https://github.com/useblocks/sphinx-needs/issues/1210
         add_doc(self.env, self.env.docname)
@@ -322,3 +335,101 @@ def render_needs(
                         ] = f"{docs_href}#{oneline_need.need['id']}"
 
         return rendered_needs
+
+    def render_marked_rst_needs(
+        self,
+        src_analyse: SourceAnalyse,
+        local_url_field: str | None,
+        remote_url_field: str | None,
+        dirs: dict[str, Path],
+    ) -> list[nodes.Node]:
+        """Render needs from marked RST blocks (``@rst ... @endrst``).
+
+        Each block is expected to contain a single need directive.
+        Warnings are emitted when the block does not contain a directive
+        or contains content outside the directive.
+        """
+        rendered_nodes: list[nodes.Node] = []
+        for marked_rst in src_analyse.marked_rst:
+            parsed = parse_single_directive(marked_rst.rst)
+            src_file = str(marked_rst.filepath)
+            src_line = marked_rst.source_map["start"]["row"] + 1
+
+            if parsed is None:
+                logger.warning(
+                    f"No directive found in marked RST block [{src_file}:{src_line}]",
+                    location=(self.env.docname, self.lineno),
+                )
+                continue
+
+            if parsed["has_extra_content"]:
+                logger.warning(
+                    "Content found outside directive in marked RST block "
+                    f"[{src_file}:{src_line}]; "
+                    "only a single directive is supported",
+                    location=(self.env.docname, self.lineno),
+                )
+
+            # Build content StringList with source mapping
+            content_lines = parsed["content"].splitlines() if parsed["content"] else []
+            content_offset = self.content_offset
+            if parsed["content_line_offset"] is not None:
+                content_offset = src_line - 1 + parsed["content_line_offset"]
+            content = StringList(content_lines, source=src_file)
+
+            # Build arguments list (title)
+            arguments = [parsed["argument"]] if parsed["argument"] else []
+
+            # Options are passed as raw strings without conversion or
+            # validation here; NeedDirective uses a DummyOptionSpec that
+            # accepts all keys as strings, and performs its own key-by-key
+            # validation inside run().
+            # NOTE: DummyOptionSpec was added in sphinx-needs v6
+            # (d09332d); earlier versions use a fixed option_spec.
+            options: dict[str, str | None] = dict(parsed["options"])
+
+            # Inject URL fields
+            filepath = src_analyse.analyse_config.src_dir / marked_rst.filepath
+            target_filepath = dirs["target_dir"] / filepath.relative_to(dirs["src_dir"])
+
+            if local_url_field:
+                target_filepath.parent.mkdir(parents=True, exist_ok=True)
+                target_filepath.write_text(filepath.read_text())
+                local_rel_path, _ = get_rel_path(
+                    Path(self.env.docname), target_filepath, dirs["out_dir"]
+                )
+                options[local_url_field] = generate_str_link_name(
+                    marked_rst, local_rel_path, dirs, local=True
+                )
+            if remote_url_field:
+                options[remote_url_field] = generate_str_link_name(
+                    marked_rst, target_filepath, dirs, local=False
+                )
+
+            directive_lineno = src_line + parsed["directive_line_offset"]
+
+            # Instantiate NeedDirective directly rather than using add_need(),
+            # so that it can process the full directive body (content, options)
+            # through its own run() logic.  We pass the real state/state_machine
+            # from the enclosing SphinxDirective — no mocking needed.
+            need_directive = NeedDirective(
+                name=parsed["name"],
+                arguments=arguments,
+                options=options,
+                content=content,
+                lineno=directive_lineno,
+                content_offset=content_offset,
+                block_text="",
+                state=self.state,
+                state_machine=self.state_machine,
+            )
+            try:
+                rendered_nodes.extend(need_directive.run())
+            except Exception as exc:
+                logger.warning(
+                    "Failed to render directive in marked RST block "
+                    f"[{src_file}:{src_line}]: {exc}",
+                    location=(self.env.docname, self.lineno),
+                )
+
+        return rendered_nodes

tests/__snapshots__/test_src_trace/test_build_html[sphinx_project4-source_code4].doctree.xml

@@ -0,0 +1,5 @@
+<document source="<source>">
+    <target anonymous="" ids="RST_IMPL_1" refid="RST_IMPL_1">
+    <Need classes="need need-impl" ids="RST_IMPL_1" refid="RST_IMPL_1">
+        <paragraph>
+            This need was defined inside an @rst block.

tests/__snapshots__/test_src_trace/test_build_html[sphinx_project5-source_code5].doctree.xml

@@ -0,0 +1,8 @@
+<document source="<source>">
+    <target anonymous="" ids="OL_IMPL_1" refid="OL_IMPL_1">
+    <Need classes="need need-impl" ids="OL_IMPL_1" refid="OL_IMPL_1">
+    <target anonymous="" ids="RST_IMPL_2" refid="RST_IMPL_2">
+    <Need classes="need need-impl" ids="RST_IMPL_2" refid="RST_IMPL_2">
+        <paragraph>
+            This is a detailed need from an @rst block,
+            coexisting with a one-line need.

tests/doc_test/rst_basic/conf.py

@@ -0,0 +1,15 @@
+# Configuration file for the Sphinx documentation builder.
+
+project = "rst-block-test"
+copyright = "2025, useblocks"
+author = "useblocks"
+
+extensions = ["sphinx_needs", "sphinx_codelinks"]
+
+templates_path = ["_templates"]
+exclude_patterns = ["_build", "Thumbs.db", ".DS_Store"]
+
+src_trace_config_from_toml = "src_trace.toml"
+
+html_theme = "alabaster"
+html_static_path = ["_static"]

tests/doc_test/rst_basic/dummy_src.cpp

@@ -0,0 +1,13 @@
+#include <iostream>
+
+/* @rst
+.. impl:: RST Block Implementation
+   :id: RST_IMPL_1
+   :status: open
+
+   This need was defined inside an @rst block.
+@endrst */
+void rst_block_function()
+{
+    std::cout << "RST block example" << std::endl;
+}

tests/doc_test/rst_basic/index.rst

@@ -0,0 +1,2 @@
+.. src-trace::
+   :project: src

tests/doc_test/rst_basic/src_trace.toml

@@ -0,0 +1,5 @@
+[codelinks.projects.src]
+remote_url_pattern = "https://github.com/useblocks/sphinx-codelinks/blob/{commit}/{path}#L{line}"
+
+[codelinks.projects.src.analyse]
+get_rst = true