trade-assistant/backend/app/services/analytics.py

from typing import Dict, Any, List, Optional
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, func, and_, extract
from app.models.customer import Customer, Conversation, Message
from app.models.quotation import Quotation
from app.models.analytics import UsageLog
from app.models.user import User
from app.models.preference import MarketingEffect
from datetime import datetime, timedelta
import logging

logger = logging.getLogger(__name__)


class AnalyticsService:
    def __init__(self, db: AsyncSession):
        self.db = db

    async def get_customer_stats(self, user_id: str) -> Dict[str, Any]:
        total = await self.db.execute(
            select(func.count(Customer.id)).where(Customer.user_id == user_id)
        )
        by_status = await self.db.execute(
            select(Customer.status, func.count(Customer.id))
            .where(Customer.user_id == user_id)
            .group_by(Customer.status)
        )
        by_country = await self.db.execute(
            select(Customer.country, func.count(Customer.id))
            .where(Customer.user_id == user_id)
            .where(Customer.country.isnot(None))
            .group_by(Customer.country)
            .order_by(func.count(Customer.id).desc())
            .limit(10)
        )

        now = datetime.utcnow()
        silent_3 = await self.db.execute(
            select(func.count(Customer.id)).where(
                and_(
                    Customer.user_id == user_id,
                    Customer.last_contact_at.isnot(None),
                    Customer.last_contact_at < now - timedelta(days=3),
                    Customer.status.in_(["lead", "negotiating"]),
                )
            )
        )

        return {
            "total": total.scalar() or 0,
            "by_status": {row[0] or "unknown": row[1] for row in by_status.all()},
            "by_country": {row[0] or "unknown": row[1] for row in by_country.all()},
            "silent_customers": silent_3.scalar() or 0,
        }

    async def get_translation_stats(self, user_id: str) -> Dict[str, Any]:
        now = datetime.utcnow()
        today_start = now.replace(hour=0, minute=0, second=0, microsecond=0)

        today_count = await self.db.execute(
            select(func.count(UsageLog.id)).where(
                and_(
                    UsageLog.user_id == user_id,
                    UsageLog.action == "translate",
                    UsageLog.created_at >= today_start,
                )
            )
        )
        total_count = await self.db.execute(
            select(func.count(UsageLog.id)).where(
                and_(UsageLog.user_id == user_id, UsageLog.action == "translate")
            )
        )

        daily_result = await self.db.execute(
            select(
                extract("year", UsageLog.created_at),
                extract("month", UsageLog.created_at),
                extract("day", UsageLog.created_at),
                func.count(UsageLog.id),
            )
            .where(
                and_(
                    UsageLog.user_id == user_id,
                    UsageLog.action == "translate",
                    UsageLog.created_at >= now - timedelta(days=30),
                )
            )
            .group_by(
                extract("year", UsageLog.created_at),
                extract("month", UsageLog.created_at),
                extract("day", UsageLog.created_at),
            )
            .order_by(
                extract("year", UsageLog.created_at),
                extract("month", UsageLog.created_at),
                extract("day", UsageLog.created_at),
            )
        )

        return {
            "today": today_count.scalar() or 0,
            "total": total_count.scalar() or 0,
            "daily": [
                {
                    "date": f"{int(r[0])}-{int(r[1]):02d}-{int(r[2]):02d}",
                    "count": r[3],
                }
                for r in daily_result.all()
            ],
        }

    async def get_quotation_stats(self, user_id: str) -> Dict[str, Any]:
        total = await self.db.execute(
            select(func.count(Quotation.id)).where(Quotation.user_id == user_id)
        )
        by_status = await self.db.execute(
            select(Quotation.status, func.count(Quotation.id))
            .where(Quotation.user_id == user_id)
            .group_by(Quotation.status)
        )
        total_value = await self.db.execute(
            select(func.sum(Quotation.total)).where(
                and_(Quotation.user_id == user_id, Quotation.status == "accepted")
            )
        )

        return {
            "total": total.scalar() or 0,
            "by_status": {row[0] or "draft": row[1] for row in by_status.all()},
            "total_accepted_value": float(total_value.scalar() or 0),
        }

    async def get_message_stats(self, user_id: str) -> Dict[str, Any]:
        now = datetime.utcnow()
        today_start = now.replace(hour=0, minute=0, second=0, microsecond=0)

        total_msgs = await self.db.execute(
            select(func.count(Message.id))
            .join(Conversation, Message.conversation_id == Conversation.id)
            .where(Conversation.user_id == user_id)
        )
        today_msgs = await self.db.execute(
            select(func.count(Message.id))
            .join(Conversation, Message.conversation_id == Conversation.id)
            .where(
                and_(
                    Conversation.user_id == user_id,
                    Message.created_at >= today_start,
                )
            )
        )

        return {
            "total": total_msgs.scalar() or 0,
            "today": today_msgs.scalar() or 0,
        }

    async def get_marketing_stats(self, user_id: str) -> Dict[str, Any]:
        total = await self.db.execute(
            select(func.count(MarketingEffect.id)).where(MarketingEffect.user_id == user_id)
        )
        copy_count = await self.db.execute(
            select(func.count(MarketingEffect.id)).where(
                and_(MarketingEffect.user_id == user_id, MarketingEffect.event_type == "copy")
            )
        )
        send_count = await self.db.execute(
            select(func.count(MarketingEffect.id)).where(
                and_(MarketingEffect.user_id == user_id, MarketingEffect.event_type == "send")
            )
        )
        top_products = await self.db.execute(
            select(MarketingEffect.product_name, func.count(MarketingEffect.id))
            .where(
                and_(
                    MarketingEffect.user_id == user_id,
                    MarketingEffect.product_name.isnot(None),
                )
            )
            .group_by(MarketingEffect.product_name)
            .order_by(func.count(MarketingEffect.id).desc())
            .limit(5)
        )

        return {
            "total_events": total.scalar() or 0,
            "copy_count": copy_count.scalar() or 0,
            "send_count": send_count.scalar() or 0,
            "top_products": [{"name": r[0], "count": r[1]} for r in top_products.all()],
        }