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

Refresh older sub first + save subscriptions

Browse source
This commit is contained in:
BreadTube 2025-09-30 23:01:30 +09:00 • committed by Corentin
commit 693564bb04
4 changed files with 128 additions and 85 deletions
Download patch file Download diff file Expand all files Collapse all files

12
breadtube_bot/youtube_manager.py
View file

@ -17,7 +17,7 @@ if TYPE_CHECKING:
class YoutubeManager:
DEFAULT_DAILY_REQUESTS = 10_000
DEFAULT_DAILY_POINTS = 10_000
SHORTS_CHECK_STATUS = 303
@dataclass
@ -28,12 +28,12 @@ class YoutubeManager:
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_REQUESTS, next_reset=time.time() + 24 * 3600)
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, 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_REQUESTS
self.rate_limit.remaining = self.DEFAULT_DAILY_POINTS
elif self.rate_limit.remaining <= 0:
sleep_time = time.time() - self.rate_limit.next_reset
self._logger.debug('No more remaining in Youtube RateLimit : sleeping for %.03fs', sleep_time)
@ -60,7 +60,9 @@ class YoutubeManager:
def is_shorts(self, connection: http.client.HTTPConnection, video_id: str) -> bool:
try:
connection.request('GET', f'/shorts/{video_id}')
endpoint = f'/shorts/{video_id}'
self._logger.debug('YoutubeManager: Checking for shorts: %s', endpoint)
connection.request('GET', endpoint)
response = connection.getresponse()
response.read()
return response.status != self.SHORTS_CHECK_STATUS
@ -71,6 +73,7 @@ class YoutubeManager:
HTTPHeaders, ChannelResult]:
url = ('https://www.googleapis.com/youtube/v3/channels?part=snippet'
f'&id={channel_id}&key={self._api_key}')
self._logger.debug('YoutubeManager: request channel info for channel %s', channel_id)
headers, info = self._request(url=url, request_timeout=request_timeout)
return headers, ChannelResult.from_dict(info)
@ -78,5 +81,6 @@ class YoutubeManager:
HTTPHeaders, SearchResult]:
url = (f'https://www.googleapis.com/youtube/v3/search?part=snippet&channelId={channel_id}'
f'&maxResults={max_results}&order=date&type=video&key={self._api_key}')
self._logger.debug('YoutubeManager: request channel videos for channel %s', channel_id)
headers, info = self._request(url=url, request_timeout=request_timeout)
return headers, SearchResult.from_dict(info)