# system_manager.py # -*- coding: utf-8 -*- from __future__ import annotations import json import os from typing import Any, Dict, List, Optional, Tuple from ai_task_manager import AITaskManager, TaskStatus from book_preparation_manager import BookPreparationManager from report_manager import ReportManager from storage_manager import StorageManager from voice_manager import VoiceManager VERSION = "system_manager_v3_2_dialogue_existing_ai_role_router" class SystemManager: """ المنسق المركزي الوحيد لمحطة Aiocr Radar. المسؤوليات: - المحافظة على عقد الـ API القديم بدون SessionManager منفصل. - توجيه عمليات الخزنة القديمة. - تنسيق تحضير الكتب وتقسيمها وتحويلها. - تنسيق الوصول إلى VoiceManager وStorageManager. - إرسال كل انتقال مهم إلى ReportManager. لا ينفذ عمل المدراء المتخصصين بدلاً منهم. """ def __init__( self, *, storage_manager: StorageManager, report_manager: ReportManager, voice_manager: VoiceManager, task_manager: AITaskManager, book_preparation_manager: BookPreparationManager, ai_manager: Optional[Any] = None, text_quality_manager: Optional[Any] = None, smart_editor_manager: Optional[Any] = None, ): self.storage = storage_manager self.reports = report_manager self.voice = voice_manager self.tasks = task_manager self.books = book_preparation_manager self.ai = ai_manager self.text_quality = text_quality_manager self.smart_editor = smart_editor_manager self.report_repo_key = os.getenv("REPORTS_REPO_KEY", "aiocr_assistant").strip() or "aiocr_assistant" def _persist_trace_report( self, trace_id: str, *, repo_key: str = "", book_id: str = "", engine: str = "", ) -> str: """يحفظ التقرير دون أن يجعل نجاح التقرير شرطاً لنجاح العملية الأصلية.""" if not trace_id: return "" target_repo = (repo_key or self.report_repo_key).strip() try: local_path = self.reports.save_local_report(trace_id=trace_id) if book_id and engine: target_path = self.storage.book_report_path(book_id, engine, trace_id) elif book_id: target_path = self.storage.book_operation_report_path(book_id, trace_id) else: target_path = self.storage.system_report_path(trace_id) self.storage.upload_file(target_repo, local_path, target_path) return target_path except Exception as exc: try: self.reports.record( event="report.persist.failed", trace_id=trace_id, manager="SystemManager", status="error", repo_key=target_repo, error=str(exc), ) except Exception: pass return "" # ------------------------------------------------------------------ # Trace / transitions # ------------------------------------------------------------------ def new_trace( self, *, book_id: str = "", task_id: str = "", metadata: Optional[Dict[str, Any]] = None, ) -> str: return self.reports.new_trace( book_id=book_id, task_id=task_id, metadata=metadata, ) def _transition( self, trace_id: str, from_manager: str, to_manager: str, **kwargs: Any, ) -> None: self.reports.transition( trace_id=trace_id, from_manager=from_manager, to_manager=to_manager, **kwargs, ) # ------------------------------------------------------------------ # API القديم - انتقل من SessionManager إلى SystemManager # ------------------------------------------------------------------ def process_api( self, text: str, engine: str, voice_name: str = "", speaker_name: str = "", *, provider_options: Optional[Dict[str, Any]] = None, trace_id: str = "", ) -> Tuple[str, Dict[str, Any]]: """ نفس وظيفة SessionManager.process_api القديمة، لكن تحت المدير المركزي. العقد الخارجي لا يتغير: يرجع (path, metadata). """ trace_id = self.reports.ensure_trace( trace_id, metadata={"operation": "api_tts", "engine": engine}, ) self._transition( trace_id, "api", "AITaskManager", event="api.task.requested", engine=engine, ) task = self.tasks.create_task( text=text, source_type="api", engine=engine, voice_name=voice_name, speaker_name=speaker_name, provider_options=dict(provider_options or {}), trace_id=trace_id, ) self.reports.record( event="task.created", trace_id=trace_id, manager="AITaskManager", task_id=task.task_id, engine=engine, metadata={"source_type": "api"}, ) path, metadata = self.tasks.execute(task) metadata = dict(metadata) metadata["input_source"] = "api" metadata["trace_id"] = trace_id self.reports.close_trace(trace_id, metadata={"operation": "api_tts"}) metadata["report_repo_path"] = self._persist_trace_report(trace_id) return path, metadata def process_vault_file( self, repo_key: str = "aiocr_assistant", text_file: Optional[str] = None, engine: str = "microsoft", voice_name: str = "", speaker_name: str = "", *, provider_options: Optional[Dict[str, Any]] = None, trace_id: str = "", ) -> Tuple[str, Dict[str, Any]]: """ نفس وظيفة SessionManager.process_vault_file القديمة. يقرأ TXT من الخزنة، ينشئ مهمة، يولد الصوت، ثم يحفظ الناتج. """ trace_id = self.reports.ensure_trace( trace_id, metadata={ "operation": "vault_tts", "engine": engine, "repo_key": repo_key, "text_file": text_file or "", }, ) lock = self.storage.get_repository_lock(repo_key) with lock, self.storage.report_context(trace_id=trace_id): self._transition( trace_id, "SystemManager", "StorageManager", event="vault.text.requested", engine=engine, repo_key=repo_key, ) text = self.storage.read_text_file(repo_key, text_file) self._transition( trace_id, "StorageManager", "AITaskManager", event="vault.task.requested", engine=engine, repo_key=repo_key, ) task = self.tasks.create_task( text=text, source_type="vault", engine=engine, voice_name=voice_name, speaker_name=speaker_name, repo_key=repo_key, provider_options=dict(provider_options or {}), trace_id=trace_id, ) self.reports.record( event="task.created", trace_id=trace_id, manager="AITaskManager", task_id=task.task_id, engine=engine, repo_key=repo_key, metadata={"source_type": "vault", "input_file": text_file or ""}, ) path, metadata = self.tasks.execute(task) task.status = TaskStatus.SAVING output_path = self.storage.resolve_output_path(repo_key, text_file) self._transition( trace_id, "AITaskManager", "StorageManager", event="vault.audio.persist.requested", task_id=task.task_id, engine=engine, repo_key=repo_key, ) saved_path = self.storage.upload_audio( repo_key, path, output_path, ) task.status = TaskStatus.DONE metadata = dict(metadata) metadata["input_source"] = "vault" metadata["input_file"] = text_file metadata["repo"] = self.storage.describe_repository(repo_key) metadata["output_file"] = saved_path metadata["status"] = TaskStatus.DONE.value metadata["trace_id"] = trace_id self.reports.close_trace(trace_id, metadata={"operation": "vault_tts"}) metadata["report_repo_path"] = self._persist_trace_report(trace_id, repo_key=repo_key) return path, metadata def fetch_vault_audio(self, repo_key: str, audio_file: str) -> str: trace_id = self.reports.new_trace( metadata={ "operation": "fetch_vault_audio", "repo_key": repo_key, "audio_file": audio_file, } ) self._transition( trace_id, "SystemManager", "StorageManager", event="vault.audio.fetch.requested", repo_key=repo_key, path=audio_file, ) with self.storage.report_context(trace_id=trace_id): path = self.storage.download_audio_file(repo_key, audio_file) self.reports.close_trace(trace_id, metadata={"operation": "fetch_vault_audio"}) self._persist_trace_report(trace_id, repo_key=repo_key) return path # ------------------------------------------------------------------ # النص / الكتاب # ------------------------------------------------------------------ def save_prepared_book( self, *, repo_key: str, title: str, original_text: str, prepared_text: str, source_manifest: Optional[List[Dict[str, Any]]] = None, book_id: Optional[str] = None, trace_id: str = "", ) -> Tuple[Dict[str, Any], str]: trace_id = self.reports.ensure_trace( trace_id, book_id=book_id or "", metadata={"operation": "prepare_book"}, ) self._transition( trace_id, "app", "BookPreparationManager", event="book.prepare.requested", book_id=book_id or "", ) with self.storage.report_context(trace_id=trace_id, book_id=book_id or ""): manifest = self.books.save_prepared_book( text=prepared_text, title=title, repo_key=repo_key, source_manifest=source_manifest or [], original_text=original_text, book_id=book_id, ) real_book_id = str(manifest.get("book_id") or book_id or "") self.reports.record( event="book.prepare.completed", trace_id=trace_id, manager="BookPreparationManager", book_id=real_book_id, metadata={"analysis": manifest.get("analysis") or {}}, ) self._transition( trace_id, "BookPreparationManager", "StorageManager", event="book.persisted", book_id=real_book_id, ) self.reports.close_trace(trace_id, metadata={"operation": "prepare_book", "book_id": real_book_id}) manifest = dict(manifest) manifest["report_repo_path"] = self._persist_trace_report( trace_id, repo_key=repo_key, book_id=real_book_id ) return manifest, trace_id def prepare_segmentation( self, *, repo_key: str, book_id: str, engine: str, trace_id: str = "", ) -> Tuple[Dict[str, Any], str]: trace_id = self.reports.ensure_trace( trace_id, book_id=book_id, metadata={"operation": "segment_book", "engine": engine}, ) self._transition( trace_id, "app", "BookPreparationManager", event="book.segmentation.requested", book_id=book_id, engine=engine, ) with self.storage.report_context(trace_id=trace_id, book_id=book_id): manifest = self.books.prepare_segmentation( repo_key=repo_key, book_id=book_id, engine=engine, ) self.reports.record( event="book.segmentation.completed", trace_id=trace_id, manager="BookPreparationManager", book_id=book_id, engine=engine, metadata={ "segments_count": len(manifest.get("segments") or []), "batches_count": manifest.get("batches_count", 0), }, ) self.reports.close_trace(trace_id, metadata={"operation": "segment_book", "book_id": book_id, "engine": engine}) manifest = dict(manifest) manifest["report_repo_path"] = self._persist_trace_report( trace_id, repo_key=repo_key, book_id=book_id ) return manifest, trace_id # ------------------------------------------------------------------ # المحرر الذكي / جودة النص / الإدخال الصوتي # ------------------------------------------------------------------ def apply_text_quality( self, operation: str, text: str, *, language: str = "ar", diacritics_mode: str = "", trace_id: str = "", ) -> Tuple[str, str]: if self.text_quality is None: raise RuntimeError("TextQualityManager غير موصول بـ SystemManager.") clean = str(text or "").strip() if not clean: raise ValueError("لا يوجد نص للمعالجة.") trace_id = self.reports.ensure_trace( trace_id, metadata={"operation": "text_quality", "quality_operation": operation} ) self._transition(trace_id, "app", "TextQualityManager", event="text.quality.requested") try: result = self.text_quality.apply( operation, clean, language=language, diacritics_mode=diacritics_mode ) self.reports.record( event="text.quality.completed", trace_id=trace_id, manager="TextQualityManager", metadata={"quality_operation": operation, "language": language}, ) self.reports.close_trace(trace_id, metadata={"operation": "text_quality"}) return str(result or ""), trace_id except Exception as exc: self.reports.record( event="text.quality.failed", trace_id=trace_id, manager="TextQualityManager", status="error", error=str(exc), metadata={"quality_operation": operation}, ) self.reports.close_trace(trace_id, status="error", error=str(exc), metadata={"operation": "text_quality"}) raise def transcribe_editor_audio( self, audio_path: str, *, language: str = "ar", trace_id: str = "", ) -> Tuple[str, str]: selected = str(audio_path or "").strip() if not selected: raise ValueError("لا يوجد تسجيل صوتي للإدخال.") trace_id = self.reports.ensure_trace( trace_id, metadata={"operation": "editor_stt", "language": language} ) self._transition(trace_id, "app", "VoiceManager", event="editor.stt.requested") try: # STT ليس TTS: نبحث أولاً في AIManager لأنه يملك وظائف النماذج، # ثم ندعم VoiceManager فقط للتوافق مع نسخة قديمة قد تحتوي STT. fn = None owner = None for manager in (self.ai, self.voice): if manager is None: continue for name in ("transcribe_audio", "transcribe", "speech_to_text", "recognize_speech"): candidate = getattr(manager, name, None) if callable(candidate): fn = candidate owner = manager.__class__.__name__ break if fn is not None: break if fn is None: raise RuntimeError( "بوابة STT غير موجودة في AIManager الحالي. " "وظيفة arabic_speech_to_text معرفة في ai_models.json، لكن يلزم أن يوفّر AIManager دالة transcribe_audio/speech_to_text لتنفيذها." ) try: result = fn(selected, language=language, trace_id=trace_id) except TypeError: try: result = fn(selected, language=language) except TypeError: result = fn(selected) if isinstance(result, tuple): text = str(result[0] or "").strip() elif isinstance(result, dict): text = str(result.get("text") or result.get("transcript") or result.get("result") or "").strip() else: text = str(result or "").strip() if not text: raise RuntimeError("لم يُرجع محرك التعرف الصوتي نصاً.") self.reports.record( event="editor.stt.completed", trace_id=trace_id, manager="VoiceManager", metadata={"language": language, "characters": len(text)}, ) self.reports.close_trace(trace_id, metadata={"operation": "editor_stt"}) return text, trace_id except Exception as exc: self.reports.record( event="editor.stt.failed", trace_id=trace_id, manager="VoiceManager", status="error", error=str(exc), metadata={"language": language}, ) self.reports.close_trace(trace_id, status="error", error=str(exc), metadata={"operation": "editor_stt"}) raise # ------------------------------------------------------------------ # تحويل الكتاب إلى صوت # ------------------------------------------------------------------ def convert_prepared_book( self, *, book_id: str, engine: str, text_repo_key: str, audio_repo_key: str, segmentation_manifest: Dict[str, Any], voice_name: str = "", speaker_name: str = "", provider_options: Optional[Dict[str, Any]] = None, trace_id: str = "", ) -> Tuple[str, Dict[str, Any], str]: trace_id = self.reports.ensure_trace( trace_id, book_id=book_id, metadata={"operation": "book_conversion", "engine": engine}, ) self._transition( trace_id, "app", "AITaskManager", event="book.conversion.requested", book_id=book_id, engine=engine, ) task = self.tasks.create_book_conversion_task( book_id=book_id, engine=engine, text_repo_key=text_repo_key, audio_repo_key=audio_repo_key, segmentation_manifest=segmentation_manifest, voice_name=voice_name, speaker_name=speaker_name, provider_options=provider_options or {}, trace_id=trace_id, ) self.reports.record( event="task.created", trace_id=trace_id, manager="AITaskManager", task_id=task.task_id, book_id=book_id, engine=engine, ) self._transition( trace_id, "AITaskManager", "VoiceManager", event="voice.execution.started", task_id=task.task_id, book_id=book_id, engine=engine, ) with self.storage.report_context( trace_id=trace_id, task_id=task.task_id, book_id=book_id, ): final_path, metadata = self.tasks.execute_prepared_book(task, self.storage) self._transition( trace_id, "AITaskManager", "StorageManager", event="book.audio.persisted", task_id=task.task_id, book_id=book_id, engine=engine, ) # نهاية خط التحويل الفعلي: من بدء المهمة حتى حفظ الصوت الكامل. # رفع ملف التقرير نفسه عملية إدارية ولا تدخل في زمن تحويل الكتاب. self.reports.close_trace( trace_id, metadata={"operation": "book_conversion", "book_id": book_id, "engine": engine}, ) # حفظ التقرير مع نفس تحويل الكتاب في خزنة الصوت. report_repo_path = self._persist_trace_report( trace_id, repo_key=audio_repo_key, book_id=book_id, engine=engine ) report_path = self.reports.save_local_report( trace_id=trace_id, filename=f"trace_{trace_id}", ) metadata = dict(metadata) metadata["trace_id"] = trace_id metadata["local_report_path"] = report_path metadata["report_repo_path"] = report_repo_path metadata["report_summary"] = self.reports.build_summary(trace_id=trace_id) return final_path, metadata, trace_id # ------------------------------------------------------------------ # الحوار والنقاش # ------------------------------------------------------------------ @staticmethod def _dialogue_character_map(dialogue_contract: Dict[str, Any]) -> Dict[str, Dict[str, Any]]: result: Dict[str, Dict[str, Any]] = {} for item in list((dialogue_contract or {}).get("characters") or []): if not isinstance(item, dict): continue character_id = str(item.get("character_id") or "").strip() if character_id: result[character_id] = dict(item) return result def validate_dialogue_contract(self, dialogue_contract: Dict[str, Any]) -> Dict[str, Any]: payload = dict(dialogue_contract or {}) config = dict(payload.get("dialogue_config") or {}) characters = list(payload.get("characters") or []) if payload.get("schema_version") not in {None, "", "aiocr_dialogue_contract_v1"}: raise ValueError("إصدار عقد الحوار غير مدعوم.") if not bool(config.get("enabled", True)): raise ValueError("الحوار غير مفعّل في العقد.") if not str(config.get("source_text") or "").strip(): raise ValueError("موضوع الحوار أو النص الأساسي فارغ.") if not characters: raise ValueError("لا توجد شخصيات في عقد الحوار.") seen: set[str] = set() normalized_characters: List[Dict[str, Any]] = [] for index, raw in enumerate(characters, start=1): item = dict(raw or {}) character_id = str( item.get("character_id") or f"character_{index:02d}" ).strip() if character_id in seen: raise ValueError(f"character_id مكرر: {character_id}") seen.add(character_id) if not str(item.get("name") or "").strip(): item["name"] = f"الشخصية {index}" item.update({ "character_id": character_id, "provider": str( item.get("provider") or "microsoft_multilingual" ).strip(), "language_mode": str( item.get("language_mode") or "عربي" ).strip(), "default_emotion": str( item.get("default_emotion") or "neutral" ).strip() or "neutral", "emotion_source": str( item.get("emotion_source") or "fixed" ).strip() or "fixed", }) normalized_characters.append(item) config["character_count"] = len(normalized_characters) payload["schema_version"] = "aiocr_dialogue_contract_v1" payload["dialogue_config"] = config payload["characters"] = normalized_characters payload["status"] = "VALIDATED" return payload @staticmethod def _dialogue_voice_for_segment( character: Dict[str, Any], language: str, ) -> str: lang = str(language or "").strip().lower() if lang.startswith("en"): return str( character.get("english_voice_id") or character.get("arabic_voice_id") or "" ).strip() return str( character.get("arabic_voice_id") or character.get("english_voice_id") or "" ).strip() @staticmethod def _dialogue_provider_for_segment( character: Dict[str, Any], language: str, ) -> str: provider = str( character.get("provider") or "microsoft_multilingual" ).strip() lang = str(language or "").strip().lower() if provider == "microsoft_normal" and lang.startswith("en"): return "microsoft_multilingual" return provider def normalize_dialogue_segments( self, dialogue_contract: Dict[str, Any], dialogue_segments: List[Dict[str, Any]], ) -> List[Dict[str, Any]]: contract = self.validate_dialogue_contract(dialogue_contract) characters = self._dialogue_character_map(contract) normalized: List[Dict[str, Any]] = [] for index, raw in enumerate( list(dialogue_segments or []), start=1, ): segment = dict(raw or {}) text = str(segment.get("text") or "").strip() if not text: continue character_id = str( segment.get("character_id") or "" ).strip() if character_id not in characters: raise ValueError( f"المقطع {index} يشير إلى شخصية غير موجودة: " f"{character_id}" ) character = characters[character_id] language = str( segment.get("language") or "ar" ).strip().lower() or "ar" emotion = str( segment.get("emotion") or character.get("default_emotion") or "neutral" ).strip() or "neutral" normalized.append({ "segment_id": str( segment.get("segment_id") or f"dialogue_{index:04d}" ), "sequence_index": index - 1, "character_id": character_id, "speaker_name": str( character.get("name") or character_id ), "text": text, "language": language, "provider": self._dialogue_provider_for_segment( character, language, ), "voice_name": self._dialogue_voice_for_segment( character, language, ), "voice_profile": str( character.get("voice_profile") or "" ).strip(), "emotion": emotion, "emotion_source": str( segment.get("emotion_source") or character.get("emotion_source") or "fixed" ), "style": str( segment.get("style") or character.get("style") or "" ).strip(), "pause_after_ms": int( segment.get("pause_after_ms") or 180 ), }) if not normalized: raise ValueError("لا توجد مقاطع حوار صالحة للتوليد.") return normalized @staticmethod def _extract_dialogue_json(value: Any) -> Dict[str, Any]: if isinstance(value, dict): return dict(value) text = str(value or "").strip() if not text: return {} if text.startswith("```"): lines = text.splitlines() if lines and lines[0].strip().startswith("```"): lines = lines[1:] if lines and lines[-1].strip() == "```": lines = lines[:-1] text = "\n".join(lines).strip() first = text.find("{") last = text.rfind("}") if first >= 0 and last > first: text = text[first:last + 1] try: parsed = json.loads(text) except Exception: return {} return parsed if isinstance(parsed, dict) else {} def generate_dialogue_segments( self, *, dialogue_contract: Dict[str, Any], trace_id: str = "", ) -> Tuple[List[Dict[str, Any]], str]: contract = self.validate_dialogue_contract(dialogue_contract) if self.ai is None: raise RuntimeError( "AIManager غير موصول بـ SystemManager؛ لا يمكن تأليف الحوار." ) config = dict(contract.get("dialogue_config") or {}) characters = list(contract.get("characters") or []) character_ids = [ str(item.get("character_id") or "").strip() for item in characters if str(item.get("character_id") or "").strip() ] ai_emotions = bool(config.get("ai_emotions", True)) trace_id = self.reports.ensure_trace( trace_id, metadata={ "operation": "dialogue_script_generation", "mode": str(config.get("mode") or ""), "characters": len(characters), }, ) compact_characters = [ { "character_id": item.get("character_id"), "name": item.get("name"), "role": item.get("role"), "language_mode": item.get("language_mode"), "default_emotion": item.get("default_emotion"), "style": item.get("style"), } for item in characters ] prompt = ( "حوّل المادة التالية إلى حوار طبيعي منظم فقط، ثم أعد JSON صالحاً فقط " "بدون Markdown وبدون شرح.\\n" f"نمط الحوار: {config.get('mode') or 'حوار طبيعي'}\\n" f"الشخصيات: {json.dumps(compact_characters, ensure_ascii=False)}\\n" f"تعليمات إضافية: {config.get('extra_instructions') or ''}\\n" "المطلوب JSON بالشكل التالي حصراً:\\n" '{"segments":[{"character_id":"character_01","text":"...",' '"language":"ar","emotion":"neutral","style":"","pause_after_ms":180}]}\\n' "القواعد:\\n" "- character_id يجب أن يكون من الشخصيات المعطاة فقط.\\n" "- حافظ على تسلسل حوار مفهوم وطبيعي، ولا تجعل المتحدث نفسه يحتكر الحوار.\\n" "- language يكون ar أو en حسب لغة الجملة.\\n" "- إذا كانت لغة الشخصية مختلطة فيجوز التبديل بين ar و en حسب السياق.\\n" "- لا تكتب أسماء المحركات أو الأصوات داخل النص.\\n" ) if ai_emotions: prompt += ( "- اختر emotion المناسب لكل مقطع من: " "neutral, happy, sad, angry, excited, calm, soft, whispering, fearful.\\n" ) else: prompt += ( "- لا تختر مشاعر جديدة؛ استخدم default_emotion للشخصية فقط.\\n" ) source_text = str(config.get("source_text") or "").strip() result: Any self._transition( trace_id, "SystemManager", "AIManager", event="dialogue.script.ai.started", ) # dialogue_generation هي وظيفة في المنصة وليست نموذجًا مستقلاً. # نعيد استخدام الدور النصي المركزي educational_generation حتى يبقى # اختيار النموذج داخل ai_models.py ولا ننشئ Role جديدًا مكررًا. ai_role = "educational_generation" if hasattr(self.ai, "run_text_function"): result = self.ai.run_text_function( ai_role, source_text, trace_id=trace_id, instruction=prompt, ) elif hasattr(self.ai, "execute_service"): result = self.ai.execute_service( role=ai_role, source_text=source_text, service_label="dialogue_generation", parameters={ "function_role": "dialogue_generation", "dialogue_contract": contract, "output_format": "json", "strict_json": True, }, user_instructions=prompt, topic=source_text[:300], resource_profile="cpu_ai", ) elif hasattr(self.ai, "generate_text"): result = self.ai.generate_text( source_text, instruction=prompt, trace_id=trace_id, ) else: raise RuntimeError( "AIManager الحالي لا يوفّر بوابة نصية مدعومة لتوليد الحوار." ) if isinstance(result, dict) and result.get("success") is False: raise RuntimeError( str(result.get("message") or result.get("error") or "فشل توليد الحوار.") ) parsed: Dict[str, Any] = {} if isinstance(result, dict): data = result.get("data") if isinstance(data, dict) and isinstance(data.get("segments"), list): parsed = data elif isinstance(result.get("segments"), list): parsed = result else: parsed = self._extract_dialogue_json( result.get("text") or result.get("output") or result.get("result") or "" ) else: parsed = self._extract_dialogue_json(result) raw_segments = parsed.get("segments") if isinstance(parsed, dict) else None if not isinstance(raw_segments, list) or not raw_segments: raise RuntimeError( "AIManager لم يُرجع dialogue segments بصيغة JSON الصالحة." ) allowed_emotions = { "neutral", "happy", "sad", "angry", "excited", "calm", "soft", "whispering", "fearful", } char_map = self._dialogue_character_map(contract) clean_segments: List[Dict[str, Any]] = [] for index, raw in enumerate(raw_segments, start=1): if not isinstance(raw, dict): continue character_id = str(raw.get("character_id") or "").strip() text = str(raw.get("text") or "").strip() if character_id not in character_ids or not text: continue character = char_map[character_id] language = str(raw.get("language") or "ar").strip().lower() if language not in {"ar", "en"}: language = "ar" emotion = str(raw.get("emotion") or "").strip().lower() if not ai_emotions: emotion = str( character.get("default_emotion") or "neutral" ).strip().lower() if emotion not in allowed_emotions: emotion = str( character.get("default_emotion") or "neutral" ).strip().lower() if emotion not in allowed_emotions: emotion = "neutral" try: pause_after_ms = int(raw.get("pause_after_ms") or 180) except Exception: pause_after_ms = 180 pause_after_ms = max(0, min(pause_after_ms, 3000)) clean_segments.append({ "segment_id": f"dialogue_{index:04d}", "character_id": character_id, "text": text, "language": language, "emotion": emotion, "emotion_source": "ai" if ai_emotions else "fixed", "style": str( raw.get("style") or character.get("style") or "" ).strip(), "pause_after_ms": pause_after_ms, }) if not clean_segments: raise RuntimeError( "تم رفض ناتج الحوار لأن المقاطع لا تطابق الشخصيات المعتمدة." ) self.reports.record( event="dialogue.script.ai.completed", trace_id=trace_id, manager="AIManager", metadata={ "segments_count": len(clean_segments), "characters_used": sorted({ item["character_id"] for item in clean_segments }), }, ) return clean_segments, trace_id def generate_and_synthesize_dialogue( self, *, dialogue_contract: Dict[str, Any], trace_id: str = "", ) -> Tuple[str, Dict[str, Any], str]: segments, trace_id = self.generate_dialogue_segments( dialogue_contract=dialogue_contract, trace_id=trace_id, ) final_path, result, trace_id = self.synthesize_dialogue( dialogue_contract=dialogue_contract, dialogue_segments=segments, trace_id=trace_id, ) result = dict(result or {}) result["generated_script"] = segments return final_path, result, trace_id def synthesize_dialogue( self, *, dialogue_contract: Dict[str, Any], dialogue_segments: List[Dict[str, Any]], trace_id: str = "", ) -> Tuple[str, Dict[str, Any], str]: contract = self.validate_dialogue_contract(dialogue_contract) segments = self.normalize_dialogue_segments( contract, dialogue_segments, ) trace_id = self.reports.ensure_trace( trace_id, metadata={ "operation": "dialogue_generation", "mode": str( (contract.get("dialogue_config") or {}).get("mode") or "" ), "characters": len(contract.get("characters") or []), }, ) self._transition( trace_id, "app", "SystemManager", event="dialogue.requested", ) generated_paths: List[str] = [] execution: List[Dict[str, Any]] = [] try: for segment in segments: provider_options = { "language": segment["language"], "emotion": segment["emotion"], "emotion_source": segment["emotion_source"], "style": segment["style"], "voice_profile": segment["voice_profile"], } speaker_name = ( segment["voice_profile"] or segment["speaker_name"] ) path, _ = self.synthesize_segment( provider=segment["provider"], text=segment["text"], voice_name=segment["voice_name"], speaker_name=speaker_name, source_type="ui", provider_options=provider_options, trace_id=trace_id, segment_id=segment["segment_id"], ) generated_paths.append(path) info = self.voice.validate_audio_file(path) execution.append({ **segment, "audio_path": path, "duration_seconds": info.get( "duration_seconds", 0, ), "status": "completed", }) final_path = ( generated_paths[0] if len(generated_paths) == 1 else self.voice.merge_audio_files( generated_paths, prefix="dialogue", ) ) final_info = self.voice.validate_audio_file(final_path) self.reports.record( event="dialogue.completed", trace_id=trace_id, manager="SystemManager", metadata={ "segments_count": len(execution), "audio_duration_seconds": final_info.get( "duration_seconds", 0, ), }, ) self.reports.close_trace( trace_id, metadata={"operation": "dialogue_generation"}, ) return final_path, { "schema_version": "aiocr_dialogue_result_v1", "dialogue_contract": contract, "segments": execution, "final_audio": final_path, "trace_id": trace_id, "status": "completed", }, trace_id except Exception as exc: self.reports.record( event="dialogue.failed", trace_id=trace_id, manager="SystemManager", status="error", error=str(exc), ) self.reports.close_trace( trace_id, status="error", error=str(exc), metadata={"operation": "dialogue_generation"}, ) raise def dialogue_runtime_capabilities(self) -> Dict[str, Any]: return { "contract_schema": "aiocr_dialogue_contract_v1", "result_schema": "aiocr_dialogue_result_v1", "ai_manager_connected": self.ai is not None, "supports_ai_script_generation": self.ai is not None, "dialogue_ai_role": "educational_generation", "dialogue_function_role": "dialogue_generation", "supports_multi_character": True, "supports_mixed_language_voices": True, "supports_segment_emotion": True, "supports_coqui_voice_variants": True, } # ------------------------------------------------------------------ # توليد صوت مباشر منخفض المستوى # ------------------------------------------------------------------ def synthesize_segment( self, *, provider: str, text: str, voice_name: str = "", speaker_name: str = "", source_type: str = "api", provider_options: Optional[Dict[str, Any]] = None, trace_id: str = "", segment_id: str = "", ) -> Tuple[str, str]: created_here = not bool(trace_id) trace_id = self.reports.ensure_trace( trace_id, metadata={"operation": "direct_voice", "provider": provider}, ) origin = "app" if (source_type or "").strip().lower() == "ui" else "AITaskManager" self._transition( trace_id, origin, "VoiceManager", event="voice.direct.requested", engine=provider, segment_id=segment_id, ) path = self.voice.synthesize_segment( provider=provider, text=text, voice_name=voice_name, speaker_name=speaker_name, source_type=source_type, provider_options=provider_options or {}, trace_id=trace_id, segment_id=segment_id, ) if created_here and (source_type or "").strip().lower() == "ui": self.reports.close_trace(trace_id, metadata={"operation": "direct_voice", "provider": provider}) return path, trace_id # ------------------------------------------------------------------ # تقارير # ------------------------------------------------------------------ def report_summary( self, *, trace_id: str = "", task_id: str = "", book_id: str = "", ) -> Dict[str, Any]: return self.reports.build_summary( trace_id=trace_id, task_id=task_id, book_id=book_id, ) def report_events( self, *, trace_id: str = "", task_id: str = "", book_id: str = "", ) -> List[Dict[str, Any]]: return self.reports.events( trace_id=trace_id, task_id=task_id, book_id=book_id, )