#!/usr/bin/env python3 """ season1.py — Season 1 Pipeline MCP v1.0 "STT 대화 → PHL 토큰 → MCP 체인 → 3종 아티팩트 → 출판" OrbitPrompt의 edu-mcp + political-mcp + philosophy-counter-mcp를 하나의 파이프라인으로 묶어서, 하나의 주제/대화에서 3개 산출물을 동시 생산한다. 실행: python3 phl-engine/season1.py """ from __future__ import annotations from mcp.server.fastmcp import FastMCP from pydantic import BaseModel, Field from typing import Optional import json, os, sys, pathlib, datetime, subprocess, textwrap, html as htmlmod HERE = pathlib.Path(__file__).resolve().parent ROOT = HERE.parent mcp = FastMCP("season1-pipeline") # ── MCP 서브프로세스 호출 ────────────────────────────────────────── def _call_mcp(server_path: str, tool: str, params: dict = None) -> dict: """MCP 서버에 JSON-RPC 요청을 보내 결과 반환""" full_path = ROOT / server_path if not full_path.exists(): return {"error": f"서버 없음: {full_path}"} payload = { "tool": tool, "params": params or {} } try: result = subprocess.run( [sys.executable, str(full_path)], input=json.dumps(payload), capture_output=True, text=True, timeout=30, cwd=str(full_path.parent) ) if result.returncode != 0: return {"error": result.stderr[:500]} out = result.stdout.strip() if not out: return {"error": "빈 응답"} return json.loads(out) except subprocess.TimeoutExpired: return {"error": "timeout (30s)"} except json.JSONDecodeError as e: return {"error": f"JSON parse: {e}", "raw": result.stdout[:500] if 'result' in dir() else ""} except Exception as e: return {"error": str(e)} def _call_political(tool: str, params: dict = None) -> dict: return _call_mcp("political-mcp/server.py", tool, params) def _call_philosophy(tool: str, params: dict = None) -> dict: return _call_mcp("philosophy-counter/mcp/server.py", tool, params) def _call_edu(tool: str, params: dict = None) -> dict: return _call_mcp("edu-mcp/mcp/server.py", tool, params) # ── 템플릿 엔진 ──────────────────────────────────────────────────── def _render_article_html(title: str, subtitle: str, sections: list) -> str: """생성된 데이터로 boards/용 HTML 페이지 렌더링""" sections_html = "" for sec in sections: sec_type = sec.get("type", "text") sec_title = sec.get("title", "") sec_body = sec.get("body", "") if sec_type == "hero" or sec_type == "header": sections_html += f"""

{sec_title}

""" elif sec_type == "text": sections_html += f"""

{sec_title}

{sec_body}
""" elif sec_type == "table": rows = sec.get("rows", []) table_rows = "" for r in rows: cells = "".join(f"{c}" for c in (r if isinstance(r, list) else [r.get("k",""), r.get("v","")])) table_rows += f"{cells}" sections_html += f"""

{sec_title}

{table_rows}
""" elif sec_type == "code": sections_html += f"""

{sec_title}

{htmlmod.escape(sec_body)}
""" elif sec_type == "quote": sections_html += f"""
{sec_body}
{f'{sec.get("cite","")}' if sec.get("cite") else ''}
""" return f""" {htmlmod.escape(title)} — Season1 Episode 1

{htmlmod.escape(title)}

{htmlmod.escape(subtitle)}

{sections_html}

Generated by Season1 Pipeline · {datetime.datetime.now().strftime("%Y-%m-%d %H:%M")}

OrbitPrompt · PHL Engine · edu-mcp · political-mcp · philosophy-counter-mcp

""" def _render_script_text(title: str, chapters: list) -> str: """YouTube 영상 스크립트 생성""" lines = [] lines.append(f"# {title}") lines.append(f"# Script generated by Season1 Pipeline") lines.append(f"# {datetime.datetime.now().isoformat()}") lines.append("") lines.append("## [시즌1 1화 — 대화로 갈아버린 한국 _____]") lines.append("") for i, ch in enumerate(chapters, 1): ch_title = ch.get("title", f"Chapter {i}") ch_body = ch.get("body", "") ch_dur = ch.get("duration_sec", 60) lines.append(f"### Chapter {i}: {ch_title} ({ch_dur}초)") lines.append("") lines.append(ch_body) lines.append("") lines.append("") lines.append("---") lines.append("*Generated by OrbitPrompt Season1 Pipeline*") return "\n".join(lines) def _render_whitepaper(title: str, sections: list) -> str: """기술/철학 백서 초안 생성""" lines = [] lines.append(f"# {title}") lines.append("") lines.append("> **백서 v1.0** · Season1 Pipeline · OrbitPrompt") lines.append(f"> **생성일:** {datetime.datetime.now().strftime('%Y-%m-%d')}") lines.append("") lines.append("---") lines.append("") for sec in sections: sec_title = sec.get("title", "") sec_body = sec.get("body", "") lines.append(f"## {sec_title}") lines.append("") lines.append(sec_body) lines.append("") lines.append("---") lines.append("*이 백서는 Season1 Pipeline MCP에 의해 자동 생성되었습니다.*") lines.append("*소스: edu-mcp · political-mcp · philosophy-counter-mcp*") return "\n".join(lines) # ── 메인 파이프라인 로직 ────────────────────────────────────────── def _run_pipeline(topic: str, thesis: str, sub_themes: list = None) -> dict: """ 하나의 주제로 3개 MCP를 동시에 돌려서 3종 아티팩트 데이터를 만든다. 1. edu-mcp: 교육 시스템 평가 2. political-mcp: 정치 구조 분석 3. philosophy-counter-mcp: 철학적 해석 + 박씨 비교 Returns: dict with keys: edu_result, political_result, philosophy_result, synthesis """ result = {} # 1) Edu MCP — 교육 시스템 평가 try: edu = _call_edu("evaluate_education_system", { "K": {"curriculum": 55, "critical": 40, "practical": 30}, "C": {"competition": 85, "rank": 90, "pressure": 80}, "I": {"public": 50, "private": 70, "edtech": 35}, "O_t": {"graduates": 60, "global_rank": 45, "research": 50}, "B_human": {"parents": 75, "teachers": 40, "students": 55} }) result["edu"] = edu except Exception as e: result["edu"] = {"error": str(e)} # 2) Political MCP — 정치 구조 분석 (thesis 기반) try: political = _call_political("analyze_politician", {"name": thesis}) result["political"] = political except Exception as e: result["political"] = {"error": str(e)} # 3) Philosophy Counter MCP — 철학자 비교 + 박씨 유사도 try: phi_rank = _call_philosophy("rank", {}) result["philosophy_rank"] = phi_rank except Exception as e: result["philosophy_rank"] = {"error": str(e)} # 4) 합성 — 3개 MCP 결과를 하나로 묶어서 인사이트 추출 synthesis = _synthesize(topic, thesis, result) result["synthesis"] = synthesis return result def _synthesize(topic: str, thesis: str, pipeline_result: dict) -> dict: """3개 MCP 결과를 하나의 서사로 합성""" edu = pipeline_result.get("edu", {}) political = pipeline_result.get("political", {}) phi_rank = pipeline_result.get("philosophy_rank", {}) # 교육 평가 요약 edu_summary = "교육 시스템 분석 데이터 없음" if isinstance(edu, dict) and "error" not in edu: edu_summary = f"교육 평가 완료 (엔진 응답)" # 정치 분석 요약 political_summary = "정치 분석 데이터 없음" if isinstance(political, dict) and "error" not in political: name = political.get("politician", thesis) phi12 = political.get("phi12_strength", "?") political_summary = f"{name} Φ12={phi12}" # 철학 랭킹 요약 phi_summary = "철학 분석 데이터 없음" if isinstance(phi_rank, dict) and "error" not in phi_rank: ranking = phi_rank.get("ranking", phi_rank.get("results", [])) if ranking and isinstance(ranking, list) and len(ranking) > 0: top = ranking[0] if isinstance(top, dict): phi_summary = f"Top: {top.get('name', top.get('philosopher','?'))}" return { "topic": topic, "thesis": thesis, "edu_summary": edu_summary, "political_summary": political_summary, "philosophy_summary": phi_summary, "generated": datetime.datetime.now().isoformat(), "pipeline_version": "1.0" } # ──────────────────────────────────────────────────────────────────── # MCP 도구 # ──────────────────────────────────────────────────────────────────── @mcp.tool() def produce_episode(topic: str, thesis: str = "", sub_themes: list = None, output_dir: str = "") -> str: """ [Season1 핵심 도구] 하나의 주제로 3개 MCP 체인을 돌려서 boards HTML + YouTube 스크립트 + 백서 초안을 동시 생산한다. Args: topic: 에피소드 주제 (예: "한국 교육의 정치적 포획") thesis: 중심 테제 (예: "이재명" — 정치인 이름 또는 철학자 키) sub_themes: 보조 주제 리스트 (선택) output_dir: 출력 디렉토리 (기본: boards/) Returns: 3종 아티팩트 경로 + 파이프라인 실행 보고서 """ # 1) 파이프라인 실행 pipeline_result = _run_pipeline(topic, thesis, sub_themes) synthesis = pipeline_result.get("synthesis", {}) # 2) 산출물 데이터 구성 sections = [ { "type": "hero", "title": f"🧠 {topic}", "body": f"테제: {thesis}" }, { "type": "text", "title": "시놉시스", "body": f"""이 에피소드는 "{topic}"를 주제로, OrbitPrompt의 3개 MCP 엔진(edu-mcp · political-mcp · philosophy-counter-mcp)이 동시에 분석한 결과를 하나의 서사로 합성한다. **핵심 질문:** {thesis}? """ }, { "type": "table", "title": "📊 MCP 체인 분석 결과", "rows": [ ["엔진", "결과", "상태"], ["🎓 edu-mcp", synthesis.get("edu_summary", "N/A"), "✅"], ["🏛️ political-mcp", synthesis.get("political_summary", "N/A"), "✅"], ["📚 philosophy-counter-mcp", synthesis.get("philosophy_summary", "N/A"), "✅"] ] }, { "type": "quote", "title": "", "body": "대화로 갈아버린 철학, 시스템이 증명하는 솔루션.", "cite": "Season1 Pipeline — OrbitPrompt" }, { "type": "text", "title": "🔬 3축 해석 프레임워크", "body": f"""**교육 축 (edu-mcp)** 한국 교육 시스템을 K(지식)·C(경쟁)·I(제도)·O(결과) 4축으로 평가. 필요한 전환: 암기 경쟁 → 비판적 창의성 + 실용 기술. **정치 축 (political-mcp)** Φ12 드라이버로 정치인/구조 분석. 철학 7축(Meta·Reverse·Modular·Language·Zoom·Spiral·Quantum)과 정치 5축(힘·범위·속도·안정성·도덕) 통합 점수. **철학 축 (philosophy-counter-mcp)** 8명의 철학자(공자·플라톤·니체·마르크스·사르트르·들뢰즈·한병철·박씨) 비교 + Φ7 7축 유사도 분석으로 세계관 해석. """ }, { "type": "text", "title": "🎬 시즌1 에피소드 구조", "body": """``` Episode 1: 철학 생산 공장의 시작 → 문제 정의: STT 대화는 왜 시스템이 되어야 하는가 → MCP 체인: edu → political → philosophy 3축 동시 가동 → 산출물: boards HTML + YouTube 스크립트 + 백서 → 출판: GitHub Pages + YouTube + Telegram Episode 2-N: (주제 선택에 따라 자동 확장) → Netflix-Log OS: 드라마/영화 해석 → 축구×교육: 한국 교육을 축구 모델로 해부 → Working-Class Leben철학: 노동계급 철학 """ }, { "type": "code", "title": "🧬 파이프라인 실행 코드", "body": f"""# produce_episode() 호출 기록 topic: "{topic}" thesis: "{thesis}" sub_themes: {json.dumps(sub_themes or [])} timestamp: {datetime.datetime.now().isoformat()} # 이 에피소드는 아래 명령으로 재현 가능: # produce_episode(topic="{topic}", thesis="{thesis}") """ }, { "type": "text", "title": "🔗 레포 연결 구조", "body": f"""``` OrbitPrompt (설계 사무소 — 이 레포) ├── phl-engine/server.py ← PHL 실행 엔진 ├── phl-engine/season1.py ← 시즌1 파이프라인 (★ 현재) ├── edu-mcp/mcp/server.py ← 교육 평가 엔진 ├── political-mcp/server.py ← 정치 분석 엔진 ├── philosophy-counter/mcp/server.py ← 철학 비교 엔진 ├── political-mcp/political_model.py ← Φ12 드라이버 ├── philosophy-counter/mcp/phi7_engine.py ← Φ7 유사도 엔진 ├── edu-mcp/mcp/engine.py ← 4축 교육 평가 ├── scripts/fab-run.sh ← FAB 라우트 실행기 ├── data/fab-manifest.json ← 공정 매니페스트 ├── boards/ ← 15개 출력 갤러리 └── docs/ ← 40+ 문서 ``` """ } ] # 에피소드별 스크립트 챕터 chapters = [ {"title": "오프닝 — 문제 제기", "body": f"\"{topic}\" — 왜 이 주제인가. STT 대화 한 줄에서 시작된 철학 생산.", "duration_sec": 60}, {"title": "교육 시스템 분석", "body": synthesis.get("edu_summary", "edu-mcp 평가 결과"), "duration_sec": 120}, {"title": "정치 구조 해부", "body": synthesis.get("political_summary", "political-mcp 분석 결과"), "duration_sec": 120}, {"title": "철학적 해석", "body": synthesis.get("philosophy_summary", "philosophy-counter-mcp 해석 결과"), "duration_sec": 90}, {"title": "3축 합성 — 솔루션", "body": "3개 MCP 결과를 하나의 서사로: 시스템 솔루션의 방향", "duration_sec": 120}, {"title": "클로징 — 다음 에피소드", "body": "이 시스템이 계속 도는 한, 매주 새로운 에피소드가 생산된다.", "duration_sec": 60} ] # 백서 섹션 wp_sections = [ {"title": "초록", "body": f"본 문서는 OrbitPrompt Season1 Pipeline의 첫 번째 에피소드 \"{topic}\"의 기술/철학 백서이다. STT 대화 입력부터 3종 아티팩트(HTML · 스크립트 · 백서) 출력까지의 전 과정을 기록한다."}, {"title": "시스템 아키텍처", "body": "3개 MCP(edu · political · philosophy)를 PHL 토큰 체계 아래 단일 파이프라인으로 묶었다. 각 MCP는 독립적 FastMCP 서버로 동작하며, season1.py가 오케스트레이터 역할을 수행한다."}, {"title": "데이터 흐름", "body": f"입력(topic=\"{topic}\") → 3개 MCP 병렬 호출 → 결과 합성 → 3종 아티팩트 동시 생성. 각 MCP의 출력을 표준 스키마로 정규화한 뒤, 템플릿 엔진이 HTML/스크립트/백서를 각각 렌더링한다."}, {"title": "철학적 기반", "body": "PHL(Protocol for Human Language) — 의미 압축 프로토콜. 사용자의 말이 토큰이 되고, 토큰이 MCP를 호출하고, MCP가 결과물을 생산한다. 이 백서 자체가 그 프로토콜의 증거물이다."}, {"title": "확장 방향", "body": "에피소드 2부터는 입력 소스를 STT(텔레그램 음성)로 전환하고, YouTube API 자동 업로드까지 파이프라인에 포함할 예정이다."} ] # 3) 산출물 저장 out_dir = ROOT / (output_dir if output_dir else "boards") out_dir.mkdir(parents=True, exist_ok=True) # 에피소드 ID 생성 ep_slug = topic.lower().replace(" ", "-")[:30] ep_slug = "".join(c for c in ep_slug if c.isalnum() or c in "-_") ep_id = f"season1-ep1-{ep_slug}" timestamp = datetime.datetime.now().strftime("%Y%m%d-%H%M%S") # HTML 저장 html_path = out_dir / f"{ep_id}.html" html_path.write_text(_render_article_html( f"S1-EP1: {topic}", f"테제: {thesis} · 생성: {datetime.datetime.now().strftime('%Y-%m-%d %H:%M')}", sections ), encoding="utf-8") # 스크립트 저장 script_path = out_dir / f"{ep_id}-script.md" script_path.write_text(_render_script_text( f"S1-EP1: {topic} — YouTube Script", chapters ), encoding="utf-8") # 백서 저장 wp_path = out_dir / f"{ep_id}-whitepaper.md" wp_path.write_text(_render_whitepaper( f"S1-EP1 백서: {topic}", wp_sections ), encoding="utf-8") # 4) 리턴 report = f"""## ✅ Season1 Episode 1 Produced ### 📋 Episode Info - **Title:** {topic} - **Thesis:** {thesis} - **ID:** {ep_id} ### 🧠 MCP Pipeline Results | Engine | Status | |--------|--------| | edu-mcp | {"✅" if "error" not in str(pipeline_result.get("edu",{})) else "❌"} | | political-mcp | {"✅" if "error" not in str(pipeline_result.get("political",{})) else "❌"} | | philosophy-counter-mcp | {"✅" if "error" not in str(pipeline_result.get("philosophy_rank",{})) else "❌"} | ### 📦 3 Artifacts Generated | # | Type | Path | |---|------|------| | 1 | 🖥️ **Gallery HTML** | `{html_path.relative_to(ROOT)}` | | 2 | 🎬 **YouTube Script** | `{script_path.relative_to(ROOT)}` | | 3 | 📄 **Whitepaper** | `{wp_path.relative_to(ROOT)}` | ### 🔄 Pipeline ``` STT/입력 → PHL 토큰 → Season1 Pipeline → 3종 아티팩트 ├── edu-mcp (교육 시스템 평가) ├── political-mcp (정치 구조 분석) └── philosophy-counter-mcp (철학 해석 + 박씨 비교) ``` ### 🚀 Next To deploy: `fab-run.sh --route RT-α` (content production) """ return report @mcp.tool() def list_episodes(output_dir: str = "boards") -> str: """생성된 시즌1 에피소드 목록을 반환한다.""" out_dir = ROOT / output_dir if not out_dir.exists(): return "에피소드 없음 (boards/ 디렉토리 없음)" files = sorted(out_dir.glob("season1-ep1-*")) if not files: return "아직 생성된 에피소드가 없습니다.\n`produce_episode(topic=..., thesis=...)`로 첫 번째 에피소드를 생성하세요." lines = ["## 📋 Season1 Episodes", ""] for f in files: size = f.stat().st_size mtime = datetime.datetime.fromtimestamp(f.stat().st_mtime).strftime("%Y-%m-%d %H:%M") lines.append(f"- {f.name} ({size:,} bytes, {mtime})") return "\n".join(lines) @mcp.tool() def pipeline_status() -> str: """시즌1 파이프라인 MCP 엔진 상태 확인""" results = [] # 각 MCP 서버 존재 확인 servers = [ ("edu-mcp", "edu-mcp/mcp/server.py"), ("political-mcp", "political-mcp/server.py"), ("philosophy-counter-mcp", "philosophy-counter/mcp/server.py"), ] lines = ["## 🔄 Season1 Pipeline Status", ""] for name, path in servers: full = ROOT / path if full.exists(): size = full.stat().st_size result = _call_mcp(path, "ping") if name == "political-mcp" else _call_mcp(path, "version") status = "✅ Online" if "error" not in str(result) else "⚠️ 응답 없음" lines.append(f"### {name}") lines.append(f"- Path: {path}") lines.append(f"- Size: {size:,} bytes") lines.append(f"- Status: {status}") else: lines.append(f"### {name}") lines.append(f"- ❌ 서버 파일 없음: {path}") lines.append("") # 생성된 에피소드 ep_count = len(list((ROOT / "boards").glob("season1-ep1-*"))) lines.append(f"### Output") lines.append(f"- 생성된 에피소드: {ep_count}개") lines.append(f"- boards/ 디렉토리: {len(list((ROOT/'boards').glob('*.html')))}개 HTML") lines.append("") return "\n".join(lines) @mcp.tool() def produce_debug(topic: str = "한국 교육의 정치적 포획", thesis: str = "교육은 왜 정치적 도구가 되었는가") -> str: """ [데모용] 실제 MCP 호출 없이 파이프라인 결과물 구조만 빠르게 생성. MCP 서버가 없거나 타임아웃 우려될 때 사용. Args: topic: 에피소드 주제 thesis: 중심 테제 """ # MCP 호출 없이 합성 데이터로 3종 아티팩트 생성 pipeline_result = { "edu": {"note": "edu-mcp simulation — 실제 서버에서 대체 필요"}, "political": {"note": "political-mcp simulation"}, "philosophy_rank": {"note": "philosophy-counter-mcp simulation"}, "synthesis": { "topic": topic, "thesis": thesis, "edu_summary": "K=45 C=82 I=55 O=48 → E_g=0.52 (한계 상태, 경쟁 축 과부하)", "political_summary": "Φ12 분석: 구조적 모순, 제도 개혁 필요", "philosophy_summary": "박씨 유사도 Top3: 한병철(71.4%) > 니체(64.3%) > 들뢰즈(57.1%)", "generated": datetime.datetime.now().isoformat() } } # produce_episode와 동일한 로직으로 아티팩트 생성 synthesis = pipeline_result["synthesis"] sections = [ {"type": "hero", "title": f"🧠 {topic}", "body": f"테제: {thesis}"}, {"type": "text", "title": "시놉시스", "body": f"이 에피소드는 \"{topic}\"를 주제로, OrbitPrompt의 3개 MCP 엔진이 동시에 분석한 결과를 하나의 서사로 합성한다."}, {"type": "table", "title": "📊 분석 결과", "rows": [ ["엔진", "결과", "상태"], ["🎓 edu-mcp", synthesis.get("edu_summary", "N/A"), "✅"], ["🏛️ political-mcp", synthesis.get("political_summary", "N/A"), "✅"], ["📚 philosophy-counter-mcp", synthesis.get("philosophy_summary", "N/A"), "✅"] ]}, {"type": "quote", "title": "", "body": "대화로 갈아버린 철학, 시스템이 증명하는 솔루션.", "cite": "Season1 Pipeline — OrbitPrompt"}, {"type": "text", "title": "🔬 3축 해석", "body": "교육 축: 암기 경쟁 → 비판적 창의성. 정치 축: Φ12 구조 분석. 철학 축: 박씨 vs 철학자 Φ7 비교."}, ] out_dir = ROOT / "boards" out_dir.mkdir(parents=True, exist_ok=True) ep_slug = topic.lower().replace(" ", "-")[:30] ep_slug = "".join(c for c in ep_slug if c.isalnum() or c in "-_") ep_id = f"season1-ep1-demo-{ep_slug}" html_path = out_dir / f"{ep_id}.html" html_path.write_text(_render_article_html( f"S1-EP1: {topic}", f"테제: {thesis} · DEMO · 생성: {datetime.datetime.now().strftime('%Y-%m-%d %H:%M')}", sections ), encoding="utf-8") return f"""## ✅ Demo Episode Produced ### 📦 Artifact - **Path:** `boards/{ep_id}.html` - **Size:** {html_path.stat().st_size:,} bytes ### 📊 Simulation Data {synthesis.get("edu_summary")} {synthesis.get("political_summary")} {synthesis.get("philosophy_summary")} > 실제 에피소드를 생산하려면: > `produce_episode(topic="{topic}", thesis="{thesis}")` """ # ── 리소스 ───────────────────────────────────────────────────────── @mcp.resource("season1://status") def resource_status() -> str: """파이프라인 상태""" return pipeline_status() @mcp.resource("season1://episodes") def resource_episodes() -> str: """에피소드 목록""" return list_episodes() # ── 메인 ──────────────────────────────────────────────────────────── if __name__ == "__main__": print("🎬 Season1 Pipeline v1.0 starting...", file=sys.stderr) print(f" Root: {ROOT}", file=sys.stderr) print(f" Tools: produce_episode / list_episodes / pipeline_status / produce_debug", file=sys.stderr) print(f" MCP chain: edu-mcp + political-mcp + philosophy-counter-mcp", file=sys.stderr) mcp.run()