Spaces:
Sleeping
Sleeping
| """Parse `## References` markdown tables on syntopicon entry files into | |
| structured reference dicts, resolved against a (session_date, row_index) | |
| lookup map produced by the lecture chunker. | |
| """ | |
| from __future__ import annotations | |
| import logging | |
| import re | |
| from pathlib import Path | |
| logger = logging.getLogger(__name__) | |
| _DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$") | |
| _PROBLEM_ID_RE = re.compile(r"^P\d+$") | |
| def parse_row_spec(rows_raw: str) -> list[int]: | |
| """Parse a Row(s) cell. Supports '159', '159, 232', '159-162'. | |
| Returns sorted unique list. Whitespace tolerated. Returns empty list on | |
| unparseable input. | |
| """ | |
| s = rows_raw.strip() | |
| if not s: | |
| return [] | |
| if "-" in s and "," not in s: | |
| # Range form | |
| try: | |
| lo, hi = s.split("-", 1) | |
| return list(range(int(lo.strip()), int(hi.strip()) + 1)) | |
| except ValueError: | |
| return [] | |
| if "," in s: | |
| try: | |
| return sorted({int(p.strip()) for p in s.split(",") if p.strip()}) | |
| except ValueError: | |
| return [] | |
| try: | |
| return [int(s)] | |
| except ValueError: | |
| return [] | |
| def parse_references( | |
| markdown_path: Path, | |
| row_lookup: dict[tuple[str, int], tuple[str, str]], | |
| ) -> list[dict]: | |
| """Parse the ## References table in a syntopicon markdown file. | |
| Returns a list of reference dicts. Each dict has session_date, speaker, | |
| row_index, how_used, notes, problem_id (if notes match /^P\\d+$/), | |
| lecture_turn_doc_id, and lecture_problem_doc_id. | |
| Unresolvable rows are logged at WARNING and skipped. Malformed rows and | |
| bad dates are also skipped, with logging. Files without ## References | |
| return [] silently. | |
| """ | |
| text = markdown_path.read_text(encoding="utf-8") | |
| lines = text.splitlines() | |
| # Find ## References heading | |
| in_section = False | |
| in_table = False | |
| skipped_separator = False | |
| refs: list[dict] = [] | |
| for line in lines: | |
| stripped = line.strip() | |
| if stripped.lower().startswith("## references"): | |
| in_section = True | |
| continue | |
| if not in_section: | |
| continue | |
| if stripped.startswith("## "): | |
| # Different section; stop | |
| break | |
| if not stripped: | |
| if in_table: | |
| # Blank line ends table | |
| break | |
| continue | |
| if not in_table: | |
| # Skip until table header | |
| if stripped.startswith("|"): | |
| in_table = True | |
| continue | |
| if not skipped_separator: | |
| # The line right after the header is the separator |---|---| | |
| if "---" in stripped: | |
| skipped_separator = True | |
| continue | |
| # No separator? still skip if it looks like one; otherwise treat as data | |
| if stripped.startswith("|"): | |
| skipped_separator = True | |
| # fall through to parse this row as data? No — skip header again. | |
| continue | |
| # Parse a data row | |
| cells = [c.strip() for c in stripped.strip("|").split("|")] | |
| if len(cells) != 5: | |
| logger.warning( | |
| "syntopicon_reference_parser: malformed row in %s (got %d cells, want 5): %r", | |
| markdown_path, len(cells), stripped, | |
| ) | |
| continue | |
| session_date, speaker, rows_raw, how_used, notes = cells | |
| if not _DATE_RE.match(session_date): | |
| logger.warning( | |
| "syntopicon_reference_parser: bad date format %r in %s", | |
| session_date, markdown_path, | |
| ) | |
| continue | |
| row_indices = parse_row_spec(rows_raw) | |
| problem_id = notes if _PROBLEM_ID_RE.match(notes.strip()) else None | |
| for row_index in row_indices: | |
| key = (session_date, row_index) | |
| if key not in row_lookup: | |
| logger.warning( | |
| "syntopicon_reference_parser: unresolvable row %s in %s (row=%d)", | |
| session_date, markdown_path, row_index, | |
| ) | |
| continue | |
| turn_id, problem_did = row_lookup[key] | |
| refs.append({ | |
| "session_date": session_date, | |
| "speaker": speaker, | |
| "row_index": row_index, | |
| "how_used": how_used, | |
| "notes": notes, | |
| "problem_id": problem_id, | |
| "lecture_turn_doc_id": turn_id, | |
| "lecture_problem_doc_id": problem_did, | |
| }) | |
| if in_section and not refs: | |
| logger.warning( | |
| "syntopicon_reference_parser: %s has '## References' heading but no resolvable rows", | |
| markdown_path, | |
| ) | |
| return refs | |
| def parse_all_references( | |
| syntopicon_dir: Path, | |
| row_lookup: dict[tuple[str, int], tuple[str, str]], | |
| ) -> dict[str, list[dict]]: | |
| """Walk the syntopicon directory; for each *.md file (excluding _registry.json | |
| and other non-term files), parse references. Returns {term_slug: refs}. | |
| The term_slug is the markdown filename stem (e.g., 'demand' for 'demand.md'). | |
| """ | |
| out: dict[str, list[dict]] = {} | |
| for md_path in sorted(syntopicon_dir.glob("*.md")): | |
| if md_path.stem.startswith("_"): | |
| continue | |
| out[md_path.stem] = parse_references(md_path, row_lookup) | |
| return out | |