""" md_io.py -------- 檔案讀寫與 build pipeline（最頂層，依賴全部其他模組）。 公開 API： build_from_config(config_path, force=False) → List[Path] """ from __future__ import annotations from pathlib import Path from typing import List import json from . import md_blocks # 注入 _ROOT_MD 用 from .md_blocks import ( parse_document, assign_ids_and_build_toc, split_blocks_into_sections, ) # --------------------------------------------------------------------------- # 副檔名 → 語言對照 # --------------------------------------------------------------------------- _EXT2LANG: dict[str, str] = { ".py": "python", ".js": "javascript", ".ts": "typescript", ".css": "css", ".html": "markup", ".md": "markdown", ".json": "json", ".yml": "yaml", ".yaml": "yaml", ".tex": "latex", ".r": "r", ".cpp": "cpp", ".c": "c", } # --------------------------------------------------------------------------- # 檔案嵌入工具 # --------------------------------------------------------------------------- def _parse_file_spec(spec: str) -> tuple[str, str | None]: spec = (spec or "").strip() if "::" in spec: p, f = spec.split("::", 1) return p.strip(), (f.strip() or None) return spec, None def _split_md_by_file_markers(md_text: str) -> list[tuple[str, str]]: import re FILE_MARKER_RE = re.compile( r'', re.IGNORECASE ) segs, pos = [], 0 for m in FILE_MARKER_RE.finditer(md_text or ""): if m.start() > pos: segs.append(("text", md_text[pos:m.start()])) gd = m.groupdict() if hasattr(m, "groupdict") else {} file_token = (gd.get("path") or "").strip() if not file_token: file_token = md_text[m.start():m.end()] segs.append(("file", file_token)) pos = m.end() if pos < len(md_text or ""): segs.append(("text", md_text[pos:])) return segs def _resolve_code_file(md_dir: Path, file_token: str) -> Path | None: p = Path(file_token) if not p.is_absolute(): p = (md_dir / file_token).resolve() if not p.exists(): alt = (Path.cwd() / "assets" / "files" / file_token).resolve() if alt.exists(): p = alt return p if p.exists() and p.is_file() else None def _read_code_for_embed(p: Path, max_bytes: int = 300_000) -> tuple[str, str]: ext = p.suffix.lower() lang = _EXT2LANG.get(ext, "text") data = p.read_bytes() note = "" if len(data) > max_bytes: data = data[:max_bytes] note = f"\n\n# [truncated] file too large; showing first {max_bytes} bytes\n" code = data.decode("utf-8", errors="replace") + note return code, lang def _materialize_file_embeds(nodes: list[dict], base_dir: Path) -> list[dict]: out: list[dict] = [] for n in nodes: if not isinstance(n, dict): out.append(n); continue if n.get("type") == "file_embed": file_token = (n.get("relpath") or "").strip() p = _resolve_code_file(base_dir, file_token) if not p: out.append({"type": "code", "filename": file_token, "lang": "text", "code": f"# [missing] cannot locate file: {file_token}"}) else: code, lang = _read_code_for_embed(p) out.append({"type": "code", "filename": p.name, "path": str(p), "lang": lang, "code": code}) elif n.get("type") == "card": n["body"] = _materialize_file_embeds(n.get("body") or [], base_dir) out.append(n) elif n.get("type") == "flow": steps = [] for s in n.get("steps") or []: s = dict(s) s["content"] = _materialize_file_embeds(s.get("content") or [], base_dir) steps.append(s) n["steps"] = steps out.append(n) elif n.get("type") == "solution": n["body"] = _materialize_file_embeds(n.get("body") or [], base_dir) out.append(n) else: out.append(n) return out # --------------------------------------------------------------------------- # 單檔轉換 # --------------------------------------------------------------------------- def _export_one(md_path: Path, out_path: Path) -> Path: text = md_path.read_text(encoding="utf-8") md_dir = md_path.parent md_blocks._ROOT_MD = text segments = _split_md_by_file_markers(text) blocks: list[dict] = [] for kind, payload in segments: if kind == "text": if payload.strip(): blocks.extend(parse_document(payload, root_md=text)) else: file_token, func = _parse_file_spec(payload) p = _resolve_code_file(md_dir, file_token) if not p: blocks.append({"type": "code", "filename": file_token, "lang": "text", "code": f"# [missing] cannot locate file: {file_token}"}) else: code, lang = _read_code_for_embed(p) if func: blocks.append({"type": "pycall", "filename": p.name, "path": str(p), "func": func, "lang": lang, "code": code}) else: blocks.append({"type": "code", "filename": p.name, "path": str(p), "lang": lang, "code": code}) blocks = _materialize_file_embeds(blocks, md_dir) toc = assign_ids_and_build_toc(blocks) sections = split_blocks_into_sections(blocks) payload = {"toc": toc, "blocks": blocks, "sections": sections} out_path.parent.mkdir(parents=True, exist_ok=True) out_path.write_text(json.dumps(payload, ensure_ascii=False, indent=4), encoding="utf-8") return out_path # --------------------------------------------------------------------------- # Build pipeline（對外 API） # --------------------------------------------------------------------------- def build_from_config( config_path: str | Path, force: bool = False, ) -> List[Path]: cfgp = Path(config_path).resolve() base = cfgp.parent cfg = json.loads(cfgp.read_text(encoding="utf-8")) up_to_date = False if force else bool(cfg.get("up_to_date", True)) outputs: List[Path] = [] for t in cfg.get("tasks", []): md = (base / t["md"]).resolve() out = (base / t["out"]).resolve() if not md.exists(): raise FileNotFoundError(f"[build] Markdown not found: {md}") need = True if not force and up_to_date and out.exists(): need = md.stat().st_mtime > out.stat().st_mtime if need or not out.exists(): p = _export_one(md, out) print(f"[build] wrote: {p}") else: print(f"[build] up-to-date: {out}") p = out outputs.append(p) return outputs if __name__ == "__main__": import sys if len(sys.argv) >= 2: build_from_config(sys.argv[1]) else: print("Usage: python md_io.py ")