import json import math import os import re import hashlib import tornado.log import tornado.web import tornado.httpserver import tornado.ioloop from tornado.options import options, parse_command_line, define from streaming_form_data import StreamingFormDataParser from streaming_form_data.targets import FileTarget import subprocess define('rootdir', '.', help="upload to direct") define('port', 8018, help="serve port") class Md5FileTarget(FileTarget): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self._hash_md5 = hashlib.md5() def on_data_received(self, chunk: bytes): if self._fd: self._fd.write(chunk) self._hash_md5.update(chunk) @property def md5_value(self): return self._hash_md5.hexdigest() @tornado.web.stream_request_body class UploadFileHandler(tornado.web.RequestHandler): def __init__(self, application, request, **kwargs): super().__init__(application, request, **kwargs) self.first_chunk_received = False self.upload_file = None self.parser = None self.name = "" self.m_file_path = None def headers_set(self): self.set_header('Access-Control-Allow-Origin', '*') self.set_header('Access-Control-Allow-Methods', 'POST, OPTIONS') self.set_header('Access-Control-Allow-Headers', '*, Content-Type, Origin') def prepare(self): return def init_parser(self, file_path): self.upload_file = Md5FileTarget(file_path) self.parser = StreamingFormDataParser(headers=self.request.headers) self.parser.register('file', self.upload_file) def parse_file_path(self, chunk): file_name_pattern = re.compile(b'filename=\\"(.+)\\"') filename = file_name_pattern.search(chunk).groups()[0] self.name = filename dir_url = self.request.path.split("/", 3)[-1] rel_path = dir_url.replace("/", os.sep) abs_dir = os.path.join(os.path.abspath(options.rootdir), rel_path) if not os.path.exists(abs_dir): os.makedirs(abs_dir) self.m_file_path = abs_dir file_path = os.path.join(abs_dir, filename.decode()) return file_path # only POST/PUT with body can call this def data_received(self, chunk): if not self.first_chunk_received: full_path = self.parse_file_path(chunk) self.init_parser(full_path) self.first_chunk_received = True self.parser.data_received(chunk) def options(self, *args, **kwargs): self.headers_set() self.set_status(204) self.finish() def parser_file(self): order = ['/usr/bin/ffprobe', '-i', self.upload_file.filename, '-print_format', 'json', '-show_streams', '-show_format', '-v', 'quiet'] order = ['ffprobe', '-i', self.upload_file.filename, '-print_format', 'json', '-show_streams', '-show_format', '-v', 'quiet'] media_info = subprocess.check_output(order) media_info = json.loads(media_info) storage_info = {} for i in media_info["streams"]: if i["codec_type"] == "video": storage_info["v_codec"] = i["codec_name"] storage_info["v_profile"] = i.get("profile", "") storage_info["width"] = int(i["width"]) storage_info["height"] = int(i["height"]) storage_info["v_bit_rate"] = int(i.get("bit_rate", 0)) storage_info["sample_aspect_ratio"] = i.get("sample_aspect_ratio", "1:1") storage_info["display_aspect_ratio"] = i.get("display_aspect_ratio", "16:9") duration = i.get("duration") or media_info["format"].get("duration", 0) storage_info["seconds"] = math.ceil(float(duration)) storage_info["dub"] = i["disposition"]["dub"] frame_rate = i.get("r_frame_rate") or i.get("avg_frame_rate", "0") storage_info["frame_rate"] = math.ceil(eval(frame_rate)) for i in media_info["streams"]: if i["codec_type"] == "audio": # 可能有多个；只取第一个 storage_info["a_codec"] = i["codec_name"] storage_info["a_channels"] = i["channels"] storage_info["a_bit_rate"] = int(i.get("bit_rate", 0)) storage_info["a_sample_rate"] = int(i["sample_rate"]) break storage_info["bit_rate"] = int(media_info["format"]["bit_rate"]) storage_info["size"] = int(media_info["format"]["size"]) # 计算视频码率 if not storage_info["v_bit_rate"]: storage_info["v_bit_rate"] = int(storage_info["bit_rate"]) - int(storage_info.get("a_bit_rate", 0)) return storage_info def post(self, *args, **kwargs): self.headers_set() cmd = self.get_argument("cmd", "") status = 403 if cmd == "upload": storage_info = self.resp_file() self.finish(storage_info) elif cmd == "m3u8": storage_info = self.resp_file() storage_info["is_hls"] = self.handle_hls() self.finish(storage_info) else: self.set_status(status) self.finish() def resp_file(self): self.set_status(200) md5 = self.upload_file.md5_value print(self.upload_file) print(self.upload_file.filename) print(self.m_file_path) tornado.log.app_log.info("upload file path is {}".format(self.upload_file.filename)) # 分析出视频信息 try: storage_info = self.parser_file() except Exception as e: tornado.log.app_log.info("upload file error is {}".format(e)) print(e) storage_info = dict() tornado.log.app_log.info("upload file name is {}, md5: {}".format(self.name, md5)) # self.write(self.response) storage_info["md5"] = md5 return storage_info def handle_hls(self): play_file = os.path.join(self.make_m3u8_path(self.upload_file.md5_value), 'playlist.m3u8') out_file = os.path.join(self.make_m3u8_path(self.upload_file.md5_value), 'output%03d.ts') url = self.upload_file.filename return self.work_cmd(url, play_file, out_file) @staticmethod def work_cmd(url, play_file, out_file): try: order = ['ffmpeg', '-i', url, '-c', 'copy', '-map', '0','-threads','20','-preset',' ultrafast' ,'-f', 'segment', '-segment_list', play_file, '-segment_time', '10', out_file] # /usr/bin/ffprobe subprocess.run(order) except Exception as e: tornado.log.app_log.info("handle_hls error is {}".format(e)) print(e) return False return True def make_m3u8_path(self, md5): file_path = os.path.join(self.m_file_path, md5) if not os.path.exists(file_path): os.makedirs(file_path) print(file_path) return file_path class BaseHandler(tornado.web.RequestHandler): def set_default_headers(self): self.set_header("Access-Control-Allow-Origin", "*") self.set_header("Access-Control-Allow-Headers", "Content-Type,x-requested-with,x-token") self.set_header("Access-Control-Allow-Methods", "POST, GET, OPTIONS") def options(self): self.set_status(204) self.finish() class HlsHandler(BaseHandler): def post(self): obj_json = json.loads(self.request.body) md5 = obj_json.get("md5") url = obj_json.get("url") if url.find("/matrix/") != -1: file_path = url.split("/matrix")[-1] file_path = "/mnt/matrix" + file_path else: file_path = url.split("/test")[-1] file_path = "/mnt/test" + file_path file_path_dir = "/".join(file_path.split("/")[:-1]) hls_file_dir = os.path.join(file_path_dir, md5) os.makedirs(hls_file_dir, exist_ok=True) play_file_path = os.path.join(hls_file_dir, 'playlist.m3u8') out_file_path = os.path.join(hls_file_dir, 'output%03d.ts') is_hls = UploadFileHandler.work_cmd(file_path, play_file_path, out_file_path) return self.finish({"code": 0, "data": {"is_hls": is_hls}}) def main(): parse_command_line() port = options.port app = tornado.web.Application([ (r'/api/(?P\d+)/(?P.*)', UploadFileHandler), (r'/api/hls', HlsHandler), ]) http_server = tornado.httpserver.HTTPServer(app, max_body_size=1 << 36) tornado.log.app_log.info("start upload server at port {}".format(port)) http_server.listen(port) tornado.ioloop.IOLoop.current().start() if __name__ == '__main__': main()