viewed-fix

services/viewed.py

@@ -5,6 +5,7 @@ import time
 from datetime import datetime, timedelta, timezone
 from typing import Dict
 
+# ga
 from google.analytics.data_v1beta import BetaAnalyticsDataClient
 from google.analytics.data_v1beta.types import DateRange, Dimension, Metric, RunReportRequest
 
@@ -12,40 +13,49 @@ from orm.author import Author
 from orm.shout import Shout, ShoutAuthor, ShoutTopic
 from orm.topic import Topic
 from services.db import local_session
-from utils.logger import root_logger as logger
+from services.logger import root_logger as logger
 
-GOOGLE_KEYFILE_PATH = os.environ.get("GOOGLE_KEYFILE_PATH") or "/dump/google-service.json"
+GOOGLE_KEYFILE_PATH = os.environ.get("GOOGLE_KEYFILE_PATH", "/dump/google-service.json")
-GOOGLE_PROPERTY_ID = os.environ.get("GOOGLE_PROPERTY_ID")
+GOOGLE_PROPERTY_ID = os.environ.get("GOOGLE_PROPERTY_ID", "")
 VIEWS_FILEPATH = "/dump/views.json"
 
 
 class ViewedStorage:
+    lock = asyncio.Lock()
     views_by_shout = {}
     shouts_by_topic = {}
     shouts_by_author = {}
+    views = None
     period = 60 * 60  # каждый час
     analytics_client: BetaAnalyticsDataClient | None = None
+    auth_result = None
     disabled = False
     start_date = datetime.now().strftime("%Y-%m-%d")
 
     @staticmethod
     async def init():
-        """Подключение к клиенту Google Analytics и инициализация данных."""
+        """Подключение к клиенту Google Analytics с использованием аутентификации"""
         self = ViewedStorage
-        self.load_precounted_views()
+        async with self.lock:
+            # Загрузка предварительно подсчитанных просмотров из файла JSON
+            self.load_precounted_views()
 
-        os.environ.setdefault("GOOGLE_APPLICATION_CREDENTIALS", GOOGLE_KEYFILE_PATH)
+            os.environ.setdefault("GOOGLE_APPLICATION_CREDENTIALS", GOOGLE_KEYFILE_PATH)
-        if GOOGLE_KEYFILE_PATH and os.path.isfile(GOOGLE_KEYFILE_PATH):
+            if GOOGLE_KEYFILE_PATH and os.path.isfile(GOOGLE_KEYFILE_PATH):
-            self.analytics_client = BetaAnalyticsDataClient()
+                # Using a default constructor instructs the client to use the credentials
-            logger.info(" * Клиент Google Analytics успешно авторизован")
+                # specified in GOOGLE_APPLICATION_CREDENTIALS environment variable.
-            asyncio.create_task(self.worker())  # Запуск фоновой задачи
+                self.analytics_client = BetaAnalyticsDataClient()
-        else:
+                logger.info(" * Клиент Google Analytics успешно авторизован")
-            logger.info(" * Пожалуйста, добавьте ключевой файл Google Analytics")
+
-            self.disabled = True
+                # Запуск фоновой задачи
+                _task = asyncio.create_task(self.worker())
+            else:
+                logger.info(" * Пожалуйста, добавьте ключевой файл Google Analytics")
+                self.disabled = True
 
     @staticmethod
     def load_precounted_views():
-        """Загрузка предварительно подсчитанных просмотров из файла JSON."""
+        """Загрузка предварительно подсчитанных просмотров из файла JSON"""
         self = ViewedStorage
         try:
             if os.path.exists(VIEWS_FILEPATH):
@@ -57,7 +67,7 @@ class ViewedStorage:
                 if now_date == self.start_date:
                     logger.info(" * Данные актуализованы!")
                 else:
-                    logger.warning(f" * Файл просмотров {VIEWS_FILEPATH} устарел: {self.start_date}")
+                    logger.warn(f" * Файл просмотров {VIEWS_FILEPATH} устарел: {self.start_date}")
 
                 with open(VIEWS_FILEPATH, "r") as file:
                     precounted_views = json.load(file)
@@ -68,42 +78,49 @@ class ViewedStorage:
         except Exception as e:
             logger.error(f"Ошибка загрузки предварительно подсчитанных просмотров: {e}")
 
+    # noinspection PyTypeChecker
     @staticmethod
     async def update_pages():
-        """Обновление данных просмотров из Google Analytics."""
+        """Запрос всех страниц от Google Analytics, отсортированных по количеству просмотров"""
         self = ViewedStorage
         logger.info(" ⎧ Обновление данных просмотров от Google Analytics ---")
         if not self.disabled:
             try:
                 start = time.time()
-                if self.analytics_client:
+                async with self.lock:
-                    request = RunReportRequest(
+                    if self.analytics_client:
-                        property=f"properties/{GOOGLE_PROPERTY_ID}",
+                        request = RunReportRequest(
-                        dimensions=[Dimension(name="pagePath")],
+                            property=f"properties/{GOOGLE_PROPERTY_ID}",
-                        metrics=[Metric(name="screenPageViews")],
+                            dimensions=[Dimension(name="pagePath")],
-                        date_ranges=[DateRange(start_date=self.start_date, end_date="today")],
+                            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):
+                        response = self.analytics_client.run_report(request)
-                        slugs = set()
+                        if response and isinstance(response.rows, list):
-                        new_views_by_shout = {}  # временное хранилище
+                            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]
+                                    views_count = int(row.metric_values[0].value)
 
-                        for row in response.rows:
+                                    # Обновление данных в хранилище
-                            page_path = row.dimension_values[0].value
+                                    self.views_by_shout[slug] = self.views_by_shout.get(slug, 0)
-                            slug = page_path.split("discours.io/")[-1]
+                                    self.views_by_shout[slug] += views_count
-                            views_count = int(row.metric_values[0].value)
+                                    self.update_topics(slug)
 
-                            # Запись путей страниц для логирования
+                                    # Запись путей страниц для логирования
-                            slugs.add(slug)
+                                    slugs.add(slug)
 
-                            # Обновление данных в временном хранилище
+                                logger.info(f" ⎪ Собрано страниц: {len(slugs)} ")
-                            new_views_by_shout[slug] = new_views_by_shout.get(slug, 0) + views_count
 
-                        self.views_by_shout = new_views_by_shout  # атомарная замена
+                        end = time.time()
-                        logger.info(f" ⎪ Собрано страниц: {len(slugs)} ")
+                        logger.info(" ⎪ Обновление страниц заняло %fs " % (end - start))
-
-                    end = time.time()
-                    logger.info(" ⎪ Обновление страниц заняло %fs " % (end - start))
             except Exception as error:
                 logger.error(error)
                 self.disabled = True
@@ -134,7 +151,7 @@ class ViewedStorage:
 
     @staticmethod
     def update_topics(shout_slug):
-        """Обновление счетчиков темы по slug shout."""
+        """Обновление счетчиков темы по slug shout"""
         self = ViewedStorage
         with local_session() as session:
             # Определение вспомогательной функции для избежания повторения кода
@@ -154,7 +171,7 @@ class ViewedStorage:
 
     @staticmethod
     async def worker():
-        """Асинхронная задача обновления."""
+        """Асинхронная задача обновления"""
         failed = 0
         self = ViewedStorage
         if self.disabled: