parse.py
python
sha256:4671b7f787ddbe63ced31c895b688c77ab495653b65a730b423329f26b3c1439
feat: K1-P1 complete — agent provenance, build-verification…
Sonnet 4.6
patch
55 days ago
| 1 | """Parse governance doc claims for drift verification (D1/D3).""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import re |
| 6 | |
| 7 | from tools.governance_hygiene.types import QueueRow |
| 8 | |
| 9 | SHA_RE = r"[0-9a-fA-F]{7,40}" |
| 10 | STATUS_RE = r"\*\*([^*]+)\*\*" |
| 11 | |
| 12 | |
| 13 | def parse_handover_github_main_sha(text: str) -> str | None: |
| 14 | """Extract claimed GitHub ``main`` sha from handover (D1 left-hand side).""" |
| 15 | claims: set[str] = set() |
| 16 | |
| 17 | patterns = [ |
| 18 | rf"GitHub\s+`main`\s*\|\s*`?({SHA_RE})`?", |
| 19 | rf"\*\*main HEAD\*\*\s*\|\s*`({SHA_RE})`", |
| 20 | rf"\*\*Repo state:\*\*\s*`main`\s+at\s+`({SHA_RE})`", |
| 21 | rf"GitHub\s+`main`\s*\|\s*`\s*({SHA_RE})\s*`", |
| 22 | ] |
| 23 | for pattern in patterns: |
| 24 | for match in re.finditer(pattern, text, flags=re.IGNORECASE): |
| 25 | claims.add(match.group(1).lower()) |
| 26 | |
| 27 | if not claims: |
| 28 | return None |
| 29 | if len(claims) > 1: |
| 30 | return None |
| 31 | return next(iter(claims)) |
| 32 | |
| 33 | |
| 34 | def parse_queue_rows(roadmap_text: str) -> list[QueueRow]: |
| 35 | """Parse build-queue table rows from roadmap (D3 claims).""" |
| 36 | rows: list[QueueRow] = [] |
| 37 | in_queue = False |
| 38 | for line in roadmap_text.splitlines(): |
| 39 | if line.strip().startswith("## Build queue"): |
| 40 | in_queue = True |
| 41 | continue |
| 42 | if in_queue and line.startswith("## "): |
| 43 | break |
| 44 | if not in_queue or not line.startswith("|"): |
| 45 | continue |
| 46 | if "---" in line or "Phase |" in line or "Phase|" in line: |
| 47 | continue |
| 48 | cells = [cell.strip() for cell in line.strip("|").split("|")] |
| 49 | if len(cells) < 4: |
| 50 | continue |
| 51 | phase_label = cells[0].strip() |
| 52 | if not phase_label or phase_label.startswith("<"): |
| 53 | continue |
| 54 | rows.append( |
| 55 | QueueRow( |
| 56 | phase_label=phase_label, |
| 57 | model=cells[1].strip(), |
| 58 | status=cells[2].strip(), |
| 59 | deliverable=cells[3].strip(), |
| 60 | raw_line=line, |
| 61 | ) |
| 62 | ) |
| 63 | return rows |
| 64 | |
| 65 | |
| 66 | def normalize_status(status: str) -> str: |
| 67 | """Normalize a queue status cell to uppercase token.""" |
| 68 | cleaned = status.upper() |
| 69 | for token in ("TODO", "WIP", "DONE", "MERGED", "BLOCKED", "NEXT"): |
| 70 | if token in cleaned: |
| 71 | return token |
| 72 | return cleaned |
| 73 | |
| 74 | |
| 75 | def phase_tokens(phase_label: str) -> list[str]: |
| 76 | """Derive searchable tokens from a phase label.""" |
| 77 | tokens: list[str] = [] |
| 78 | bold = re.search(r"\*\*([^*]+)\*\*", phase_label) |
| 79 | label = bold.group(1) if bold else phase_label |
| 80 | tokens.append(label.strip()) |
| 81 | for part in re.split(r"[\s/—\-]+", label): |
| 82 | part = part.strip() |
| 83 | if part and part not in tokens: |
| 84 | tokens.append(part) |
| 85 | return tokens |
| 86 | |
| 87 | |
| 88 | def pr_matches_row(pr_title: str, row: QueueRow) -> bool: |
| 89 | """Return whether a merged PR title plausibly belongs to a queue row.""" |
| 90 | title_lower = pr_title.lower() |
| 91 | for token in phase_tokens(row.phase_label): |
| 92 | if token.lower() in title_lower: |
| 93 | return True |
| 94 | return False |
File History
1 commit
sha256:4671b7f787ddbe63ced31c895b688c77ab495653b65a730b423329f26b3c1439
feat: K1-P1 complete — agent provenance, build-verification…
Sonnet 4.6
patch
55 days ago