50 lines
1.9 KiB
Python
50 lines
1.9 KiB
Python
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)
|