#!/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()