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_worker_model import FFMpegSliceSegment
from BowongModalFunctions.utils.VideoUtils import VideoUtils, FFMPEGSliceOptions


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

        stream_url = "https://cdn.roasmax.cn/test/records/hls/fc-01JXY1AS1HDGS300EQ54ATAKHF/playlist.m3u8"
        adapter = TypeAdapter(List[FFMpegSliceSegment])
        segments = adapter.validate_json("""[
        {
          "start": 920.425,
          "end": 921.688
        },
        {
          "start": 922.425,
          "end": 923.688
        },
        {
          "start": 940.425,
          "end": 941.688
        }
      ]""")
        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()

    if __name__ == '__main__':
        unittest.main()
