1import httpx
2
3async def download_douyin_video(video_url: str, save_path: str) -> bool:
4    """
5    下载抖音视频
6
7    Args:
8        video_url: 视频播放地址
9        save_path: 保存路径
10
11    Returns:
12        是否下载成功
13    """
14    headers = {
15        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
16        "Referer": "https://www.douyin.com/",  # 必须
17        "Accept": "*/*",
18        "Accept-Encoding": "identity",  # 避免压缩，方便流式下载
19        "Range": "bytes=0-",  # 支持断点续传
20    }
21
22    async with httpx.AsyncClient(follow_redirects=True, timeout=60) as client:
23        async with client.stream("GET", video_url, headers=headers) as response:
24            if response.status_code in [200, 206]:
25                with open(save_path, "wb") as f:
26                    async for chunk in response.aiter_bytes(chunk_size=8192):
27                        f.write(chunk)
28                return True
29    return False

1import httpx
2
3async def download_tiktok_video_web(
4    video_url: str,
5    save_path: str,
6    tt_chain_token: str  # 必须提供
7) -> bool:
8    """
9    下载 TikTok Web 端视频
10
11    注意:
12    - Web 端链接必须携带 tt_chain_token Cookie
13    - tt_chain_token 可从浏览器或 API 响应中获取
14    """
15    headers = {
16        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/142.0.0.0 Safari/537.36",
17        "Referer": "https://www.tiktok.com/",
18        "Accept": "video/webm,video/ogg,video/*;q=0.9,*/*;q=0.8",
19        "Accept-Language": "en-US,en;q=0.5",
20        "Cookie": f"tt_chain_token={tt_chain_token}",  # 必须
21        "Sec-Fetch-Dest": "video",
22        "Sec-Fetch-Mode": "no-cors",
23        "Sec-Fetch-Site": "cross-site",
24    }
25
26    async with httpx.AsyncClient(follow_redirects=True, timeout=60) as client:
27        async with client.stream("GET", video_url, headers=headers) as response:
28            if response.status_code in [200, 206]:
29                with open(save_path, "wb") as f:
30                    async for chunk in response.aiter_bytes(chunk_size=8192):
31                        f.write(chunk)
32                return True
33            elif response.status_code == 403:
34                print("下载失败: tt_chain_token 无效或缺失")
35    return False

1import httpx
2import subprocess
3from pathlib import Path
4
5async def download_bilibili_video(
6    video_url: str,
7    audio_url: str,
8    save_path: str,
9    cookie: str = None
10) -> bool:
11    """
12    下载 B站 视频 (需要合并音视频)
13
14    Args:
15        video_url: 视频流地址
16        audio_url: 音频流地址
17        save_path: 最终保存路径
18        cookie: 可选，高清视频需要 SESSDATA
19    """
20    headers = {
21        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/142.0.0.0 Safari/537.36",
22        "Referer": "https://www.bilibili.com/",  # 必须，否则 403
23        "Accept": "*/*",
24        "Accept-Encoding": "identity",
25        "Origin": "https://www.bilibili.com",
26    }
27
28    if cookie:
29        headers["Cookie"] = cookie
30
31    # 临时文件路径
32    video_temp = save_path + ".video.m4s"
33    audio_temp = save_path + ".audio.m4s"
34
35    async with httpx.AsyncClient(follow_redirects=True, timeout=120) as client:
36        # 下载视频流
37        async with client.stream("GET", video_url, headers=headers) as response:
38            if response.status_code not in [200, 206]:
39                return False
40            with open(video_temp, "wb") as f:
41                async for chunk in response.aiter_bytes(chunk_size=8192):
42                    f.write(chunk)
43
44        # 下载音频流
45        async with client.stream("GET", audio_url, headers=headers) as response:
46            if response.status_code not in [200, 206]:
47                return False
48            with open(audio_temp, "wb") as f:
49                async for chunk in response.aiter_bytes(chunk_size=8192):
50                    f.write(chunk)
51
52    # 使用 FFmpeg 合并音视频
53    try:
54        subprocess.run([
55            "ffmpeg", "-y",
56            "-i", video_temp,
57            "-i", audio_temp,
58            "-c:v", "copy",
59            "-c:a", "copy",
60            save_path
61        ], check=True, capture_output=True)
62
63        # 删除临时文件
64        Path(video_temp).unlink()
65        Path(audio_temp).unlink()
66        return True
67    except subprocess.CalledProcessError:
68        return False

1import httpx
2
3async def download_xiaohongshu_video(video_url: str, save_path: str) -> bool:
4    """
5    下载小红书视频
6
7    小红书对 Referer 检查较严格
8    """
9    headers = {
10        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
11        "Referer": "https://www.xiaohongshu.com/",  # 必须
12        "Accept": "*/*",
13        "Origin": "https://www.xiaohongshu.com",
14        "Sec-Fetch-Dest": "video",
15        "Sec-Fetch-Mode": "cors",
16        "Sec-Fetch-Site": "cross-site",
17    }
18
19    async with httpx.AsyncClient(follow_redirects=True, timeout=60) as client:
20        async with client.stream("GET", video_url, headers=headers) as response:
21            if response.status_code in [200, 206]:
22                with open(save_path, "wb") as f:
23                    async for chunk in response.aiter_bytes(chunk_size=8192):
24                        f.write(chunk)
25                return True
26    return False

1import httpx
2
3async def download_weibo_video(video_url: str, save_path: str) -> bool:
4    """
5    下载微博视频
6
7    微博支持多个 Referer 域名
8    """
9    headers = {
10        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
11        "Referer": "https://weibo.com/",  # 或 https://m.weibo.cn/
12        "Accept": "*/*",
13        "Accept-Encoding": "identity",
14    }
15
16    async with httpx.AsyncClient(follow_redirects=True, timeout=60) as client:
17        async with client.stream("GET", video_url, headers=headers) as response:
18            if response.status_code in [200, 206]:
19                with open(save_path, "wb") as f:
20                    async for chunk in response.aiter_bytes(chunk_size=8192):
21                        f.write(chunk)
22                return True
23    return False

1import httpx
2
3async def download_youtube_video(
4    video_url: str,
5    save_path: str,
6    proxy: str = None  # 必须是美国加州洛杉矶代理
7) -> bool:
8    """
9    下载 YouTube 视频
10
11    重要:
12    - 必须使用美国加利福尼亚州洛杉矶地区的代理 IP
13    - 其他地区 IP 会返回 403 Forbidden
14    - 视频流 URL 有时效性，通常 6 小时内有效
15    """
16    headers = {
17        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/142.0.0.0 Safari/537.36",
18        "Referer": "https://www.youtube.com/",
19        "Origin": "https://www.youtube.com",
20        "Accept": "*/*",
21        "Accept-Language": "en-US,en;q=0.9",
22        "Accept-Encoding": "identity",  # 避免压缩
23        "Range": "bytes=0-",  # 支持断点续传
24        "Sec-Fetch-Dest": "video",
25        "Sec-Fetch-Mode": "no-cors",
26        "Sec-Fetch-Site": "cross-site",
27    }
28
29    # 代理配置 - 必须是洛杉矶 IP
30    proxies = None
31    if proxy:
32        proxies = {
33            "http://": proxy,
34            "https://": proxy,
35        }
36
37    async with httpx.AsyncClient(
38        follow_redirects=True,
39        timeout=120,
40        proxies=proxies
41    ) as client:
42        async with client.stream("GET", video_url, headers=headers) as response:
43            if response.status_code in [200, 206]:
44                with open(save_path, "wb") as f:
45                    async for chunk in response.aiter_bytes(chunk_size=8192):
46                        f.write(chunk)
47                return True
48            elif response.status_code == 403:
49                print("下载失败: IP 地区不符合要求，请使用美国加州洛杉矶的代理")
50            elif response.status_code == 410:
51                print("下载失败: 视频链接已过期，请重新获取")
52    return False

1import subprocess
2from pathlib import Path
3
4async def download_youtube_video_with_audio(
5    video_url: str,
6    audio_url: str,
7    save_path: str,
8    proxy: str = None
9) -> bool:
10    """
11    下载 YouTube 视频并合并音频
12
13    YouTube 1080p 及以上清晰度的视频，音频和视频是分离的
14    需要分别下载后使用 FFmpeg 合并
15    """
16    video_temp = save_path + ".video.mp4"
17    audio_temp = save_path + ".audio.m4a"
18
19    # 下载视频流
20    success = await download_youtube_video(video_url, video_temp, proxy)
21    if not success:
22        return False
23
24    # 下载音频流
25    success = await download_youtube_video(audio_url, audio_temp, proxy)
26    if not success:
27        return False
28
29    # 使用 FFmpeg 合并
30    try:
31        subprocess.run([
32            "ffmpeg", "-y",
33            "-i", video_temp,
34            "-i", audio_temp,
35            "-c:v", "copy",
36            "-c:a", "aac",
37            save_path
38        ], check=True, capture_output=True)
39
40        # 删除临时文件
41        Path(video_temp).unlink()
42        Path(audio_temp).unlink()
43        return True
44    except subprocess.CalledProcessError as e:
45        print(f"FFmpeg 合并失败: {e}")
46        return False

1# 推荐的代理配置 - 必须是洛杉矶地区
2LA_PROXIES = [
3    "http://user:pass@la-proxy1.example.com:8080",  # 洛杉矶代理 1
4    "http://user:pass@la-proxy2.example.com:8080",  # 洛杉矶代理 2
5    "socks5://user:pass@la-socks.example.com:1080", # SOCKS5 代理
6]
7
8# 验证代理 IP 是否在洛杉矶
9async def verify_proxy_location(proxy: str) -> bool:
10    """验证代理 IP 是否在洛杉矶地区"""
11    async with httpx.AsyncClient(proxies={"https://": proxy}) as client:
12        response = await client.get("https://ipapi.co/json/")
13        data = response.json()
14
15        city = data.get("city", "")
16        region = data.get("region", "")
17        country = data.get("country_code", "")
18
19        # 检查是否在加州洛杉矶
20        is_la = (
21            country == "US" and
22            region == "California" and
23            "Los Angeles" in city
24        )
25
26        if is_la:
27            print(f"✓ 代理 IP 位于: {city}, {region}, {country}")
28        else:
29            print(f"✗ 代理 IP 位于: {city}, {region}, {country} (不符合要求)")
30
31        return is_la

1from fastapi import FastAPI, Query, Response
2from fastapi.responses import StreamingResponse
3import httpx
4
5app = FastAPI()
6
7# 各平台的请求头配置
8PLATFORM_HEADERS = {
9    "douyin": {
10        "Referer": "https://www.douyin.com/",
11        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
12    },
13    "tiktok": {
14        "Referer": "https://www.tiktok.com/",
15        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
16    },
17    "bilibili": {
18        "Referer": "https://www.bilibili.com/",
19        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
20    },
21    "xiaohongshu": {
22        "Referer": "https://www.xiaohongshu.com/",
23        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
24    },
25    "weibo": {
26        "Referer": "https://weibo.com/",
27        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
28    },
29}
30
31
32@app.get("/api/download/video")
33async def proxy_video(
34    url: str = Query(..., description="视频 URL"),
35    platform: str = Query(..., description="平台标识"),
36):
37    """
38    视频代理下载接口
39
40    解决前端跨域和防盗链问题
41    """
42    headers = PLATFORM_HEADERS.get(platform, {})
43
44    async def video_stream():
45        async with httpx.AsyncClient(follow_redirects=True, timeout=120) as client:
46            async with client.stream("GET", url, headers=headers) as response:
47                async for chunk in response.aiter_bytes(chunk_size=8192):
48                    yield chunk
49
50    return StreamingResponse(
51        video_stream(),
52        media_type="video/mp4",
53        headers={
54            "Content-Disposition": "attachment; filename=video.mp4",
55            "Access-Control-Allow-Origin": "*",  # 允许跨域
56        }
57    )
58
59
60@app.get("/api/download/image")
61async def proxy_image(
62    url: str = Query(..., description="图片 URL"),
63    platform: str = Query(..., description="平台标识"),
64):
65    """图片代理下载接口"""
66    headers = PLATFORM_HEADERS.get(platform, {})
67
68    async with httpx.AsyncClient(follow_redirects=True) as client:
69        response = await client.get(url, headers=headers)
70
71        return Response(
72            content=response.content,
73            media_type=response.headers.get("content-type", "image/jpeg"),
74            headers={
75                "Access-Control-Allow-Origin": "*",
76            }
77        )