core/services/viewed.py

import asyncio
import os
import time
from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import ClassVar

# ga
from google.analytics.data_v1beta import BetaAnalyticsDataClient
from google.analytics.data_v1beta.types import (
    DateRange,
    Dimension,
    Metric,
    RunReportRequest,
)
from google.analytics.data_v1beta.types import Filter as GAFilter

from orm.author import Author
from orm.shout import Shout, ShoutAuthor, ShoutTopic
from orm.topic import Topic
from storage.db import local_session
from storage.redis import redis
from utils.logger import root_logger as logger

GOOGLE_KEYFILE_PATH = os.environ.get("GOOGLE_KEYFILE_PATH", "/dump/google-service.json")
GOOGLE_PROPERTY_ID = os.environ.get("GOOGLE_PROPERTY_ID", "")


class ViewedStorage:
    """
    Класс для хранения и доступа к данным о просмотрах.
    Использует Redis в качестве основного хранилища и Google Analytics для сбора новых данных.
    """

    lock = asyncio.Lock()
    views_by_shout: ClassVar[dict] = {}
    shouts_by_topic: ClassVar[dict] = {}
    shouts_by_author: ClassVar[dict] = {}
    views = None
    period = 60 * 60  # каждый час
    analytics_client: BetaAnalyticsDataClient | None = None
    auth_result = None
    running = False
    redis_views_key = None
    last_update_timestamp = 0
    start_date = datetime.now(tz=UTC).strftime("%Y-%m-%d")
    _background_task: asyncio.Task | None = None

    @staticmethod
    async def init() -> None:
        """Подключение к клиенту Google Analytics и загрузка данных о просмотрах из Redis"""
        self = ViewedStorage
        async with self.lock:
            # Загрузка предварительно подсчитанных просмотров из Redis
            await self.load_views_from_redis()

            os.environ.setdefault("GOOGLE_APPLICATION_CREDENTIALS", GOOGLE_KEYFILE_PATH)
            if GOOGLE_KEYFILE_PATH and Path(GOOGLE_KEYFILE_PATH).is_file():
                # Using a default constructor instructs the client to use the credentials
                # specified in GOOGLE_APPLICATION_CREDENTIALS environment variable.
                self.analytics_client = BetaAnalyticsDataClient()
                logger.info(" * Google Analytics credentials accepted")

                # Запуск фоновой задачи
                task = asyncio.create_task(self.worker())
                # Store reference to prevent garbage collection
                self._background_task = task
            else:
                logger.warning(" * please, add Google Analytics credentials file")
                self.running = False

    @staticmethod
    async def load_views_from_redis() -> None:
        """Загрузка предварительно подсчитанных просмотров из Redis"""
        self = ViewedStorage

        # Подключаемся к Redis если соединение не установлено
        try:
            if not await redis.ping():
                await redis.connect()
        except Exception as e:
            logger.warning(f"Redis connection check failed: {e}")
            # Try to connect anyway
            await redis.connect()

        # Логируем настройки Redis соединения
        logger.info("* Redis connected")

        # Получаем список всех ключей migrated_views_* и находим самый последний
        keys = await redis.execute("KEYS", "migrated_views_*")
        if keys is None:
            keys = []
            logger.warning("Redis KEYS command returned None, treating as empty list")
        logger.info("Raw Redis result for 'KEYS migrated_views_*': %d", len(keys))

        # Декодируем байтовые строки, если есть
        if keys and isinstance(keys[0], bytes):
            keys = [k.decode("utf-8") for k in keys]
            logger.info("Decoded keys: %s", keys)

        if not keys:
            logger.info(" * No migrated_views keys found in Redis - views will be 0")
            return

        # Фильтруем только ключи timestamp формата (исключаем migrated_views_slugs)
        timestamp_keys = [k for k in keys if k != "migrated_views_slugs"]
        logger.info("Timestamp keys after filtering: %s", timestamp_keys)

        if not timestamp_keys:
            logger.info(" * No migrated_views timestamp keys found in Redis - views will be 0")
            return

        # Сортируем по времени создания (в названии ключа) и берем последний
        timestamp_keys.sort()
        latest_key = timestamp_keys[-1]
        self.redis_views_key = latest_key
        logger.info("Selected latest key: %s", latest_key)

        # Получаем метку времени создания для установки start_date
        timestamp = await redis.execute("HGET", latest_key, "_timestamp")
        if timestamp:
            self.last_update_timestamp = int(timestamp)
            timestamp_dt = datetime.fromtimestamp(int(timestamp), tz=UTC)
            self.start_date = timestamp_dt.strftime("%Y-%m-%d")

            # Если данные сегодняшние, считаем их актуальными
            now_date = datetime.now(tz=UTC).strftime("%Y-%m-%d")
            if now_date == self.start_date:
                logger.info(" * Views data is up to date!")
            else:
                logger.warning("Views data is from %s, may need update", self.start_date)

        # 🔎 ЗАГРУЖАЕМ ДАННЫЕ из Redis в views_by_shout
        logger.info("🔍 Loading views data from Redis key: %s", latest_key)

        # Получаем все данные из hash
        views_data = await redis.execute("HGETALL", latest_key)

        if views_data and len(views_data) > 0:
            # Преобразуем список [key1, value1, key2, value2] в словарь
            views_dict = {}
            try:
                # Проверяем что views_data это словарь или список
                if isinstance(views_data, dict):
                    # Если это уже словарь
                    for key, value in views_data.items():
                        key_str = key.decode("utf-8") if isinstance(key, bytes) else str(key)
                        value_str = value.decode("utf-8") if isinstance(value, bytes) else str(value)

                        if not key_str.startswith("_"):
                            try:
                                views_dict[key_str] = int(value_str)
                            except (ValueError, TypeError):
                                logger.warning(f"🔍 Invalid views value for {key_str}: {value_str}")

                elif isinstance(views_data, list | tuple):
                    # Если это список [key1, value1, key2, value2]
                    for i in range(0, len(views_data), 2):
                        if i + 1 < len(views_data):
                            key = (
                                views_data[i].decode("utf-8")
                                if isinstance(views_data[i], bytes)
                                else str(views_data[i])
                            )
                            value = (
                                views_data[i + 1].decode("utf-8")
                                if isinstance(views_data[i + 1], bytes)
                                else str(views_data[i + 1])
                            )

                            # Пропускаем служебные ключи
                            if not key.startswith("_"):
                                try:
                                    views_dict[key] = int(value)
                                except (ValueError, TypeError):
                                    logger.warning(f"🔍 Invalid views value for {key}: {value}")
                else:
                    logger.warning(f"🔍 Unexpected Redis data format: {type(views_data)}")

                # Загружаем данные в класс
                self.views_by_shout.update(views_dict)
                logger.info("🔍 Loaded %d shouts with views from Redis", len(views_dict))

                # Показываем образцы загруженных данных только если есть данные
                if views_dict:
                    sample_items = list(views_dict.items())[:3]
                    logger.info("🔍 Sample loaded data: %s", sample_items)
                else:
                    logger.debug("🔍 No valid views data found in Redis hash - views will be 0")

            except Exception as e:
                logger.warning(f"🔍 Error parsing Redis views data: {e} - views will be 0")
        else:
            logger.debug("🔍 Redis hash is empty for key: %s - views will be 0", latest_key)

        # Выводим информацию о количестве загруженных записей
        total_entries = await redis.execute("HGET", latest_key, "_total")
        if total_entries:
            logger.info("%s shouts with views loaded from Redis key: %s", total_entries, latest_key)

        logger.info("Found migrated_views keys: %s", keys)

    # noinspection PyTypeChecker
    @staticmethod
    async def update_pages() -> None:
        """Запрос всех страниц от Google Analytics, отсортированных по количеству просмотров"""
        self = ViewedStorage
        logger.info(" ⎧ views update from Google Analytics ---")
        if self.running:
            try:
                start = time.time()
                async with self.lock:
                    if self.analytics_client:
                        request = RunReportRequest(
                            property=f"properties/{GOOGLE_PROPERTY_ID}",
                            dimensions=[Dimension(name="pagePath")],
                            metrics=[Metric(name="screenPageViews")],
                            date_ranges=[DateRange(start_date=self.start_date, end_date="today")],
                        )
                        response = self.analytics_client.run_report(request)
                        if response and isinstance(response.rows, list):
                            slugs = set()
                            for row in response.rows:
                                print(
                                    row.dimension_values[0].value,
                                    row.metric_values[0].value,
                                )
                                # Извлечение путей страниц из ответа Google Analytics
                                if isinstance(row.dimension_values, list):
                                    page_path = row.dimension_values[0].value
                                    slug = page_path.split("discours.io/")[-1]
                                    fresh_views = int(row.metric_values[0].value)

                                    # Обновление данных в хранилище
                                    self.views_by_shout[slug] = self.views_by_shout.get(slug, 0)
                                    self.views_by_shout[slug] += fresh_views
                                    self.update_topics(slug)

                                    # Запись путей страниц для логирования
                                    slugs.add(slug)

                                logger.info("collected pages: %d", len(slugs))

                        end = time.time()
                        logger.info("views update time: %.2fs", end - start)
            except (ConnectionError, TimeoutError, ValueError) as error:
                logger.error(error)
                self.running = False

    @staticmethod
    def get_shout(shout_slug: str = "", shout_id: int = 0) -> int:
        """
        🔎 Синхронное получение метрики просмотров shout по slug или id из кеша.

        Использует кешированные данные из views_by_shout (in-memory кеш).
        Для обновления данных используется асинхронный фоновый процесс.

        Args:
            shout_slug: Slug публикации
            shout_id: ID публикации

        Returns:
            int: Количество просмотров из кеша
        """
        self = ViewedStorage

        # 🔍 DEBUG: Логируем только если кеш пустой и это первый запрос
        cache_size = len(self.views_by_shout)
        if cache_size == 0 and shout_slug:
            logger.debug(f"🔍 ViewedStorage cache is empty for slug '{shout_slug}'")

        # 🔎 Используем только in-memory кеш для быстрого доступа
        if shout_slug:
            views = self.views_by_shout.get(shout_slug, 0)
            if views > 0:
                logger.debug(f"🔍 Found {views} views for slug '{shout_slug}'")
            return views

        # 🔎 Для ID ищем slug в БД и затем получаем views_count
        if shout_id:
            try:
                with local_session() as session:
                    from orm.shout import Shout

                    shout = session.query(Shout).where(Shout.id == shout_id).first()
                    if shout and shout.slug:
                        views = self.views_by_shout.get(shout.slug, 0)
                        logger.debug(f"🔍 Found slug '{shout.slug}' for id {shout_id}, views: {views}")
                        return views
                    logger.debug(f"🔍 No shout found with id {shout_id} or missing slug")
            except Exception as e:
                logger.warning(f"Failed to get shout slug for id {shout_id}: {e}")
            return 0

        logger.debug("🔍 get_shout called without slug or id")
        return 0

    @staticmethod
    async def get_shout_media(shout_slug: str) -> dict[str, int]:
        """Получение метрики воспроизведения shout по slug."""
        self = ViewedStorage

        # TODO: get media plays from Google Analytics

        return self.views_by_shout.get(shout_slug, 0)

    @staticmethod
    def get_topic(topic_slug: str) -> int:
        """Получение суммарного значения просмотров темы."""
        self = ViewedStorage
        views_count = 0
        for shout_slug in self.shouts_by_topic.get(topic_slug, []):
            views_count += self.get_shout(shout_slug=shout_slug)
        return views_count

    @staticmethod
    def get_author(author_slug: str) -> int:
        """Получение суммарного значения просмотров автора."""
        self = ViewedStorage
        views_count = 0
        for shout_slug in self.shouts_by_author.get(author_slug, []):
            views_count += self.get_shout(shout_slug=shout_slug)
        return views_count

    @staticmethod
    def update_topics(shout_slug: str) -> None:
        """Обновление счетчиков темы по slug shout"""
        self = ViewedStorage
        with local_session() as session:
            # Определение вспомогательной функции для избежания повторения кода
            def update_groups(dictionary: dict, key: str, value: str) -> None:
                dictionary[key] = list({*dictionary.get(key, []), value})

            # Обновление тем и авторов с использованием вспомогательной функции
            for [_st, topic] in (
                session.query(ShoutTopic, Topic).join(Topic).join(Shout).where(Shout.slug == shout_slug).all()
            ):
                update_groups(self.shouts_by_topic, topic.slug, shout_slug)

            for [_st, author] in (
                session.query(ShoutAuthor, Author).join(Author).join(Shout).where(Shout.slug == shout_slug).all()
            ):
                update_groups(self.shouts_by_author, author.slug, shout_slug)

    @staticmethod
    async def stop() -> None:
        """Остановка фоновой задачи"""
        self = ViewedStorage
        async with self.lock:
            self.running = False
            logger.info("ViewedStorage worker was stopped.")

    @staticmethod
    async def worker() -> None:
        """Асинхронная задача обновления"""
        failed = 0
        self = ViewedStorage

        while self.running:
            try:
                await self.update_pages()
                failed = 0
            except (ConnectionError, TimeoutError, ValueError) as exc:
                failed += 1
                logger.debug(exc)
                logger.info("update failed #%d, wait 10 secs", failed)
                if failed > 3:
                    logger.info(" - views update failed, not trying anymore")
                    self.running = False
                    break
            if failed == 0:
                when = datetime.now(UTC) + timedelta(seconds=self.period)
                t = format(when.astimezone().isoformat())
                logger.info("       ⎩ next update: %s", t.split("T")[0] + " " + t.split("T")[1].split(".")[0])
                await asyncio.sleep(self.period)
            else:
                await asyncio.sleep(10)
                logger.info(" - try to update views again")

    @staticmethod
    async def update_slug_views(slug: str) -> int:
        """
        Получает fresh статистику просмотров для указанного slug.

        Args:
            slug: Идентификатор страницы

        Returns:
            int: Количество просмотров
        """
        self = ViewedStorage
        if not self.analytics_client:
            logger.warning("Google Analytics client not initialized")
            return 0

        try:
            # Создаем фильтр для точного совпадения конца URL
            request = RunReportRequest(
                property=f"properties/{GOOGLE_PROPERTY_ID}",
                date_ranges=[DateRange(start_date=self.start_date, end_date="today")],
                dimensions=[Dimension(name="pagePath")],
                dimension_filter=GAFilter(
                    field_name="pagePath",
                    string_filter=GAFilter.StringFilter(
                        value=f".*/{slug}$",  # Используем регулярное выражение для точного совпадения конца URL
                        match_type=GAFilter.StringFilter.MatchType.FULL_REGEXP,
                        case_sensitive=False,  # Включаем чувствительность к регистру для точности
                    ),
                ),
                metrics=[Metric(name="screenPageViews")],
            )

            response = self.analytics_client.run_report(request)

            if not response.rows:
                return 0

            views = int(response.rows[0].metric_values[0].value)
        except (ConnectionError, ValueError, AttributeError):
            logger.exception("Google Analytics API Error")
            return 0
        else:
            # Кэшируем результат
            self.views_by_shout[slug] = views
            return views