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()