Youtube Subscription Implementation (WIP)

breadtube_bot/youtube_subscription.py

@@ -0,0 +1,50 @@
+from dataclasses import dataclass
+
+from .yt_objects import ChannelSnippet
+
+
+@dataclass
+class SubscriptionInfo:
+    name: str
+    channel_id: str
+    last_update: float
+    channel_info: ChannelSnippet | None = None
+
+
+Subscriptions = dict[str, SubscriptionInfo]
+
+
+SUBSCRIPTION_FILE_COLUMNS: list[bytes] = [b'Channel Name', b'Channel ID']
+
+
+class SubscriptionHelper:
+    @staticmethod
+    def read_text(content: bytes) -> Subscriptions:
+        content_lines = content.strip().splitlines()
+        if content_lines[0] != b';'.join(SUBSCRIPTION_FILE_COLUMNS):
+            raise RuntimeError(f"Unexpected header: {content_lines[0]} != {b','.join(SUBSCRIPTION_FILE_COLUMNS)}")
+
+        subscriptions: Subscriptions = {}
+        for line_number, line in enumerate(content_lines[1:], start=1):
+            values = line.split(b';')
+            if len(values) != len(SUBSCRIPTION_FILE_COLUMNS):
+                raise RuntimeError(f'Unexpected value count (columns) at line {line_number}'
+                                f', got {len(values)} but expected {len(SUBSCRIPTION_FILE_COLUMNS)}')
+            name = values[0].decode()
+            channel_id = values[1].decode()
+            subscriptions[channel_id] = SubscriptionInfo(name=name, channel_id=channel_id, last_update=0)
+        return subscriptions
+
+    @staticmethod
+    def update_subscriptions(new: Subscriptions, previous: Subscriptions):
+        for channel_id, previous_info in previous.items():
+            if channel_id in new:
+                new[channel_id].last_update = previous_info.last_update
+                new[channel_id].channel_info = previous_info.channel_info
+
+    @staticmethod
+    def generate_text(subscriptions: Subscriptions) -> bytes:
+        content_lines: list[bytes] = [b';'.join(SUBSCRIPTION_FILE_COLUMNS)]
+        content_lines.extend([
+            f"{sub_info.name.replace(';', ',')};{sub_info.channel_id}".encode() for sub_info in subscriptions.values()])
+        return b'\n'.join(content_lines)