#!/usr/bin/env python3 """Curriculum ingestion pipeline — extracts Math, ELA, Science topics from curriculum sources and generates wiki entity pages + brain events. Usage: python3 wiki_curriculum_ingest.py # Generate wiki pages python3 wiki_curriculum_ingest.py --brain # Also ingest to brain python3 wiki_curriculum_ingest.py --query # Run test queries """ import os import re import sys import yaml from pathlib import Path from datetime import date WIKI_DIR = Path(os.path.expanduser("~/wiki")) CURRICULUM_DIR = Path(os.path.expanduser("~/Home_School/2nd_Grade")) TODAY = date.today().isoformat() # --------------------------------------------------------------------------- # Slugging helpers # --------------------------------------------------------------------------- def slugify(text): """Convert text to lowercase-hyphen slug.""" text = text.lower().strip() text = re.sub(r'[^\w\s-]', '', text) text = re.sub(r'[\s_]+', '-', text) return text.strip('-') # --------------------------------------------------------------------------- # MATH PARSER — Curriculum_32_Weeks.md # --------------------------------------------------------------------------- def parse_math(): """Parse Curriculum_32_Weeks.md → list of (week, topic, skills, phase, standard)""" plan_path = CURRICULUM_DIR / "Math" / "Curriculum_32_Weeks.md" if not plan_path.exists(): print(f"[warn] Math plan not found: {plan_path}") return [] content = plan_path.read_text() # Extract phase blocks — capture phase number phase_blocks = re.findall(r'^## Phase (\d+):\s*(.+?)\n\n(.+?)(?=^## Phase |\Z)', content, re.MULTILINE | re.DOTALL) results = [] for phase_num_str, phase_name, phase_body in phase_blocks: phase_num = int(phase_num_str) phase_name = phase_name.strip() # Extract week rows from markdown table table_rows = re.findall(r'^\|\s*(\d+)\s*\|\s*(.+?)\s*\|\s*(.+?)\s*\|', phase_body, re.MULTILINE) # Skip header row for week_str, topic, skills in table_rows: try: week = int(week_str) except ValueError: continue # Clean topic/skills topic = topic.strip() skills = skills.strip().replace('|', ';') # Determine phase range phase_ranges = { 1: range(1, 7), 2: range(7, 15), 3: range(15, 21), 4: range(21, 27), 5: range(27, 33), } # Determine standard standards = { 1: "2.NBT.A, 2.OA.A", 2: "2.NBT.A, 2.NBT.B, 2.OA.A", 3: "2.MD.A, 2.MD.B, 2.MD.C, 2.MD.D", 4: "2.G.A", 5: "2.OA.A", } results.append({ "week": week, "topic": topic, "skills": skills, "phase": phase_num, "phase_name": phase_name, "standard": standards.get(phase_num, ""), "subject": "math", }) return results # --------------------------------------------------------------------------- # ELA PARSER — generate_weekN.py docstrings # --------------------------------------------------------------------------- def parse_ela(): """Parse ELA week generator docstrings → list of week topic data.""" ela_dir = CURRICULUM_DIR / "English_Language_Arts" if not ela_dir.exists(): print(f"[warn] ELA dir not found: {ela_dir}") return [] results = [] for week_dir in sorted(ela_dir.glob("Week_*")): if not week_dir.is_dir(): continue # Extract week number week_match = re.search(r'Week_(\d+)', week_dir.name) if not week_match: continue week = int(week_match.group(1)) # Find generator file gens = list(week_dir.glob("generate_week*.py")) if not gens: continue gen_file = gens[0] content = gen_file.read_text() # Extract docstring using ast (handles all quote styles) import ast try: tree = ast.parse(content) docstring = ast.get_docstring(tree) or "" except SyntaxError: docstring = "" # Parse docstring fields phonics_match = re.search(r'Phonics:\s*(.+)', docstring) spelling_match = re.search(r'Spelling:\s*(.+)', docstring) grammar_match = re.search(r'Grammar:\s*(.+)', docstring) # Extract title from docstring first line title_match = re.search(r'Week\s+\d+\s+Generator\s*-\s*(.+)', docstring) topics = [] if phonics_match: topics.append(("phonics", phonics_match.group(1).strip())) if spelling_match: topics.append(("spelling", spelling_match.group(1).strip())) if grammar_match: topics.append(("grammar", grammar_match.group(1).strip())) week_title = title_match.group(1).strip() if title_match else f"Week {week}" results.append({ "week": week, "title": week_title, "topics": topics, "subject": "ela", "source": f"Home_School/2nd_Grade/English_Language_Arts/{week_dir.name}/{gen_file.name}", }) return results # --------------------------------------------------------------------------- # SCIENCE PARSER — 32_week_course_plan.md # --------------------------------------------------------------------------- def parse_science(): """Parse 32_week_course_plan.md → list of week topic data.""" plan_path = CURRICULUM_DIR / "Science" / "32_week_course_plan.md" if not plan_path.exists(): print(f"[warn] Science plan not found: {plan_path}") return [] content = plan_path.read_text() # Split by season/phase blocks — only match FALL, WINTER, SPRING, YEAR-END season_blocks = re.findall(r'^## (FALL|WINTER|SPRING|YEAR-END):\s*(.+?)\s*\n\n(.+?)(?=^## |\Z)', content, re.MULTILINE | re.DOTALL) results = [] season_to_phase = {"FALL": 1, "WINTER": 2, "SPRING": 3, "YEAR-END": 4} for season, phase_name, body in season_blocks: phase_num = season_to_phase.get(season.strip(), 0) phase_name = phase_name.strip() # Extract week entries: ### Week N: Topic week_entries = re.findall(r'^### Week (\d+):\s*(.+)', body, re.MULTILINE) for week_str, topic in week_entries: try: week = int(week_str) except ValueError: continue topic = topic.strip() # Extract focus and objectives week_section = f"### Week {week_str}: {topic}\n" focus_match = re.search(r'\*\*Focus:\*\*\s*(.+)', body[body.find(week_section):], re.DOTALL) focus = focus_match.group(1).strip() if focus_match else "" # Determine standard standards = { 1: "2-PS1, 2-PS4", # Physical Science 2: "2-LS1, 2-LS2", # Life Science 3: "2-ESS2, 2-ESS3", # Earth Science 4: "2-ETS1", # Engineering } results.append({ "week": week, "topic": topic, "skills": focus, "phase": phase_num, "phase_name": f"{season.strip()}: {phase_name}", "standard": standards.get(phase_num, ""), "subject": "science", }) return results # --------------------------------------------------------------------------- # HISTORY PARSER — 32_week_course_plan.md # --------------------------------------------------------------------------- def parse_history(): """Parse 32_week_course_plan.md → list of week topic data.""" plan_path = CURRICULUM_DIR / "History" / "32_week_course_plan.md" if not plan_path.exists(): print(f"[warn] History plan not found: {plan_path}") return [] content = plan_path.read_text() # Split by season/phase blocks — only match FALL, WINTER, SPRING, YEAR-END season_blocks = re.findall(r'^## (FALL|WINTER|SPRING|YEAR-END):\s*(.+?)\s*\n\n(.+?)(?=^## |\Z)', content, re.MULTILINE | re.DOTALL) results = [] season_to_phase = {"FALL": 1, "WINTER": 2, "SPRING": 3, "YEAR-END": 4} for season, phase_name, body in season_blocks: phase_num = season_to_phase.get(season.strip(), 0) phase_name = phase_name.strip() # Extract week entries: ### Week N: Topic week_entries = re.findall(r'^### Week (\d+):\s*(.+)', body, re.MULTILINE) for week_str, topic in week_entries: try: week = int(week_str) except ValueError: continue topic = topic.strip() # Extract focus line week_section = f"### Week {week_str}: {topic}\n" focus_match = re.search(r'\*\*Focus:\*\*\s*(.+)', body[body.find(week_section):], re.DOTALL) focus = focus_match.group(1).strip() if focus_match else "" results.append({ "week": week, "topic": topic, "skills": focus, "phase": phase_num, "phase_name": f"{season.strip()}: {phase_name}", "standard": "", "subject": "history", }) return results # --------------------------------------------------------------------------- # WIKI PAGE GENERATION # --------------------------------------------------------------------------- def generate_lesson_topic_page(topic_data, subject, existing_topics): """Generate a single lesson topic wiki page.""" topic_name = topic_data["topic"] slug = slugify(topic_name) # Deduplicate — check if this topic already has a page if slug in existing_topics: existing = existing_topics[slug] existing["weeks"].append(topic_data["week"]) existing["week_entries"].append(topic_data) return existing_topics[slug] skills = topic_data.get("skills", "") standard = topic_data.get("standard", "") # Build wikilinks to related topics (same phase/subject) related = [] for other_slug, other_data in existing_topics.items(): if other_data["subject"] == subject and other_data.get("phase") == topic_data.get("phase"): if other_slug != slug: related.append(other_slug) # Determine tags tags = [subject, "grade-2", "homeschool"] if standard: tags.append("ccss" if subject == "math" else "ngss") if topic_data.get("phase") in [6, 14, 20, 26, 32]: tags.append("review") # Build description based on subject if subject == "math": description = (f"Second grade math topic covering {topic_name.lower()}. " f"Part of Phase {topic_data.get('phase', '?')} ({topic_data.get('phase_name', 'N/A')}).") key_concepts = skills.replace(";", ", ") elif subject == "ela": description = f"Second grade ELA topic: {topic_name}." key_concepts = skills if skills else "" else: # science or history description = (f"Second grade {subject} topic covering {topic_name.lower()}. " f"Part of Phase {topic_data.get('phase', '?')} ({topic_data.get('phase_name', 'N/A')}).") key_concepts = skills.replace(";", ", ") frontmatter = yaml.dump({ "title": topic_name, "created": TODAY, "updated": TODAY, "type": "lesson_topic", "subject": subject, "tags": tags, "standard": standard, "phase": topic_data.get("phase"), "phase_name": topic_data.get("phase_name", ""), "weeks": [topic_data["week"]], "source": topic_data.get("source", ""), }, default_flow_style=False, sort_keys=False) body = f"""# {topic_name} {description} ## Key Concepts {key_concepts} ## Standards {standard or "Not specified"} """ # Prerequisites: link to previous phase topics prev_phase = topic_data.get("phase", 1) - 1 if prev_phase > 0: prereq_links = [] for other_slug, other_data in existing_topics.items(): if other_data["subject"] == subject and other_data.get("phase") == prev_phase: prereq_links.append(f"[[{other_slug}]]") if prereq_links: body += "## Prerequisites\n" + ", ".join(prereq_links[-5:]) + "\n\n" # Limit to 5 # Related topics if related: body += "## Related Topics\n" + ", ".join(f"[[{s}]]" for s in related[:10]) + "\n\n" body += f"---\n*Generated from curriculum data* ^[[raw/curriculum/{subject}]]\n" page_data = { "slug": slug, "subject": subject, "phase": topic_data.get("phase"), "content": f"---\n{frontmatter.strip()}\n---\n\n{body}", "weeks": [topic_data["week"]], "week_entries": [topic_data], } existing_topics[slug] = page_data return page_data def generate_week_schedule_page(week_num, subject, week_topics): """Generate a week schedule page linking all topics for that week.""" slug = f"week-{week_num}-{subject}" # Build topic links topic_links = [] for topic_name in week_topics: topic_slug = slugify(topic_name) topic_links.append(f"[[{topic_slug}]]") frontmatter = yaml.dump({ "title": f"Week {week_num} — {subject.title()}", "created": TODAY, "updated": TODAY, "type": "week_schedule", "subject": subject, "tags": [subject, "grade-2", "homeschool", "week-schedule"], "weeks": [week_num], }, default_flow_style=False, sort_keys=False) body = f"""# Week {week_num} — {subject.title()} ## Topics Covered """ for i, topic_name in enumerate(week_topics, 1): body += f"{i}. [[{slugify(topic_name)}]]\n" body += f""" ## Week Structure - **Monday:** New concept introduction - **Wednesday:** Mixed practice - **Friday:** Review & challenge --- *Generated from curriculum data* ^[[raw/curriculum/{subject}]] """ return slug, f"---\n{frontmatter.strip()}\n---\n\n{body}" def generate_skill_category_pages(all_topics_by_subject): """Generate skill category pages grouping topics by skill area.""" pages = {} # Math skill categories math_skills = { "number-sense": ["Counting", "Place Value", "Number Sense", "Comparing", "Ordering"], "addition": ["Addition", "Adding", "Repeated Addition"], "subtraction": ["Subtraction", "Subtracting"], "measurement": ["Length", "Measuring", "Measurement", "Time", "Money", "Data"], "geometry": ["Shapes", "Geometry", "Lines", "Angles", "Symmetry", "Area", "Perimeter"], "fractions": ["Fractions", "Equal Parts"], "multiplication": ["Multiplication", "Arrays", "Equal Groups"], "division": ["Division", "Equal Sharing"], } for skill, keywords in math_skills.items(): matching = [s for s, d in all_topics_by_subject.get("math", {}).items() if any(kw.lower() in s.lower() for kw in keywords)] if not matching: continue frontmatter = yaml.dump({ "title": f"Math: {skill.replace('-', ' ').title()}", "created": TODAY, "updated": TODAY, "type": "skill_category", "subject": "math", "tags": ["math", "grade-2", "homeschool", skill], }, default_flow_style=False, sort_keys=False) body = f"""# Math: {skill.replace('-', ' ').title()} Topics that develop {skill.replace('-', ' ')} skills in second grade. ## Topics """ for topic_slug in sorted(matching): body += f"- [[{topic_slug}]]\n" body += f"\n---\n*Generated from curriculum data* ^[[raw/curriculum/math]]\n" pages[skill] = f"---\n{frontmatter.strip()}\n---\n\n{body}" # ELA skill categories ela_skills = { "phonics": ["Phonics", "Vowel", "Digraph", "Silent E", "Soft C", "R-Controlled"], "grammar": ["Grammar", "Sentence", "Verb", "Adjective", "Adverb", "Punctuation", "Capitalization"], "vocabulary": ["Synonym", "Antonym", "Multiple Meaning", "Word Family"], } for skill, keywords in ela_skills.items(): matching = [s for s, d in all_topics_by_subject.get("ela", {}).items() if any(kw.lower() in s.lower() or kw.lower() in (d.get("description", "")).lower() for kw in keywords)] if not matching: continue frontmatter = yaml.dump({ "title": f"ELA: {skill.title()}", "created": TODAY, "updated": TODAY, "type": "skill_category", "subject": "ela", "tags": ["ela", "grade-2", "homeschool", skill], }, default_flow_style=False, sort_keys=False) body = f"""# ELA: {skill.title()} Topics that develop {skill} skills in second grade. ## Topics """ for topic_slug in sorted(matching): body += f"- [[{topic_slug}]]\n" body += f"\n---\n*Generated from curriculum data* ^[[raw/curriculum/ela]]\n" pages[skill] = f"---\n{frontmatter.strip()}\n---\n\n{body}" # Science skill categories science_skills = { "physical-science": ["Physical Science", "Matter", "Energy", "Sound", "Light"], "life-science": ["Life Science", "Plants", "Animals", "Habitat", "Ecosystem"], "earth-science": ["Earth Science", "Weather", "Water", "Erosion", "Rocks", "Earth"], "engineering": ["Engineering", "Design", "Problem Solving"], } for skill, keywords in science_skills.items(): matching = [s for s, d in all_topics_by_subject.get("science", {}).items() if any(kw.lower() in s.lower() for kw in keywords)] if not matching: continue frontmatter = yaml.dump({ "title": f"Science: {skill.replace('-', ' ').title()}", "created": TODAY, "updated": TODAY, "type": "skill_category", "subject": "science", "tags": ["science", "grade-2", "homeschool", skill], }, default_flow_style=False, sort_keys=False) body = f"""# Science: {skill.replace('-', ' ').title()} Topics that develop {skill.replace('-', ' ')} understanding in second grade. ## Topics """ for topic_slug in sorted(matching): body += f"- [[{topic_slug}]]\n" body += f"\n---\n*Generated from curriculum data* ^[[raw/curriculum/science]]\n" pages[skill] = f"---\n{frontmatter.strip()}\n---\n\n{body}" return pages # --------------------------------------------------------------------------- # BRAIN EVENTS # --------------------------------------------------------------------------- def build_curriculum_events(all_topics_by_subject, week_schedules): """Build brain events for curriculum data.""" events = [] for subject, topics in all_topics_by_subject.items(): for slug, page_data in topics.items(): for entry in page_data.get("week_entries", []): events.append({ "type": "lesson_topic", "subject": subject, "slug": slug, "topic": entry["topic"], "week": entry["week"], "skills": entry.get("skills", ""), "phase": entry.get("phase"), "phase_name": entry.get("phase_name", ""), "standard": entry.get("standard", ""), "tags": [subject, "grade-2", "homeschool", slug], }) for slug, content in week_schedules.items(): events.append({ "type": "week_schedule", "slug": slug, "tags": ["week-schedule", "grade-2", "homeschool"], }) return events # --------------------------------------------------------------------------- # MAIN # --------------------------------------------------------------------------- def main(): mode = sys.argv[1] if len(sys.argv) > 1 else "--generate" # Parse all subjects print("=== Parsing Curriculum Data ===") math_data = parse_math() print(f" Math: {len(math_data)} week topics") ela_data = parse_ela() print(f" ELA: {len(ela_data)} weeks") science_data = parse_science() print(f" Science: {len(science_data)} week topics") history_data = parse_history() print(f" History: {len(history_data)} week topics") # Generate lesson topic pages print("\n=== Generating Lesson Topic Pages ===") all_topics_by_subject = {"math": {}, "ela": {}, "science": {}, "history": {}} total_topics = 0 # Math topics for entry in math_data: page = generate_lesson_topic_page(entry, "math", all_topics_by_subject["math"]) total_topics += 1 # ELA topics (each week has multiple sub-topics: phonics, spelling, grammar) for entry in ela_data: for topic_type, topic_desc in entry["topics"]: topic_entry = { "topic": topic_desc.split(":")[0].strip() if ":" in topic_desc else topic_desc.strip(), "skills": topic_desc, "week": entry["week"], "source": entry["source"], "phase": 0, } page = generate_lesson_topic_page(topic_entry, "ela", all_topics_by_subject["ela"]) total_topics += 1 # Science topics for entry in science_data: page = generate_lesson_topic_page(entry, "science", all_topics_by_subject["science"]) total_topics += 1 # History topics for topic_data in history_data: generate_lesson_topic_page(topic_data, "history", all_topics_by_subject["history"]) total_topics += 1 print(f" Total unique topics: {sum(len(v) for v in all_topics_by_subject.values())}") print(f" Total unique topics: {sum(len(v) for v in all_topics_by_subject.values())}") # Generate week schedule pages print("\n=== Generating Week Schedule Pages ===") week_schedules = {} # Math week schedules math_weeks = {} for entry in math_data: w = entry["week"] if w not in math_weeks: math_weeks[w] = [] math_weeks[w].append(entry["topic"]) for week, topics in sorted(math_weeks.items()): slug, content = generate_week_schedule_page(week, "math", topics) week_schedules[slug] = content # ELA week schedules for entry in ela_data: topics = [t[1].split(":")[0].strip() for t in entry["topics"]] slug, content = generate_week_schedule_page(entry["week"], "ela", topics) week_schedules[slug] = content # Science week schedules science_weeks = {} for entry in science_data: w = entry["week"] if w not in science_weeks: science_weeks[w] = [] science_weeks[w].append(entry["topic"]) for week, topics in sorted(science_weeks.items()): slug, content = generate_week_schedule_page(week, "science", topics) week_schedules[slug] = content # History week schedules history_weeks = {} for entry in history_data: w = entry["week"] if w not in history_weeks: history_weeks[w] = [] history_weeks[w].append(entry["topic"]) for week, topics in sorted(history_weeks.items()): slug, content = generate_week_schedule_page(week, "history", topics) week_schedules[slug] = content print(f" Week schedules: {len(week_schedules)}") # Generate skill category pages print("\n=== Generating Skill Category Pages ===") skill_pages = generate_skill_category_pages(all_topics_by_subject) print(f" Skill categories: {len(skill_pages)}") # Write all pages if mode in ("--generate", "--brain", ""): print("\n=== Writing Wiki Pages ===") entities_dir = WIKI_DIR / "entities" entities_dir.mkdir(parents=True, exist_ok=True) written = 0 for subject, topics in all_topics_by_subject.items(): for slug, page_data in topics.items(): path = entities_dir / f"{slug}.md" path.write_text(page_data["content"]) written += 1 for slug, content in week_schedules.items(): path = entities_dir / f"{slug}.md" path.write_text(content) written += 1 for slug, content in skill_pages.items(): path = entities_dir / f"{slug}.md" path.write_text(content) written += 1 print(f" Written: {written} pages to {entities_dir}") # Ingest to brain if mode == "--brain": print("\n=== Ingesting to Brain ===") sys.path.insert(0, os.path.expanduser("~/.hermes/brain")) from brain import log_event events = build_curriculum_events(all_topics_by_subject, week_schedules) for event in events: log_event( event_type="curriculum_ingest", args=event, tags=event["tags"], skip_synapse=True, ) print(f" Ingested: {len(events)} events to brain") elif mode == "--query": print("\n=== Test Queries ===") run_test_queries() print("\nDone.") def run_test_queries(): """Run test queries against the generated wiki pages.""" import sqlite3 # Build simple search index from entity files entities_dir = WIKI_DIR / "entities" if not entities_dir.exists(): print(" No entities dir found. Run --generate first.") return # Collect all pages pages = {} for p in entities_dir.glob("*.md"): content = p.read_text() fm_match = re.match(r'^---\n(.*?)\n---\n', content, re.DOTALL) if fm_match: fm = yaml.safe_load(fm_match.group(1)) pages[p.stem] = fm print(f" Loaded {len(pages)} pages") # Query tests tests = [ ("Math topics in Phase 1", lambda: [s for s, d in pages.items() if d.get("subject") == "math" and d.get("phase") == 1]), ("ELA phonics topics", lambda: [s for s, d in pages.items() if d.get("subject") == "ela" and "phonics" in d.get("tags", [])]), ("Science Phase 2 topics", lambda: [s for s, d in pages.items() if d.get("subject") == "science" and d.get("phase") == 2]), ("History topics", lambda: [s for s, d in pages.items() if d.get("subject") == "history"]), ("History Phase 1 topics", lambda: [s for s, d in pages.items() if d.get("subject") == "history" and d.get("phase") == 1]), ("All week schedules", lambda: [s for s, d in pages.items() if d.get("type") == "week_schedule"]), ("Math topics with standard 2.MD", lambda: [s for s, d in pages.items() if d.get("subject") == "math" and "2.MD" in d.get("standard", "")]), ("Skill categories", lambda: [s for s, d in pages.items() if d.get("type") == "skill_category"]), ("Week 1 topics (all subjects)", lambda: [s for s, d in pages.items() if d.get("type") == "week_schedule" and "week-1" in s]), ] for test_name, test_fn in tests: results = test_fn() status = "✅" if results else "⚠️" print(f" {status} {test_name}: {len(results)} results") if results and len(results) <= 5: for r in results: print(f" - {r}") if __name__ == "__main__": main()