266 lines
17 KiB
Python
266 lines
17 KiB
Python
import json
|
||
import os
|
||
import unittest
|
||
from typing import List, Optional
|
||
import httpx
|
||
from loguru import logger
|
||
from tqdm import tqdm
|
||
from BowongModalFunctions.models.ffmpeg_tasks.models import FFMpegSliceSegment, FFMPEGSliceOptions
|
||
from BowongModalFunctions.utils.VideoUtils import VideoUtils
|
||
|
||
|
||
class FFMPEGTestCase(unittest.IsolatedAsyncioTestCase):
|
||
temp_dir = f"./videos"
|
||
douyin_cookies = "ttwid=1%7CB1qls3GdnZhUov9o2NxOMxxYS2ff6OSvEWbv0ytbES4%7C1680522049%7C280d802d6d478e3e78d0c807f7c487e7ffec0ae4e5fdd6a0fe74c3c6af149511; my_rd=1; passport_csrf_token=3ab34460fa656183fccfb904b16ff742; passport_csrf_token_default=3ab34460fa656183fccfb904b16ff742; d_ticket=9f562383ac0547d0b561904513229d76c9c21; n_mh=hvnJEQ4Q5eiH74-84kTFUyv4VK8xtSrpRZG1AhCeFNI; store-region=cn-fj; store-region-src=uid; LOGIN_STATUS=1; __security_server_data_status=1; FORCE_LOGIN=%7B%22videoConsumedRemainSeconds%22%3A180%7D; pwa2=%223%7C0%7C3%7C0%22; download_guide=%223%2F20230729%2F0%22; volume_info=%7B%22isUserMute%22%3Afalse%2C%22isMute%22%3Afalse%2C%22volume%22%3A0.6%7D; strategyABtestKey=%221690824679.923%22; stream_recommend_feed_params=%22%7B%5C%22cookie_enabled%5C%22%3Atrue%2C%5C%22screen_width%5C%22%3A1536%2C%5C%22screen_height%5C%22%3A864%2C%5C%22browser_online%5C%22%3Atrue%2C%5C%22cpu_core_num%5C%22%3A8%2C%5C%22device_memory%5C%22%3A8%2C%5C%22downlink%5C%22%3A10%2C%5C%22effective_type%5C%22%3A%5C%224g%5C%22%2C%5C%22round_trip_time%5C%22%3A150%7D%22; VIDEO_FILTER_MEMO_SELECT=%7B%22expireTime%22%3A1691443863751%2C%22type%22%3Anull%7D; home_can_add_dy_2_desktop=%221%22; __live_version__=%221.1.1.2169%22; device_web_cpu_core=8; device_web_memory_size=8; xgplayer_user_id=346045893336; csrf_session_id=2e00356b5cd8544d17a0e66484946f28; odin_tt=724eb4dd23bc6ffaed9a1571ac4c757ef597768a70c75fef695b95845b7ffcd8b1524278c2ac31c2587996d058e03414595f0a4e856c53bd0d5e5f56dc6d82e24004dc77773e6b83ced6f80f1bb70627; __ac_nonce=064caded4009deafd8b89; __ac_signature=_02B4Z6wo00f01HLUuwwAAIDBh6tRkVLvBQBy9L-AAHiHf7; ttcid=2e9619ebbb8449eaa3d5a42d8ce88ec835; webcast_leading_last_show_time=1691016922379; webcast_leading_total_show_times=1; webcast_local_quality=sd; live_can_add_dy_2_desktop=%221%22; msToken=1JDHnVPw_9yTvzIrwb7cQj8dCMNOoesXbA_IooV8cezcOdpe4pzusZE7NB7tZn9TBXPr0ylxmv-KMs5rqbNUBHP4P7VBFUu0ZAht_BEylqrLpzgt3y5ne_38hXDOX8o=; msToken=jV_yeN1IQKUd9PlNtpL7k5vthGKcHo0dEh_QPUQhr8G3cuYv-Jbb4NnIxGDmhVOkZOCSihNpA2kvYtHiTW25XNNX_yrsv5FN8O6zm3qmCIXcEe0LywLn7oBO2gITEeg=; tt_scid=mYfqpfbDjqXrIGJuQ7q-DlQJfUSG51qG.KUdzztuGP83OjuVLXnQHjsz-BRHRJu4e986"
|
||
|
||
def setUp(self) -> None:
|
||
os.makedirs(self.temp_dir, exist_ok=True)
|
||
|
||
async def test_ffmpeg_concat(self):
|
||
videos = {
|
||
"urls": [
|
||
"https://d2nj71io21vkj2.cloudfront.net/vod/ap-seoul/1500034234/1397757909715120723.mp4",
|
||
"https://d2nj71io21vkj2.cloudfront.net/vod/ap-seoul/1500034234/1397757909695745707.mp4",
|
||
"https://d2nj71io21vkj2.cloudfront.net/test/slice/outputs/fc-01JWDYP2H5B547XSNM2VC9ZGF4/output_0.mp4",
|
||
"https://d2nj71io21vkj2.cloudfront.net/test/slice/outputs/fc-01JWDYP2H5B547XSNM2VC9ZGF4/output_0.mp4",
|
||
"https://d2nj71io21vkj2.cloudfront.net/test/slice/outputs/fc-01JWDYP2H5B547XSNM2VC9ZGF4/output_0.mp4",
|
||
"https://d2nj71io21vkj2.cloudfront.net/test/slice/outputs/fc-01JWDYP0KJPKHBSEFFPT9KY4DY/output_0.mp4",
|
||
"https://d2nj71io21vkj2.cloudfront.net/test/slice/outputs/fc-01JWDYP1XYBQ1A34XJM566WPZ8/output_0.mp4",
|
||
"https://d2nj71io21vkj2.cloudfront.net/test/slice/outputs/fc-01JWDYP1C13J3CDQESJF99QT5K/output_0.mp4",
|
||
"https://d2nj71io21vkj2.cloudfront.net/test/slice/outputs/fc-01JWDYP1PXBQ0MQGM9YBNPZQ1E/output_0.mp4",
|
||
"https://d2nj71io21vkj2.cloudfront.net/test/slice/outputs/fc-01JWDYP4A7J7RPRFHA1KGWW3S6/output_0.mp4",
|
||
"https://d2nj71io21vkj2.cloudfront.net/test/slice/outputs/fc-01JWDYP2RD3MTRJ5D54KX6T9D3/output_0.mp4",
|
||
"https://d2nj71io21vkj2.cloudfront.net/test/slice/outputs/fc-01JWDYP3Y6JZMFK2WDP2A4KF3N/output_0.mp4",
|
||
"https://d2nj71io21vkj2.cloudfront.net/test/slice/outputs/fc-01JWDYP3PFTM1A2Y7PFQ96HQJJ/output_0.mp4",
|
||
"https://d2nj71io21vkj2.cloudfront.net/test/slice/outputs/fc-01JWDYP4JQVHFWBBN0JEJ5HBHX/output_0.mp4",
|
||
"https://d2nj71io21vkj2.cloudfront.net/test/slice/outputs/fc-01JWDYP54TK35VP8E659AP2201/output_0.mp4",
|
||
"https://d2nj71io21vkj2.cloudfront.net/test/slice/outputs/fc-01JWDYP1PXBQ0MQGM9YBNPZQ1E/output_0.mp4",
|
||
"https://d2nj71io21vkj2.cloudfront.net/test/slice/outputs/fc-01JWDYP6GM06MZEPB5Y12CA2ZX/output_0.mp4",
|
||
"https://d2nj71io21vkj2.cloudfront.net/test/slice/outputs/fc-01JWDYP3A7MSSSAKF4EJAG8H70/output_0.mp4",
|
||
"https://d2nj71io21vkj2.cloudfront.net/test/slice/outputs/fc-01JWDYP4TPNGHW9W12JCTTA80V/output_0.mp4"
|
||
]
|
||
}
|
||
outputs: List[str] = []
|
||
with httpx.Client() as client:
|
||
for i, video in enumerate(videos.get("urls")):
|
||
output_path = f"{self.temp_dir}/input_{i}.mp4"
|
||
with client.stream("GET", video) as response:
|
||
response.raise_for_status()
|
||
file_size = int(response.headers.get("content-length", 0))
|
||
progress_bar = tqdm(
|
||
total=file_size,
|
||
unit='iB',
|
||
unit_scale=True,
|
||
desc='Downloading'
|
||
)
|
||
# 以二进制写模式打开文件
|
||
with open(output_path, 'wb') as file:
|
||
# 分块下载，每次读取1MB
|
||
chunk_size = 1024 * 1024 # 1MB
|
||
downloaded_size = 0
|
||
for chunk in response.iter_bytes(chunk_size=chunk_size):
|
||
if chunk:
|
||
file.write(chunk)
|
||
downloaded_size += len(chunk)
|
||
|
||
progress_bar.update(len(chunk))
|
||
# 每下载100MB记录一次日志
|
||
if downloaded_size % (100 * 1024 * 1024) == 0:
|
||
logger.info(
|
||
f"Downloaded: {downloaded_size / (1024 * 1024 * 1024):.2f} GB")
|
||
progress_bar.close()
|
||
self.assertTrue(os.path.exists(output_path))
|
||
outputs.append(output_path)
|
||
final_output_path = f"{self.temp_dir}/output.mp4"
|
||
await VideoUtils.ffmpeg_concat_medias(media_paths=outputs,
|
||
output_path=final_output_path)
|
||
self.assertTrue(os.path.exists(final_output_path))
|
||
|
||
async def test_ffmpeg_extract_frame(self):
|
||
video_url = "https://d2nj71io21vkj2.cloudfront.net/test/slice/outputs/fc-01JW84BFYAH0F9BS9T9DHDRKKH/output_0.mp4"
|
||
with httpx.Client() as client:
|
||
output_path = f"{self.temp_dir}/input.mp4"
|
||
with client.stream("GET", video_url) as response:
|
||
response.raise_for_status()
|
||
file_size = int(response.headers.get("content-length", 0))
|
||
progress_bar = tqdm(
|
||
total=file_size,
|
||
unit='iB',
|
||
unit_scale=True,
|
||
desc='Downloading'
|
||
)
|
||
# 以二进制写模式打开文件
|
||
with open(output_path, 'wb') as file:
|
||
# 分块下载，每次读取1MB
|
||
chunk_size = 1024 * 1024 # 1MB
|
||
downloaded_size = 0
|
||
for chunk in response.iter_bytes(chunk_size=chunk_size):
|
||
if chunk:
|
||
file.write(chunk)
|
||
downloaded_size += len(chunk)
|
||
|
||
progress_bar.update(len(chunk))
|
||
# 每下载100MB记录一次日志
|
||
if downloaded_size % (100 * 1024 * 1024) == 0:
|
||
logger.info(
|
||
f"Downloaded: {downloaded_size / (1024 * 1024 * 1024):.2f} GB")
|
||
progress_bar.close()
|
||
self.assertTrue(os.path.exists(output_path))
|
||
image_path, metadata = await VideoUtils.ffmpeg_extract_frame_image(video_path=output_path, frame_index=1)
|
||
logger.info(f"Extracted {image_path} and metadata: {metadata}")
|
||
|
||
async def test_ffmpeg_slice_stream(self):
|
||
from pydantic import TypeAdapter
|
||
# #EXT-X-PROGRAM-DATE-TIME:2025-06-23T06:20:30.720+0000
|
||
# #EXT-X-PROGRAM-DATE-TIME:2025-06-23T10:09:38.476+0000
|
||
|
||
# stream_url = "https://cdn.roasmax.cn/test/records/hls/fc-01JXY1AS1HDGS300EQ54ATAKHF/playlist.m3u8"
|
||
stream_url = "https://cdn.roasmax.cn/test/records/hls/fc-01JYDQ3RVMPAEKMYHA0HQ14WAJ/playlist.m3u8"
|
||
adapter = TypeAdapter(List[FFMpegSliceSegment])
|
||
segments = adapter.validate_json("""[
|
||
{
|
||
"start": 12600,
|
||
"end": 13500
|
||
}
|
||
]""")
|
||
for segment in segments:
|
||
logger.info(f"{segment.start.toFormatStr()} --> {segment.end.toFormatStr()}")
|
||
options = FFMPEGSliceOptions()
|
||
outputs = await VideoUtils.ffmpeg_slice_media(media_path=stream_url,
|
||
media_markers=segments,
|
||
options=options,
|
||
is_streams=True,
|
||
output_path=f"{self.temp_dir}/output_stream.mp4")
|
||
for (output, metadata) in outputs:
|
||
self.assertTrue(os.path.exists(output))
|
||
|
||
async def test_ffmpeg_streamlink(self):
|
||
webcast_id = "333555252930"
|
||
segment_duration = 5
|
||
title = None
|
||
live_url: Optional[str] = None
|
||
quality: Optional[str] = None
|
||
|
||
with httpx.Client() as client:
|
||
cookie_getter_response = client.get(url='https://live.douyin.com/')
|
||
cookie_getter_response.raise_for_status()
|
||
stream_url_response = client.get(url='https://live.douyin.com/webcast/room/web/enter/',
|
||
params={
|
||
'aid': 6383,
|
||
'device_platform': 'web',
|
||
'browser_language': 'zh-CN',
|
||
'browser_platform': 'Win32',
|
||
'browser_name': 'Chrome',
|
||
'browser_version': '100.0.0.0',
|
||
'web_rid': webcast_id
|
||
})
|
||
stream_url_response.raise_for_status()
|
||
stream_info_json = stream_url_response.json()
|
||
if data := stream_info_json['data']['data']:
|
||
data = data[0]
|
||
if data['status'] == 2:
|
||
title = data['title']
|
||
live_url = ''
|
||
stream_data = json.loads(data['stream_url']['live_core_sdk_data']['pull_data']['stream_data'])
|
||
for quality_code in ('origin', 'uhd', 'hd', 'sd', 'md', 'ld'):
|
||
if quality_data := stream_data['data'].get(quality_code):
|
||
logger.info(f"质量={quality_code}")
|
||
live_url = quality_data['main']['flv']
|
||
quality = quality_code
|
||
break
|
||
else:
|
||
raise ValueError("直播间没有开播")
|
||
logger.info(f"{title} url = {live_url}")
|
||
if live_url:
|
||
output_dir = f"{self.temp_dir}/{title}"
|
||
os.makedirs(output_dir, exist_ok=True)
|
||
# output_file_pattern = "%10d.ts"
|
||
# output_pattern = f"{output_dir}/{output_file_pattern}"
|
||
logger.info(f"开始录制 [{title}|{quality}] {live_url} 到 {output_dir}")
|
||
# await VideoUtils.ffmpeg_stream_record_as_hls(stream_url=live_url,
|
||
# segment_duration=segment_duration,
|
||
# stream_content_timeout=stream_content_timeout,
|
||
# segments_output_dir=output_dir,
|
||
# playlist_output_dir=manifest_output_dir,
|
||
# playlist_output_method='PUT',
|
||
# playlist_output_headers={
|
||
# "Content-Type": "application/vnd.apple.mpegurl",
|
||
# "Authorization": f"Bearer bowong7777"
|
||
# },
|
||
# manifest_segment_prefix=manifest_segment_prefix)
|
||
logger.info(f'停止录制：{output_dir}')
|
||
|
||
async def test_ffprobe_read_video(self):
|
||
# video = "./videos/input_0.mp4"
|
||
video = "./videos/fc-01JXKK5VWK6B924TDB9WZETACW/playlist.m3u8"
|
||
video = "https://cdn.roasmax.cn/prod/records/hls/fc-01JXKK5VWK6B924TDB9WZETACW/playlist.m3u8"
|
||
video_metadata = VideoUtils.ffprobe_media_metadata(video)
|
||
logger.info(video_metadata)
|
||
self.assertIsNotNone(video_metadata)
|
||
|
||
async def test_ffprobe_read_audio(self):
|
||
audio = "./videos/output.wav"
|
||
audio_metadata = VideoUtils.ffprobe_media_metadata(audio)
|
||
self.assertIsNotNone(audio_metadata)
|
||
|
||
async def test_ffprobe_read_hls(self):
|
||
# hls = "./fc-01JWX1CD01JTV0DECWVWX1X9FM/playlist.m3u8"
|
||
hls = "./fc-01JWZS8954RZP4B13H3TA6TKMM/playlist.m3u8"
|
||
hls_metadata = VideoUtils.ffprobe_media_metadata(hls)
|
||
self.assertIsNotNone(hls_metadata)
|
||
|
||
async def test_ffprobe_read_image(self):
|
||
image = "./1928302183527354368.png"
|
||
image_metadata = VideoUtils.ffprobe_media_metadata(image)
|
||
self.assertIsNotNone(image_metadata)
|
||
|
||
async def test_ffprobe_read_gif(self):
|
||
gif = "./flow_frame.gif"
|
||
gif_metadata = VideoUtils.ffprobe_media_metadata(gif)
|
||
self.assertIsNotNone(gif_metadata)
|
||
|
||
async def test_ffmpeg_convert_stream(self):
|
||
stream_url = "https://cdn.roasmax.cn/prod/records/hls/fc-01JXKJ2JCTDKWJ6J97FN6ABQPM/playlist.m3u8"
|
||
output_path = f"{self.temp_dir}/fc-01JXKJ2JCTDKWJ6J97FN6ABQPM/output.mp4"
|
||
options = FFMPEGSliceOptions()
|
||
local_output, metadata = await VideoUtils.ffmpeg_convert_stream_media(media_stream_url=stream_url,
|
||
options=options,
|
||
output_path=output_path)
|
||
self.assertIsNotNone(local_output)
|
||
|
||
async def test_ffmpeg_best_convert(self):
|
||
media_stream_url = "https://cdn.roasmax.cn/test/records/hls/fc-01JY0ZRKCG6K7WSD9H4Q0X922K/playlist.m3u8"
|
||
local_m3u8_path, temp_dir, diff = await VideoUtils.convert_m3u8_to_local_source(
|
||
media_stream_url=media_stream_url, temp_dir=f"./videos/large")
|
||
|
||
ffmpeg_cmd = VideoUtils.async_ffmpeg_init()
|
||
ffmpeg_cmd.input(local_m3u8_path,
|
||
protocol_whitelist="file,http,https,tcp,tls")
|
||
h264_options = {
|
||
"reset_timestamps": "1",
|
||
"vcodec": "libx264",
|
||
"movflags": "+faststart", # 优化MP4文件结构
|
||
"pix_fmt": "yuv420p",
|
||
"acodec": "aac",
|
||
"aac_coder": "fast",
|
||
"ar": 44100,
|
||
"ac": 2,
|
||
"ab": "192k",
|
||
"crf": 18,
|
||
"r": 30
|
||
}
|
||
ffmpeg_cmd.output("./videos/output_large_h264.mp4", options=h264_options)
|
||
await ffmpeg_cmd.execute()
|
||
|
||
async def test_ffmpeg_subtitle_apply(self):
|
||
video = "./videos/subtitle_test/A美洋MEIYANG 方圆牛仔马甲 率性哲学！八支精棉牛仔无袖上衣-周二.mp4"
|
||
subtitle = "./videos/subtitle_test/A美洋MEIYANG 方圆牛仔马甲 率性哲学！八支精棉牛仔无袖上衣-周二.vtt"
|
||
output = "./videos/subtitle_test/output.mp4"
|
||
output_video, metadata = await VideoUtils.ffmpeg_subtitle_apply(media_path=video, subtitle_path=None,
|
||
embed_subtitle_path=subtitle, font_dir=None,
|
||
output_path=output)
|
||
self.assertEqual(len(metadata.streams), 3)
|
||
|
||
if __name__ == '__main__':
|
||
unittest.main()
|