|
| 1 | +"""Direct quick maps on the pages kb init routes to. |
| 2 | +
|
| 3 | +Direct's product is the embedded map: for a repository routed by a |
| 4 | +knowledge-side ``_routes.yaml``, every owner page must carry a ``## … Direct …`` |
| 5 | +section (``direct_routing._direct_quick_map_text``: the first such heading up to |
| 6 | +the next ``## ``, non-empty, at most 3500 characters), and |
| 7 | +``test_every_routed_page_yields_a_quick_map`` fails CI on any page that does |
| 8 | +not. The first real run (afd-plugin skeleton, PR #265) routed pages that had |
| 9 | +none. |
| 10 | +
|
| 11 | +kb init therefore renders that section deterministically on every owner page |
| 12 | +of the routes file it writes or extends — the rules on the page as rows, or one |
| 13 | +row for a map card or an entry page — keeps it the first section after the |
| 14 | +title, regenerates it whenever it appends rules to the page (deepen), and |
| 15 | +blocks a change whose routed pages would not yield a map. A map init did not |
| 16 | +write (no ``MARKER``) is somebody's hand-written map and is never rewritten. A |
| 17 | +repository routed by its adapter manifest carries no routes file and is exempt: |
| 18 | +Direct's manifest fallback reports ``read_required`` for its pages. |
| 19 | +
|
| 20 | +Standard library plus the knowledge format; nothing here names a repository. |
| 21 | +""" |
| 22 | + |
| 23 | +from __future__ import annotations |
| 24 | + |
| 25 | +import re |
| 26 | +from dataclasses import replace |
| 27 | +from typing import Iterable, Mapping, Sequence |
| 28 | + |
| 29 | +from ..knowledge_service.facts import CODE_SPAN, PATH |
| 30 | +from ..knowledge_service.lifecycle import LifecycleError, Page, Section |
| 31 | + |
| 32 | +# The ID scanner of the knowledge format (``lifecycle.ANY_RULE_HEADING``) reads |
| 33 | +# ``## Word ...`` as the ID ``Word`` and L1 refuses one ID on two pages, so the |
| 34 | +# heading opens with a word that scanner cannot take for an ID; Direct only |
| 35 | +# needs "Direct" somewhere on the heading line. |
| 36 | +QUICK_MAP_HEADING = "## 代码快速入口（Direct）" |
| 37 | +MARKER = "<!-- kb-init:quick-map -->" |
| 38 | +MAX_CHARS = 3000 # what init renders (rows are dropped past it) |
| 39 | +HARD_CAP = 3500 # what Direct serves whole (``_direct_quick_map`` cap) |
| 40 | +MAX_CELL = 80 |
| 41 | +MAX_SOURCES = 3 |
| 42 | +_DIRECT = re.compile(r"^##\s+.*Direct", re.IGNORECASE) |
| 43 | +_TRIGGER = re.compile(r"^\s*[-*]\s*(?:\*\*)?触发(?:\*\*)?\s*[:：]\s*(?P<text>.+?)\s*$") |
| 44 | +_TITLE = re.compile(r"^#{2,3}\s+[A-Za-z0-9][A-Za-z0-9-]{1,40}\s+[—-]\s+(?P<title>.+?)\s*$") |
| 45 | +_HEADER = ("| PR 描述在做什么 | 精确规则 | 第一批 live 源码 |", "|---|---|---|") |
| 46 | +_MORE = "| … | … | … |" |
| 47 | + |
| 48 | + |
| 49 | +def _cell(text: str, limit: int = MAX_CELL) -> str: |
| 50 | + text = " ".join(str(text).split()).replace("|", "\\|") |
| 51 | + return text if len(text) <= limit else text[:limit - 1].rstrip() + "…" |
| 52 | + |
| 53 | + |
| 54 | +def _paths_in(text: str, limit: int = MAX_SOURCES) -> list[str]: |
| 55 | + """The repository paths (``path`` or ``path::Symbol``) ``text`` names in |
| 56 | + backticks, in order of first appearance.""" |
| 57 | + out: list[str] = [] |
| 58 | + for match in CODE_SPAN.finditer(text): |
| 59 | + token = match.group("tok").strip() |
| 60 | + if PATH.fullmatch(token.split("::", 1)[0]) and token not in out: |
| 61 | + out.append(token) |
| 62 | + if len(out) >= limit: |
| 63 | + break |
| 64 | + return out |
| 65 | + |
| 66 | + |
| 67 | +def _trigger(section: Section) -> str: |
| 68 | + for line in section.text.splitlines()[1:]: |
| 69 | + match = _TRIGGER.match(line) |
| 70 | + if match: |
| 71 | + return match.group("text") |
| 72 | + match = _TITLE.match(section.heading) |
| 73 | + return match.group("title") if match else section.rule_id |
| 74 | + |
| 75 | + |
| 76 | +def _sources(paths: Sequence[str], fallback: Sequence[str]) -> str: |
| 77 | + chosen = list(paths) or list(fallback)[:MAX_SOURCES] |
| 78 | + return "、".join(f"`{p}`" for p in chosen) if chosen else "—" |
| 79 | + |
| 80 | + |
| 81 | +def render_quick_map(page_text: str, *, signals: Sequence[str], prefixes: Sequence[str], |
| 82 | + key_files: Sequence[str] = ()) -> str: |
| 83 | + """The Direct section for ``page_text``: one row per active rule on the |
| 84 | + page (its 触发 line or title, its ID, the paths it names), or one row for |
| 85 | + a page without rules (a map card or an entry page: the owner's signals, |
| 86 | + its key files or scope prefixes). Deterministic; at most ``MAX_CHARS`` |
| 87 | + (rows are dropped from the end behind a ``…`` row), never over |
| 88 | + ``HARD_CAP``.""" |
| 89 | + try: |
| 90 | + rules = [s for s in Page.parse(page_text).rules() if s.footer.status == "active"] |
| 91 | + except LifecycleError: |
| 92 | + rules = [] |
| 93 | + rows: list[str] = [] |
| 94 | + for section in rules: |
| 95 | + rows.append(f"| {_cell(_trigger(section))} | {section.rule_id} | " |
| 96 | + f"{_sources(_paths_in(section.text), prefixes)} |") |
| 97 | + if not rows: |
| 98 | + files = list(key_files) or _paths_in(page_text, MAX_SOURCES) |
| 99 | + rows.append(f"| {_cell('、'.join(signals) or '（无触发词）')} | 入口 | {_sources(files, prefixes)} |") |
| 100 | + lead = (("触发词：" + "、".join(_cell(s, 40) for s in list(signals)[:12]) + "。") if signals else "") |
| 101 | + lead += "PR 描述或改动命中下表一行时，先读该行的规则，再读页面其余部分。" |
| 102 | + |
| 103 | + def render(kept: list[str], more: bool) -> str: |
| 104 | + return "\n".join([QUICK_MAP_HEADING, MARKER, "", lead, "", *_HEADER, *kept, *([_MORE] if more else [])]) |
| 105 | + |
| 106 | + kept, more = list(rows), False |
| 107 | + text = render(kept, more) |
| 108 | + while len(text) > MAX_CHARS and kept: |
| 109 | + kept.pop() |
| 110 | + more = True |
| 111 | + text = render(kept, more) |
| 112 | + if len(text) > HARD_CAP: # pragma: no cover - the lead alone is far under the cap |
| 113 | + raise ValueError(f"quick map cannot fit under {HARD_CAP} characters") |
| 114 | + return text + "\n" |
| 115 | + |
| 116 | + |
| 117 | +def _after_table(section_text: str) -> str: |
| 118 | + """The lines of init's map section that follow ITS table (content the |
| 119 | + page gained after the map was written: an index line, prose, even a |
| 120 | + second table); "" when there are none. The generated table is the run of |
| 121 | + ``|`` lines right after the ``|---|---|---|`` separator, so anything past |
| 122 | + that run — including another table behind prose — is the page's.""" |
| 123 | + lines = section_text.splitlines() |
| 124 | + separator = next((i for i, line in enumerate(lines) if line.strip() == _HEADER[1]), None) |
| 125 | + if separator is None: |
| 126 | + return "" |
| 127 | + end = separator + 1 |
| 128 | + while end < len(lines) and lines[end].startswith("|"): |
| 129 | + end += 1 |
| 130 | + return "\n".join(lines[end:]) |
| 131 | + |
| 132 | + |
| 133 | +def has_map(page_text: str) -> bool: |
| 134 | + """The page carries a Direct section (init's or hand-written).""" |
| 135 | + return any(_DIRECT.match(s.heading) for s in Page.parse(page_text).sections) |
| 136 | + |
| 137 | + |
| 138 | +def has_hand_written_map(page_text: str) -> bool: |
| 139 | + """A Direct section that init did not write (no ``MARKER``).""" |
| 140 | + for section in Page.parse(page_text).sections: |
| 141 | + if _DIRECT.match(section.heading): |
| 142 | + return MARKER not in section.text |
| 143 | + return False |
| 144 | + |
| 145 | + |
| 146 | +def with_quick_map(page_text: str, section_text: str) -> str: |
| 147 | + """``page_text`` with ``section_text`` as its Direct section: init's own |
| 148 | + map replaced in place, a page without one gets it as the first section |
| 149 | + after the title, a hand-written map is left exactly as it is.""" |
| 150 | + page = Page.parse(page_text) |
| 151 | + body = section_text.rstrip("\n") + "\n\n" |
| 152 | + sections = list(page.sections) |
| 153 | + for index, section in enumerate(sections): |
| 154 | + if _DIRECT.match(section.heading): |
| 155 | + if MARKER not in section.text: |
| 156 | + return page_text |
| 157 | + # what the page gained after the map (an index line appended to an |
| 158 | + # entry page, say) belongs to the page, not to the map: keep it |
| 159 | + tail = _after_table(section.text) |
| 160 | + last = index == len(sections) - 1 |
| 161 | + trailing = section.text[len(section.text.rstrip("\n")):] if last else "\n\n" |
| 162 | + rendered = section_text.rstrip("\n") + ("\n\n" + tail.rstrip("\n") if tail.strip() else "") |
| 163 | + sections[index] = Section(rendered + (trailing or "\n")) |
| 164 | + return replace(page, sections=tuple(sections)).render() |
| 165 | + head = page.head if page.head.endswith("\n\n") else page.head.rstrip("\n") + "\n\n" |
| 166 | + if not sections: |
| 167 | + body = section_text.rstrip("\n") + "\n" |
| 168 | + return replace(page, head=head, sections=(Section(body), *sections)).render() |
| 169 | + |
| 170 | + |
| 171 | +def mapped(page_text: str, *, signals: Sequence[str], prefixes: Sequence[str]) -> str | None: |
| 172 | + """``page_text`` with its Direct quick map, or None when the map would |
| 173 | + push the page over the knowledge format's capacity (``page_over_capacity``: |
| 174 | + the caller keeps the page as it is and says so).""" |
| 175 | + from ..knowledge_service.ops import page_over_capacity |
| 176 | + |
| 177 | + text = with_quick_map(page_text, render_quick_map(page_text, signals=signals, prefixes=prefixes)) |
| 178 | + return None if page_over_capacity(text) else text |
| 179 | + |
| 180 | + |
| 181 | +def map_inputs(routes_text: str | None, page: str) -> tuple[list[str], list[str]]: |
| 182 | + """The (signals, scope_prefixes) the routes file gives ``page``'s first |
| 183 | + owner — what its map is rendered with — or empty lists.""" |
| 184 | + import yaml |
| 185 | + |
| 186 | + try: |
| 187 | + doc = yaml.safe_load(routes_text or "") or {} |
| 188 | + except yaml.YAMLError: |
| 189 | + return [], [] |
| 190 | + for owner in (doc.get("owners") or []) if isinstance(doc, dict) else []: |
| 191 | + if isinstance(owner, dict) and str(owner.get("path") or "") == page: |
| 192 | + return ([str(x) for x in owner.get("signals") or []], |
| 193 | + [str(x) for x in owner.get("scope_prefixes") or []]) |
| 194 | + return [], [] |
| 195 | + |
| 196 | + |
| 197 | +def refresh_quick_maps(head: Mapping[str, str], base: Mapping[str, str], routes_text: str | None, *, |
| 198 | + briefing_docs: Iterable[str]) -> tuple[dict[str, str], list[str], list[str]]: |
| 199 | + """``head`` with init's Direct quick map on every owner page of the |
| 200 | + routes file that this change touched or that has no map yet, plus the |
| 201 | + checklist lines and the BLOCKING problems that produced. An owner page |
| 202 | + unchanged since ``base`` that already has a map is left alone (routing |
| 203 | + drift makes no churn); a briefing doc is never written (checklist; the |
| 204 | + presence check then blocks); a page whose regenerated map no longer |
| 205 | + fits the format's capacity is a problem, never a stale map.""" |
| 206 | + out, checklist, problems = dict(head), [], [] |
| 207 | + off_limits = set(briefing_docs) |
| 208 | + try: |
| 209 | + pages = owner_pages(routes_text) |
| 210 | + except ValueError: |
| 211 | + return out, checklist, problems # the routes file is reported by the validators |
| 212 | + for page in pages: |
| 213 | + text = out.get(page) |
| 214 | + if text is None or not page.endswith(".md"): |
| 215 | + continue |
| 216 | + if page in off_limits: |
| 217 | + checklist.append(f"{page} is a briefing doc (knowledge.briefing_docs): init does not write its " |
| 218 | + "Direct quick map; add one by hand or route another page") |
| 219 | + continue |
| 220 | + if text == base.get(page) and has_map(text): |
| 221 | + continue |
| 222 | + signals, prefixes = map_inputs(routes_text, page) |
| 223 | + with_map = mapped(text, signals=signals, prefixes=prefixes) |
| 224 | + if with_map is None: |
| 225 | + problems.append(f"quick map: {page} has no room for its Direct quick map (page capacity): " |
| 226 | + "split the page by hand, or route another page") |
| 227 | + continue |
| 228 | + out[page] = with_map |
| 229 | + return out, checklist, problems |
| 230 | + |
| 231 | + |
| 232 | +def owner_pages(routes_text: str | None) -> list[str]: |
| 233 | + """The owner pages a routes file names, in routing order, each once.""" |
| 234 | + from .init_coverage import load_owners |
| 235 | + |
| 236 | + if routes_text is None: |
| 237 | + return [] |
| 238 | + return list(dict.fromkeys(o.path for o in load_owners(routes_text))) |
| 239 | + |
| 240 | + |
| 241 | +def quick_map_problems(files: Mapping[str, str], pages: Iterable[str], |
| 242 | + base: Mapping[str, str] | None = None) -> list[str]: |
| 243 | + """Why Direct could not serve the routed ``pages`` of ``files``: the same |
| 244 | + extraction the server runs (``_direct_quick_map_text``), page by page. |
| 245 | + Blocking: a route without a map hands the host nothing, and a map the |
| 246 | + server would truncate is not whole — CI refuses newly truncated maps, |
| 247 | + so a page this change wrote or changed (``base``) may not truncate.""" |
| 248 | + from ..direct_routing import _direct_quick_map_text |
| 249 | + |
| 250 | + problems = [] |
| 251 | + for page in pages: |
| 252 | + text = files.get(page) |
| 253 | + if text is None: |
| 254 | + problems.append(f"quick map: routed page {page} does not exist") |
| 255 | + continue |
| 256 | + _, status = _direct_quick_map_text(text) |
| 257 | + if status == "unavailable": |
| 258 | + problems.append(f"quick map: {page} yields no Direct quick map (a routed page needs a " |
| 259 | + f"`{QUICK_MAP_HEADING[3:]}` section with a non-empty body)") |
| 260 | + elif status == "truncated" and (base is None or base.get(page) != text): |
| 261 | + problems.append(f"quick map: {page}: the Direct section as served exceeds {HARD_CAP} characters " |
| 262 | + "(truncated); keep the map first and short, and start the rules with a `## ` " |
| 263 | + "heading so level-three rules are not part of it") |
| 264 | + return problems |
0 commit comments