→コード: v4.3.8 動画の取得ロジックを修正
>Fet-Fe (→コード: v4.3.7.post1 コメント修正) |
>Fet-Fe (→コード: v4.3.8 動画の取得ロジックを修正) |
||
11行目: | 11行目: | ||
"""Twitter自動収集スクリプト | """Twitter自動収集スクリプト | ||
ver4.3. | ver4.3.8 2024/8/30恒心 | ||
当コードは恒心停止してしまった https://rentry.co/7298g の降臨ショーツイート自動収集スクリプトの復刻改善版です。 | 当コードは恒心停止してしまった https://rentry.co/7298g の降臨ショーツイート自動収集スクリプトの復刻改善版です。 | ||
1,291行目: | 1,291行目: | ||
def _download_m3u8( | def _download_m3u8( | ||
self, | self, | ||
media_url: str, | |||
ts_filename: str, | ts_filename: str | None, | ||
mp4_filename: str, | mp4_filename: str, | ||
proxies: dict[str, str] | None) -> FfmpegStatus: | proxies: dict[str, str] | None) -> FfmpegStatus: | ||
1,298行目: | 1,298行目: | ||
Args: | Args: | ||
media_url (str): 動画のm3u8/mp4ファイルのURL。 | |||
ts_filename (str): m3u8から取得したtsファイルのパス。 | ts_filename (str | None): m3u8から取得したtsファイルのパス。\ | ||
mp4_filename (str): | `None` の場合はtsファイルをダウンロードせず、直接mp4をダウンロードする。 | ||
proxies (dict[str, str] | None): | mp4_filename (str): 取得したmp4ファイルのパス。 | ||
proxies (dict[str, str] | None): ffmpegでの通信に用いるプロキシ設定。 | |||
Returns: | Returns: | ||
FfmpegStatus: ffmpegでの保存ステータス。 | FfmpegStatus: ffmpegでの保存ステータス。 | ||
""" | """ | ||
os.makedirs(self.MEDIA_DIR, exist_ok=True) | |||
if ts_filename is not None: | |||
ts_returncode: Final[int] = subprocess.run( | |||
[ | |||
'ffmpeg', '-y', | |||
'-http_proxy', 'proxies["http"]', | |||
'-i', media_url, | |||
'-c', 'copy', ts_filename | |||
] if proxies is not None else [ | |||
'ffmpeg', '-y', | |||
], | '-i', media_url, | ||
'-c', 'copy', ts_filename | |||
], | |||
stdout=subprocess.DEVNULL).returncode | |||
# 取得成功したらtsをmp4に変換 | |||
if ts_returncode == 0: | |||
ts2mp4_returncode: Final[int] = subprocess.run( | |||
[ | |||
'ffmpeg', '-y', '-i', ts_filename, | |||
'-acodec', 'copy', '-vcodec', 'copy', mp4_filename | |||
], | |||
stdout=subprocess.DEVNULL | |||
).returncode | |||
if ts2mp4_returncode == 0: | |||
return FfmpegStatus.MP4 | |||
else: | |||
return FfmpegStatus.TS | |||
else: | |||
return FfmpegStatus.FAILED | |||
else: | |||
returncode: Final[int] = subprocess.run( | |||
[ | [ | ||
'ffmpeg', '-y', '-i', | 'ffmpeg', '-y', | ||
'- | '-http_proxy', 'proxies["http"]', | ||
'-i', media_url, | |||
'-c', 'copy', mp4_filename | |||
] if proxies is not None else [ | |||
'ffmpeg', '-y', | |||
'-i', media_url, | |||
'-c', 'copy', mp4_filename | |||
], | ], | ||
stdout=subprocess.DEVNULL | stdout=subprocess.DEVNULL).returncode | ||
if returncode == 0: | |||
if | |||
return FfmpegStatus.MP4 | return FfmpegStatus.MP4 | ||
else: | else: | ||
return FfmpegStatus.FAILED | |||
def _tweet_date(self, url: str) -> datetime: | def _tweet_date(self, url: str) -> datetime: | ||
1,412行目: | 1,433行目: | ||
continue | continue | ||
# videoタグのdata-url属性またはvideoタグ直下のsourceタグからURLが取得できる | |||
data_url: Final[str | list[str] | None] = video.get('data-url') | data_url: Final[str | list[str] | None] = video.get('data-url') | ||
source_tag: Final[Tag | None] = video.select_one('source') | |||
src_url: Final[str | list[str] | None] = \ | |||
source_tag.get('src') if source_tag is not None else None | |||
video_url: Final[str | list[str] | None] = data_url or src_url | |||
assert isinstance(video_url, str) | |||
tweet_id: Final[str] = tweet_url.split('/')[-1] | tweet_id: Final[str] = tweet_url.split('/')[-1] | ||
if data_url is not None: | |||
f'{os.path.join(self.MEDIA_DIR, tweet_id)}_{i}.ts' | # data-url属性からURLを取得した場合 | ||
video_matched: Final[re.Match[str] | None] = re.search( | |||
r'[^/]+$', video_url) | |||
assert video_matched is not None | |||
media_path: Final[str] = unquote(video_matched.group()) | |||
media_url: str = urljoin(self.NITTER_INSTANCE, media_path) | |||
ts_filename: str | None = ( | |||
f'{os.path.join(self.MEDIA_DIR, tweet_id)}_{i}.ts' | |||
) | |||
else: | |||
# sourceタグからURLを取得した場合 | |||
media_url: str = video_url | |||
ts_filename: str | None = None | |||
mp4_filename: Final[str] = ( | mp4_filename: Final[str] = ( | ||
f'{os.path.join(self.MEDIA_DIR, tweet_id)}_{i}.mp4' | f'{os.path.join(self.MEDIA_DIR, tweet_id)}_{i}.mp4' | ||
1,427行目: | 1,460行目: | ||
match self._download_m3u8( | match self._download_m3u8( | ||
media_url, | |||
ts_filename, | ts_filename, | ||
mp4_filename, | mp4_filename, |