kick

README.MD

@@ -6,11 +6,11 @@ Simple LiveStreams notifier in telegram based on [aiogram](https://github.com/ai
 
 ### Available platforms
 
-| Platform        | Based on                                        | Status                                                               |
-|-----------------|-------------------------------------------------|----------------------------------------------------------------------|
-| Youtube         | [youtube-dlp](https://github.com/yt-dlp/yt-dlp) |  ✅                                                 |
-| Twitch          | [twitch-api](https://github.com/Teekeks/pyTwitchAPI)                                  | ✅                                             |
-
+| Platform | Based on                                             | Status                                               |
+|----------|------------------------------------------------------|------------------------------------------------------|
+| Youtube  | [youtube-dlp](https://github.com/yt-dlp/yt-dlp)      | ✅                                                    |
+| Twitch   | [twitch-api](https://github.com/Teekeks/pyTwitchAPI) | ✅                                                    |
+| Kick     | [aiohttp](https://github.com/aio-libs/aiohttp)          | ❌ (Until cloudflare-403 fix or creating official api |
 Use this bot to receive periodic reports on live broadcasts on stream platforms and generate and send the report to telegram. 
 
 The current version of the bot works in the telegram channel [НАСРАНО](https://t.me/HACPAH1).

src/bot/dialogs/user/channel/add/constants.py

@@ -12,7 +12,7 @@
 url_examples: dict = {
     ChannelType.YOUTUBE: "https://www.youtube.com/@username",
     ChannelType.TWITCH: "https://www.twitch.tv/username",
-    ChannelType.KICK: "empty",
+    ChannelType.KICK: "https://www.kick.com/username",
 }
 
 __all__ = ["url_examples", "url_validators"]

src/constants.py

@@ -8,7 +8,7 @@
 SQLITE_DATABASE_FILE_PATH: str = os.environ.get(
     "SQLITE_DATABASE_FILE_PATH", os.path.join(ROOT_DIR, "youtube-notifier-bot.db")
 )
-VERSION: str = "2024-09-26.07"
+VERSION: str = "2024-09-26.11"
 
 __all__ = [
     "CONFIG_FILE_PATH",

src/scheduler/jobs/telegram_notify_job/data_fetcher/__init__.py

@@ -1,4 +1,9 @@
+from .kick import async_kick_fetch_livestreams
 from .twitch import async_twitch_fetch_livestreams
 from .youtube import async_youtube_fetch_livestreams
 
-__all__ = ["async_twitch_fetch_livestreams", "async_youtube_fetch_livestreams"]
+__all__ = [
+    "async_kick_fetch_livestreams",
+    "async_twitch_fetch_livestreams",
+    "async_youtube_fetch_livestreams",
+]

src/scheduler/jobs/telegram_notify_job/data_fetcher/kick/__init__.py

@@ -0,0 +1,4 @@
+from .fetcher import async_kick_fetch_livestreams
+
+
+__all__ = ["async_kick_fetch_livestreams"]

src/scheduler/jobs/telegram_notify_job/data_fetcher/kick/constants.py

@@ -0,0 +1,4 @@
+KICK_API_URL = "https://kick.com/api/v2/channels/"
+
+
+__all__ = ["KICK_API_URL"]

src/scheduler/jobs/telegram_notify_job/data_fetcher/kick/fetcher.py

@@ -0,0 +1,97 @@
+import asyncio
+import operator
+from datetime import datetime
+from typing import Optional
+
+from aiohttp import ClientSession
+from dateutil.tz import tzutc
+
+from .constants import KICK_API_URL
+from src.db.models import ChannelModel
+from src.logger import logger
+from src.scheduler.jobs.telegram_notify_job.data_fetcher.utils import make_time_readable
+from src.scheduler.jobs.telegram_notify_job.dto import ErrorVideoInfo
+from src.scheduler.jobs.telegram_notify_job.dto import VideoInfo
+from src.utils import extract_kick_username
+
+
+async def async_fetch_livestream(
+    channel: ChannelModel, session: ClientSession
+) -> Optional[VideoInfo] | ErrorVideoInfo:
+    """
+    :param session:
+    :param channel:
+    :return:
+    """
+    await logger.ainfo(channel.model_dump_json())
+
+    live_stream = None
+    try:
+
+        username = extract_kick_username(channel.url)
+        if not username:
+            raise Exception(f"Cannot extract username for {channel.url}")
+
+        async with session.get(f"{KICK_API_URL}{username}") as resp:
+
+            raw_data = await resp.json()
+
+            livestream: Optional[dict] = raw_data.get("livestream", None)
+            if livestream:
+                is_live = livestream.get("is_live", None)
+
+                if is_live:
+
+                    concurrent_view_count = livestream["viewer_count"]
+
+                    duration = make_time_readable(
+                        (
+                            datetime.now(tz=tzutc())
+                            - datetime.strptime(
+                                livestream["start_time"], "%Y-%m-%d %H:%M:%S"
+                            )
+                        ).seconds
+                    )
+
+                    live_stream = VideoInfo(
+                        url=channel.url,
+                        label=channel.label,
+                        concurrent_view_count=concurrent_view_count,
+                        duration=duration,
+                    )
+
+    except Exception as ex:
+        await logger.aerror(f"Fetching info error: {channel.url} {ex}")
+        return ErrorVideoInfo(channel=channel.model_dump(), ex_message=str(ex))
+
+    return live_stream
+
+
+async def async_kick_fetch_livestreams(
+    channels: list[ChannelModel],
+) -> tuple[list[VideoInfo], list[ErrorVideoInfo]]:
+    """
+    :param channels:
+    :return:
+    """
+
+    async with ClientSession() as client:
+        tasks = [
+            async_fetch_livestream(channel=channel, session=client)
+            for channel in channels
+        ]
+
+        data = await asyncio.gather(*tasks)
+
+    errors = [stream for stream in data if isinstance(stream, ErrorVideoInfo)]
+
+    live_streams = [stream for stream in data if isinstance(stream, VideoInfo)]
+
+    live_streams = sorted(
+        live_streams, key=operator.attrgetter("concurrent_view_count"), reverse=True
+    )
+
+    return live_streams, errors
+
+
+__all__ = ["async_kick_fetch_livestreams"]