diff --git a/.gitignore b/.gitignore
index 279916a..a079d69 100644
--- a/.gitignore
+++ b/.gitignore
@@ -39,6 +39,7 @@ assets/
# Uploaded catalogs (user-specific)
catalog/uploaded_*.json
+catalog/textbook_*.json
# Test & Coverage
.pytest_cache/
diff --git a/api_server.py b/api_server.py
index f279f36..4329af2 100644
--- a/api_server.py
+++ b/api_server.py
@@ -14,7 +14,7 @@
from queue import Queue
from threading import Thread
-from fastapi import FastAPI, BackgroundTasks, HTTPException, UploadFile, File, Header
+from fastapi import FastAPI, BackgroundTasks, HTTPException, UploadFile, File, Header, Form
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse, JSONResponse, StreamingResponse
from fastapi.staticfiles import StaticFiles
@@ -26,6 +26,7 @@
from src import __version__
from src.pdf_processor import PDFSlideProcessor
from src.ADDIE_optimize import ADDIEOptimizer
+from src.textbook_reference import TextbookReferenceBuilder
import tempfile
import shutil
@@ -67,6 +68,7 @@ class OptimizeRequest(BaseModel):
model_name: str = Field(default="gpt-4o-mini", description="OpenAI model to use")
exp_name: str = Field(default="default", description="Experiment name for output")
chapter_name: Optional[str] = Field(default=None, description="Specific chapter to optimize (None = all)")
+ mode: str = Field(default="regenerate", description="Improvement strategy: 'regenerate' (per-slide full rewrite) or 'refine' (localized frame-level rewrite)")
class TaskStatus(BaseModel):
task_id: str
@@ -391,6 +393,58 @@ async def upload_catalog(
except Exception as e:
raise HTTPException(status_code=500, detail=f"Error uploading catalog: {str(e)}")
+@app.post("/api/textbook/catalog")
+async def build_textbook_catalog(
+ file: UploadFile = File(...),
+ course_name: str = Form(default="")
+):
+ """
+ Build catalog-compatible reference data from an uploaded textbook file.
+
+ This proof-of-concept extracts a bounded amount of text from PDF/TXT/MD
+ input and converts it into the existing catalog schema. It does not call
+ an LLM, so it can be used before starting a paid generation task.
+ """
+ allowed_suffixes = {".pdf", ".txt", ".md"}
+ original_name = file.filename or "textbook.pdf"
+ suffix = Path(original_name).suffix.lower()
+ if suffix not in allowed_suffixes:
+ raise HTTPException(status_code=400, detail="Only PDF, TXT, and Markdown textbook files are supported")
+
+ MAX_FILE_SIZE = 50 * 1024 * 1024 # 50 MB
+ content = await file.read()
+ if len(content) > MAX_FILE_SIZE:
+ raise HTTPException(status_code=400, detail="Textbook file is too large (max 50 MB)")
+
+ try:
+ with tempfile.TemporaryDirectory() as temp_dir:
+ temp_path = Path(temp_dir) / original_name
+ temp_path.write_bytes(content)
+
+ builder = TextbookReferenceBuilder()
+ catalog_data = builder.build_catalog(temp_path, course_name=course_name)
+
+ catalog_dir = Path("catalog")
+ catalog_dir.mkdir(exist_ok=True)
+ catalog_name = f"textbook_{uuid.uuid4().hex[:8]}"
+ catalog_filename = f"{catalog_name}.json"
+ catalog_path = catalog_dir / catalog_filename
+ with open(catalog_path, "w", encoding="utf-8") as f:
+ json.dump(catalog_data, f, indent=2, ensure_ascii=False)
+
+ return {
+ "catalog_name": catalog_name,
+ "filename": catalog_filename,
+ "saved_path": str(catalog_path),
+ "catalog_data": catalog_data,
+ "message": "Textbook reference catalog generated successfully"
+ }
+
+ except ValueError as e:
+ raise HTTPException(status_code=400, detail=str(e))
+ except Exception as e:
+ raise HTTPException(status_code=500, detail=f"Error processing textbook: {str(e)}")
+
@app.get("/api/tasks/list")
async def list_tasks():
"""
@@ -900,6 +954,8 @@ async def run_optimization_task(task_id: str, request: OptimizeRequest, api_key:
sys.stdout.flush()
print(f"Experiment: {request.exp_name}")
sys.stdout.flush()
+ print(f"Mode: {request.mode}")
+ sys.stdout.flush()
if request.chapter_name:
print(f"Chapter: {request.chapter_name}")
sys.stdout.flush()
@@ -916,6 +972,7 @@ async def run_optimization_task(task_id: str, request: OptimizeRequest, api_key:
model_name=request.model_name,
exp_name=request.exp_name,
chapter_name=request.chapter_name,
+ mode=request.mode,
)
print("\n" + "=" * 60)
diff --git a/frontend/app.js b/frontend/app.js
index 628f894..4c054d8 100644
--- a/frontend/app.js
+++ b/frontend/app.js
@@ -96,6 +96,10 @@ const translations = {
modeGenerate: '生成新课程',
modeOptimize: '优化已有材料',
pptxLabel: '同时生成 PPTX 幻灯片',
+ optimizeModeLabel: '优化方式',
+ optimizeModeRegenerate: '逐张重新生成（整章重写）',
+ optimizeModeRefine: '局部精修（仅改相关帧）',
+ optimizeModeTip: '重新生成：每张幻灯片整体重写。局部精修：只定位并改写与需求相关的帧，其余保持不变。',
optimizeSectionTitle: '优化配置',
optimizeSubmitButton: '🔧开始优化',
optimizeProgressTitle: '优化进度',
@@ -165,6 +169,8 @@ const translations = {
modelOptionGpt4Turbo: 'GPT-4 Turbo',
expNameLabel: 'Experiment Name',
expNamePlaceholder: 'Default: default',
+ textbookFileLabel: 'Textbook Reference File (optional)',
+ textbookFileTip: 'Supports PDF/TXT/MD. The system extracts a bounded excerpt and saves a textbook catalog JSON for review.',
copilotLabel: 'Enable Copilot Mode (Interactive Feedback)',
catalogModeLabel: 'Catalog Mode',
catalogOptionNone: 'Do not use',
@@ -234,9 +240,15 @@ const translations = {
catalogListFailed: 'Failed to load catalog list',
catalogSelectDefault: 'Select Catalog...',
uploadCatalogFailed: 'Failed to upload catalog file',
+ uploadTextbookFailed: 'Failed to process textbook reference file',
+ textbookCatalogReady: 'Textbook catalog generated: {filename}',
modeGenerate: 'Generate Course',
modeOptimize: 'Optimize Materials',
pptxLabel: 'Also generate PPTX slides',
+ optimizeModeLabel: 'Optimization Method',
+ optimizeModeRegenerate: 'Regenerate (rewrite every slide)',
+ optimizeModeRefine: 'Localized Refine (edit only relevant frames)',
+ optimizeModeTip: 'Regenerate: rewrite each slide wholesale. Localized Refine: locate and rewrite only the frames relevant to your requirements, leaving the rest untouched.',
optimizeSectionTitle: 'Optimization Settings',
optimizeSubmitButton: '🔧Start Optimization',
optimizeProgressTitle: 'Optimization Progress',
@@ -503,6 +515,13 @@ function setupEventListeners() {
const catalogMode = document.getElementById('catalog-mode');
catalogMode.addEventListener('change', handleCatalogModeChange);
+ // Textbook file input: when a file is selected, disable catalog mode dropdown
+ // since the textbook will generate its own catalog automatically.
+ const textbookFile = document.getElementById('textbook-file');
+ if (textbookFile) {
+ textbookFile.addEventListener('change', handleTextbookFileChange);
+ }
+
// API Key management
document.getElementById('save-api-key').addEventListener('click', saveApiKey);
document.getElementById('toggle-api-key').addEventListener('click', toggleApiKeyVisibility);
@@ -648,6 +667,27 @@ function handleCatalogModeChange(e) {
}
}
+function handleTextbookFileChange() {
+ const textbookInput = document.getElementById('textbook-file');
+ const catalogMode = document.getElementById('catalog-mode');
+ const note = document.getElementById('textbook-catalog-note');
+ const hasFile = textbookInput && textbookInput.files.length > 0;
+
+ if (hasFile) {
+ catalogMode.value = 'none';
+ catalogMode.disabled = true;
+ if (note) note.style.display = 'block';
+ // Hide any open catalog sub-groups
+ ['catalog-upload-group', 'catalog-select-group', 'catalog-json-group'].forEach(id => {
+ const el = document.getElementById(id);
+ if (el) el.style.display = 'none';
+ });
+ } else {
+ catalogMode.disabled = false;
+ if (note) note.style.display = 'none';
+ }
+}
+
async function handleFormSubmit(e) {
e.preventDefault();
@@ -675,32 +715,48 @@ async function handleFormSubmit(e) {
generate_pptx: document.getElementById('pptx-mode').checked
};
- // Handle catalog
- const catalogMode = document.getElementById('catalog-mode').value;
- if (catalogMode === 'default') {
- formData.catalog = 'default_catalog';
- } else if (catalogMode === 'select') {
- const selected = document.getElementById('catalog-select').value;
- if (selected) {
- formData.catalog = selected;
- }
- } else if (catalogMode === 'upload') {
- // Handle file upload or JSON input
- const fileInput = document.getElementById('catalog-file');
- const jsonInput = document.getElementById('catalog-json').value;
-
- if (fileInput.files.length > 0) {
- // Upload file first
- const uploadResponse = await uploadCatalogFile(fileInput.files[0]);
- formData.catalog = uploadResponse.filename.replace('.json', '');
- } else if (jsonInput.trim()) {
- // Use JSON input directly
- try {
- formData.catalog_data = JSON.parse(jsonInput);
- } catch (err) {
- alert(t('invalidCatalogJson'));
- setSubmitButtonLoading(false);
- return;
+ // Handle optional textbook reference upload first.
+ // When a textbook file is provided, it takes priority — the catalog mode
+ // dropdown is ignored because the textbook generates its own catalog.
+ const textbookInput = document.getElementById('textbook-file');
+ if (textbookInput && textbookInput.files.length > 0) {
+ const textbookResponse = await uploadTextbookCatalog(
+ textbookInput.files[0],
+ formData.course_name
+ );
+
+ formData.catalog = textbookResponse.catalog_name;
+
+ console.log(t('textbookCatalogReady', { filename: textbookResponse.filename }));
+ await loadCatalogs();
+ } else {
+ // Handle catalog mode (only when no textbook is uploaded)
+ const catalogMode = document.getElementById('catalog-mode').value;
+ if (catalogMode === 'default') {
+ formData.catalog = 'default_catalog';
+ } else if (catalogMode === 'select') {
+ const selected = document.getElementById('catalog-select').value;
+ if (selected) {
+ formData.catalog = selected;
+ }
+ } else if (catalogMode === 'upload') {
+ // Handle file upload or JSON input
+ const fileInput = document.getElementById('catalog-file');
+ const jsonInput = document.getElementById('catalog-json').value;
+
+ if (fileInput.files.length > 0) {
+ // Upload file first
+ const uploadResponse = await uploadCatalogFile(fileInput.files[0]);
+ formData.catalog = uploadResponse.filename.replace('.json', '');
+ } else if (jsonInput.trim()) {
+ // Use JSON input directly
+ try {
+ formData.catalog_data = JSON.parse(jsonInput);
+ } catch (err) {
+ alert(t('invalidCatalogJson'));
+ setSubmitButtonLoading(false);
+ return;
+ }
}
}
}
@@ -794,6 +850,24 @@ async function uploadCatalogFile(file) {
return await response.json();
}
+async function uploadTextbookCatalog(file, courseName) {
+ const formData = new FormData();
+ formData.append('file', file);
+ formData.append('course_name', courseName || '');
+
+ const response = await fetch(`${API_BASE_URL}/api/textbook/catalog`, {
+ method: 'POST',
+ body: formData
+ });
+
+ if (!response.ok) {
+ const errorText = await response.text();
+ throw new Error(`${t('uploadTextbookFailed')}: ${errorText}`);
+ }
+
+ return await response.json();
+}
+
function startStatusPolling() {
if (statusCheckInterval) {
clearInterval(statusCheckInterval);
@@ -1414,6 +1488,8 @@ async function handleOptimizeSubmit(e) {
const modelName = document.getElementById('optimize-model-name').value;
const expName = document.getElementById('optimize-exp-name').value.trim() || 'default';
const chapterName = document.getElementById('chapter-name').value.trim() || null;
+ const modeSelect = document.getElementById('optimize-mode-select');
+ const mode = modeSelect ? modeSelect.value : 'regenerate';
const submitBtn = document.getElementById('optimize-submit-btn');
submitBtn.disabled = true;
@@ -1439,7 +1515,8 @@ async function handleOptimizeSubmit(e) {
user_requirements: userRequirements,
model_name: modelName,
exp_name: expName,
- chapter_name: chapterName
+ chapter_name: chapterName,
+ mode: mode
})
});
diff --git a/frontend/index.html b/frontend/index.html
index 11be75a..861ea14 100644
--- a/frontend/index.html
+++ b/frontend/index.html
@@ -78,6 +78,13 @@ 课程配置
+
+
+
+
+ 支持 PDF/TXT/MD；系统会先提取前若干页并保存为 textbook catalog JSON 供检查。
+
+
@@ -179,6 +189,15 @@
优化配置
+
+
+
+ 重新生成：每张幻灯片整体重写。局部精修：只定位并改写与需求相关的帧，其余保持不变。
+
+
Dict[str, Any]:
"""
Run the optimize workflow.
@@ -139,12 +140,14 @@ def run(
output_dir: Base output directory
exp_name: Experiment name
chapter_name: Specific chapter to optimize (None = all chapters)
+ mode: "regenerate" (per-slide full rewrite) or "refine" (localized
+ frame-level rewrite via SlideRefiner)
Returns:
Results dict with per-chapter outcomes and overall summary
"""
runner = OptimizeRunner(self, output_dir)
- return runner.run(storage_id, user_requirements, exp_name, chapter_name)
+ return runner.run(storage_id, user_requirements, exp_name, chapter_name, mode)
class OptimizeRunner:
@@ -166,6 +169,7 @@ def run(
user_requirements: str,
exp_name: Optional[str] = None,
chapter_name: Optional[str] = None,
+ mode: str = "regenerate",
) -> Dict[str, Any]:
"""
Run the full optimization workflow.
@@ -175,6 +179,7 @@ def run(
user_requirements: User's requirements for improvement
exp_name: Experiment name for organizing outputs
chapter_name: Specific chapter to optimize (None = all)
+ mode: "regenerate" or "refine" (localized frame-level rewrite)
Returns:
Results dict
@@ -221,7 +226,7 @@ def run(
try:
chapter_result = self._optimize_chapter(
- storage_id, ch_name, user_requirements, exp_name
+ storage_id, ch_name, user_requirements, exp_name, mode
)
results["chapters"].append(chapter_result)
except Exception as e:
@@ -265,6 +270,7 @@ def _optimize_chapter(
chapter_name: str,
user_requirements: str,
exp_name: Optional[str] = None,
+ mode: str = "regenerate",
) -> Dict[str, Any]:
"""
Optimize a single chapter.
@@ -327,7 +333,7 @@ def _optimize_chapter(
knowledge_base=kb,
)
- result = deliberation.run(chapter_slides, user_requirements)
+ result = deliberation.run(chapter_slides, user_requirements, mode=mode)
result["chapter"] = chapter_name
result["knowledge_base_name"] = kb_name
diff --git a/src/optimize.py b/src/optimize.py
index bfc9048..1316f97 100644
--- a/src/optimize.py
+++ b/src/optimize.py
@@ -16,6 +16,7 @@
from src.agents import Agent, Deliberation, LLM
from src.slides import SlideUtils
from src.slide_knowledge_base import SlideKnowledgeBase
+from src.slide_refiner import SlideRefiner
class OptimizeSlidesDeliberation:
@@ -65,6 +66,7 @@ def run(
chapter_slides: List[Dict[str, Any]],
user_requirements: str,
user_feedback: Optional[Dict[str, Any]] = None,
+ mode: str = "regenerate",
) -> Dict[str, Any]:
"""
Run the optimization deliberation for a chapter.
@@ -74,6 +76,12 @@ def run(
(each has: title, content, slide_number, etc.)
user_requirements: User's requirements for improvement
user_feedback: Optional user feedback dict (e.g. {"slides": "...", "overall": "..."})
+ mode: Improvement strategy:
+ - "regenerate" (default): per-slide multi-agent deliberation that
+ rewrites every slide from scratch.
+ - "refine": localized, frame-level rewrite via SlideRefiner. Builds a
+ baseline deck from the existing slides, then surgically edits only the
+ frames most relevant to the feedback, leaving the rest untouched.
Returns:
Dict with success status, file paths, and statistics
@@ -81,8 +89,12 @@ def run(
if user_feedback is None:
user_feedback = {"slides": "", "overall": ""}
+ if mode not in ("regenerate", "refine"):
+ raise ValueError(f"Unknown optimize mode: {mode!r} (expected 'regenerate' or 'refine')")
+
print(f"\n{'='*60}")
print(f"Starting Optimize Deliberation: {self.name}")
+ print(f"Mode: {mode}")
print(f"{'='*60}\n")
print(f"Slides to optimize: {len(chapter_slides)}")
print(f"User requirements: {user_requirements[:200]}...")
@@ -107,44 +119,55 @@ def run(
f.write(f"# Content Analysis\n\n{analysis_result}")
print(f"Analysis saved to: {analysis_path}")
- # ── Phase 2: Per-slide Enhancement Deliberation ─────────────
- print(f"\n{'#'*50}")
- print(f"Phase 2: Slide Enhancement & LaTeX Generation")
- print(f"{'#'*50}\n")
-
- # Get LaTeX template
+ # Get LaTeX template (shared by both modes)
latex_template = SlideUtils.get_latex_template(catalog=False)
latex_prefix, latex_suffix = SlideUtils.parse_latex_template(latex_template)
- enhanced_frames = []
- enhanced_content_list = []
+ # ── Phase 2: Improve slides (mode-dependent) ────────────────
+ if mode == "refine":
+ print(f"\n{'#'*50}")
+ print(f"Phase 2: Localized Frame-level Refinement")
+ print(f"{'#'*50}\n")
- for idx, slide in enumerate(chapter_slides):
- slide_title = slide.get("title", f"Slide {idx + 1}")
- print(f"\n{'-'*50}")
- print(f"Enhancing Slide {idx + 1}/{len(chapter_slides)}: {slide_title}")
- print(f"{'-'*50}\n")
+ full_latex, enhanced_frames, enhanced_content_list, refine_info = self._refine_slides(
+ chapter_slides, analysis_result, user_requirements, user_feedback,
+ latex_prefix, latex_suffix,
+ )
+ else:
+ print(f"\n{'#'*50}")
+ print(f"Phase 2: Slide Enhancement & LaTeX Generation")
+ print(f"{'#'*50}\n")
- # Search knowledge base for related content
- relevant_content = self.knowledge_base.search(slide_title, top_k=3)
+ refine_info = None
+ enhanced_frames = []
+ enhanced_content_list = []
- enhanced, frames, enh_time, enh_tokens = self._run_enhancement_deliberation(
- slide, idx, analysis_result, user_requirements, user_feedback, relevant_content
- )
- total_time += enh_time
- total_tokens += enh_tokens
+ for idx, slide in enumerate(chapter_slides):
+ slide_title = slide.get("title", f"Slide {idx + 1}")
+ print(f"\n{'-'*50}")
+ print(f"Enhancing Slide {idx + 1}/{len(chapter_slides)}: {slide_title}")
+ print(f"{'-'*50}\n")
- enhanced_content_list.append(enhanced)
- enhanced_frames.extend(frames)
+ # Search knowledge base for related content
+ relevant_content = self.knowledge_base.search(slide_title, top_k=3)
- # ── Phase 3: Compile and Save ───────────────────────────────
+ enhanced, frames, enh_time, enh_tokens = self._run_enhancement_deliberation(
+ slide, idx, analysis_result, user_requirements, user_feedback, relevant_content
+ )
+ total_time += enh_time
+ total_tokens += enh_tokens
+
+ enhanced_content_list.append(enhanced)
+ enhanced_frames.extend(frames)
+
+ # Compile full LaTeX document from the regenerated frames
+ full_latex = SlideUtils.compile_latex_document(latex_prefix, enhanced_frames, latex_suffix)
+
+ # ── Phase 3: Save ───────────────────────────────────────────
print(f"\n{'#'*50}")
print(f"Phase 3: Compiling Results")
print(f"{'#'*50}\n")
- # Compile full LaTeX document
- full_latex = SlideUtils.compile_latex_document(latex_prefix, enhanced_frames, latex_suffix)
-
# Save LaTeX file
latex_file = os.path.join(self.output_dir, "enhanced_slides.tex")
with open(latex_file, "w", encoding="utf-8") as f:
@@ -155,8 +178,10 @@ def run(
with open(content_file, "w", encoding="utf-8") as f:
json.dump({
"enhanced_at": datetime.now().isoformat(),
+ "mode": mode,
"original_slides_count": len(chapter_slides),
"enhanced_slides": enhanced_content_list,
+ "refine_info": refine_info,
"user_requirements": user_requirements,
}, f, indent=2, ensure_ascii=False)
@@ -164,6 +189,7 @@ def run(
stats_file = os.path.join(self.output_dir, f"statistics_{self.id}.json")
with open(stats_file, "w", encoding="utf-8") as f:
json.dump({
+ "mode": mode,
"elapsed_time": total_time,
"token_usage": total_tokens,
"total_slides": len(chapter_slides),
@@ -178,15 +204,160 @@ def run(
return {
"success": True,
+ "mode": mode,
"latex_file": latex_file,
"content_file": content_file,
"analysis_file": analysis_path,
"total_slides": len(chapter_slides),
"total_frames": len(enhanced_frames),
+ "refine_info": refine_info,
"elapsed_time": total_time,
"token_usage": total_tokens,
}
+ # Maximum localized-refine repair attempts per chapter deck.
+ REFINE_MAX_RETRIES = 2
+
+ def _refine_slides(
+ self,
+ chapter_slides: List[Dict[str, Any]],
+ analysis_result: str,
+ user_requirements: str,
+ user_feedback: Dict[str, Any],
+ latex_prefix: str,
+ latex_suffix: str,
+ ) -> tuple:
+ """
+ Localized refinement path (mode="refine").
+
+ Builds a baseline LaTeX deck from the existing slides, then uses SlideRefiner
+ to locate only the frames relevant to the feedback and rewrite just their
+ bodies, leaving every other frame byte-identical.
+
+ Returns:
+ (full_latex, frames, content_list, refine_info)
+ """
+ # 1. Build a clean, escaped baseline deck from the existing slides.
+ baseline_frames = [
+ self._build_baseline_frame(slide, idx)
+ for idx, slide in enumerate(chapter_slides)
+ ]
+ baseline_latex = SlideUtils.compile_latex_document(
+ latex_prefix, baseline_frames, latex_suffix
+ )
+
+ # 2. Assemble the feedback string the refiner will act on.
+ feedback_text = self._build_refine_feedback(
+ analysis_result, user_requirements, user_feedback
+ )
+
+ # 3. Surgical, frame-level refinement with deterministic validation + retries.
+ refiner = SlideRefiner(self.llm)
+ result = refiner.refine_slides(
+ content=baseline_latex,
+ feedback_text=feedback_text,
+ max_retries=self.REFINE_MAX_RETRIES,
+ )
+
+ full_latex = result["refined_content"]
+ frames = SlideUtils.extract_latex_frames(full_latex)
+
+ edited_indexes = {f["index"] for f in result.get("edited_frames", [])}
+ print(
+ f"Refine complete: status={result['slide_validation_status']}, "
+ f"targeted={result.get('target_indexes')}, "
+ f"edited={sorted(edited_indexes)}, retries={result.get('retries_used')}"
+ )
+ if result["slide_validation_status"] != "PASS":
+ print("Refine validation did not fully pass:")
+ for err in result.get("slide_validation_errors", []):
+ print(f" - {err}")
+
+ content_list = [
+ {
+ "index": idx,
+ "title": slide.get("title", f"Slide {idx + 1}"),
+ "edited": idx in edited_indexes,
+ }
+ for idx, slide in enumerate(chapter_slides)
+ ]
+
+ refine_info = {
+ "validation_status": result["slide_validation_status"],
+ "validation_errors": result.get("slide_validation_errors", []),
+ "target_indexes": result.get("target_indexes", []),
+ "edited_frames": result.get("edited_frames", []),
+ "retries_used": result.get("retries_used"),
+ "locator_response": result.get("locator_response"),
+ }
+
+ return full_latex, frames, content_list, refine_info
+
+ def _build_baseline_frame(self, slide: Dict[str, Any], idx: int) -> str:
+ """Build a single valid Beamer frame from an existing slide dict."""
+ title = slide.get("title", f"Slide {idx + 1}")
+ content = slide.get("content", slide.get("text", "")) or ""
+
+ title_tex = self._escape_latex(title)
+
+ # Turn the slide content into a small, escaped itemize body.
+ lines = [ln.strip() for ln in content.splitlines() if ln.strip()]
+ lines = lines[:8] # keep frames presentation-sized
+ if lines:
+ items = "\n".join(
+ f" \\item {self._escape_latex(ln[:300])}" for ln in lines
+ )
+ body = f" \\begin{{itemize}}\n{items}\n \\end{{itemize}}"
+ else:
+ body = f" {self._escape_latex(content[:300]) or '~'}"
+
+ return (
+ f"\\begin{{frame}}[fragile]\n"
+ f" \\frametitle{{{title_tex}}}\n"
+ f"{body}\n"
+ f"\\end{{frame}}"
+ )
+
+ @staticmethod
+ def _escape_latex(text: str) -> str:
+ """Escape LaTeX special characters so baseline frames compile cleanly."""
+ if not text:
+ return ""
+ # Stash backslashes behind a sentinel so the braces in their replacement
+ # (\textbackslash{}) don't get re-escaped by the {}-escaping below.
+ sentinel = "\x00BSLASH\x00"
+ text = text.replace("\\", sentinel)
+ for ch, esc in (
+ ("&", r"\&"), ("%", r"\%"), ("$", r"\$"), ("#", r"\#"),
+ ("_", r"\_"), ("{", r"\{"), ("}", r"\}"),
+ ):
+ text = text.replace(ch, esc)
+ # These insert braces too, but only after {}-escaping has run.
+ text = text.replace("~", r"\textasciitilde{}")
+ text = text.replace("^", r"\textasciicircum{}")
+ text = text.replace(sentinel, r"\textbackslash{}")
+ return text
+
+ def _build_refine_feedback(
+ self,
+ analysis_result: str,
+ user_requirements: str,
+ user_feedback: Dict[str, Any],
+ ) -> str:
+ """Combine requirements, analysis, and human feedback into one feedback string."""
+ parts = []
+ if user_requirements:
+ parts.append(f"USER REQUIREMENTS:\n{user_requirements}")
+ slides_fb = (user_feedback or {}).get("slides", "")
+ overall_fb = (user_feedback or {}).get("overall", "")
+ if slides_fb:
+ parts.append(f"USER FEEDBACK ON SLIDES:\n{slides_fb}")
+ if overall_fb:
+ parts.append(f"OVERALL USER FEEDBACK:\n{overall_fb}")
+ if analysis_result:
+ parts.append(f"ANALYSIS & RECOMMENDATIONS:\n{analysis_result[:3000]}")
+ return "\n\n".join(parts)
+
def _run_analysis_deliberation(
self,
chapter_slides: List[Dict[str, Any]],
diff --git a/src/slide_refiner.py b/src/slide_refiner.py
new file mode 100644
index 0000000..a072273
--- /dev/null
+++ b/src/slide_refiner.py
@@ -0,0 +1,665 @@
+"""SlideRefiner - localized, frame-level refinement of an existing Beamer LaTeX deck.
+
+Extracted from the IA-aarsh refiner pipeline (src/refinement.py). Unlike the
+"regenerate" path in optimize.py, this performs surgical edits: it locates only
+the frames most relevant to the feedback, rewrites just their bodies, anchors the
+replacement on the original frame text so untouched frames stay byte-identical,
+and guards the result with deterministic body- and document-level validation plus
+retries.
+
+The LLM is only trusted to (1) pick which frames to edit and (2) rewrite a single
+frame body; everything else is regex parsing + string replacement + checks.
+
+Input contract: refine_slides(content, feedback_text, max_retries) where
+`content` is a full LaTeX document string and `feedback_text` is free text
+(evaluator metrics OR human feedback -- it is just a string).
+"""
+
+import re
+
+from src.latex_to_pptx import LaTeXParser
+
+
+class SlideRefiner:
+ def __init__(self, llm):
+ self.llm = llm
+
+ def refine_slides(self, content, feedback_text, max_retries=1):
+
+ frames = self.parse_frames(content)
+ frame_summary = self.build_frame_summary(frames)
+ locator_response = self.locate_frames(feedback_text, frame_summary)
+ target_indexes = self.parse_target_frame_indexes(locator_response)
+ target_frames = self.get_target_frames(frames, target_indexes)
+
+ edited_frames = []
+ validation_history = []
+ refined_content = content
+
+ for attempt in range(max_retries + 1):
+ working_frames = self.parse_frames(refined_content)
+
+ for target_frame in target_frames:
+ frame_index = target_frame["index"]
+ current_target_frame = working_frames[frame_index]
+ frame_context = self.build_target_frame_context(
+ working_frames,
+ current_target_frame
+ )
+
+ if attempt == 0:
+ revised_body = self.refine_frame_body(
+ frame_context,
+ feedback_text
+ )
+
+ else:
+ previous_validation = validation_history[-1]
+
+ validation_errors = "\n".join(
+ previous_validation.get("errors", [])
+ )
+
+ revised_body = self.retry_refine_frame_body(
+ frame_context,
+ feedback_text,
+ validation_errors
+ )
+
+ body_validation = self.validate_frame_body(revised_body)
+ body_retries_used = 0
+
+ while (
+ body_validation["status"] == "FAIL"
+ and body_retries_used < max_retries
+ ):
+ revised_body = self.retry_refine_frame_body(
+ frame_context,
+ feedback_text,
+ "\n".join(body_validation["errors"])
+ )
+ body_validation = self.validate_frame_body(revised_body)
+ body_retries_used += 1
+
+ if body_validation["status"] == "FAIL":
+ continue
+
+ rebuilt_frame = self.rebuild_frame(
+ current_target_frame,
+ revised_body
+ )
+
+ working_frames = self.replace_frames(
+ working_frames,
+ frame_index,
+ rebuilt_frame
+ )
+
+ if not any(
+ frame["index"] == frame_index
+ for frame in edited_frames
+ ):
+ edited_frames.append({
+ "index": frame_index,
+ "title": current_target_frame["title"]
+ })
+
+ refined_content = self.reassemble_slides(
+ refined_content,
+ working_frames
+ )
+
+ validation_result = self.validate_slide_patch(
+ original_latex=content,
+ refined_latex=refined_content,
+ edited_frames=edited_frames
+ )
+
+ validation_history.append(validation_result)
+
+ if validation_result["status"] == "PASS":
+ return {
+ "refined_content": refined_content,
+ "locator_response": locator_response,
+ "target_indexes": target_indexes,
+ "edited_frames": edited_frames,
+ "slide_validation_status": "PASS",
+ "slide_validation_errors": [],
+ "validation_history": validation_history,
+ "retries_used": attempt
+ }
+
+ final_validation = validation_history[-1]
+
+ return {
+ "refined_content": refined_content,
+ "locator_response": locator_response,
+ "target_indexes": target_indexes,
+ "edited_frames": edited_frames,
+ "slide_validation_status": "FAIL",
+ "slide_validation_errors": final_validation.get("errors", []),
+ "validation_history": validation_history,
+ "retries_used": max_retries
+ }
+ def validate_slide_patch(self, original_latex, refined_latex, edited_frames):
+
+ errors = []
+
+ required_markers = [
+ "\\documentclass",
+ "\\begin{document}",
+ "\\end{document}"
+ ]
+
+ for marker in required_markers:
+ if marker not in refined_latex:
+ errors.append(f"Missing required LaTeX marker: {marker}")
+
+ original_begin_frames = len(re.findall(r"\\begin\{frame", original_latex))
+ refined_begin_frames = len(re.findall(r"\\begin\{frame", refined_latex))
+
+ original_end_frames = len(re.findall(r"\\end\{frame\}", original_latex))
+ refined_end_frames = len(re.findall(r"\\end\{frame\}", refined_latex))
+
+ if original_begin_frames != refined_begin_frames:
+ errors.append(
+ f"Frame begin count changed unexpectedly: "
+ f"original={original_begin_frames}, refined={refined_begin_frames}"
+ )
+
+ if original_end_frames != refined_end_frames:
+ errors.append(
+ f"Frame end count changed unexpectedly: "
+ f"original={original_end_frames}, refined={refined_end_frames}"
+ )
+
+ try:
+ parsed_frames = LaTeXParser().parse(refined_latex)
+ if not parsed_frames:
+ errors.append(
+ "Refined slides could not be parsed by PPTX parser"
+ )
+ except Exception as e:
+ errors.append(f"PPTX parser failed: {str(e)}")
+
+ if "```" in refined_latex:
+ errors.append("Markdown code fences detected in refined slides")
+
+ if "TARGET_FRAMES:" in refined_latex:
+ errors.append("Locator prompt text leaked into refined slides")
+
+ original_frames = self.parse_frames(original_latex)
+ refined_frames = self.parse_frames(refined_latex)
+
+ edited_indexes = {
+ frame["index"]
+ for frame in edited_frames
+ }
+
+ for frame in original_frames:
+ idx = frame.get("index")
+ title = frame.get("title")
+
+ if idx in edited_indexes:
+ continue
+
+ if title not in refined_latex:
+ errors.append(
+ f"Unedited frame title missing after refinement: {title}"
+ )
+
+ for frame in refined_frames:
+ idx = frame.get("index")
+ body = frame.get("body")
+
+ if idx not in edited_indexes:
+ continue
+
+ if not body or not body.strip():
+ errors.append(
+ f"Edited frame body is empty for frame index {idx}"
+ )
+ continue
+
+ body_validation = self.validate_frame_body(body)
+ for error in body_validation["errors"]:
+ errors.append(
+ f"Edited frame {idx} body failed validation: {error}"
+ )
+
+ environments = [
+ "itemize",
+ "enumerate",
+ "block",
+ "columns",
+ "figure",
+ "equation"
+ ]
+
+ for env in environments:
+ begin_count = len(
+ re.findall(rf"\\begin\{{{env}\}}", refined_latex)
+ )
+
+ end_count = len(
+ re.findall(rf"\\end\{{{env}\}}", refined_latex)
+ )
+
+ if begin_count != end_count:
+ errors.append(
+ f"Unbalanced LaTeX environment '{env}': "
+ f"begin={begin_count}, end={end_count}"
+ )
+
+ if errors:
+ return {
+ "status": "FAIL",
+ "errors": errors
+ }
+
+ return {
+ "status": "PASS",
+ "errors": []
+ }
+
+ def validate_frame_body(self, body):
+ errors = []
+
+ if not body or not body.strip():
+ errors.append("Frame body is empty")
+ return {
+ "status": "FAIL",
+ "errors": errors
+ }
+
+ forbidden_patterns = [
+ ("\\begin{frame", "Frame body includes frame wrapper"),
+ ("\\end{frame}", "Frame body includes frame wrapper"),
+ ("\\frametitle", "Frame body includes frame title"),
+ ("```", "Markdown code fence detected"),
+ ("**", "Markdown bold syntax detected"),
+ ("###", "Markdown heading syntax detected"),
+ ("[Author", "Placeholder citation detected"),
+ ("[Cite", "Placeholder citation detected"),
+ ("needed_reference", "Placeholder citation key detected"),
+ ("\\cite{", "Citation command detected"),
+ ("\\footnote{", "Footnote attribution detected")
+ ]
+
+ for pattern, message in forbidden_patterns:
+ if pattern in body:
+ errors.append(message)
+
+ if self.get_max_list_nesting_depth(body) > 2:
+ errors.append("List nesting is too deep for a Beamer slide")
+
+ for env in ["itemize", "enumerate"]:
+ pattern = re.compile(
+ rf"\\begin\{{{env}\}}(.*?)\\end\{{{env}\}}",
+ re.DOTALL
+ )
+ for match in pattern.finditer(body):
+ if "\\item" not in match.group(1):
+ errors.append(
+ f"LaTeX environment '{env}' has no \\item entries"
+ )
+
+ for env in ["itemize", "enumerate", "block", "columns", "figure", "equation"]:
+ begin_count = len(re.findall(rf"\\begin\{{{env}\}}", body))
+ end_count = len(re.findall(rf"\\end\{{{env}\}}", body))
+
+ if begin_count != end_count:
+ errors.append(
+ f"Unbalanced LaTeX environment '{env}': "
+ f"begin={begin_count}, end={end_count}"
+ )
+
+ if not self.has_balanced_braces(body):
+ errors.append("Unbalanced curly braces detected")
+
+ if self.has_unescaped_ampersand(body):
+ errors.append("Unescaped ampersand detected")
+
+ if errors:
+ return {
+ "status": "FAIL",
+ "errors": errors
+ }
+
+ return {
+ "status": "PASS",
+ "errors": []
+ }
+
+ def get_max_list_nesting_depth(self, text):
+ max_depth = 0
+ current_depth = 0
+ token_pattern = re.compile(r"\\(begin|end)\{(itemize|enumerate)\}")
+
+ for match in token_pattern.finditer(text):
+ action = match.group(1)
+
+ if action == "begin":
+ current_depth += 1
+ max_depth = max(max_depth, current_depth)
+ else:
+ current_depth = max(0, current_depth - 1)
+
+ return max_depth
+
+ def has_balanced_braces(self, text):
+ cleaned = re.sub(r"\\[{}]", "", text)
+ return cleaned.count("{") == cleaned.count("}")
+
+ def has_unescaped_ampersand(self, text):
+ for line in text.splitlines():
+ if "\\begin{tabular" in line or "\\end{tabular" in line:
+ continue
+
+ for match in re.finditer("&", line):
+ if match.start() == 0 or line[match.start() - 1] != "\\":
+ return True
+
+ return False
+
+
+ def retry_refine_frame_body(
+ self,
+ frame_context,
+ feedback_text,
+ validation_errors
+ ):
+
+ prompt = f"""
+You are revising a previously edited Beamer slide frame body.
+Your previous refinement attempt failed deterministic validation.
+You must fix ONLY the validation issues while preserving useful edits.
+---
+EVALUATOR FEEDBACK:
+{feedback_text}
+---
+VALIDATION ERRORS:
+{validation_errors}
+---
+FRAME CONTEXT:
+{frame_context}
+---
+RULES:
+- Edit ONLY the TARGET FRAME body.
+- Preserve valid existing edits whenever possible.
+- Fix ONLY the reported validation failures.
+- Preserve valid Beamer LaTeX syntax.
+- Do NOT add unverifiable external-evidence claims or footnotes.
+- Do NOT invent outside materials, authors, dates, organizations, or locator keys.
+- Do NOT return frame wrappers.
+- Do NOT include markdown fences.
+- Return ONLY valid Beamer body content.
+---
+OUTPUT:
+
+Return ONLY the corrected TARGET FRAME body content.
+"""
+
+ messages = [{"role": "user", "content": prompt}]
+ response = self.llm.generate_response(messages)[0]
+ return response
+
+ def parse_frames(self, latex_content):
+ frame_pattern = r"\\begin{frame}.*?\\end{frame}"
+
+ frames = re.findall(
+ frame_pattern,
+ latex_content,
+ re.DOTALL
+ )
+
+ parsed_frames = []
+
+ for idx, frame in enumerate(frames):
+ title_match = re.search(
+ r"\\frametitle\{(.*?)\}",
+ frame
+ )
+ title = title_match.group(1) if title_match else "Untitled"
+ structure_match = re.search(
+ r"(\\begin\{frame\}(?:\[.*?\])?)\s*(\\frametitle\{.*?\})(.*?)(\\end\{frame\})",
+ frame,
+ re.DOTALL
+ )
+
+ if structure_match:
+ frame_start = structure_match.group(1).strip()
+ title_line = structure_match.group(2).strip()
+ body = structure_match.group(3).strip()
+ frame_end = structure_match.group(4).strip()
+
+ else:
+ frame_start = None
+ title_line = None
+ body = None
+ frame_end = None
+
+ parsed_frames.append({
+ "index": idx,
+ "title": title,
+ "content": frame,
+ "original_content": frame,
+ "frame_start": frame_start,
+ "title_line": title_line,
+ "body": body,
+ "frame_end": frame_end
+ })
+
+ return parsed_frames
+
+ def build_frame_summary(self, frames):
+ frame_summary_text = ""
+
+ for frame in frames:
+ index = frame.get("index")
+ title = frame.get("title")
+
+ frame_summary = f"Frame {index}: {title}\n"
+ frame_summary_text += frame_summary
+
+ return frame_summary_text
+
+
+ def locate_frames(self, feedback_text, frame_summary):
+ prompt = f"""
+You are a slide-deck reviewer.
+Your job is to identify which slide frames are MOST LIKELY responsible
+for the evaluator feedback.
+You are NOT rewriting slides.
+You are NOT evaluating the entire deck.
+You are ONLY locating likely problem regions.
+---
+FEEDBACK:
+{feedback_text}
+---
+FRAME SUMMARY:
+{frame_summary}
+---
+RULES:
+- Use the frame titles to infer which frames are most related to the feedback.
+- Select ONLY the frames most likely connected to the reported weaknesses.
+- Prefer precision over recall.
+- Do NOT select frames unless there is a reasonable connection to the feedback.
+- Keep the list compact.
+- Return a maximum of 5 frames.
+- If multiple adjacent frames appear related, include only the most relevant ones.
+- Use short reasoning phrases, not long explanations.
+---
+OUTPUT FORMAT (STRICT):
+TARGET_FRAMES:
+- Frame :
+- Frame :
+If no strong match exists:
+TARGET_FRAMES:
+- None confidently identified
+---
+Return ONLY the output.
+"""
+ messages = [{"role": "user", "content": prompt}]
+ response = self.llm.generate_response(messages)[0]
+ return response
+
+
+ def parse_target_frame_indexes(self, locator_response):
+ if not locator_response:
+ return []
+ index_pattern = r"Frame\s+(\d+)"
+ indexes = re.findall(index_pattern, locator_response)
+ return sorted(set(int(idx) for idx in indexes))
+
+ def get_target_frames(self, frames, target_indexes):
+
+ target_frames = []
+ for frame in frames:
+ index = frame.get("index")
+
+ if index in target_indexes:
+ target_frames.append(frame)
+
+ return target_frames
+
+ def build_target_frame_context(self, frames, target_frame):
+
+ idx = target_frame.get("index")
+
+ context_text = ""
+ # Previous frame
+ if idx > 0:
+ prev_frame = frames[idx - 1]
+ context_text += f"""
+ PREVIOUS FRAME:
+ Frame {prev_frame.get("index")}: {prev_frame.get("title")}
+ """
+
+ # Target frame
+ context_text += f"""
+ TARGET FRAME:
+ Frame {target_frame.get("index")}: {target_frame.get("title")}
+ {target_frame.get("content")}
+ """
+
+ # Next frame
+ if idx < len(frames) - 1:
+ next_frame = frames[idx + 1]
+ context_text += f"""
+ NEXT FRAME:
+ Frame {next_frame.get("index")}: {next_frame.get("title")}
+ """
+ return context_text.strip()
+
+
+
+ def refine_frame_body(self, frame_context, feedback_text):
+
+ prompt = f"""
+ You are a careful Beamer LaTeX slide editor.
+
+ Your job is to repair ONLY the BODY of one target frame using evaluator feedback.
+
+ You are NOT rewriting the slide deck.
+ You are NOT rewriting neighboring frames.
+ You are ONLY editing the body content of the TARGET FRAME.
+
+ ---
+
+ EVALUATOR FEEDBACK:
+ {feedback_text}
+
+ ---
+
+ FRAME CONTEXT:
+ {frame_context}
+
+ ---
+
+ RULES:
+
+ - Edit ONLY the TARGET FRAME body.
+ - Do NOT edit neighboring frames.
+ - Preserve useful existing body content whenever possible.
+ - Make the smallest useful changes needed to address the feedback.
+ - Keep the slide concise and presentation-friendly.
+ - Preserve valid Beamer LaTeX syntax.
+ - Preserve existing formatting structure when possible.
+ - Do NOT add unverifiable external-evidence claims or footnotes.
+ - Do NOT invent outside materials, authors, dates, organizations, or locator keys.
+ - Ignore feedback that requires unavailable outside evidence.
+ - Focus on clarity, alignment, examples, depth, structure, and learner accessibility.
+ - If reducing density, simplify or condense content instead of expanding it.
+ - Do not add unnecessary sections or filler content.
+
+ ---
+
+ IMPORTANT OUTPUT RULES:
+
+ - Do NOT return \\begin{{frame}}
+ - Do NOT return \\frametitle{{...}}
+ - Do NOT return \\end{{frame}}
+ - Do NOT include markdown fences.
+ - Return ONLY valid Beamer frame BODY content.
+ - Return ONLY the revised body for the TARGET FRAME.
+
+ ---
+
+ OUTPUT:
+
+ Return ONLY the revised TARGET FRAME body content.
+ """
+
+ messages = [{"role": "user", "content": prompt}]
+
+ response = self.llm.generate_response(messages)[0]
+
+ return response
+
+
+ def rebuild_frame(self, frame, new_body):
+ frame_start = frame.get("frame_start")
+ title_line = frame.get("title_line")
+ frame_end = frame.get("frame_end")
+
+ if not frame_start or not title_line or not frame_end:
+ return frame.get("content")
+
+ new_body = new_body.strip()
+
+ new_frame = f"""{frame_start}
+{title_line}
+{new_body}
+{frame_end}"""
+
+ return new_frame
+
+ def replace_frames(self, frames, frame_index, new_frame_content):
+ for frame in frames:
+ idx = frame.get("index")
+
+ if idx == frame_index:
+ frame["content"] = new_frame_content
+ return frames
+
+ def reassemble_slides(self, original_latex, frames):
+ updated_latex = original_latex
+
+ for frame in frames:
+ original_frame = frame.get("original_content")
+ current_frame = frame.get("content")
+
+ if not original_frame or not current_frame:
+ continue
+ if original_frame == current_frame:
+ continue
+
+ updated_latex = updated_latex.replace(
+ original_frame,
+ current_frame,
+ 1
+ )
+ return updated_latex
+
+
diff --git a/src/textbook_reference.py b/src/textbook_reference.py
new file mode 100644
index 0000000..5d5eea3
--- /dev/null
+++ b/src/textbook_reference.py
@@ -0,0 +1,312 @@
+"""
+Utilities for turning lightweight textbook references into catalog data.
+
+This module intentionally avoids LLM calls. It extracts a bounded amount of
+text from a user-provided textbook PDF/text file and converts it into the
+existing catalog schema so the ADDIE workflow can consume it without changing
+the core generation pipeline.
+"""
+from __future__ import annotations
+
+import re
+from collections import Counter
+from pathlib import Path
+from typing import Any, Dict, List
+#处理PDF
+try:
+ import pdfplumber
+except ImportError:
+ pdfplumber = None
+
+try:
+ import PyPDF2
+except ImportError:
+ PyPDF2 = None
+
+try:
+ import pypdf
+except ImportError:
+ pypdf = None
+
+#提取时过滤无意义词
+STOPWORDS = {
+ "about", "after", "again", "also", "because", "before", "being", "between",
+ "chapter", "course", "could", "during", "example", "first", "from", "have",
+ "into", "learn", "learning", "more", "most", "other", "page", "section",
+ "should", "students", "that", "their", "these", "this", "through", "using",
+ "with", "would", "your", "或者", "以及", "学习", "学生", "课程", "章节",
+}
+
+
+class TextbookReferenceBuilder:
+ """Build catalog-compatible reference data from a small textbook excerpt."""
+
+ def __init__(self, max_pages: int = 20, max_chars: int = 40000):#限制最大处理页数、最大提取字符数，轻量
+ self.max_pages = max_pages
+ self.max_chars = max_chars
+
+ def build_catalog(self, source_path: Path, course_name: str = "") -> Dict[str, Any]:
+ text_by_page = self.extract_text(source_path)
+ full_text = "\n".join(page["text"] for page in text_by_page).strip()
+ bounded_text = full_text[: self.max_chars]#提取文本
+ #提取章节
+ chapters = self.extract_chapters_from_toc(text_by_page)
+ if not chapters:
+ chapters = self.extract_chapters(bounded_text)
+ key_topics = self.extract_key_topics(bounded_text)
+ summary = self.summarize_excerpt(bounded_text, key_topics)
+ weekly_outline = self.build_weekly_outline(chapters, key_topics)
+ source_name = source_path.name
+ #返回适配catalog
+ return {
+ "student_profile": {
+ "student_background": "Students are assumed to be new to the course topic and will benefit from textbook-aligned explanations.",
+ "aggregate_academic_performance": "Readiness is inferred from the uploaded textbook scope rather than historical grade data.",
+ "anticipated_learner_needs_and_barriers": (
+ "Materials should introduce concepts progressively, reuse textbook terminology, "
+ "and provide examples connected to the reference chapters."
+ ),
+ },
+ "instructor_preferences": {
+ "instructor_emphasis_intent": "Use the uploaded textbook as the primary reference material for topic selection and examples.",
+ "instructor_style_preferences": "Keep generated materials aligned with textbook chapter order and terminology.",
+ "instructor_focus_for_assessment": "Assess whether students can explain and apply the key textbook concepts.",
+ },
+ "course_structure": {
+ "course_learning_outcomes": self.build_learning_outcomes(key_topics),
+ "total_number_of_weeks": str(max(1, min(len(chapters) or 6, 12))),
+ "weekly_schedule_outline": weekly_outline,
+ "textbook_reference_summary": summary,
+ "required_readings": self.build_required_readings(source_name, chapters),
+ },
+ "assessment_design": {
+ "assessment_format_preferences": "Short concept checks, applied exercises, and a final synthesis task grounded in the textbook reference.",
+ "assessment_delivery_constraints": "Assessments should cite or refer back to the uploaded textbook chapters where relevant.",
+ },
+ "teaching_constraints": {
+ "platform_policy_constraints": "Generated content should avoid quoting long textbook passages verbatim.",
+ "ta_support_availability": "No additional teaching assistant support is assumed for this prototype.",
+ "instructional_delivery_context": "Textbook-guided course generation prototype.",
+ "max_slide_count": "10",
+ },
+ "institutional_requirements": {
+ "program_learning_outcomes": "Materials should align textbook concepts with practical learning outcomes.",
+ "academic_policies_and_institutional_standards": "Respect copyright by summarizing and paraphrasing reference material.",
+ "department_syllabus_requirements": "Include textbook-derived topics, learning objectives, and assessment alignment.",
+ },
+ "prior_feedback": {
+ "historical_course_evaluation_results": "No prior feedback supplied; this catalog was generated from textbook reference content.",
+ },
+ "textbook_reference": {
+ "source": source_name,
+ "course_name": course_name,
+ "pages_processed": len(text_by_page),
+ "characters_used": len(bounded_text),
+ "key_topics": key_topics,
+ "detected_chapters": chapters,
+ "excerpt_summary": summary,
+ "sample_excerpt": bounded_text[:1500],
+ },
+ }
+
+ def extract_text(self, source_path: Path) -> List[Dict[str, Any]]:
+ suffix = source_path.suffix.lower()
+ if suffix == ".pdf":
+ return self.extract_pdf_text(source_path)
+ if suffix in {".txt", ".md"}:
+ text = source_path.read_text(encoding="utf-8", errors="ignore")
+ return [{"page": 1, "text": text[: self.max_chars]}]
+ raise ValueError("Only PDF, TXT, and Markdown textbook references are supported.")
+
+ def extract_pdf_text(self, pdf_path: Path) -> List[Dict[str, Any]]:
+ pages: List[Dict[str, Any]] = []
+
+ if pdfplumber is not None:
+ try:
+ with pdfplumber.open(str(pdf_path)) as pdf:
+ for index, page in enumerate(pdf.pages[: self.max_pages], start=1):
+ text = page.extract_text() or ""
+ if text.strip():
+ pages.append({"page": index, "text": text})
+ if pages:
+ return pages
+ except Exception as exc:
+ print(f"[textbook_reference] pdfplumber extraction failed, falling back to PyPDF2: {exc}")
+ pages = []
+
+ if PyPDF2 is not None:
+ with pdf_path.open("rb") as file:
+ reader = PyPDF2.PdfReader(file)
+ for index, page in enumerate(reader.pages[: self.max_pages], start=1):
+ text = page.extract_text() or ""
+ if text.strip():
+ pages.append({"page": index, "text": text})
+
+ if not pages and pypdf is not None:
+ with pdf_path.open("rb") as file:
+ reader = pypdf.PdfReader(file)
+ for index, page in enumerate(reader.pages[: self.max_pages], start=1):
+ text = page.extract_text() or ""
+ if text.strip():
+ pages.append({"page": index, "text": text})
+
+ if not pages:
+ raise ValueError("Could not extract readable text from the textbook file.")
+
+ return pages
+ #优先扫描前6页找目录，提取效率更高
+ def extract_chapters_from_toc(self, text_by_page: List[Dict[str, Any]]) -> List[Dict[str, str]]:
+ """Scan each page for TOC-like patterns before falling back to full-text extraction."""
+ toc_chapter_pattern = re.compile(
+ r"(?:Chapter\s*\d+|第\s*[一二三四五六七八九十\d]+\s*章|\d+\.\d+(?:\.\d+)?)\s*.{3,80}",
+ flags=re.IGNORECASE,
+ )#正则覆盖常见的目录标题模式
+ seen = set()
+ chapters: List[Dict[str, str]] = []
+
+ for page in text_by_page[:6]: # TOC is usually in the first few pages
+ lines = [re.sub(r"\s+", " ", line).strip() for line in page["text"].splitlines()]
+ toc_candidates = [
+ line for line in lines
+ if toc_chapter_pattern.fullmatch(line)
+ and not line.endswith((".", ",", ";", ":"))
+ and len(line.split()) <= 14
+ ]
+ if len(toc_candidates) >= 3: # Looks like a TOC page
+ for line in toc_candidates:
+ # Strip trailing page number (e.g., "Chapter 1: Introduction 42" → "Chapter 1: Introduction")
+ cleaned = re.sub(r"\s+\d{1,4}\s*$", "", line).strip()
+ if cleaned and len(cleaned) >= 5:
+ normalized = cleaned.lower()
+ if normalized not in seen:
+ seen.add(normalized)
+ chapters.append({"title": cleaned})
+ if chapters:
+ break # Found a good TOC page, stop scanning
+
+ return chapters[:12]
+ #如果没找到目录页就从全文中提取章节标题，兜底方案
+ def extract_chapters(self, text: str) -> List[Dict[str, str]]:
+ seen = set()
+ chapters: List[Dict[str, str]] = []
+ lines = [re.sub(r"\s+", " ", line).strip() for line in text.splitlines()]
+
+ for index, line in enumerate(lines):
+ title = None
+
+ if line.upper() == "CHAPTER" and index > 0 and index + 1 < len(lines):
+ previous_line = lines[index - 1] #单独的CHAPTER行+ 前后行
+ next_line = lines[index + 1]
+ if re.fullmatch(r"\d{1,3}", previous_line) and self._looks_like_heading(next_line):
+ title = f"Chapter {previous_line}: {next_line}"
+
+ if title is None:#数字层级标题
+ section_match = re.fullmatch(r"(\d+\.\d+(?:\.\d+)?)\s+(.{3,80})", line)
+ section_title = section_match.group(2) if section_match else ""
+ if (
+ section_match
+ and not re.search(r"\s\d{2,4}$", section_title)
+ and self._looks_like_heading(section_title)
+ ):
+ title = line
+
+ if title is None:#英文章节标题
+ chapter_match = re.fullmatch(r"(Chapter\s*\d+)\s*[:.\-\s]\s*(.{3,80})", line, flags=re.IGNORECASE)
+ chapter_title = chapter_match.group(2) if chapter_match else ""
+ if (
+ chapter_match
+ and self._looks_like_heading(chapter_title)
+ and self._looks_like_chapter_title(chapter_title)
+ ):
+ title = f"{chapter_match.group(1)}: {chapter_title}"
+
+ if title is None:#中文章节标题
+ chinese_match = re.fullmatch(r"(第\s*[一二三四五六七八九十\d]+\s*章)\s*(.{0,80})", line)
+ if chinese_match:
+ suffix = chinese_match.group(2).strip()
+ title = f"{chinese_match.group(1)} {suffix}".strip()
+
+ if title:
+ normalized = title.lower()
+ if normalized not in seen:
+ seen.add(normalized)
+ chapters.append({"title": title})
+ if len(chapters) >= 12:
+ break
+
+ return chapters
+
+ def _looks_like_heading(self, text: str) -> bool:#通过文本特征过滤掉非标题行，提升章节提取的准确性
+ text = text.strip()
+ if not 3 <= len(text) <= 100:
+ return False
+ if text.endswith((".", ",", ";", ":")):
+ return False
+ words = text.split()
+ if len(words) > 12: #单词数限制最多12个词，标题简洁
+ return False
+ alpha_chars = re.findall(r"[A-Za-z\u4e00-\u9fff]", text)
+ if len(alpha_chars) < 3:
+ return False
+ return True
+
+ def _looks_like_chapter_title(self, text: str) -> bool:
+ first_word = text.split()[0] if text.split() else ""
+ if first_word and first_word[0].islower():
+ return False #首单词首字母不能小写，标题通常首字母大写
+ if re.search(r"\b(discussed|are|is|was|were|will|can|cannot|should|examples?)\b", text, flags=re.IGNORECASE):
+ return False #排除包含无意义动词的行
+ if re.search(r"\s\d{2,4}$", text):
+ return False
+ return True
+ #基于于词频提取关键词，出现次数越多，越可能是核心主题
+ def extract_key_topics(self, text: str, limit: int = 12) -> List[str]:
+ words = re.findall(r"[A-Za-z][A-Za-z\-]{3,}|[\u4e00-\u9fff]{2,}", text.lower())
+ candidates = [
+ word.strip("-")
+ for word in words
+ if len(word) > 3 and word not in STOPWORDS and not word.isdigit()
+ ]
+ counts = Counter(candidates)
+ return [word for word, _ in counts.most_common(limit)]
+
+ def summarize_excerpt(self, text: str, key_topics: List[str]) -> str:
+ sentences = re.split(r"(?<=[.!?。！？])\s+", text.replace("\n", " "))
+ clean_sentences = [re.sub(r"\s+", " ", sentence).strip() for sentence in sentences]
+ clean_sentences = [sentence for sentence in clean_sentences if 40 <= len(sentence) <= 260]
+
+ selected = clean_sentences[:3]
+ topic_text = ", ".join(key_topics[:6]) if key_topics else "core textbook concepts"
+ if selected:
+ return f"Main detected topics include {topic_text}. Representative excerpt themes: " + " ".join(selected)
+ return f"Main detected topics include {topic_text}."
+
+ def build_learning_outcomes(self, key_topics: List[str]) -> str:
+ topics = key_topics[:6] or ["core concepts", "key methods", "applications"]
+ return (
+ "By the end of the course, students should be able to explain, compare, "
+ f"and apply textbook concepts including {', '.join(topics)}."
+ )
+
+ def build_weekly_outline(self, chapters: List[Dict[str, str]], key_topics: List[str]) -> str:
+ if chapters:
+ outline_items = [
+ f"Module {index}: {chapter['title']}"
+ for index, chapter in enumerate(chapters[:12], start=1)
+ ]
+ else: #无章节按关键词生成
+ topics = key_topics[:8] or ["Textbook overview", "Core concepts", "Applications"]
+ outline_items = [
+ f"Module {index}: {topic.title()}"
+ for index, topic in enumerate(topics, start=1)
+ ]
+ return "; ".join(outline_items)
+
+ def build_required_readings(self, source_name: str, chapters: List[Dict[str, str]]) -> str:
+ if chapters:
+ readings = [
+ f"{source_name}, {chapter['title']}"
+ for chapter in chapters[:8]
+ ]
+ return "; ".join(readings)
+ return f"Selected excerpts from {source_name}"