build_notebooklm_sources.py｜完整原始碼

"""Build clean, refreshable text mirrors for the CM108G NotebookLM sources.

The learner-facing HTML remains the web reading experience.  This script builds
one plain-text/Markdown-like source per week for stable native Google Docs files.
It deliberately removes instructor-only blocks and internal governance comments.
"""
from __future__ import annotations

import csv
import hashlib
import re
from datetime import datetime, timezone
from pathlib import Path


ROOT = Path(__file__).resolve().parents[1]
MODULE_ROOT = ROOT / "04_modules"
OUTPUT_ROOT = ROOT / "01_sources" / "notebooklm" / "generated"
MANIFEST = ROOT / "01_sources" / "notebooklm" / "notebooklm_drive_source_manifest.csv"

INSTRUCTOR_ONLY_BLOCK = re.compile(
    r"<!--\s*instructor-only:start\s*-->.*?"
    r"<!--\s*instructor-only:end\s*-->",
    flags=re.I | re.S,
)
HTML_COMMENT = re.compile(r"<!--.*?-->", flags=re.S)
LINK = re.compile(r"\[([^\]]+)\]\(([^)]+)\)")


def between(text: str, start: str, end: str) -> str:
    """Return the text between exact delimiters, failing loudly if absent."""
    if start not in text or end not in text:
        raise ValueError(f"Missing required delimiters: {start!r}, {end!r}")
    return text.split(start, 1)[1].split(end, 1)[0].strip()


def clean(markdown: str) -> str:
    """Remove internal comments and make links readable to a text indexer."""
    cleaned = INSTRUCTOR_ONLY_BLOCK.sub("", markdown)
    cleaned = HTML_COMMENT.sub("", cleaned)
    cleaned = LINK.sub(r"\1 (\2)", cleaned)
    cleaned = re.sub(r"\n{3,}", "\n\n", cleaned)
    return cleaned.strip()


def read_title_and_question(manuscript: str) -> tuple[str, str]:
    title_match = re.search(
        r"^#\s+Week\s+\d+\s*(?::|\|)\s*(.+?)\s*$", manuscript, re.M
    )
    question_match = re.search(
        r"^##\s+Core question\s*\n+\s*(.+?)\s*$", manuscript, re.M
    )
    if not title_match:
        raise ValueError("Could not read unit title")
    if question_match:
        return title_match.group(1).strip(), question_match.group(1).strip()
    # Weeks 2 and 3 predate the later explicit heading.  Reuse the already
    # published page metadata so the source mirror stays aligned with the web
    # unit without inventing a new question.
    week_number = re.search(r"^#\s+Week\s+(\d+)", manuscript, re.M)
    if week_number:
        page = ROOT / "06_web" / "pages" / f"week-{int(week_number.group(1)):02d}.html"
        if page.is_file():
            page_text = page.read_text(encoding="utf-8")
            page_question = re.search(
                r'<p class="dek">(.*?)</p>', page_text, flags=re.S
            )
            if page_question:
                return title_match.group(1).strip(), re.sub(
                    r"<[^>]+>", "", page_question.group(1)
                ).strip()
    raise ValueError("Could not read unit core question")


def build_week(week: int) -> tuple[Path, dict[str, str]]:
    module = MODULE_ROOT / f"week-{week:02d}"
    manuscript = (module / "manuscript.md").read_text(encoding="utf-8")
    title, question = read_title_and_question(manuscript)
    sections = {
        "Reading": clean(
            between(manuscript, "<!-- main-body:start -->", "<!-- main-body:end -->")
        ),
        "Glossary": clean((module / "glossary.md").read_text(encoding="utf-8")),
        "Assessment and reflection": clean(
            (module / "assessment.md").read_text(encoding="utf-8")
        ),
        "Resources and APA 7 references": clean(
            (module / "resources.md").read_text(encoding="utf-8")
        ),
    }
    document = [
        f"# CM108G Become Judy — Week {week}: {title}",
        "",
        f"Core question: {question}",
        "",
        "This is the refreshable NotebookLM reading source for this week. It is "
        "derived from the approved learner-facing course material. The web page "
        "and this source are maintained as separate representations of the same "
        "unit; internal project controls and instructor-only answer material are "
        "not included.",
        "",
    ]
    for heading, body in sections.items():
        document.extend([f"## {heading}", "", body, ""])
    output = OUTPUT_ROOT / f"week-{week:02d}.md"
    output.parent.mkdir(parents=True, exist_ok=True)
    output.write_text("\n".join(document).rstrip() + "\n", encoding="utf-8")
    digest = hashlib.sha256(output.read_bytes()).hexdigest()
    words = len(re.findall(r"\b[\w’'-]+\b", output.read_text(encoding="utf-8")))
    return output, {
        "week": f"week-{week:02d}",
        "title": title,
        "core_question": question,
        "local_mirror": str(output.relative_to(ROOT)).replace("\\", "/"),
        "content_sha256": digest,
        "word_count": str(words),
    }


def preserve_drive_columns(rows: list[dict[str, str]]) -> list[dict[str, str]]:
    """Keep Drive and NotebookLM IDs across rebuilds."""
    existing: dict[str, dict[str, str]] = {}
    if MANIFEST.is_file():
        with MANIFEST.open(encoding="utf-8-sig", newline="") as handle:
            for row in csv.DictReader(handle):
                if row.get("week"):
                    existing[row["week"]] = row
    now = datetime.now(timezone.utc).isoformat(timespec="seconds")
    merged: list[dict[str, str]] = []
    for row in rows:
        old = existing.get(row["week"], {})
        row.update(
            {
                "drive_folder_id": old.get("drive_folder_id", ""),
                "drive_document_id": old.get("drive_document_id", ""),
                "drive_url": old.get("drive_url", ""),
                "notebook_id": old.get("notebook_id", ""),
                "notebook_source_id": old.get("notebook_source_id", ""),
                "last_drive_sync": old.get("last_drive_sync", ""),
                "last_notebooklm_refresh": old.get("last_notebooklm_refresh", ""),
                "status": old.get("status", "pending-drive-provisioning"),
                "last_manifest_build": now,
            }
        )
        merged.append(row)
    return merged


def write_manifest(rows: list[dict[str, str]]) -> None:
    MANIFEST.parent.mkdir(parents=True, exist_ok=True)
    fields = [
        "week",
        "title",
        "core_question",
        "local_mirror",
        "content_sha256",
        "word_count",
        "drive_folder_id",
        "drive_document_id",
        "drive_url",
        "notebook_id",
        "notebook_source_id",
        "last_drive_sync",
        "last_notebooklm_refresh",
        "status",
        "last_manifest_build",
    ]
    with MANIFEST.open("w", encoding="utf-8", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=fields, lineterminator="\n")
        writer.writeheader()
        writer.writerows(rows)


def main() -> None:
    rows = [build_week(week)[1] for week in range(1, 19)]
    write_manifest(preserve_drive_columns(rows))
    print(f"Built {len(rows)} NotebookLM source mirrors")
    print(f"Manifest: {MANIFEST}")


if __name__ == "__main__":
    main()