AYO/breadtube-bot
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add channel id by name message

Browse source
This commit is contained in:
BreadTube 2026-01-09 19:56:27 +09:00
commit 4cd8fb80da
5 changed files with 45 additions and 12 deletions
Download patch file Download diff file Expand all files Collapse all files

17
breadtube_bot/youtube_manager.py
View file

@ -2,6 +2,7 @@ from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime
import json
import time
from typing import TYPE_CHECKING
import urllib.error
@ -26,11 +27,12 @@ class YoutubeManager:
remaining: int
next_reset: float
def __init__(self, logger: logging.Logger):
def __init__(self, api_key: str, logger: logging.Logger):
self._api_key = api_key
self._logger = logger
self.rate_limit = self.RateLimit(remaining=self.DEFAULT_DAILY_POINTS, next_reset=time.time() + 24 * 3600)
def _request(self, url: str, request_timeout: float, expected_status: int = 200) -> tuple[HTTPHeaders, str]:
def _request(self, url: str, request_timeout: float, expected_status: int = 200) -> tuple[HTTPHeaders, dict]:
if time.time() >= self.rate_limit.next_reset:
self.rate_limit.next_reset = time.time() + 24 * 3600
self.rate_limit.remaining = self.DEFAULT_DAILY_POINTS
@ -41,14 +43,14 @@ class YoutubeManager:
self.rate_limit.remaining -= 1
request = urllib.request.Request(url)
# request.add_header('Accept', 'application/json')
request.add_header('Accept', 'application/json')
try:
with urllib.request.urlopen(request, timeout=request_timeout) as response:
if response.status != expected_status:
raise RuntimeError(
f'Unexpected YT status {response.status} (expected: {expected_status})'
f' -> {response.read().decode()}')
return dict(response.getheaders()), response.read().decode()
return dict(response.getheaders()), json.loads(response.read().decode())
except urllib.error.HTTPError as error:
raise RuntimeError(
f'HTTP error calling API ({url}): {error}:\n'
@ -69,6 +71,13 @@ class YoutubeManager:
except Exception as error:
raise RuntimeError(f'Exception calling YouTube shorts ({video_id}): {error}') from error
def request_channel_id(self, channel_name: str, request_timeout: float) -> str:
url = ('https://www.googleapis.com/youtube/v3/channels?part=snippet'
f'&forHandle%40={channel_name}&key={self._api_key}')
self._logger.debug('YoutubeManager: request channel id for channel %s', channel_name)
_, info = self._request(url=url, request_timeout=request_timeout)
return info['items'][0]['id']
@staticmethod
def _parse_rss_data(data) -> tuple[ChannelInfo, list[VideoInfo]]:
videos: list[VideoInfo] = []