backend/src/testgen/services/analytics.py - master - gr.ev.vl/TestGen

backend/src/testgen/services/analytics.py
75 строк2 KB
gr.ev.vl
Initial commit
06 июн 2026, 02:41
06 июн 2026, 02:4102fb1fb
"""
Сервис аналитики и сбора статистики.
"""

import logging
from uuid import UUID

from sqlalchemy.ext.asyncio import AsyncSession

from ..domain.repositories.operator_decision import OperatorDecisionRepository

logger = logging.getLogger(__name__)


class AnalyticsService:
    """Сервис аналитики."""

    def __init__(self, db: AsyncSession):
        self.db = db
        self.decision_repo = OperatorDecisionRepository(db)

    async def get_operator_stats(self, operator_id: UUID) -> dict:
        """Получить статистику оператора."""
        decisions = await self.decision_repo.get_by_operator(operator_id)

        total = len(decisions)
        accepted = sum(1 for d in decisions if "accept" in d.decision_type)
        rejected = sum(1 for d in decisions if "reject" in d.decision_type)
        edited = sum(1 for d in decisions if "edit" in d.decision_type)

        return {
            "total_decisions": total,
            "accepted": accepted,
            "rejected": rejected,
            "edited": edited,
            "acceptance_rate": (accepted / total * 100) if total > 0 else 0,
        }

    async def get_questions_stats(self) -> dict:
        """Получить статистику по вопросам."""
        # Здесь можно добавить запросы к БД для агрегации
        return {
            "total_generated": 0,
            "total_approved": 0,
            "total_exported": 0,
        }

    async def get_model_usage(self) -> list[dict]:
        """Получить статистику использования моделей."""
        return []

    async def record_decision(
        self,
        operator_id: UUID,
        decision_type: str,
        entity_type: str,
        entity_id: UUID,
        context: dict,
        original_data: dict | None = None,
        modified_data: dict | None = None,
        llm_model: str | None = None,
        generation_run_id: UUID | None = None,
    ) -> None:
        """Записать решение оператора."""
        await self.decision_repo.create(
            operator_id=operator_id,
            decision_type=decision_type,
            entity_type=entity_type,
            entity_id=entity_id,
            context=context,
            original_data=original_data,
            modified_data=modified_data,
            llm_model=llm_model,
            generation_run_id=generation_run_id,
        )