#!/usr/bin/env python3
# coding: utf-8
import os
import logging
import pathlib
import requests
import threading
from urllib import parse
def to_logger(file_name):
logger = logging.getLogger(file_name)
logger.setLevel(logging.INFO)
fmt = logging.Formatter("%(asctime)s - %(levelname)s: %(message)s")
fh = logging.FileHandler(file_name, encoding="utf-8")
fh.setFormatter(fmt)
if not logger.handlers:
logger.addHandler(fh)
return logger
def down2video(video_url, timeout=3, retry=5, error_log="error.log"):
r = None
ct = 0
# 请求头部
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/102.1.4753.73 Safari/527.56 Edg/102.1.4753.73",
"Accept-Encoding": "identity"
}
# 请求URL,例如http://www.example.com/video/to1.mp4
url = video_url
# 请求超时时间,默认3秒
timeout = timeout
# 失败重试次数。默认5次
retry = retry
# 错误日志文件,默认error.log
error_log = error_log
to_string = parse.urlparse(url)
# 存放目录,使用绝对路径,例如/root/video
video_dir = os.getcwd() + os.path.dirname(to_string.path)
# 创建目录,目录不存在则创建,即使目录已存在也不会抛出异常
pathlib.Path(video_dir).mkdir(parents=True, exist_ok=True)
# 文件名,使用绝对路径,例如/root/video/to1.mp4
video_name = video_dir + "/" + os.path.basename(to_string.path)
# 错误日志
logger = to_logger(error_log)
while ct < retry:
try:
r = requests.get(url=url, headers=headers, timeout=timeout)
r.raise_for_status()
if r.status_code == 200:
# 将body写入文件,适合小文件,大文件可以用chunk方式
with open(video_name, "wb") as fp:
fp.write(r.content)
return r
except Exception as e:
logger.info(e)
ct += 1
logger.info(url)
logger.info(r)
return r
def main(urls):
urls = urls
th_list = []
th_num = 4
for url in urls:
t = threading.Thread(target=down2video, args=[url])
th_list.append(t)
t.start()
while len(th_list) > th_num:
th_list = [th for th in th_list if th.is_alive()]
if __name__ == "__main__":
import time
start_time = time.time()
import sys
url_file = sys.argv[1]
urls = []
with open(url_file, "r") as fp:
for url in fp.readlines():
urls.append(url.strip("\n"))
main(urls)
end_time = time.time()
print("Total time:{}(s)".format(end_time-start_time))
# import sys, m3u8
# m3u8_url = sys.argv[1]
# play_list = m3u8.load(m3u8_url)
# urls = []
# for line in play_list.files:
# urls.append(play_list.base_uri + line)
# main(urls)
# url = ["http://192.168.3.120/video/video10/0000.ts"]
# # 根据URI自动创建指定的目录,并将视频存放在此目录下
# main(url)
root@ubuntu:~/py# cat url_list.txt
http://192.168.3.120/video/video10/0000.ts
http://192.168.3.120/video/video10/0001.ts
http://192.168.3.120/video/video10/0002.ts
http://192.168.3.120/video/video10/0003.ts
http://192.168.3.120/video/video10/0004.ts
http://192.168.3.120/video/video10/0005.ts
root@ubuntu:~/py#
root@ubuntu:~/py# ./down2video.py url_list.txt
Total time:0.03328990936279297(s)
root@ubuntu:~/py#
root@ubuntu:~/py# tree
.
├── down2video.py
├── error.log
├── url_list.txt
└── video
└── video10
├── 0000.ts
├── 0001.ts
├── 0002.ts
├── 0003.ts
├── 0004.ts
└── 0005.ts
2 directories, 9 files
root@ubuntu:~/py#
本人并不会写代码,基本上都是网上抄来改的。
目前也一直在用,还是可以的。
网友评论