diff options
Diffstat (limited to 'src/portal/py/modules')
| -rw-r--r-- | src/portal/py/modules/__init__.py | 16 | ||||
| -rw-r--r-- | src/portal/py/modules/common.py | 2 | ||||
| -rw-r--r-- | src/portal/py/modules/fanbox.py | 32 | ||||
| -rw-r--r-- | src/portal/py/modules/instagram.py | 10 | ||||
| -rw-r--r-- | src/portal/py/modules/patreon.py | 104 | ||||
| -rw-r--r-- | src/portal/py/modules/pixiv_app.py | 12 | ||||
| -rw-r--r-- | src/portal/py/modules/pixiv_web.py | 25 | ||||
| -rw-r--r-- | src/portal/py/modules/twitter.py | 6 | ||||
| -rw-r--r-- | src/portal/py/modules/youtube.py | 118 |
9 files changed, 212 insertions, 113 deletions
diff --git a/src/portal/py/modules/__init__.py b/src/portal/py/modules/__init__.py index 994b25a..1a9e9ca 100644 --- a/src/portal/py/modules/__init__.py +++ b/src/portal/py/modules/__init__.py @@ -1,11 +1,11 @@ import config from modules.youtube import YoutubeModule -from modules.twitter import TwitterScrapeModule -from modules.pixiv_app import PixivAppModule -from modules.pixiv_web import PixivWebModule +#from modules.twitter import TwitterScrapeModule +#from modules.pixiv_app import PixivAppModule +#from modules.pixiv_web import PixivWebModule from modules.fanbox import FanboxModule -from modules.instagram import InstagramModule +#from modules.instagram import InstagramModule from modules.patreon import PatreonModule ALL_MODULES = { @@ -13,12 +13,12 @@ ALL_MODULES = { # 'twitter': (TwitterModule( # config.TWITTER_ACCESS_TOKEN, config.TWITTER_ACCESS_TOKEN_SECRET, # config.TWITTER_CONSUMER_TOKEN, config.TWITTER_CONSUMER_TOKEN_SECRET), []), - 'twitter': (TwitterScrapeModule(config.TWITTER_COOKIES_PATH), []), - 'pixiv_web': (PixivWebModule(config.PIXIV_SESSID, config.PIXIV_USERID), []), +# 'twitter': (TwitterScrapeModule(config.TWITTER_COOKIES_PATH), []), +# 'pixiv_web': (PixivWebModule(config.PIXIV_SESSID, config.PIXIV_USERID), []), # 'pixiv_app': (PixivAppModule(config.PIXIV_REFRESH_TOKEN), []), # 'fanbox': (FanboxModule(config.FANBOX_SESSID), []), # 'instagram': (InstagramModule( # config.INSTAGRAM_USER_AGENT, config.INSTAGRAM_SETTINGS_PATH, -# config.INSTAGRAM_SESSION_ID), []) - 'patreon': (PatreonModule(config.PATREON_SESSION_ID, config.PATREON_UUID, config.PATREON_USER_ID), []) +# config.INSTAGRAM_SESSION_ID), []), +# 'patreon': (PatreonModule(config.PATREON_SESSION_ID, config.PATREON_UUID, config.PATREON_USER_ID), []) } diff --git a/src/portal/py/modules/common.py b/src/portal/py/modules/common.py index 62c3d5d..3b87884 100644 --- a/src/portal/py/modules/common.py +++ b/src/portal/py/modules/common.py @@ -1,6 +1,6 @@ from datetime import datetime, timezone -def parse_pixiv_date(date_str: str) -> datetime: +def reformat_pixiv_date(date_str: str) -> datetime: rindex = date_str.rfind(':') date_str = date_str[:rindex] + date_str[rindex + 1:] return datetime.strptime(date_str, "%Y-%m-%dT%H:%M:%S%z") diff --git a/src/portal/py/modules/fanbox.py b/src/portal/py/modules/fanbox.py index 4323d6b..bf55b9d 100644 --- a/src/portal/py/modules/fanbox.py +++ b/src/portal/py/modules/fanbox.py @@ -3,9 +3,9 @@ import httpx from typing import Optional, Any from json.decoder import JSONDecodeError from base import USER_AGENT, Search, Module, Method, ParsedJson -from post import MediaUrl, PostType, DateType, Post, User, Image, File, Tag +from post import Url, PostType, DateType, Post, User, Image, File, Tag from query_parser import QueryParser -from modules.common import parse_pixiv_date, get_current_utc_time +from modules.common import reformat_pixiv_date, get_current_utc_time BASE_URL = "https://api.fanbox.cc" @@ -23,8 +23,8 @@ class FanboxBase(Search): return None return obj['body'] - def create_media_url(self, url: str) -> MediaUrl: - return MediaUrl(url, url.split('.')[-1]) + def create_media_url(self, url: str) -> Url: + return Url(url, url.split('.')[-1]) def create_post(self, data: ParsedJson) -> Optional[Post]: kwargs = {} @@ -33,12 +33,12 @@ class FanboxBase(Search): kwargs['unique_id'] = unique_id kwargs['raw_responses'] = {} kwargs['raw_responses']['api'] = json.dumps(data) - publish_date = parse_pixiv_date(data['publishedDatetime']).timestamp() + publish_date = reformat_pixiv_date(data['publishedDatetime']).timestamp() kwargs['dates'] = { DateType.CREATED: publish_date, DateType.RETRIEVED: get_current_utc_time().timestamp() } - update_date = parse_pixiv_date(data['updatedDatetime']).timestamp() + update_date = reformat_pixiv_date(data['updatedDatetime']).timestamp() if update_date != publish_date: kwargs['dates'][DateType.EDITED] = update_date user = User(unique_id=f'fanbox:u:{data['user']['userId']}', username=data['user']['name']) @@ -57,11 +57,11 @@ class FanboxBase(Search): if data['type'] == 'image': text = data['body']['text'] for i, value in enumerate(data['body']['images']): - media[str(i)] = Image(url=MediaUrl(value['originalUrl'], value['extension']), thumbnail_url=self.create_media_url(value['thumbnailUrl'])) + media[str(i)] = Image(url=Url(value['originalUrl'], value['extension']), thumbnail_url=self.create_media_url(value['thumbnailUrl'])) elif data['type'] == 'file': text = data['body']['text'] for i, value in enumerate(data['body']['files']): - media[str(i)] = File(url=MediaUrl(value['url'], value['extension']), name=value['name']) + media[str(i)] = File(url=Url(value['url'], value['extension']), name=value['name']) elif data['type'] == 'article': for block in data['body']['blocks']: if block['type'] == 'p': @@ -71,9 +71,9 @@ class FanboxBase(Search): elif block['type'] == 'url_embed': text += f'embed::{block['urlEmbedId']}[]' + '\n' for key, value in data['body']['imageMap'].items(): - media[key] = Image(url=MediaUrl(value['originalUrl'], value['extension']), thumbnail_url=self.create_media_url(value['thumbnailUrl'])) + media[key] = Image(url=Url(value['originalUrl'], value['extension']), thumbnail_url=self.create_media_url(value['thumbnailUrl'])) for key, value in data['body']['fileMap'].items(): - media[key] = File(url=MediaUrl(value['url'], value['extension']), name=value['name']) + media[key] = File(url=Url(value['url'], value['extension']), name=value['name']) kwargs['text'] = text kwargs['media'] = media post = Post(**kwargs) @@ -138,7 +138,7 @@ class FanboxUser(FanboxBase): self.errored = True return False self.pages[num] = [] - for item in obj['items']: + for item in obj: post = self.request_post(item['id']) if post: self.pages[num].append(post.unique_id) @@ -169,12 +169,20 @@ class FanboxModule(Module): self.parser.add_command('post', FanboxPost) self.parser.add_command('user', FanboxUser) self.parser.add_command('supporting', FanboxSupporting) - self.headers['Accept-Encoding'] = 'gzip, deflate, br' + self.headers['Accept-Encoding'] = 'gzip, deflate, br, zstd' self.headers['Accept-Language'] = 'en-US,en;q=0.5' + self.headers['Alt-Used'] = 'api.fanbox.cc' self.headers['Origin'] = 'https://www.fanbox.cc' self.headers['Referer'] = 'https://www.fanbox.cc/' self.headers['User-Agent'] = USER_AGENT self.cookies['FANBOXSESSID'] = sessid + self.cookies['privacy_policy_agreement'] = '7' + self.cookies['privacy_policy_notification'] = '0' + self.cookies['p_ab_id'] = '0' + self.cookies['p_ab_id_2'] = '6' + self.cookies['p_ab_d_id'] = '251960935' + self.cookies['cf_clearance'] = '' + self.cookies['__cf_bm'] = '' def search(self, query: str, *extra_args: Any) -> Optional[Search]: return self.parser.parse_query(query) diff --git a/src/portal/py/modules/instagram.py b/src/portal/py/modules/instagram.py index d107566..28f7def 100644 --- a/src/portal/py/modules/instagram.py +++ b/src/portal/py/modules/instagram.py @@ -1,10 +1,9 @@ import os import log -import email.utils from typing import Optional, Any from pathlib import Path from base import USER_AGENT, Module, Search -from post import MediaUrl, PostType, DateType, Post, User, Image, Video +from post import Url, PostType, DateType, Post, User, Image, Video from modules.common import get_current_utc_time from instagrapi import Client from instagrapi.exceptions import UserNotFound, LoginRequired, ChallengeRequired @@ -19,13 +18,13 @@ class InstagramSearch(Search): self.page: int = 0 self.cursor: Any = None - def create_media_url(self, url: str) -> MediaUrl: + def create_media_url(self, url: str) -> Url: question = url.find('?') if question >= 0: ext = url[:question].split('.')[-1] else: ext = url.split('.')[-1] - return MediaUrl(url, ext) + return Url(url, ext) def request_page(self, num: int) -> bool: if num >= self.page + self.REACH_LIMIT: @@ -34,6 +33,7 @@ class InstagramSearch(Search): for i in range(self.page, num + 1): try: media, self.cursor = self.module.cl.user_medias_paginated_v1(self.pk, 0, end_cursor=self.cursor) + #media, self.cursor = self.module.cl.user_clips_paginated_v1(self.pk, amount=12, end_cursor=self.cursor) except LoginRequired: self.errored = True return False @@ -47,10 +47,12 @@ class InstagramSearch(Search): kwargs = {} kwargs['type'] = PostType.POST unique_id = 'instagram:p:{}'.format(m.id) + #unique_id = 'instagram:r:{}'.format(m.id) kwargs['unique_id'] = unique_id kwargs['raw_responses'] = {} kwargs['raw_responses']['tile'] = m.json() kwargs['url'] = 'https://www.instagram.com/p/{}'.format(m.code) + #kwargs['url'] = 'https://www.instagram.com/reel/{}'.format(m.code) kwargs['dates'] = { DateType.CREATED: m.taken_at.timestamp(), DateType.RETRIEVED: get_current_utc_time().timestamp() diff --git a/src/portal/py/modules/patreon.py b/src/portal/py/modules/patreon.py index 40c3255..f33ec18 100644 --- a/src/portal/py/modules/patreon.py +++ b/src/portal/py/modules/patreon.py @@ -1,9 +1,12 @@ -import json +import log import httpx +import sys from json import JSONDecodeError from typing import Optional, Any +from datetime import datetime from base import USER_AGENT, Search, Module, Method, ParsedJson -from post import MediaUrl, PostType, DateType, Post, User, Media, Image, Animation, Tag, TagType +from modules.common import get_current_utc_time +from post import Date, DateType, DateMeta, Post, User, Url, Media, File from query_parser import QueryParser BASE_URL = 'https://www.patreon.com/api' @@ -26,7 +29,7 @@ class PatreonBase(Search): return None return obj - def search_user(self, query: str) -> Optional[User]: + def search_user(self, query: str) -> tuple[Optional[User], Optional[str]]: url = f'{BASE_URL}/search' params = { 'q': query, @@ -37,32 +40,76 @@ class PatreonBase(Search): } obj = self.check_api_response(self.module.do_request(Method.GET, url, params=params)) if not obj: - return None + return None, None for result in obj['data']: + username = result['attributes']['creator_name'] + display_name = result['attributes']['name'] + # Only accept exact matches. + if username.lower() != query.lower() and display_name.lower() != query.lower(): + continue prefix = result['id'].find('campaign_') if prefix >= 0: campaign = result['id'][prefix + len('campaign_'):] unique_id = 'patreon:c:' + campaign - user = User(unique_id=unique_id, username=result['attributes']['creator_name'], display_name=result['attributes']['name']) - return user - return None + user = User(unique_id=unique_id, username=username, display_name=display_name) + return user, campaign + return None, None class PatreonUser(PatreonBase): def __init__(self, userdata: Any, arg: str): super().__init__(userdata) self.username: str = arg self.user: Optional[User] = None - self.campaign: str + self.campaign: Optional[str] = None + self.cursor: Optional[str] = None + self.attachments: dict[str, Media] = {} + + # post_type + # image_file: https://www.patreon.com/posts/emma-again-29391136 + # text_only: https://www.patreon.com/posts/hello-update-29391184 + + def create_post(self, data: ParsedJson, hash: str) -> Optional[Post]: + kwargs = {} + unique_id = f'patreon:p:{data['id']}' + kwargs['unique_id'] = unique_id + kwargs['raw_responses'] = {} + kwargs['raw_responses']['api'] = hash + kwargs['url'] = data['attributes']['url'] + published_at = datetime.strptime(data['attributes']['published_at'], "%Y-%m-%dT%H:%M:%S.%f%z").timestamp() + current_time = get_current_utc_time().timestamp() + kwargs['dates'] = [ + Date(DateType.CREATED, (published_at, published_at), DateMeta.NONE), + Date(DateType.RETRIEVED, (current_time, current_time), DateMeta.NONE), + ] + kwargs['author'] = self.user + kwargs['title'] = data['attributes']['title'] + kwargs['likes'] = data['attributes']['like_count'] + kwargs['comments'] = data['attributes']['comment_count'] + kwargs['text'] = data['attributes']['content'] + post_type = data['attributes']['post_type'] + media = {} + #if 'access_rules' in data['relationships']: + # if not (len(data['relationships']['access_rules']['data']) == 1 and data['relationships']['access_rules']['data'][0]['id'] == '9816126'): + # return None + if 'media' in data['relationships']: + for m in data['relationships']['media']['data']: + media[m['id']] = self.attachments[m['id']] + if 'attachments' in data['relationships']: + for a in data['relationships']['attachments']['data']: + media[a['id']] = self.attachments[a['id']] + kwargs['media'] = media + post = Post(**kwargs) + self.module.add_to_map(unique_id, post) + return post def request_page(self, num: int) -> bool: if num >= self.page + self.REACH_LIMIT: return False if not self.user: - user = self.search_user(self.username) - if not user: + self.user, self.campaign = self.search_user(self.username) + if not self.user: self.errored = True return False - self.campaign = user.unique_id[len('patreon:c:'):] request_satisfied = False for i in range(self.page, num + 1): url = f'{BASE_URL}/posts' @@ -81,7 +128,7 @@ class PatreonUser(PatreonBase): 'filter[contains_exclusive_posts]': 'true', 'filter[is_draft]': 'false', 'filter[accessible_by_user_id]': self.module.user_id, - 'sort': '-published_at', + 'sort': 'published_at', 'json-api-version': '1.0', 'json-api-use-default-includes': 'false' } @@ -91,8 +138,27 @@ class PatreonUser(PatreonBase): if not obj: self.errored = True return False - print(json.dumps(obj, indent=4)) - return True + hash = self.module.add_raw_response(obj) + self.cursor = obj['meta']['pagination']['cursors']['next'] + self.pages[i] = [] + for inc in obj['included']: + if inc['type'] == 'attachment': + self.attachments[inc['id']] = File(url=Url(inc['attributes']['url']), name=inc['attributes']['name']) + elif inc['type'] == 'media': + self.attachments[inc['id']] = Media(url=Url(inc['attributes']['download_url']), thumbnail_url=Url(inc['attributes']['image_urls']['thumbnail'])) + else: + log.error(f'Unhandled inc type {inc['type']}.') + for entry in obj['data']: + if entry['type'] != 'post': + log.error(f'Unhandled entry type {entry['type']}.') + continue + post = self.create_post(entry, hash) + if post: + self.pages[i].append(post.unique_id) + self.page += 1 + if i == num: + request_satisfied = True + return request_satisfied class PatreonModule(Module): def __init__(self, session_id: str, uuid: str, user_id: str): @@ -100,7 +166,8 @@ class PatreonModule(Module): self.user_id: str = user_id self.parser: QueryParser = QueryParser(self, 'user') self.parser.add_command('user', PatreonUser) - self.headers['Accept-Encoding'] = 'gzip, deflate, br' + self.headers['User-Agent'] = USER_AGENT + self.headers['Accept-Encoding'] = 'gzip, deflate, br, zstd' self.headers['Accept-Language'] = 'en-US,en;q=0.5' self.headers['x-patreon-uuid'] = uuid self.cookies['patreon_locale_code'] = 'en-US' @@ -116,9 +183,4 @@ class PatreonModule(Module): post = self.unique_id_map[unique_id] if key not in post.media: return None - return { - 'urls': [post.media[key].url], - 'headers': { - 'User-Agent': USER_AGENT - } - } + return { 'urls': [post.media[key].url], 'headers': self.headers, 'cookies': self.cookies } diff --git a/src/portal/py/modules/pixiv_app.py b/src/portal/py/modules/pixiv_app.py index c84030d..ef5886d 100644 --- a/src/portal/py/modules/pixiv_app.py +++ b/src/portal/py/modules/pixiv_app.py @@ -1,10 +1,10 @@ import json from typing import Optional, Any from base import USER_AGENT, Search, Module, ParsedJson -from post import MediaUrl, PostType, DateType, User, Post, Image, Tag, TagType +from post import Url, PostType, DateType, User, Post, Image, Tag, TagType from query_parser import QueryParser -from pixivpy3 import * -from modules.common import get_current_utc_time, parse_pixiv_date +from pixivpy3 import AppPixivAPI +from modules.common import get_current_utc_time, reformat_pixiv_date class PixivAppBase(Search): def __init__(self, userdata: Any): @@ -33,8 +33,8 @@ class PixivAppBase(Search): self.page += 1 return request_satisfied - def create_media_url(self, url: str) -> MediaUrl: - return MediaUrl(url, url.split('.')[-1]) + def create_media_url(self, url: str) -> Url: + return Url(url, url.split('.')[-1]) def create_tag(self, data: ParsedJson) -> Tag: tag = Tag(type=TagType.GENERAL, name=data['name']) @@ -58,7 +58,7 @@ class PixivAppBase(Search): kwargs['raw_responses']['api'] = json.dumps(data) kwargs['url'] = 'https://www.pixiv.net/en/artworks/{}'.format(data['id']) kwargs['dates'] = { - DateType.CREATED: parse_pixiv_date(data['create_date']).timestamp(), + DateType.CREATED: reformat_pixiv_date(data['create_date']).timestamp(), DateType.RETRIEVED: get_current_utc_time().timestamp() } user = User(unique_id=f'pixiv:u:{data['user']['id']}', username=data['user']['account'], display_name=data['user']['name']) diff --git a/src/portal/py/modules/pixiv_web.py b/src/portal/py/modules/pixiv_web.py index be68b7e..7dbb8bb 100644 --- a/src/portal/py/modules/pixiv_web.py +++ b/src/portal/py/modules/pixiv_web.py @@ -5,9 +5,9 @@ import urllib.parse from typing import Optional, Any from json.decoder import JSONDecodeError from base import USER_AGENT, Search, Module, Method, ParsedJson -from post import MediaUrl, PostType, DateType, Post, User, Media, Image, Animation, Tag, TagType +from post import Url, PostType, DateType, Post, User, Media, Image, Animation, Tag, TagType from query_parser import QueryParser -from modules.common import parse_pixiv_date, get_current_utc_time +from modules.common import reformat_pixiv_date, get_current_utc_time BASE_URL = 'https://www.pixiv.net/ajax' LANG = 'en' @@ -30,8 +30,8 @@ class PixivWebBase(Search): return None return obj['body'] - def create_media_url(self, url: str) -> MediaUrl: - return MediaUrl(url, url.split('.')[-1]) + def create_media_url(self, url: str) -> Url: + return Url(url, url.split('.')[-1]) def parse_pages(self, data: ParsedJson) -> dict[str, Media]: media = {} @@ -45,7 +45,7 @@ class PixivWebBase(Search): animation.frames.append((frame['file'], frame['delay'])) return animation - def seek_profile_picture_url(self, user_illusts: ParsedJson) -> Optional[MediaUrl]: + def seek_profile_picture_url(self, user_illusts: ParsedJson) -> Optional[Url]: for illust in user_illusts.values(): if illust is None: continue @@ -71,14 +71,15 @@ class PixivWebBase(Search): kwargs['raw_responses'] = {} kwargs['raw_responses']['api'] = json.dumps(data) kwargs['url'] = f'https://www.pixiv.net/{LANG}/artworks/{data['illustId']}' - create_date = parse_pixiv_date(data['createDate']).timestamp() + create_date = reformat_pixiv_date(data['createDate']).timestamp() + current_date = get_current_utc_time().timestamp() kwargs['dates'] = { - DateType.CREATED: create_date, - DateType.RETRIEVED: get_current_utc_time().timestamp() + DateType.CREATED: (create_date, create_date, 0), + DateType.RETRIEVED: (current_date, current_date, 0) } - upload_date = parse_pixiv_date(data['uploadDate']).timestamp() + upload_date = reformat_pixiv_date(data['uploadDate']).timestamp() if upload_date != create_date: - kwargs['dates'][DateType.EDITED] = upload_date + kwargs['dates'][DateType.EDITED] = (upload_date, upload_date, 0) user = User(unique_id=f'pixiv:u:{data['userId']}', username=data['userAccount'], display_name=data['userName']) profile_picture_url = self.seek_profile_picture_url(data['userIllusts']) if profile_picture_url: @@ -242,10 +243,10 @@ class PixivWebModule(Module): self.parser.add_command('illust', PixivWebIllust) self.parser.add_command('user', PixivWebUser) self.parser.add_command('following', PixivWebFollowing) - self.headers['Accept-Encoding'] = 'gzip, deflate, br' # add 'zstd' when httpx update + self.headers['User-Agent'] = USER_AGENT + self.headers['Accept-Encoding'] = 'gzip, deflate, br, zstd' self.headers['Accept-Language'] = 'en-US,en;q=0.5' self.headers['Referer'] = 'https://www.pixiv.net/' - self.headers['User-Agent'] = USER_AGENT self.headers['x-user-id'] = user_id self.cookies['PHPSESSID'] = sessid diff --git a/src/portal/py/modules/twitter.py b/src/portal/py/modules/twitter.py index 2996d0e..386bc5b 100644 --- a/src/portal/py/modules/twitter.py +++ b/src/portal/py/modules/twitter.py @@ -3,7 +3,7 @@ import http.cookiejar from typing import Optional, Any, Iterator from datetime import timezone from base import USER_AGENT, Search, Module -from post import MediaUrl, PostType, DateType, PostRef, Post, User, Media, Image, Video +from post import Url, PostType, DateType, PostRef, Post, User, Media, Image, Video from query_parser import QueryParser from modules.common import get_current_utc_time @@ -26,7 +26,7 @@ class TwitterScrapeBase(Search): self.iterator: Iterator[Tweet | TweetRef | Tombstone] self.page: int = 0 - def create_media_url(self, url: str) -> MediaUrl: + def create_media_url(self, url: str) -> Url: format = url.find('format=') if format >= 0: ext = url[format + 7:format + 10] @@ -35,7 +35,7 @@ class TwitterScrapeBase(Search): if question >= 0: url = url[0:question] ext = url.split('.')[-1] - return MediaUrl(url, ext) + return Url(url, ext) def parse_media(self, data: list[twitter.Medium]) -> dict[str, Media]: media = {} diff --git a/src/portal/py/modules/youtube.py b/src/portal/py/modules/youtube.py index d14dfe8..f809ff6 100644 --- a/src/portal/py/modules/youtube.py +++ b/src/portal/py/modules/youtube.py @@ -1,7 +1,8 @@ import log +import json from typing import Optional, Any from base import Search, Module, ParsedJson -from post import MediaUrl, Post, PostType, Media, Video +from post import Url, Post, PostType, Media, Video from query_parser import QueryParser from yt_dlp import YoutubeDL @@ -25,65 +26,53 @@ ydl_opts = { 'quiet': False, 'logger': YDLLogger(), 'cachedir': False, -# 'cookiefile': '', + 'socket_timeout': 10 +# 'cookiefile': '' } -ydl = YoutubeDL(ydl_opts) - -def get_playback_url(data, video=True): - if 'entries' in data: - if len(data['entries']) == 0: - return None - data = data['entries'][0] +# https://github.com/yt-dlp/yt-dlp/issues/4103 - if 'formats' not in data: - if 'url' in data: - return data['url'] - return None +ydl = YoutubeDL(ydl_opts) - # Filter out hls temporarily. - data['formats'] = list(filter(lambda f: not f['protocol'].startswith('m3u8'), data['formats'])) +def get_playback_url(data, video=True) -> Optional[str]: + data['formats'] = list(filter( + lambda f: not f['protocol'].startswith('m3u8'), data['formats'])) if len(data['formats']) == 0: return None - url = data['formats'][0]['url'] + result = data['formats'][-1]['url'] - # audio_ext? - has_audio = list(filter(lambda f: 'acodec' not in f or f['acodec'] != 'none', data['formats'])) + has_audio = list(filter(lambda f: + (f['acodec'] and f['acodec'] != 'none') or + (f['audio_ext'] and f['audio_ext'] != 'none') or + (f['abr']), data['formats'])) if video: - if len(has_audio) > 0: - data['formats'] = has_audio - try: - data['formats'] = list(filter(lambda f: 'quality' in f, data['formats'])) - url = max(data['formats'], key=lambda f: f['quality'])['url'] - except: - pass + data['formats'] = list(filter(lambda f: + (f['vcodec'] and f['vcodec'] != 'none') or + (f['video_ext'] and f['video_ext'] != 'none') or + (f['vbr']), has_audio)) else: - if len(has_audio) == 0: - return None - data['formats'] = has_audio - try: - audio_only = list(filter(lambda f: f['vcodec'] == 'none', data['formats'])) - if len(audio_only) > 0: - data['formats'] = audio_only - else: - data['formats'] = list(filter(lambda f: f['ext'] in ['mp4'], data['formats'])) - except: - pass - try: - data['formats'] = list(filter(lambda f: 'abr' in f, data['formats'])) - url = max(data['formats'], key=lambda f: f['abr'])['url'] - except: - pass + data['formats'] = list(filter(lambda f: + (f['vcodec'] and f['vcodec'] == 'none') and + (f['video_ext'] and f['video_ext'] == 'none') and + (not f['vbr']), has_audio)) + + if len(data['formats']) > 0: + selection = max(data['formats'], key=lambda f: 0 if 'quality' not in f else f['quality']) + log.info(json.dumps(selection, indent=4)) + result = selection['url'] + else: + log.warn('Defaulting in Youtube get_playback_url().') - return url + return result class YoutubeBase(Search): def __init__(self, userdata: Any): super().__init__() self.module: YoutubeModule = userdata + self.guessed_index: int = 0 def get_info(self) -> Optional[ParsedJson]: return None @@ -92,11 +81,25 @@ class YoutubeBase(Search): info = self.get_info() if not info: return False - url = get_playback_url(info) + if 'entries' in info: + if len(info['entries']) == 0: + return False + if len(info['entries']) <= self.guessed_index: + self.guessed_index = 0 + info = info['entries'][self.guessed_index] + if 'formats' not in info and 'url' in info: + self.link = info['url'] + info = self.get_info() + if not info: + return False + if 'direct' in info and info['direct']: + url = info['url'] + else: + url = get_playback_url(info, video=False) if not url: return False - media: dict[str, Media] = { '0': Video(url=MediaUrl(url=url)) } unique_id = f'youtube:v:{info['id']}' + media: dict[str, Media] = { '0': Video(url=Url(url)) } self.module.add_to_map(unique_id, Post(type=PostType.POST, unique_id=unique_id, url='', title=info['title'], text='', media=media)) self.pages[num] = [unique_id] return True @@ -107,7 +110,13 @@ class YoutubeSearch(YoutubeBase): self.query: str = arg def get_info(self) -> Optional[ParsedJson]: - return ydl.extract_info(f'ytsearch1:{self.query}', download=False) + ydl.params['extract_flat'] = False + try: + info = ydl.extract_info(f'ytsearch1:{self.query}', download=False) + except Exception as e: + log.error(repr(e)) + return None + return info class YoutubeLink(YoutubeBase): def __init__(self, userdata: Any, arg: str): @@ -115,7 +124,24 @@ class YoutubeLink(YoutubeBase): self.link: str = arg def get_info(self) -> Optional[ParsedJson]: - return ydl.extract_info(self.link, download=False) + index = self.link.find('&index=') + if index >= 0: + sub = self.link[index + 7:] + end = sub.find('&') + if end >= 0: + sub = sub[:end] + try: + self.guessed_index = int(sub, 10) - 1 + except ValueError: + pass + log.info(f'Guessed index: {self.guessed_index}') + ydl.params['extract_flat'] = 'in_playlist' + try: + info = ydl.extract_info(self.link, download=False) + except Exception as e: + log.error(repr(e)) + return None + return info class YoutubeModule(Module): def __init__(self): |