Spaces:

Asem75
/

Aiocr_Radar

Sleeping

Aiocr_Radar / system_manager.py
Upload 2 files
de8c3fa verified 24 days ago
46.7 kB
# 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,
        )