backend/services/processor.py - master - NoWay1co/AudioAnalytics

backend/services/processor.py
763 строки28 KB
Ivan
Коммит для решения проблем с несовпадением фильтров каналов и активацией блоков соблюдения скриптов. Обновлены компоненты фронтенда для исправления отображения каналов и активации данных о соблюдении. Реализованы новые эндпоинты для повторного анализа записей и добавлены поля для распределения соблюдения в аналитике дашборда. Все изменения протестированы и задокументированы.
07 фев 2026, 06:21
07 фев 2026, 06:21b99e062
"""
Processor service - orchestrator for audio processing pipeline.
Coordinates transcription (T-one) and analysis (GigaChat ScriptAnalyzer) services.

Replaces the previous Whisper + Gemini pipeline (Phase 4 integration).
"""

from uuid import UUID
from typing import Any, Dict, List, Optional

from sqlalchemy.orm import Session, joinedload
from sqlalchemy import func

from config import settings
from database import SessionLocal
from models import Recording, Transcript, Analysis, Employee, Script
from services.tone_client import ToneClient, ToneAPIError
from services.gigachat_client import GigaChatClient, GigaChatAPIError
from services.script_analyzer import ScriptAnalyzerService, ScriptAnalyzerError
from services.minio_client import get_minio_service
from services.logger import (
    processing_logger,
    log_processing_start,
    log_processing_step,
    log_processing_success,
    log_processing_error,
)


class ProcessingError(Exception):
    """Error during recording processing."""

    def __init__(self, message: str, stage: str):
        self.message = message
        self.stage = stage
        super().__init__(self.message)


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _create_gigachat_client() -> GigaChatClient:
    """Create a GigaChatClient from application settings."""
    return GigaChatClient(
        client_id=settings.GIGACHAT_CLIENT_ID,
        auth_key=settings.GIGACHAT_AUTH_KEY,
        scope=settings.GIGACHAT_SCOPE,
        auth_url=settings.GIGACHAT_AUTH_URL,
        api_url=settings.GIGACHAT_API_URL,
        model=settings.GIGACHAT_MODEL,
        timeout=settings.GIGACHAT_TIMEOUT,
    )


def _load_script_for_recording(db: Session, recording: Recording) -> Optional[Script]:
    """Load the active script for a recording's salon or organization.

    Resolution order:
    1. Salon-specific active script (preferred).
    2. Organization-level active script.
    3. Any default active script.

    Returns:
        The best matching Script, or None if none found.
    """
    script = (
        db.query(Script)
        .options(joinedload(Script.steps))
        .filter(
            Script.is_active == True,
            (Script.salon_id == recording.salon_id)
            | (Script.organization_id == recording.organization_id),
        )
        .order_by(Script.salon_id.desc().nulls_last())
        .first()
    )

    if not script:
        script = (
            db.query(Script)
            .options(joinedload(Script.steps))
            .filter(
                Script.is_default == True,
                Script.is_active == True,
            )
            .first()
        )

    return script


async def _analyze_without_script(
    gigachat: GigaChatClient,
    transcript_text: str,
) -> Dict[str, Any]:
    """Perform basic analysis without a script (no compliance scoring).

    Used as a fallback when no active script is found for the recording.
    Returns quality, sentiment, risk, summary, topics, and recommendations
    but no checklist or compliance_score.

    Args:
        gigachat: Active GigaChatClient instance.
        transcript_text: Full transcription text.

    Returns:
        Analysis result dict compatible with the Analysis model.
    """
    system_prompt = (
        "Ты - AI-аналитик качества обслуживания клиентов в салонах красоты.\n"
        "Проанализируй транскрипт телефонного разговора между сотрудником и клиентом.\n"
        "Верни результат СТРОГО в формате JSON."
    )

    analysis_prompt = (
        "Проанализируй разговор менеджера с клиентом.\n\n"
        "ВАЖНО: Верни ТОЛЬКО JSON без пояснений. Формат:\n"
        "{\n"
        '  "quality_score": 0-100,\n'
        '  "sentiment_score": от -1.0 до 1.0,\n'
        '  "sentiment_class": "positive|neutral|negative",\n'
        '  "summary": "краткое резюме разговора в 2-3 предложениях",\n'
        '  "key_moments": [\n'
        '    {"timestamp": "MM:SS", "label": "название момента", "note": "описание"}\n'
        '  ],\n'
        '  "risk_type": "conflict|complaint_threat|legal_threat|lost_client" или null,\n'
        '  "risk_severity": "low|medium|high|critical" или null,\n'
        '  "recommendations": "текст рекомендаций по улучшению",\n'
        '  "topics": ["тема1", "тема2"]\n'
        "}\n\n"
        "Разговор:\n"
        f"{transcript_text}\n\n"
        "Ответ (только JSON):"
    )

    result = await gigachat.generate_json(
        [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": analysis_prompt},
        ],
        temperature=0.1,
    )

    # -- Normalize and validate returned values --

    quality_score = result.get("quality_score", 50)
    if not isinstance(quality_score, (int, float)):
        quality_score = 50
    quality_score = max(0, min(100, int(quality_score)))

    sentiment_score = result.get("sentiment_score", 0.0)
    if not isinstance(sentiment_score, (int, float)):
        sentiment_score = 0.0
    sentiment_score = max(-1.0, min(1.0, float(sentiment_score)))

    sentiment_class = result.get("sentiment_class", "neutral")
    if sentiment_class not in ("positive", "neutral", "negative"):
        if sentiment_score > 0.3:
            sentiment_class = "positive"
        elif sentiment_score < -0.3:
            sentiment_class = "negative"
        else:
            sentiment_class = "neutral"

    summary = result.get("summary", "")
    if not isinstance(summary, str):
        summary = str(summary)

    key_moments = result.get("key_moments", [])
    if not isinstance(key_moments, list):
        key_moments = []

    risk_type = result.get("risk_type")
    valid_risk_types = ("conflict", "complaint_threat", "legal_threat", "lost_client")
    if risk_type and risk_type not in valid_risk_types:
        risk_type = None

    risk_severity = result.get("risk_severity")
    valid_severities = ("low", "medium", "high", "critical")
    if risk_severity and risk_severity not in valid_severities:
        risk_severity = None

    recommendations = result.get("recommendations", "")
    if not isinstance(recommendations, str):
        recommendations = str(recommendations) if recommendations else ""

    topics = result.get("topics", [])
    if not isinstance(topics, list):
        topics = []
    topics = [str(t) for t in topics if t]

    return {
        "checklist": [],
        "compliance_score": None,
        "quality_score": quality_score,
        "sentiment_score": sentiment_score,
        "sentiment_class": sentiment_class,
        "summary": summary,
        "key_moments": key_moments,
        "risk_type": risk_type,
        "risk_severity": risk_severity,
        "recommendations": recommendations,
        "topics": topics,
        "conversation_count": 1,
    }


# ---------------------------------------------------------------------------
# Main pipeline
# ---------------------------------------------------------------------------


async def process_recording(recording_id: UUID) -> bool:
    """Full processing pipeline for a recording.

    Pipeline steps:
    1.  Update status -> transcribing
    2.  Download audio bytes from MinIO
    3.  POST bytes to T-one microservice for transcription
    4.  Save Transcript to DB (with T-one phrases as segments)
    5.  Update status -> analyzing
    6.  Load active Script for salon/org from DB
    7.  Call ScriptAnalyzerService.analyze_transcription() via GigaChat
    8.  Save Analysis to DB (ALL merged metrics)
    9.  Update recording: status=done, is_analyzed=True, has_risk
    10. Update employee aggregates (incl. compliance_score)

    Args:
        recording_id: UUID of the recording to process.

    Returns:
        True if processing completed successfully, False otherwise.
    """
    db: Session = SessionLocal()
    recording: Optional[Recording] = None

    try:
        log_processing_start(str(recording_id))

        # Get recording from DB
        recording = db.query(Recording).filter(Recording.id == recording_id).first()

        if not recording:
            log_processing_error(
                str(recording_id), "Recording not found in database", "NotFoundError"
            )
            return False

        if not recording.file_url:
            log_processing_error(
                str(recording_id), "Recording has no file_url", "ValidationError"
            )
            _set_error_status(db, recording, "No file URL associated with recording")
            return False

        # ========== STEP 1-4: TRANSCRIPTION (T-one) ==========
        recording.processing_status = "transcribing"
        db.commit()
        log_processing_step(str(recording_id), "Status", "transcribing")

        # Step 2: Download audio bytes from MinIO
        try:
            minio_service = get_minio_service()
            audio_bytes = minio_service.download_file(recording.file_url)
            log_processing_step(
                str(recording_id),
                "MinIO",
                f"Downloaded {len(audio_bytes)} bytes from {recording.file_url}",
            )
        except Exception as e:
            error_msg = f"Failed to download audio: {str(e)}"
            log_processing_error(str(recording_id), error_msg, "MinIOError")
            _set_error_status(db, recording, error_msg)
            return False

        # Step 3: Transcribe audio via T-one microservice
        try:
            tone_client = ToneClient(
                base_url=settings.TONE_SERVICE_URL,
                timeout=settings.TONE_TIMEOUT,
            )
            transcript_result = await tone_client.transcribe(
                audio_data=audio_bytes,
                filename=recording.file_url,
            )
            log_processing_step(
                str(recording_id),
                "T-one",
                (
                    f"Transcription complete: {transcript_result.word_count} words, "
                    f"{len(transcript_result.phrases)} phrases, "
                    f"{transcript_result.duration_seconds:.1f}s duration"
                ),
            )
        except ToneAPIError as e:
            error_msg = f"T-one API: {e.message}"
            log_processing_error(str(recording_id), error_msg, "ToneAPIError")
            _set_error_status(db, recording, error_msg)
            return False

        # Step 4: Save transcript to DB
        transcript = Transcript(
            recording_id=recording_id,
            full_text=transcript_result.full_text,
            segments=transcript_result.segments,
            language=transcript_result.language,
            word_count=transcript_result.word_count,
            model_version="T-one",
            confidence=None,
        )
        db.add(transcript)
        db.commit()
        db.refresh(transcript)
        log_processing_step(
            str(recording_id), "DB", f"Transcript saved, id={transcript.id}"
        )

        # Update recording duration if T-one provides it
        if transcript_result.duration_seconds:
            recording.duration_seconds = int(transcript_result.duration_seconds)
            db.commit()

        # ========== STEP 5-8: ANALYSIS (GigaChat) ==========
        recording.processing_status = "analyzing"
        db.commit()
        log_processing_step(str(recording_id), "Status", "analyzing")

        # Step 6: Load active script for the salon or organization
        script = _load_script_for_recording(db, recording)

        if script:
            log_processing_step(
                str(recording_id),
                "Script",
                (
                    f"Using script '{script.name}' "
                    f"(id={script.id}, v{script.current_version})"
                ),
            )
        else:
            log_processing_step(
                str(recording_id),
                "Script",
                "No active script found; analyzing without compliance",
            )

        # Step 7: Analyze transcript via GigaChat
        try:
            gigachat = _create_gigachat_client()

            if script:
                analyzer = ScriptAnalyzerService(gigachat_client=gigachat)
                analysis_result = await analyzer.analyze_transcription(
                    transcription=transcript_result.full_text,
                    script=script,
                    auto_split=True,
                )
            else:
                # No script -- analyze without compliance scoring
                analysis_result = await _analyze_without_script(
                    gigachat=gigachat,
                    transcript_text=transcript_result.full_text,
                )

            log_processing_step(
                str(recording_id),
                "GigaChat",
                (
                    f"Analysis complete: "
                    f"quality={analysis_result.get('quality_score')}, "
                    f"compliance={analysis_result.get('compliance_score')}, "
                    f"sentiment={analysis_result.get('sentiment_class')}"
                ),
            )
        except (ScriptAnalyzerError, GigaChatAPIError) as e:
            error_msg = f"GigaChat analysis: {e.message}"
            log_processing_error(str(recording_id), error_msg, type(e).__name__)
            _set_error_status(db, recording, error_msg)
            return False

        # Step 8: Save analysis to DB (all merged metrics)
        analysis = Analysis(
            recording_id=recording_id,
            transcript_id=transcript.id,
            script_id=script.id if script else None,
            script_version=script.current_version if script else None,
            quality_score=analysis_result.get("quality_score", 50),
            compliance_score=analysis_result.get("compliance_score"),
            sentiment_score=analysis_result.get("sentiment_score", 0.0),
            sentiment_class=analysis_result.get("sentiment_class", "neutral"),
            summary=analysis_result.get("summary", ""),
            key_moments=analysis_result.get("key_moments", []),
            risk_type=analysis_result.get("risk_type"),
            risk_severity=analysis_result.get("risk_severity"),
            checklist=analysis_result.get("checklist", []),
            recommendations=analysis_result.get("recommendations", ""),
            topics=analysis_result.get("topics", []),
            conversation_count=analysis_result.get("conversation_count", 1),
            model_version="gigachat",
            tokens_used=analysis_result.get("tokens_used", 0),
        )
        db.add(analysis)
        db.commit()
        db.refresh(analysis)
        log_processing_step(
            str(recording_id), "DB", f"Analysis saved, id={analysis.id}"
        )

        # ========== STEP 9: UPDATE RECORDING STATUS ==========
        recording.processing_status = "done"
        recording.is_analyzed = True
        recording.has_risk = analysis_result.get("risk_type") is not None
        db.commit()
        log_processing_step(str(recording_id), "Status", "done")

        # ========== STEP 10: UPDATE EMPLOYEE AGGREGATES ==========
        if recording.employee_id:
            await _update_employee_stats(db, recording.employee_id)
            log_processing_step(
                str(recording_id),
                "Employee",
                f"Stats updated for {recording.employee_id}",
            )

        log_processing_success(
            str(recording_id),
            (
                f"quality={analysis_result.get('quality_score')}, "
                f"compliance={analysis_result.get('compliance_score')}, "
                f"sentiment={analysis_result.get('sentiment_class')}, "
                f"has_risk={recording.has_risk}"
            ),
        )

        return True

    except Exception as e:
        error_msg = f"Unexpected error: {type(e).__name__}: {str(e)}"
        log_processing_error(str(recording_id), error_msg, "UnexpectedError")

        if recording:
            _set_error_status(db, recording, error_msg)

        return False

    finally:
        db.close()


# ---------------------------------------------------------------------------
# Status helpers
# ---------------------------------------------------------------------------


def _set_error_status(
    db: Session, recording: Recording, error_message: str
) -> None:
    """Set recording status to error with message."""
    recording.processing_status = "error"
    recording.processing_error = error_message[:500]
    db.commit()


async def _update_employee_stats(db: Session, employee_id: UUID) -> None:
    """Update employee aggregate statistics.

    Recalculates:
    - total_recordings: count of all done recordings
    - avg_quality_score: average quality score from analyses
    - avg_sentiment: average sentiment score from analyses
    - avg_compliance (computed for logging; Employee model column pending)
    """
    try:
        stats = db.query(
            func.count(Analysis.id).label("total"),
            func.avg(Analysis.quality_score).label("avg_quality"),
            func.avg(Analysis.sentiment_score).label("avg_sentiment"),
            func.avg(Analysis.compliance_score).label("avg_compliance"),
        ).join(
            Recording, Recording.id == Analysis.recording_id
        ).filter(
            Recording.employee_id == employee_id,
            Recording.processing_status == "done",
        ).first()

        employee = db.query(Employee).filter(Employee.id == employee_id).first()

        if employee and stats:
            employee.total_recordings = stats.total or 0
            employee.avg_quality_score = (
                round(stats.avg_quality, 2) if stats.avg_quality else None
            )
            employee.avg_sentiment = (
                round(stats.avg_sentiment, 2) if stats.avg_sentiment else None
            )
            db.commit()

            # Log compliance stats (Employee model column for avg_compliance pending)
            if stats.avg_compliance is not None:
                processing_logger.info(
                    "Employee %s stats: total=%d, avg_quality=%s, "
                    "avg_sentiment=%s, avg_compliance=%.1f",
                    employee_id,
                    stats.total or 0,
                    stats.avg_quality,
                    stats.avg_sentiment,
                    stats.avg_compliance,
                )

    except Exception as e:
        processing_logger.warning(
            "Failed to update employee stats for %s: %s", employee_id, str(e)
        )
        # Don't fail the whole pipeline for stats update failure


# ---------------------------------------------------------------------------
# Reprocess & status
# ---------------------------------------------------------------------------


async def reanalyze_recording(recording_id: UUID) -> bool:
    """Re-analyze a recording with the current active script.

    Unlike reprocess_recording, this keeps the existing transcript and only
    re-runs the LLM analysis phase.  Useful when a script was added or
    updated after the original analysis.

    Args:
        recording_id: UUID of the recording to re-analyze.

    Returns:
        True if re-analysis completed successfully.
    """
    db: Session = SessionLocal()

    try:
        recording = db.query(Recording).filter(Recording.id == recording_id).first()
        if not recording:
            processing_logger.error("Re-analyze: recording %s not found", recording_id)
            return False

        if recording.processing_status != "done":
            processing_logger.warning(
                "Re-analyze: recording %s status is '%s', expected 'done'",
                recording_id, recording.processing_status,
            )
            return False

        transcript = db.query(Transcript).filter(
            Transcript.recording_id == recording_id
        ).first()
        if not transcript or not transcript.full_text:
            processing_logger.error("Re-analyze: no transcript for recording %s", recording_id)
            return False

        log_processing_start(str(recording_id))
        log_processing_step(str(recording_id), "Re-analyze", "Starting re-analysis (keeping transcript)")

        # Update status
        recording.processing_status = "analyzing"
        db.commit()

        # Load active script
        script = _load_script_for_recording(db, recording)
        if not script:
            log_processing_step(
                str(recording_id), "Script",
                "No active script found; skipping re-analysis",
            )
            recording.processing_status = "done"
            db.commit()
            return False

        log_processing_step(
            str(recording_id), "Script",
            f"Using script '{script.name}' (id={script.id}, v{script.current_version})",
        )

        # Run analysis via GigaChat
        try:
            gigachat = _create_gigachat_client()
            analyzer = ScriptAnalyzerService(gigachat_client=gigachat)
            analysis_result = await analyzer.analyze_transcription(
                transcription=transcript.full_text,
                script=script,
                auto_split=True,
            )
            log_processing_step(
                str(recording_id), "GigaChat",
                f"Re-analysis complete: quality={analysis_result.get('quality_score')}, "
                f"compliance={analysis_result.get('compliance_score')}, "
                f"sentiment={analysis_result.get('sentiment_class')}",
            )
        except (ScriptAnalyzerError, GigaChatAPIError) as e:
            error_msg = f"GigaChat re-analysis: {e.message}"
            log_processing_error(str(recording_id), error_msg, type(e).__name__)
            recording.processing_status = "done"  # Restore status
            db.commit()
            return False

        # Update or create Analysis record
        existing_analysis = db.query(Analysis).filter(
            Analysis.recording_id == recording_id
        ).first()

        if existing_analysis:
            # Update existing analysis
            existing_analysis.script_id = script.id
            existing_analysis.script_version = script.current_version
            existing_analysis.quality_score = analysis_result.get("quality_score", 50)
            existing_analysis.compliance_score = analysis_result.get("compliance_score")
            existing_analysis.sentiment_score = analysis_result.get("sentiment_score", 0.0)
            existing_analysis.sentiment_class = analysis_result.get("sentiment_class", "neutral")
            existing_analysis.summary = analysis_result.get("summary", "")
            existing_analysis.key_moments = analysis_result.get("key_moments", [])
            existing_analysis.risk_type = analysis_result.get("risk_type")
            existing_analysis.risk_severity = analysis_result.get("risk_severity")
            existing_analysis.checklist = analysis_result.get("checklist", [])
            existing_analysis.recommendations = analysis_result.get("recommendations", "")
            existing_analysis.topics = analysis_result.get("topics", [])
            existing_analysis.conversation_count = analysis_result.get("conversation_count", 1)
            existing_analysis.model_version = "gigachat"
            existing_analysis.tokens_used = analysis_result.get("tokens_used", 0)
        else:
            # Create new analysis
            new_analysis = Analysis(
                recording_id=recording_id,
                transcript_id=transcript.id,
                script_id=script.id,
                script_version=script.current_version,
                quality_score=analysis_result.get("quality_score", 50),
                compliance_score=analysis_result.get("compliance_score"),
                sentiment_score=analysis_result.get("sentiment_score", 0.0),
                sentiment_class=analysis_result.get("sentiment_class", "neutral"),
                summary=analysis_result.get("summary", ""),
                key_moments=analysis_result.get("key_moments", []),
                risk_type=analysis_result.get("risk_type"),
                risk_severity=analysis_result.get("risk_severity"),
                checklist=analysis_result.get("checklist", []),
                recommendations=analysis_result.get("recommendations", ""),
                topics=analysis_result.get("topics", []),
                conversation_count=analysis_result.get("conversation_count", 1),
                model_version="gigachat",
                tokens_used=analysis_result.get("tokens_used", 0),
            )
            db.add(new_analysis)

        # Update recording flags
        recording.processing_status = "done"
        recording.is_analyzed = True
        recording.has_risk = analysis_result.get("risk_type") is not None
        db.commit()

        # Update employee aggregates
        if recording.employee_id:
            await _update_employee_stats(db, recording.employee_id)

        log_processing_success(
            str(recording_id),
            f"Re-analysis: quality={analysis_result.get('quality_score')}, "
            f"compliance={analysis_result.get('compliance_score')}, "
            f"sentiment={analysis_result.get('sentiment_class')}, "
            f"has_risk={recording.has_risk}",
        )
        return True

    except Exception as e:
        error_msg = f"Re-analyze unexpected error: {type(e).__name__}: {str(e)}"
        processing_logger.error(error_msg)
        # Restore status to done so the recording is not stuck
        try:
            recording = db.query(Recording).filter(Recording.id == recording_id).first()
            if recording and recording.processing_status == "analyzing":
                recording.processing_status = "done"
                db.commit()
        except Exception:
            pass
        return False

    finally:
        db.close()


async def reprocess_recording(recording_id: UUID) -> bool:
    """Reprocess a recording (e.g., after an error or script update).

    Deletes existing transcript and analysis, then runs the full pipeline.

    Args:
        recording_id: UUID of the recording to reprocess.

    Returns:
        True if reprocessing completed successfully.
    """
    db: Session = SessionLocal()

    try:
        # Delete existing transcript and analysis
        db.query(Analysis).filter(Analysis.recording_id == recording_id).delete()
        db.query(Transcript).filter(Transcript.recording_id == recording_id).delete()

        # Reset recording status
        recording = db.query(Recording).filter(
            Recording.id == recording_id
        ).first()
        if recording:
            recording.processing_status = "pending"
            recording.processing_error = None
            recording.is_analyzed = False
            recording.has_risk = False
            db.commit()

        db.close()

        # Run full pipeline
        return await process_recording(recording_id)

    except Exception as e:
        processing_logger.error(
            "Failed to reprocess recording %s: %s", recording_id, str(e)
        )
        db.close()
        return False


def get_processing_status(recording_id: UUID) -> dict:
    """Get current processing status for a recording.

    Returns:
        Dict with status info: {status, error, has_transcript, has_analysis}
    """
    db: Session = SessionLocal()

    try:
        recording = db.query(Recording).filter(
            Recording.id == recording_id
        ).first()

        if not recording:
            return {"status": "not_found", "error": "Recording not found"}

        has_transcript = db.query(Transcript).filter(
            Transcript.recording_id == recording_id
        ).first() is not None

        has_analysis = db.query(Analysis).filter(
            Analysis.recording_id == recording_id
        ).first() is not None

        return {
            "status": recording.processing_status,
            "error": recording.processing_error,
            "has_transcript": has_transcript,
            "has_analysis": has_analysis,
            "is_analyzed": recording.is_analyzed,
            "has_risk": recording.has_risk,
        }

    finally:
        db.close()