#!/usr/bin/env python3
"""
Ingest 2nd Grade History curriculum into the brain system.
Parses the 32_week_course_plan.md and wiki entity pages, creates:
- 1 umbrella entity (history)
- 4 phase entities (linked to umbrella)
- 32 topic entities with rich content (linked to phases)
- Edges: part_of, precedes, co_occurs
Usage:
python3 ~/wiki/ingest_history_knowledge.py # Full ingest
python3 ~/wiki/ingest_history_knowledge.py --dry # Preview only
python3 ~/wiki/ingest_history_knowledge.py --query # Run test queries
"""
import sys
import os
import re
import json
from pathlib import Path
from datetime import date
sys.path.insert(0, str(Path.home() / '.hermes' / 'brain'))
import brain
brain._db_connection = None
brain.init_db()
TODAY = date.today().isoformat()
PLAN_PATH = Path.home() / 'Home_School' / '2nd_Grade' / 'History' / '32_week_course_plan.md'
WIKI_DIR = Path.home() / 'wiki' / 'entities'
# ---------------------------------------------------------------------------
# Parsing
# ---------------------------------------------------------------------------
def parse_plan():
"""Parse the 32_week_course_plan.md → dict of week -> topic data."""
if not PLAN_PATH.exists():
print(f"[warn] Plan not found: {PLAN_PATH}")
return {}
content = PLAN_PATH.read_text()
topics = {}
# Split by phase blocks
phase_blocks = re.findall(
r'^## (FALL|WINTER|SPRING|YEAR-END):\s*(.+?)\s*\n\n(.+?)(?=^## |\Z)',
content, re.MULTILINE | re.DOTALL
)
season_to_phase = {"FALL": 1, "WINTER": 2, "SPRING": 3, "YEAR-END": 4}
for season, phase_name, body in phase_blocks:
phase_num = season_to_phase.get(season.strip(), 0)
phase_name = phase_name.strip()
# Extract week entries — split by ### Week markers
week_parts = re.split(r'^(?=### Week \d+:)', body, flags=re.MULTILINE)
for part in week_parts:
m = re.match(r'^### Week (\d+):\s*(.+?)\s*\n', part, re.MULTILINE)
if not m:
continue
week_str = m.group(1)
topic_name = m.group(2).strip()
week = int(week_str)
details = part[m.end():]
# Extract focus
focus_match = re.search(r'\*\*Focus:\*\*\s*(.+)', details)
focus = focus_match.group(1).strip() if focus_match else ""
# Extract objectives
obj_match = re.search(r'\*\*Objectives:\*\*\s*(.+)', details)
objectives = obj_match.group(1).strip() if obj_match else ""
# Extract activities — everything between **Activities:** and **Assessment:**
act_match = re.search(r'\*\*Activities:\*\*\s*\n((?:(?!\*\*Assessment:).)*?)', details, re.DOTALL)
activities = act_match.group(1).strip() if act_match else ""
# Extract assessment
assess_match = re.search(r'\*\*Assessment:\*\*\s*(.+)', details)
assessment = assess_match.group(1).strip() if assess_match else ""
slug = re.sub(r'[^\w\s-]', '', topic_name.lower().strip())
slug = re.sub(r'[\s_]+', '-', slug).strip('-')
topics[week] = {
"week": week,
"topic": topic_name,
"slug": slug,
"phase": phase_num,
"phase_name": f"{season.strip()}: {phase_name}",
"focus": focus,
"objectives": objectives,
"activities": activities,
"assessment": assessment,
}
return topics
def get_wiki_content(slug):
"""Load corresponding wiki file for extra content."""
wiki_file = WIKI_DIR / f"{slug}.md"
if wiki_file.exists():
return wiki_file.read_text()
return None
# ---------------------------------------------------------------------------
# Entity creation
# ---------------------------------------------------------------------------
def create_entity(content, category, entity_type="memory", pinned=False, strength="moderate", metadata=None):
"""Create entity in brain and return its ID."""
eid = brain.create_entity(
content=content,
entity_type=entity_type,
category=category,
strength=strength,
pinned=pinned,
metadata=metadata or {},
log=True,
)
return eid
# ---------------------------------------------------------------------------
# Ingestion
# ---------------------------------------------------------------------------
def ingest_umbrella():
"""Create the history curriculum umbrella entity."""
content = f"""## History Curriculum — 2nd Grade
Comprehensive 32-week history and social studies curriculum for 2nd grade homeschooling.
Covers: time/place foundations, American history, world geography, and historical thinking.
### Structure
- **Phase 1 (FALL):** Time, Place & Community Foundations (Weeks 1–8)
- **Phase 2 (WINTER):** Early America (Weeks 9–16)
- **Phase 3 (SPRING):** American Government & World Geography (Weeks 17–24)
- **Phase 4 (YEAR-END):** Inventions, Famous People, & Civil Rights (Weeks 25–32)
### Pedagogical Approach
- Age-appropriate historical concepts
- Primary source introduction
- Hands-on activities and projects
- Assessment through creation and storytelling
**Ingested:** {TODAY}
**Source:** Home_School/2nd_Grade/History/32_week_course_plan.md
"""
eid = create_entity(
content=content,
category="curriculum/history",
entity_type="curriculum_umbrella",
pinned=True,
strength="strong",
metadata={"source": "history_curriculum_ingest", "date": TODAY},
)
print(f" ✅ Umbrella: curriculum/history → {eid}")
return eid
def ingest_phases(topics, umbrella_id):
"""Create 4 phase entities and link to umbrella."""
phase_weeks = {}
for week, data in sorted(topics.items()):
pn = data["phase"]
if pn not in phase_weeks:
phase_weeks[pn] = {"name": data["phase_name"], "weeks": []}
phase_weeks[pn]["weeks"].append(week)
phase_ids = {}
for pn, pdata in sorted(phase_weeks.items()):
week_range = f"Weeks {min(pdata['weeks'])}–{max(pdata['weeks'])}"
topics_list = ""
for w in sorted(pdata['weeks']):
topics_list += f"- Week {w}: {topics[w]['topic']}\n"
content = f"""## History Phase {pn}: {pdata['name']}
{week_range} — {len(pdata['weeks'])} weeks of 2nd grade history curriculum.
### Topics
{topics_list}
"""
category = f"curriculum/history/phase-{pn}"
eid = create_entity(
content=content,
category=category,
entity_type="curriculum_phase",
pinned=pn <= 2,
strength="strong" if pn <= 2 else "moderate",
metadata={"phase": pn, "weeks": sorted(pdata['weeks']), "source": "history_curriculum_ingest"},
)
phase_ids[pn] = eid
print(f" ✅ Phase {pn}: {pdata['name']} → {eid}")
# Link to umbrella
brain.add_edge(eid, "part_of", umbrella_id)
return phase_ids
def ingest_topics(topics, phase_ids):
"""Create topic entities with rich content."""
topic_ids = {}
for week, data in sorted(topics.items()):
phase = data["phase"]
slug = data["slug"]
# Build entity content
content = f"""## Week {week}: {data['topic']}
**Phase:** {data['phase_name']}
**Focus:** {data['focus']}
### Objectives
{data['objectives']}
### Activities
{data['activities']}
### Assessment
{data['assessment']}
"""
# Check for wiki file with additional content
wiki_content = get_wiki_content(slug)
if wiki_content:
# Extract wikilinks
wikilinks = re.findall(r'\[\[([^\]]+)\]\]', wiki_content)
if wikilinks:
content += "\n### Related Topics\n"
for link in wikilinks[:10]:
content += f"- {link}\n"
category = f"curriculum/history/topic/week-{week}"
# Pin foundational and review topics
pinned = week <= 8 or week in [16, 24, 26, 27, 28, 32]
eid = create_entity(
content=content,
category=category,
entity_type="curriculum_topic",
pinned=pinned,
strength="moderate",
metadata={
"week": week,
"phase": phase,
"topic": data["topic"],
"focus": data["focus"],
"source": "history_curriculum_ingest",
},
)
topic_ids[week] = eid
print(f" ✅ Wk {week}: {data['topic']} → {eid[:12]}...")
return topic_ids
def create_edges(topics, phase_ids, topic_ids):
"""Create relationship edges between entities."""
edge_count = 0
# part_of: topic -> phase
for week, data in topics.items():
phase = data["phase"]
if phase in phase_ids and week in topic_ids:
if brain.add_edge(topic_ids[week], "part_of", phase_ids[phase]):
edge_count += 1
# precedes: sequential weeks
for week in range(1, 32):
if week in topic_ids and (week + 1) in topic_ids:
if brain.add_edge(topic_ids[week], "precedes", topic_ids[week + 1]):
edge_count += 1
# phase precedes
for p in range(1, 4):
if p in phase_ids and (p + 1) in phase_ids:
if brain.add_edge(phase_ids[p], "precedes", phase_ids[p + 1]):
edge_count += 1
print(f" ✅ {edge_count} edges created")
return edge_count
# ---------------------------------------------------------------------------
# Test queries
# ---------------------------------------------------------------------------
def run_test_queries():
"""Verify history knowledge is queryable."""
queries = [
("What are the 4 phases of history curriculum?", "phases"),
("history week 16 american revolution", "revolution"),
("What does the kid learn about timelines?", "timelines"),
("community helpers history", "community"),
("winter history phase", "winter"),
("native american tribes", "native"),
("civil rights movement second grade", "civil rights"),
("maps and directions assessment", "maps"),
("inventions that changed the world", "inventions"),
("history underground railroad", "underground"),
]
print("\n=== TEST QUERIES ===")
passed = 0
for q, keyword in queries:
result = brain.query_events(q, limit=3)
hits = len(result.get('events', [])) if isinstance(result, dict) else 0
status = "✓" if hits > 0 else "✗"
print(f" {status} '{q}' → {hits} hits")
if hits > 0:
passed += 1
print(f"\n Results: {passed}/{len(queries)} queries returned hits")
return passed >= len(queries) * 0.8 # 80% threshold
# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
def main():
mode = sys.argv[1] if len(sys.argv) > 1 else "--ingest"
if mode == "--query":
run_test_queries()
return
print("=== History Curriculum Brain Ingestion ===")
# Parse source plan
print("\n[1] Parsing 32_week_course_plan.md...")
topics = parse_plan()
print(f" Found {len(topics)} topics across 4 phases")
if not topics:
print("[error] No topics found — check plan file path")
return
if mode == "--dry":
print("\n--- DRY RUN ---")
phases_seen = set()
for week, data in sorted(topics.items()):
phase_label = f"P{data['phase']}"
if data['phase'] not in phases_seen:
print(f"\n Phase {data['phase']}: {data['phase_name']}")
phases_seen.add(data['phase'])
print(f" Wk {week}: {data['topic']}")
print(f"\n Total: {len(topics)} topics")
return
# Create umbrella entity
print("\n[2] Creating umbrella entity...")
umbrella_id = ingest_umbrella()
# Ingest phases
print("\n[3] Ingesting phase entities...")
phase_ids = ingest_phases(topics, umbrella_id)
# Ingest topics
print("\n[4] Ingesting topic entities...")
topic_ids = ingest_topics(topics, phase_ids)
# Create edges
print("\n[5] Creating relationship edges...")
edge_count = create_edges(topics, phase_ids, topic_ids)
# Summary
total = 1 + 4 + len(topics) # umbrella + phases + topics
print(f"\n=== DONE ===")
print(f" Entities: {total} (1 umbrella + 4 phases + {len(topics)} topics)")
print(f" Edges: {edge_count}")
# Run tests
print("\n[6] Running test queries...")
run_test_queries()
if __name__ == "__main__":
main()