douyin/tool/video_get.py

import json
import os
import re  
import httpx  
import base64  
import os
import sys
sys.path.append(os.path.dirname(os.path.dirname(__file__)))
from douyin.video_get_iframe_by_item import get_iframe_by_item_id
from config import logger,ASR_EXE,ASR_ADDRESS,TEMP_DIR

'''
url = "4.30 03/30 bNJ:/ L@w.FH 踩中风口真的可以翻身吗？从一穷二白到AI逆袭，他怎么做到的？ # 东方对话 # 人工智能 # ai  https://v.douyin.com/i8Df1CC/ 复制此链接，打开Dou音搜索，直接观看视频！"
ret = {
    "code": 0,
    "msg": "",
    "data": [
        "https://v26-web.douyinvod.com/b96458639649189a611ac504a98ab1a2/65f2e8f7/video/tos/cn/tos-cn-ve-15/oMP2rHcp5Qffk5bH3QI29eZ02DRMIsAlfugfAE/?a=6383&ch=5&cr=3&dr=0&lr=all&cd=0%7C0%7C0%7C3&cv=1&br=1307&bt=1307&cs=0&ds=3&ft=bvTKJbQQqUYSfJ.ZPo0OqY8hFgpis3kQejKJ4cbTYG0P3-I&mime_type=video_mp4&qs=0&rc=OTY1aDQzPDg4aTo4PDs8N0Bpand4ZWU6ZmZ5cTMzNGkzM0A2Mi81LzYxNWAxNTE1YDM1YSNtLmpmcjRncmNgLS1kLS9zcw%3D%3D&btag=e00008000&cquery=100a&dy_q=1710414563&feature_id=f0150a16a324336cda5d6dd0b69ed299&l=20240314190922833D1C0045FB7109049F",
        "https://v3-web.douyinvod.com/fe13b075a0f1596fd800244131f91107/65f2e8f7/video/tos/cn/tos-cn-ve-15/oMP2rHcp5Qffk5bH3QI29eZ02DRMIsAlfugfAE/?a=6383&ch=5&cr=3&dr=0&lr=all&cd=0%7C0%7C0%7C3&cv=1&br=1307&bt=1307&cs=0&ds=3&ft=bvTKJbQQqUYSfJ.ZPo0OqY8hFgpis3kQejKJ4cbTYG0P3-I&mime_type=video_mp4&qs=0&rc=OTY1aDQzPDg4aTo4PDs8N0Bpand4ZWU6ZmZ5cTMzNGkzM0A2Mi81LzYxNWAxNTE1YDM1YSNtLmpmcjRncmNgLS1kLS9zcw%3D%3D&btag=e00008000&cquery=100a&dy_q=1710414563&feature_id=f0150a16a324336cda5d6dd0b69ed299&l=20240314190922833D1C0045FB7109049F",
        "https://www.douyin.com/aweme/v1/play/?video_id=v0200fg10000cnfvucjc77ubibmu1tpg&line=0&file_id=8469c88945874fc5854990218e225396&sign=f5a2d2c7a6e4cb83ce41a254a73f27a2&is_play_url=1&source=PackSourceEnum_FEED&aid=6383"
    ]
}
'''
async def get_video_download_urls(url, video_type="", platform="", server_addr="http://10.0.0.12:9082/video-get"):  
    payload = {  
        "url": url,  
        "type": video_type,  
        "platform": platform  
    }  
      
    async with httpx.AsyncClient() as client:  
        logger.debug(f"{json.dumps(payload)  }")
        response = await client.post(  
            server_addr,  
            headers={"Content-Type": "application/json"},  
            content=json.dumps(payload)  
        )  
          
        res_json = response.json()  
          
        # 假设响应中包含'err_no'和可能的'err_msg'字段来表示错误  
        # err_no = res_json.get("code")  
          
        return res_json


async def download_video(url, save_file_path) ->str:
    if type(url) is list:
        urls = url
    else:
        urls = [url]
    async with httpx.AsyncClient() as client:  
        for url in urls:
            response = await client.get(url)  
            # 检查是否成功获取响应  
            if response.status_code == 200:  
                # 以二进制写模式打开一个文件  
                with open(save_file_path, 'wb') as file:  
                    # 异步迭代响应的内容块并写入文件  
                    async for chunk in response.aiter_bytes():  
                        file.write(chunk)  
                return save_file_path
            else:  
                logger.info(f"try to download video faild,status: {response.status_code},url: {url}")    
                continue
        # raise Exception(f"遍历所有 urls 都无法下载 {urls}")

async def download_video_by_item_id(item_id, save_file_path):
    response = await get_iframe_by_item_id(item_id)
    # # iframe = '<iframe width="1080" height="1920" frameborder="0" src="https://open.douyin.com/player/video?vid=7259290547288870144&autoplay=0" referrerpolicy="unsafe-url" allowfullscreen></iframe>',
    iframe = response.get("data").get("iframe_code")
    urls = await get_video_download_urls(iframe)
    return await download_video(urls, save_file_path)

async def download_video_by_share_link(share_str, save_file_path) ->str:
    urls = await get_video_download_urls(share_str)
    return await download_video(urls, save_file_path)