| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166 |
- """
- @author: luojunhui
- """
- import os
- import oss2
- import aiohttp
- import aiofiles
- import requests
- from hashlib import md5
- from uuid import uuid4
- from fake_useragent import FakeUserAgent
- async def download_cover(file_path, platform, cover_url):
- """
- 下载视频封面
- :param platform:
- :param cover_url:
- :param file_path:
- :return:
- """
- headers = request_header(platform=platform, url=cover_url, download_type="cover")
- response = requests.get(url=cover_url, headers=headers)
- if b"<html>" in response.content:
- return None
- elif response.status_code != 200:
- return None
- else:
- with open(file_path, "wb") as f:
- f.write(response.content)
- return file_path
- def request_header(platform, url, download_type="video"):
- """
- 请求头
- :return:
- """
- if platform == "xg_search":
- if "v9-xg-web-pc.ixigua.com" in url:
- headers = {
- "Accept": "*/*",
- "Accept-Language": "zh-CN,zh;q=0.9",
- "Host": "v9-xg-web-pc.ixigua.com",
- "User-Agent": FakeUserAgent().chrome,
- "Origin": "https://www.ixigua.com/",
- "Referer": "https://www.ixigua.com/"
- }
- elif "v3-xg-web-pc.ixigua.com" in url:
- headers = {
- "Accept": "*/*",
- "Accept-Language": "zh-CN,zh;q=0.9",
- "Host": "v3-xg-web-pc.ixigua.com",
- "User-Agent": FakeUserAgent().chrome,
- "Origin": "https://www.ixigua.com/",
- "Referer": "https://www.ixigua.com/"
- }
- elif download_type == "cover":
- headers = {
- 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7',
- 'Accept-Language': 'en,zh;q=0.9,zh-CN;q=0.8',
- 'Cache-Control': 'max-age=0',
- 'Proxy-Connection': 'keep-alive',
- 'Upgrade-Insecure-Requests': '1',
- 'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36'
- }
- else:
- headers = {
- "Accept": "*/*",
- "Accept-Language": "zh-CN,zh;q=0.9",
- "Host": "v3-xg-web-pc.ixigua.com",
- "User-Agent": FakeUserAgent().chrome,
- "Origin": "https://www.ixigua.com/",
- "Referer": "https://www.ixigua.com/"
- }
- elif platform == "baidu_search":
- headers = {
- "Accept": "*/*",
- "Accept-Language": "zh-CN,zh;q=0.9",
- "User-Agent": FakeUserAgent().chrome,
- }
- elif platform == "wx_search":
- headers = {
- "Accept": "*/*",
- "Accept-Language": "zh-CN,zh;q=0.9",
- "User-Agent": FakeUserAgent().chrome,
- "Origin": "https://mp.weixin.qq.com",
- "Referer": "https://mp.weixin.qq.com"
- }
- elif platform == "dy_search":
- headers = {
- 'accept': '*/*',
- 'accept-language': 'en,zh;q=0.9,zh-CN;q=0.8',
- 'priority': 'i',
- 'range': 'bytes=0-',
- 'referer': 'https://v11-coldf.douyinvod.com/',
- 'user-agent': FakeUserAgent().chrome
- }
- else:
- headers = {}
- return headers
- async def download_video(file_path, platform, video_url, download_type="video"):
- """
- :param download_type:
- :param video_url:
- :param platform:
- :param file_path:
- :return:
- """
- headers = request_header(platform=platform, url=video_url, download_type=download_type)
- if os.path.exists(file_path):
- file_size = os.path.getsize(file_path)
- headers["Range"] = f"bytes={file_size}-"
- else:
- file_size = 0
- async with aiohttp.ClientSession() as session:
- async with session.get(video_url, headers=headers) as response:
- if response.status in [200, 206]:
- if file_size > 0:
- async with aiofiles.open(file_path, "ab+") as f:
- # 以1MB为单位分块下载
- async for chunk in response.content.iter_chunked(1024 * 1024):
- await f.write(chunk)
- else:
- async with aiofiles.open(file_path, "wb") as f:
- # 以1MB为单位分块下载
- async for chunk in response.content.iter_chunked(1024 * 1024):
- await f.write(chunk)
- else:
- print(response.status)
- return file_path
- def generate_video_path(platform, video_id):
- """
- 通过视频信息生成唯一视频地址
- :return:
- """
- index = "{}-{}-{}".format(platform, video_id, uuid4())
- index = md5(index.encode()).hexdigest()
- file_name = "{}.mp4".format(index)
- cover_name = "{}.png".format(index)
- file_path = os.path.join(os.getcwd(), "static", file_name)
- cover_path = os.path.join(os.getcwd(), "static", cover_name)
- return file_path, cover_path
- async def upload_to_oss(local_video_path, download_type):
- """
- 把视频上传到 oss
- :return:
- """
- oss_video_key = "long_articles/{}/".format(download_type) + str(uuid4())
- access_key_id = "LTAIP6x1l3DXfSxm"
- access_key_secret = "KbTaM9ars4OX3PMS6Xm7rtxGr1FLon"
- endpoint = "oss-cn-hangzhou.aliyuncs.com"
- bucket_name = "art-pubbucket"
- bucket = oss2.Bucket(
- oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name
- )
- bucket.put_object_from_file(key=oss_video_key, filename=local_video_path)
- return oss_video_key
|
