# app/core/books/content_services.py import json from copy import deepcopy from concurrent.futures import ThreadPoolExecutor from typing import Dict, Any, List, Tuple from .content_crew_config import safe_run_once, build_rag_block from agents.books.book_content import make_writer_crew, make_quality_crew from agents.books.book_content_gpt import make_writer_crew_gpt, make_quality_crew_gpt # from agents.books import make_writer_editor_crew MAX_CHAPTER_WORKERS = 3 # لو True: quality يقدر يولد محتوى حتى لو writer رجّع فاضي # لو False: quality يراجع فقط، ولو draft فاضي يفضل فاضي ALLOW_QUALITY_FROM_EMPTY_DRAFT = False # مفاتيح لازمة للـ writer prompt WRITER_REQUIRED_KEYS = [ "book_title", "book_description", "target_audience", "chapter_title", "chapter_outcome", "section_title", "section_overview", "subsection_title", "subsection_purpose", "style_profile_json", "writing_mode", "tone", "target_pages", "core_idea", "key_points", "scope_limit", "avoid_overlap_with", "depth_level", # "citations_required", "planner_notes", "rag_block", ] def _iter_section_plans(section: dict): """ يدعم الشكل الجديد: section["plans"] = { "عنوان المحور": { ...plan... }, ... } + fallback للشكل القديم: section["subsections"] = [ {...}, ... ] """ plans = section.get("plans") if isinstance(plans, dict) and plans: for subsection_title, plan in plans.items(): yield subsection_title, plan if isinstance(plan, dict) else {} return # fallback legacy for sub in section.get("subsections", []): subsection_title = sub.get("title") or sub.get("subsection_title") or "" yield subsection_title, sub if isinstance(sub, dict) else {} def _normalize_plan(raw_plan: dict) -> dict: """ يوحّد شكل plan سواء جاي بالشكل الجديد أو القديم. """ if not isinstance(raw_plan, dict): return {} # الشكل القديم: subsection فيه planning + rag_context if "planning" in raw_plan and isinstance(raw_plan.get("planning"), dict): planning = deepcopy(raw_plan.get("planning", {})) planning.setdefault("purpose", raw_plan.get("purpose", "")) planning.setdefault("rag_context", raw_plan.get("rag_context", {})) planning.setdefault("status", raw_plan.get("status", "ready")) return planning # الشكل الجديد: plan جاهز مباشرة return deepcopy(raw_plan) def _as_json_string(value, fallback=None) -> str: if value is None: value = fallback if fallback is not None else {} if isinstance(value, str): return value try: return json.dumps(value, ensure_ascii=False) except Exception: return str(value) def _as_text(value) -> str: if value is None: return "" return str(value).strip() def _as_text_list(value) -> str: """ تحويل list/string إلى نص ثابت للـ interpolation. """ if value is None: return "" if isinstance(value, list): cleaned = [str(x).strip() for x in value if str(x).strip()] return " | ".join(cleaned) return str(value).strip() def _ensure_required_inputs(inputs: dict, required_keys: List[str], ctx: str = ""): missing = [k for k in required_keys if k not in inputs] if missing: prefix = f"[{ctx}] " if ctx else "" raise ValueError(f"{prefix}Missing required input keys: {missing}") def build_inputs( book_meta: Dict[str, Any], chapter: Dict[str, Any], section: Dict[str, Any], subsection_title: str, plan: Dict[str, Any], ) -> Dict[str, Any]: chapter_intro = chapter.get("chapter_intro", {}) if not isinstance(chapter_intro, dict): chapter_intro = {} rag_context = plan.get("rag_context", {}) if not isinstance(rag_context, dict): rag_context = {} rag_block = build_rag_block(rag_context.get("chunks", [])) # style_profile غالبًا على مستوى chapter style_profile = chapter.get("style_profile") if style_profile is None: style_profile = section.get("style_profile") # chapter_outcome fallback من chapter_intro chapter_outcome = chapter.get("outcome") or chapter_intro.get( "why_this_chapter_now", "" ) safe_subsection_title = ( _as_text(subsection_title) or _as_text(plan.get("subsection_title", "")) or "محور فرعي" ) return { "book_title": _as_text(book_meta.get("book_title", "")), "book_description": _as_text(book_meta.get("book_description", "")), "target_audience": _as_text(book_meta.get("target_audience", "")), "chapter_title": _as_text(chapter.get("chapter_title", "")), "chapter_outcome": _as_text(chapter_outcome), "section_title": _as_text( section.get("section_title", "") or section.get("title", "") ), "section_overview": _as_text(section.get("overview", "")), "subsection_title": safe_subsection_title, "subsection_purpose": _as_text(plan.get("purpose", "")), "style_profile_json": _as_json_string(style_profile, fallback={}), "writing_mode": _as_text(plan.get("writing_mode", "")), "tone": _as_text(plan.get("tone", "")), "target_pages": plan.get("estimated_pages", 1) or 1, "core_idea": _as_text(plan.get("core_idea", "")), "key_points": _as_text_list(plan.get("key_points", [])), "scope_limit": _as_text(plan.get("scope_limit", "")), "avoid_overlap_with": _as_text_list(plan.get("avoid_overlap_with", [])), "depth_level": _as_text(plan.get("depth_level", "")), # "citations_required": bool(plan.get("citations_required", False)), "planner_notes": _as_text(plan.get("notes", "")), "rag_block": _as_text(rag_block), # اختياري "chapter_intro_json": _as_json_string(chapter_intro, fallback={}), } def process_chapter(chapter: dict, book_meta: dict, chapter_idx: int) -> dict: """ معالجة فصل كامل (sequential داخله) — تُستدعى داخل Thread. """ print(f"\n📘 Processing Chapter {chapter_idx}") chapter_out = { "chapter_title": chapter.get("chapter_title", ""), "chapter_intro": deepcopy(chapter.get("chapter_intro", {})), "style_profile": deepcopy(chapter.get("style_profile", {})), "sections": [], } if chapter.get("outcome") is not None: chapter_out["outcome"] = chapter.get("outcome", "") for section in chapter.get("sections", []): section_title = section.get("section_title", "") or section.get("title", "") section_out = { "section_title": section_title, "overview": section.get("overview", ""), "section_intro_paragraph": section.get("section_intro_paragraph", ""), "section_closing_paragraph": section.get("section_closing_paragraph", ""), "plans": {}, } writer_crew = make_writer_crew() quality_crew = make_quality_crew() # writer_editor_crew = make_writer_editor_crew() for subsection_title, raw_plan in _iter_section_plans(section): plan = _normalize_plan(raw_plan) print(f"🟢 Writing: {subsection_title}") plan_out = deepcopy(plan) try: inputs = build_inputs( book_meta, chapter, section, subsection_title, plan ) _ensure_required_inputs( inputs, WRITER_REQUIRED_KEYS, ctx=f"chapter={chapter_idx}, section={section_title}, subsection={subsection_title}", ) # 1) Writer + retry (داخل safe_run_once) # generated, sources = safe_run_once(writer_editor_crew, inputs) generated, sources = safe_run_once(writer_crew, inputs) content_before_review = generated sources_before_review = sources # 2) Quality gate reviewed_generated, reviewed_sources = "", [] if generated.strip() or ALLOW_QUALITY_FROM_EMPTY_DRAFT: # if reviewed_generated.strip() or ALLOW_QUALITY_FROM_EMPTY_DRAFT: draft_json = json.dumps( {"generated_content": generated, "sources": sources}, ensure_ascii=False, ) quality_inputs = dict(inputs) quality_inputs["draft_json"] = draft_json reviewed_generated, reviewed_sources = safe_run_once( quality_crew, quality_inputs ) if reviewed_generated.strip(): generated = reviewed_generated sources = reviewed_sources plan_out["generated_content"] = content_before_review plan_out["sources"] = sources_before_review plan_out["generated_content_reviewed"] = generated plan_out["sources_reviewed"] = sources plan_out["content_status"] = ( "generated" if generated.strip() else "empty" ) except Exception as e: plan_out["generated_content"] = "" plan_out["sources"] = [] plan_out["generated_content_reviewed"] = "" plan_out["sources_reviewed"] = [] plan_out["content_status"] = "error" plan_out["error"] = str(e) section_out["plans"][subsection_title] = plan_out chapter_out["sections"].append(section_out) return chapter_out def process_book(data: dict) -> dict: """ - parallel على مستوى chapters - الحفاظ على ترتيب chapters """ book_meta = { "book_title": data.get("book_title", ""), "book_description": data.get("book_description", ""), "target_audience": data.get("target_audience", ""), } chapters = data.get("chapters", []) chapters_out = [None] * len(chapters) futures = [] with ThreadPoolExecutor(max_workers=MAX_CHAPTER_WORKERS) as executor: for idx, chapter in enumerate(chapters): future = executor.submit(process_chapter, chapter, book_meta, idx + 1) futures.append((idx, future)) for idx, future in futures: try: chapters_out[idx] = future.result() except Exception as e: ch = chapters[idx] if idx < len(chapters) else {} chapters_out[idx] = { "chapter_title": ch.get("chapter_title", f"Chapter {idx+1}"), "sections": [], "error": str(e), } return { "message": "Book Content Generated Successfully 🚀", "content_phase": "generated", **book_meta, "chapters": chapters_out, }