1import json
2from collections.abc import Generator
3from dataclasses import dataclass
4from pathlib import Path
5from typing import Any, cast
6
7from tree_sitter import Node as TreeSitterNode
8
9from sphinx_codelinks.analyse import multiline_parser, utils
10from sphinx_codelinks.analyse.models import (
11 MarkedContentType,
12 MultilineNeed,
13 NeedIdRefs,
14 OneLineNeed,
15 SourceComment,
16 SourceFile,
17 SourceMap,
18)
19from sphinx_codelinks.analyse.oneline_parser import (
20 OnelineParserInvalidWarning,
21 oneline_parser,
22)
23from sphinx_codelinks.analyse.references import _relative_posix
24from sphinx_codelinks.config import (
25 UNIX_NEWLINE,
26 OneLineCommentStyle,
27 SourceAnalyseConfig,
28)
29from sphinx_codelinks.logger import get_logger
30from sphinx_codelinks.source_discover.config import CommentType
31
32logger = get_logger(__name__)
33
34
35def _count(n: int, noun: str) -> str:
36 """Format ``n noun`` with a naive (append-s) plural for progress summaries."""
37 return f"{n} {noun}" if n == 1 else f"{n} {noun}s"
38
39
40def _row_count(src_comment: SourceComment) -> int:
41 """The rows a comment covers: a Rust ``///`` node's text ends with its newline, which
42 starts no row of its own."""
43 return (src_comment.node.text or b"").rstrip(b"\n").count(b"\n") + 1
44
45
46@dataclass
47class AnalyseWarning:
48 file_path: str
49 lineno: int
50 msg: str
51 type: str
52 sub_type: str
53
54
55class SourceAnalyse:
56 def __init__(
57 self,
58 analyse_config: SourceAnalyseConfig,
59 *,
60 name: str = "",
61 ) -> None:
62 self.name = name
63 self.analyse_config = analyse_config
64 self.src_files: list[SourceFile] = []
65 self.src_comments: list[SourceComment] = []
66 self.need_id_refs: list[NeedIdRefs] = []
67 self.oneline_needs: list[OneLineNeed] = []
68 self.multiline_needs: list[MultilineNeed] = []
69 self.all_marked_content: list[NeedIdRefs | OneLineNeed | MultilineNeed] = []
70 # Use explicitly configured git_root if provided, otherwise auto-detect
71 if self.analyse_config.git_root is not None:
72 self.git_root: Path | None = self.analyse_config.git_root.resolve()
73 else:
74 self.git_root = utils.locate_git_root(self.analyse_config.src_dir)
75 self.git_remote_url: str | None = (
76 utils.get_remote_url(self.git_root) if self.git_root else None
77 )
78 self.git_commit_rev: str | None = (
79 utils.get_current_rev(self.git_root) if self.git_root else None
80 )
81 self.project_path: Path = self.git_root or self.analyse_config.src_dir
82 self.warnings: list[AnalyseWarning] = []
83 """The analysis' warnings, one-line and multi-line alike, in extraction order."""
84 # Per-run memo of parsed compile_commands.json, keyed by DB path, so the
85 # database is read once per run instead of once per source file.
86 self._flags_map_cache: dict[Path, dict[Path, list[str]] | None] = {}
87
88 @property
89 def oneline_warnings(self) -> list[AnalyseWarning]:
90 """The old name of :attr:`warnings`, kept for one release; read-only."""
91 return self.warnings
92
93 def get_src_strings(self) -> Generator[tuple[Path, bytes], Any, None]:
94 """Load source files and extract their content."""
95 for src_path in self.analyse_config.src_files:
96 if not utils.is_text_file(src_path):
97 continue
98 with src_path.open("r", encoding="utf-8", newline="") as f:
99 # Normalize all line endings to Unix LF
100 text = f.read()
101 text = text.replace("\r\n", "\n").replace("\r", "\n")
102 yield src_path, text.encode("utf-8")
103
104 def create_src_objects(self) -> None:
105 parser, query = utils.init_tree_sitter(self.analyse_config.comment_type)
106
107 for src_path, src_string in self.get_src_strings():
108 comments: list[TreeSitterNode] | None = utils.extract_comments(
109 src_string, parser, query
110 )
111 if not comments:
112 continue
113 src_comments: list[SourceComment] = [
114 SourceComment(node) for node in comments
115 ]
116
117 src_file = SourceFile(src_path.absolute())
118 src_file.add_comments(src_comments)
119 if self.analyse_config.get_multiline_needs:
120 src_file.lines = src_string.decode("utf-8").split(UNIX_NEWLINE)
121 self.src_files.append(src_file)
122 self.src_comments.extend(src_comments)
123
124 def _flags_map_for(self, db_path: Path) -> dict[Path, list[str]] | None:
125 """Return the parsed compile_commands.json for ``db_path``, memoized.
126
127 The database is read and parsed once per run instead of once per source
128 file (O(files x entries) -> O(entries)). A present-but-malformed or
129 unreadable database is warned once and cached as ``None`` so callers fall
130 back to the configured defines without re-reading or re-warning per file.
131 """
132 from sphinx_codelinks.analyse.preproc import compile_db
133
134 if db_path in self._flags_map_cache:
135 return self._flags_map_cache[db_path]
136 try:
137 flags = compile_db.load_flags_map(db_path)
138 except (OSError, ValueError, TypeError) as exc:
139 logger.warning(
140 f"codelinks: failed to read {db_path} ({exc}); "
141 f"falling back to the configured defines"
142 )
143 self._flags_map_cache[db_path] = None
144 return None
145 self._flags_map_cache[db_path] = flags
146 return flags
147
148 def _resolve_preproc_args(self, src_path: Path) -> list[str] | None:
149 from sphinx_codelinks.analyse.preproc import compile_db
150
151 # `run()` calls this (via create_src_objects_libclang) only when
152 # `preprocessor is not None`, but keep the guard: it narrows the type for
153 # the checker and is a cheap defense (an `assert` would trip bandit S101).
154 preproc = self.analyse_config.preprocessor
155 if preproc is None:
156 return []
157 db_path = preproc.compile_commands
158 if db_path is None:
159 db_path = compile_db.find_compile_db(src_path, self.project_path)
160 if db_path is not None and db_path.is_file():
161 flags = self._flags_map_for(db_path)
162 if flags is None:
163 # Present-but-malformed/unreadable DB (warned once in the helper):
164 # fall back to the configured defines so extraction still runs.
165 return compile_db.defines_to_args(
166 preproc.defines, preproc.includes, preproc.std
167 )
168 args = flags.get(src_path.absolute().resolve())
169 if args is not None:
170 return args
171 # Absent from the DB. compile_commands.json lists only compiled
172 # translation units, never headers — so a header here is parsed
173 # standalone with the global defines (one run = one variant). A
174 # compiled source absent from the build is skipped (spec §3.3).
175 if compile_db.is_translation_unit_source(src_path):
176 return None
177 return compile_db.defines_to_args(
178 preproc.defines, preproc.includes, preproc.std
179 )
180 # No readable DB. `find_compile_db` only ever returns a real file, so a
181 # non-None `db_path` that reaches here is an explicit `compile_commands`
182 # path that is not a readable file (typo/missing): warn and fall back
183 # rather than skip silently. A genuinely absent DB (db_path is None) just
184 # falls back to the manual defines applied globally.
185 if db_path is not None:
186 logger.warning(
187 f"codelinks: compile_commands path {db_path} is not a readable "
188 f"file; falling back to the configured defines"
189 )
190 return compile_db.defines_to_args(
191 preproc.defines, preproc.includes, preproc.std
192 )
193
[docs]194 # @Extract traceability objects with the preprocessor-aware libclang engine, IMPL_PREPROC_1, impl, [FE_PREPROC]
195 def create_src_objects_libclang(self) -> None:
196 from sphinx_codelinks.analyse.preproc import (
197 libclang_parser,
198 loader,
199 )
200
201 # Resolve the exception via the loader (never a direct ``import
202 # clang.cindex``) so a missing ``libclang`` extra still surfaces the
203 # loader's friendly install hint rather than a bare ImportError.
204 translation_unit_load_error = (
205 loader.load_clang_cindex().TranslationUnitLoadError
206 )
207
208 for src_path in self.analyse_config.src_files:
209 if not utils.is_text_file(src_path):
210 continue
211 args = self._resolve_preproc_args(src_path)
212 if args is None:
213 logger.debug(
214 f"codelinks: skipping {src_path} — not found in compile_commands.json"
215 )
216 continue
217 try:
218 comments = libclang_parser.extract_active_comments(src_path, args)
219 except translation_unit_load_error:
220 # Last-resort guard. Standalone parses pin ``-x`` to match the
221 # ``-std`` (see defines_to_args), so the common case — a ``.h``/
222 # ``.c`` header handed a C++ ``-std`` — now parses as C++ and its
223 # markers extract. This only fires if libclang still cannot load
224 # the file as a translation unit at all; skip it rather than
225 # aborting the whole run. Surfaced as a ``warning`` so the silent
226 # data-loss (a file's markers dropped) is visible — accepting that
227 # this fails ``sphinx-build -W``.
228 logger.warning(
229 f"codelinks: skipping {src_path} — libclang could not load it "
230 f"as a translation unit"
231 )
232 continue
233 if not comments:
234 continue
235 # ``c`` is a LibclangComment duck-typing the tree-sitter Node
236 # interface SourceComment reads (``.text`` / ``.start_point.row``);
237 # the Node-only path (find_associated_scope) is guarded by
238 # ``is_libclang`` so it never runs on these.
239 src_comments = [SourceComment(cast("TreeSitterNode", c)) for c in comments]
240 src_file = SourceFile(src_path.absolute())
241 src_file.add_comments(src_comments)
242 if self.analyse_config.get_multiline_needs:
243 # a libclang comment has no column: multi-line needs read what
244 # precedes it on its row from the row itself
245 text = src_path.read_text(encoding="utf-8", errors="replace")
246 text = text.replace("\r\n", "\n").replace("\r", "\n")
247 src_file.lines = text.split(UNIX_NEWLINE)
248 self.src_files.append(src_file)
249 self.src_comments.extend(src_comments)
250
251 def extract_marker(
252 self,
253 text: str,
254 ) -> Generator[tuple[str, list[str], int, int, int], None, None]:
255 lines = text.splitlines()
256 row_offset = 0
257 for line in lines:
258 for marker in self.analyse_config.need_id_refs_config.markers:
259 marker_idx = line.find(marker)
260 if marker_idx == -1:
261 continue
262 markered_text = line[marker_idx + len(marker) :].strip()
263 need_ids = markered_text.replace(",", " ").split()
264 start_column = marker_idx + len(marker)
265 end_column = start_column + len(markered_text)
266 yield marker, need_ids, row_offset, start_column, end_column
267 row_offset += 1
268
[docs]269 # @Extract need ID references from code comments, IMPL_LNK_1, impl, [FE_LNK]
270 def extract_anchors(
271 self,
272 text: str,
273 filepath: Path,
274 tagged_scope: TreeSitterNode | None,
275 src_comment: SourceComment,
276 ) -> list[NeedIdRefs]:
277 """Extract need-ids-refs from a comment."""
278 anchors: list[NeedIdRefs] = []
279 for (
280 marker,
281 need_ids,
282 row_offset,
283 start_column,
284 end_column,
285 ) in self.extract_marker(text):
286 lineno = src_comment.node.start_point.row + row_offset + 1
287 remote_url = self.git_remote_url
288 if self.git_remote_url and self.git_commit_rev:
289 remote_url = utils.form_https_url(
290 self.git_remote_url,
291 self.git_commit_rev,
292 self.project_path,
293 filepath,
294 lineno,
295 )
296 source_map: SourceMap = {
297 "start": {
298 "row": lineno - 1,
299 "column": start_column,
300 },
301 "end": {
302 "row": lineno - 1,
303 "column": end_column,
304 },
305 }
306 anchors.append(
307 NeedIdRefs(
308 filepath,
309 remote_url,
310 source_map,
311 src_comment,
312 tagged_scope,
313 need_ids,
314 marker,
315 )
316 )
317 return anchors
318
319 def _is_need_id_refs_line(self, line: str) -> bool:
320 """Whether a comment line is an ``@need-ids:`` reference rather than a need.
321
322 A line whose text, after comment decoration and whitespace, starts with a
323 configured need-id-refs marker is a reference, and never a one-line need --
324 ubCode's precedence. Without it the default one-line start sequence ``@`` also
325 matched ``// @need-ids: A, B`` and made a need titled ``need-ids: A`` with the id
326 ``B``. A line with anything alphanumeric before the marker is not a reference
327 line here (``// see @need-ids: X`` -- the one-line parser ignores it anyway, and
328 the references on it are still extracted), and a line holding a one-line need
329 before the marker (``[[…]] @need-ids: X``) keeps its need: only a line that
330 STARTS with the marker is withheld from the one-line parser.
331 """
332 for marker in self.analyse_config.need_id_refs_config.markers:
333 marker_idx = line.find(marker)
334 if marker_idx != -1 and not any(
335 char.isalnum() for char in line[:marker_idx]
336 ):
337 return True
338 return False
339
340 def extract_oneline_need(
341 self,
342 text: str,
343 src_comment: SourceComment,
344 oneline_comment_style: OneLineCommentStyle,
345 ) -> Generator[tuple[dict[str, str | list[str] | int], int]]:
346 lines = text.splitlines(keepends=True)
347 row_offset = 0
348 if len(lines) == 1:
349 # single line comment has no newline char in the extracted comment
350 lines[0] = f"{lines[0]}{UNIX_NEWLINE}"
351
352 for line in lines:
353 if self._is_need_id_refs_line(line):
354 row_offset += 1
355 continue
356 resolved = oneline_parser(line, oneline_comment_style)
357 if not resolved:
358 row_offset += 1
359 continue
360 if isinstance(resolved, OnelineParserInvalidWarning):
361 if not src_comment.source_file:
362 row_offset += 1
363 continue
364 lineno = src_comment.node.start_point.row + row_offset + 1
365 warning = AnalyseWarning(
366 str(src_comment.source_file.filepath),
367 lineno,
368 resolved.msg,
369 MarkedContentType.need,
370 resolved.sub_type.value,
371 )
372 self.warnings.append(warning)
373 row_offset += 1
374 continue
375 yield resolved, row_offset
376 row_offset += 1
377
[docs]378 # @Extract one-line traceability needs from comments, IMPL_ONE_1, impl, [FE_DEF, FE_CMT]
379 def extract_oneline_needs(
380 self,
381 text: str,
382 filepath: Path,
383 tagged_scope: TreeSitterNode | None,
384 src_comment: SourceComment,
385 oneline_comment_style: OneLineCommentStyle,
386 ) -> list[OneLineNeed]:
387 row_offset = 0
388 oneline_needs = []
389 for resolved, row_offset in self.extract_oneline_need(
390 text, src_comment, oneline_comment_style
391 ):
392 lineno = src_comment.node.start_point.row + row_offset + 1
393 remote_url = self.git_remote_url
394 if self.git_remote_url and self.git_commit_rev:
395 remote_url = utils.form_https_url(
396 self.git_remote_url,
397 self.git_commit_rev,
398 self.project_path,
399 filepath,
400 lineno,
401 )
402 source_map: SourceMap = {
403 "start": {
404 "row": lineno - 1,
405 "column": resolved[
406 "start_column"
407 ], # dynamic keys # ty: ignore[invalid-argument-type]
408 },
409 "end": {
410 "row": lineno - 1,
411 "column": resolved[
412 "end_column"
413 ], # dynamic keys # ty: ignore[invalid-argument-type]
414 },
415 }
416 del resolved["start_column"]
417 del resolved["end_column"]
418 oneline_needs.append(
419 OneLineNeed(
420 filepath,
421 remote_url,
422 source_map,
423 src_comment,
424 tagged_scope,
425 resolved, # int arguments were deleted # ty: ignore[invalid-argument-type]
426 )
427 )
428 return oneline_needs
429
[docs]430 # @Extract multi-line needs from comment runs, IMPL_MLN_1, impl, [FE_MULTILINE_NEEDS]
431 def extract_multiline_needs(self) -> dict[int, set[int]]:
432 """Extract the multi-line needs of every file, and the rows their blocks claim.
433
434 Runs before the other extractors: the rows a block claims (and those of a
435 refused header, consumed to its close) are hidden from them.
436
437 :return: For each comment holding claimed rows (keyed by ``id()`` of its
438 :class:`SourceComment`), those 0-based rows.
439 """
440 claimed: dict[int, set[int]] = {}
441 config = self.analyse_config.multiline_needs_config
442 for src_file in self.src_files:
443 for run in multiline_parser.form_runs(
444 src_file.src_comments, src_file.lines
445 ):
446 result = multiline_parser.parse_run(
447 run.lines,
448 config,
449 leaderless_block=run.leaderless_block,
450 mixed_leaders=run.mixed_leaders,
451 )
452 for issue in result.issues:
453 self.warnings.append(
454 AnalyseWarning(
455 str(src_file.filepath),
456 issue.row + 1,
457 issue.msg,
458 MarkedContentType.multiline_need,
459 issue.kind.value,
460 )
461 )
462 for block in result.blocks:
463 self.multiline_needs.append(
464 self._multiline_need(block, run, src_file.filepath)
465 )
466 if not result.claimed_rows:
467 continue
468 for src_comment in run.comments:
469 first = src_comment.node.start_point.row
470 rows = result.claimed_rows.intersection(
471 range(first, first + _row_count(src_comment))
472 )
473 if rows:
474 claimed.setdefault(id(src_comment), set()).update(rows)
475 return claimed
476
477 def _multiline_need(
478 self,
479 block: multiline_parser.ParsedBlock,
480 run: multiline_parser.CommentRun,
481 filepath: Path,
482 ) -> MultilineNeed:
483 """The record of one parsed block (see :class:`MultilineNeed`)."""
484 src_comment = next(
485 (
486 comment
487 for comment in run.comments
488 if comment.node.start_point.row
489 <= block.open_row
490 < comment.node.start_point.row + _row_count(comment)
491 ),
492 run.comments[0],
493 )
494 tagged_scope: TreeSitterNode | None = None
495 if not getattr(src_comment.node, "is_libclang", False):
496 tagged_scope = utils.find_associated_scope(
497 src_comment.node, self.analyse_config.comment_type
498 )
499 lineno = block.open_row + 1
500 remote_url = self.git_remote_url
501 if self.git_remote_url and self.git_commit_rev:
502 remote_url = utils.form_https_url(
503 self.git_remote_url,
504 self.git_commit_rev,
505 self.project_path,
506 filepath,
507 lineno,
508 )
509 source_map: SourceMap = {
510 "start": {"row": block.open_row, "column": block.open_col},
511 "end": {"row": block.close_row, "column": block.end_col},
512 }
513 content_start = (
514 {"line": block.content_start[0] + 1, "col": block.content_start[1]}
515 if block.content_start is not None
516 else None
517 )
518 scope = (
519 {
520 "kind": tagged_scope.type,
521 "start": tagged_scope.start_point.row + 1,
522 "end": tagged_scope.end_point.row + 1,
523 }
524 if tagged_scope is not None
525 else None
526 )
527 source: dict[str, Any] = {
528 "project": self.name,
529 "path": _relative_posix(filepath, self.project_path),
530 "root": "git" if self.git_root is not None else "src_dir",
531 "commit": self.git_commit_rev,
532 "start": {"line": lineno, "col": block.open_col},
533 "end": {"line": block.close_row + 1, "col": block.end_col},
534 "content_start": content_start,
535 "option_lines": dict(block.option_rows),
536 "scope": scope,
537 }
538 return MultilineNeed(
539 filepath,
540 remote_url,
541 source_map,
542 src_comment,
543 tagged_scope,
544 block.need,
545 block.markup,
546 source,
547 )
548
549 def extract_marked_content(self) -> None:
550 claimed: dict[int, set[int]] = {}
551 if self.analyse_config.get_multiline_needs:
552 claimed = self.extract_multiline_needs()
553 for src_comment in self.src_comments:
554 text = (
555 src_comment.node.text.decode("utf-8") if src_comment.node.text else None
556 )
557 if not text:
558 continue
559 filepath = (
560 src_comment.source_file.filepath if src_comment.source_file else None
561 )
562 if not filepath:
563 continue
564 claimed_rows = claimed.get(id(src_comment))
565 if claimed_rows:
566 # a block's lines are its own: no one-line need and no reference in them
567 text = multiline_parser.blank_rows(
568 text, src_comment.node.start_point.row, claimed_rows
569 )
570 if getattr(src_comment.node, "is_libclang", False):
571 tagged_scope: TreeSitterNode | None = None
572 else:
573 tagged_scope = utils.find_associated_scope(
574 src_comment.node, self.analyse_config.comment_type
575 )
576 if self.analyse_config.get_need_id_refs:
577 anchors = self.extract_anchors(
578 text, filepath, tagged_scope, src_comment
579 )
580 self.need_id_refs.extend(anchors)
581
582 if self.analyse_config.get_oneline_needs:
583 oneline_needs = self.extract_oneline_needs(
584 text,
585 filepath,
586 tagged_scope,
587 src_comment,
588 self.analyse_config.oneline_comment_style,
589 )
590 self.oneline_needs.extend(oneline_needs)
591
592 def merge_marked_content(self) -> None:
593 self.all_marked_content.extend(self.need_id_refs)
594 self.oneline_needs.sort(key=lambda x: x.source_map["start"]["row"])
595 self.all_marked_content.extend(self.oneline_needs)
596 self.all_marked_content.extend(self.multiline_needs)
597 self.all_marked_content.sort(
598 key=lambda x: (x.filepath, x.source_map["start"]["row"])
599 )
600
601 def dump_marked_content(self, outdir: Path) -> None:
602 output_path = outdir / "marked_content.json"
603 if not output_path.parent.exists():
604 output_path.parent.mkdir(parents=True)
605 to_dump = [
606 marked_content.to_dict() for marked_content in self.all_marked_content
607 ]
608 with output_path.open("w") as f:
609 json.dump(to_dump, f)
610
611 def run(self, *, log_summary: bool = True) -> None:
612 """Parse the source files and extract their markers.
613
614 :param log_summary: Whether to print the per-project summary line; a caller
615 that prints its own (the configuration pass of the Sphinx extension) passes
616 ``False``.
617 """
618 if (
619 self.analyse_config.preprocessor is not None
620 and self.analyse_config.comment_type == CommentType.cpp
621 ):
622 self.create_src_objects_libclang()
623 else:
624 self.create_src_objects()
625 self.extract_marked_content()
626 self.merge_marked_content()
627 if log_summary:
628 self.log_summary()
629
630 def log_summary(self, extra: str = "") -> None:
631 """Emit a per-project marker (default-visible) plus a -v breakdown.
632
633 :param extra: Appended to the default-visible line (the ``src-trace``
634 directive's count of the one-line needs it did not create).
635 """
636 label = f"codelinks [{self.name}]" if self.name else "codelinks"
637 logger.info(
638 f"{label}: {_count(len(self.src_files), 'file')}, "
639 f"{_count(len(self.all_marked_content), 'marker')}{extra}"
640 )
641 logger.debug(
642 f"{label}: {_count(len(self.src_comments), 'comment')}, "
643 f"{_count(len(self.oneline_needs), 'oneline need')}, "
644 f"{_count(len(self.need_id_refs), 'id-ref')}, "
645 f"{_count(len(self.multiline_needs), 'multi-line need')}"
646 )
Back to top