add_channel_history + add_video_history

Author: mithmith

TODO.md

@@ -1,13 +1,13 @@
 # TODO List для проекта Youtube Parser Downloader
 
-## Основные функции
+## Основные функции (backend)
 
 - [x] **Парсинг информации**:
   - [x] Сбор информации о каналах и видео с помощью `yt-dlp`.
   - [x] Сбор информации с использованием YouTube API.
 
 - [ ] **Мониторинг и статистика**:
-  - [ ] Получение новых видео и мониторинг каналов.
+  - [x] Получение новых видео и мониторинг каналов.
   - [ ] Добавление и обновление статистической информации по каналам и видео.
   - [ ] Выборка статистической информацией по каналам и видео, включая просмотры, лайки, количество подписчиков, количество видео, новые видео и другие метрики.
 
@@ -21,7 +21,7 @@
 - [ ] **Локализация контента**:
   - [ ] Перевод описаний, названий, аудио дорожек и субтитров.
   - [ ] Перевод аудио дорожки с помощью внешнего сервиса.
-  - [ ] Определение языка аудио дорожки.
+  - [ ] Определение языка аудио дорожки (присылает youtube API).
 
 - [ ] **Создание субтитров и краткого пересказа**:
   - [ ] Автоматическое создание субтитров в случае их отсутствия с помощью внешнего сервиса.
@@ -34,4 +34,9 @@
 
 - [ ] **Интеграция с сервером Peertube**:
   - [ ] Зеркальное размещение видео на децентрализованном хостинге.
-  
+  
+## Frontend:
+- [ ] **Сделать фронт-дашбоард для визуализации данных**
+
+## Docker-контейнер:
+- [ ] **Собрать контейнер с postgres, peertube и всем необходимым для работы**

app/__main__.py

@@ -1,5 +1,3 @@
-import json
-
 from loguru import logger
 
 from app.const import channels_list

app/db/repository.py

@@ -7,7 +7,7 @@
 from sqlalchemy.exc import IntegrityError, NoResultFound, SQLAlchemyError
 
 from app.db.base import BaseRepository
-from app.db.data_table import Channel, Tag, Thumbnail, Video, VideoTag, YTFormat
+from app.db.data_table import Channel, ChannelHistory, Tag, Thumbnail, Video, VideoHistory, VideoTag, YTFormat
 from app.schema import ChannelAPIInfoSchema, ChannelInfoSchema, ThumbnailSchema, VideoSchema, YTFormatSchema
 
 
@@ -39,7 +39,7 @@ def add_video(
                 video = Video(video_id=video_schema.id, channel_id=channel_id)
                 self._session.add(video)
 
-            upload_date = datetime.utcfromtimestamp(video_schema.timestamp) if video_schema.timestamp else None
+            upload_date = datetime.fromtimestamp(video_schema.timestamp) if video_schema.timestamp else None
             video.title = video_schema.title
             video.description = video_schema.description
             video.url = video_schema.url
@@ -346,3 +346,21 @@ def update_thumbnail_path(self, video_id: UUID, thumbnail_url: str, thumbnail_pa
             self._session.commit()
         else:
             logger.warning(f"Thumbnail not found for video ID {video_id} with URL {thumbnail_url}.")
+
+    def add_channel_history(self, channel_info: Channel):
+        history: ChannelHistory = ChannelHistory(
+            channel_id=channel_info.id,
+            follower_count=channel_info.channel_follower_count,
+            view_count=channel_info.viewCount,
+            video_count=channel_info.videoCount,
+        )
+        self.add(history)
+
+    def add_video_history(self, video_info: Video):
+        history: VideoHistory = VideoHistory(
+            video_id=video_info.id,
+            view_count=video_info.view_count,
+            like_count=video_info.like_count,
+            comment_count=video_info.comment_count,
+        )
+        self.add(history)

app/service/yt_monitor.py

@@ -1,30 +1,48 @@
+from time import sleep
+
 from loguru import logger
 
 from app.db.data_table import Channel, Thumbnail, Video
+from app.db.repository import YoutubeDataRepository
 from app.integrations.ytapi import YTApiClient
 from app.integrations.ytdlp import YTChannelDownloader
-from app.schema import ChannelAPIInfoSchema, ChannelInfoSchema
+from app.schema import ChannelAPIInfoSchema, ChannelInfoSchema, VideoSchema
 
 
 class YTMonitorService:
-    def __init__(self, channels_list: list[str]) -> None:
+    def __init__(self, channels_list: list[str], timeout: int = 300) -> None:
         self._channels_list = channels_list
         self._api_client: YTApiClient
         self._yt_dlp_client: YTChannelDownloader
+        self._repository = YoutubeDataRepository()
+        self._timeout = timeout
+
+    def run(self):
+        while True:
+            self.monitor_channels_for_newold_videos()
+            sleep(self._timeout)
 
-    def monitor_channels_for_new_videos(self) -> list[Video]:
-        channels_new_videos: list[Video] = []
+    def monitor_channels_for_newold_videos(self) -> None:
         for channel_url in self._channels_list:
             logger.info(f"Getting video from: {channel_url}")
             self._yt_dlp_client = YTChannelDownloader(channel_url)
+            self._api_client = YTApiClient()
+
             # Получаем список видео и id канала
-            # ytdlp_channel_info: ChannelInfoSchema = self._yt_dlp_client.get_channel_info()
+            ytdlp_channel_info: ChannelInfoSchema = self._yt_dlp_client.get_channel_info()
+            ytapi_channels_info: ChannelAPIInfoSchema = self._api_client.get_channel_info(
+                [ytdlp_channel_info.channel_id]
+            )
+            channel_info: Channel = self._combine_channel_info(ytdlp_channel_info, ytapi_channels_info[0])
             video_list, channel_id = self._yt_dlp_client.get_video_list()
             new_videos, old_videos = self._yt_dlp_client.filter_new_old(video_list, channel_id)
-            if new_videos:
-                channels_new_videos.extend(new_videos)
 
-        return channels_new_videos
+            if channel_info:
+                self._process_channel_info(channel_info)
+            if new_videos:
+                self._process_new_videos(new_videos, channel_id)
+            if old_videos:
+                self._process_old_videos(old_videos, channel_id)
 
     def _combine_channel_info(
         self, ytdlp_channel_info: ChannelInfoSchema, ytapi_channel_info: ChannelAPIInfoSchema
@@ -44,3 +62,19 @@ def _combine_channel_info(
             thumbnails=[Thumbnail(**thumb.model_dump()) for thumb in ytdlp_channel_info.thumbnails],
         )
         return combined_channel
+
+    def _process_channel_info(self, channel_info):
+        existing_channel = self._repository.get_channel_by_id(channel_info.channel_id)
+        if not existing_channel:
+            self._repository.add_channel(channel_info)
+        self._repository.add_channel_history(channel_info)
+
+    def _process_new_videos(self, new_videos, channel_id):
+        for video in new_videos:
+            self._repository.add_video(video)
+            self._repository.add_video_history(video, channel_id)
+
+    def _process_old_videos(self, old_videos, channel_id):
+        for video in old_videos:
+            self._repository.add_video_history(video, channel_id)
+