Madiator2011/lbry-sdk
1
0
Fork 0
forked from LBRYCommunity/lbry-sdk
Code Pull requests Activity

ensure only webm-type MKV files

Browse source
This commit is contained in:
Brannon King 2020-03-23 14:21:22 -06:00 committed by Jack Robison
parent 15eb5d47eb
commit 66857e72a4
No known key found for this signature in database
GPG key ID: DF25C68FE0239BB2
1 changed files with 31 additions and 14 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

45
lbry/file_analysis.py
View file

@ -113,9 +113,24 @@ class VideoFileAnalyzer:
def _verify_container(scan_data: json): def _verify_container(scan_data: json):
container = scan_data["format"]["format_name"] container = scan_data["format"]["format_name"]
log.debug(" Detected container is %s", container) log.debug(" Detected container is %s", container)
if not {"webm", "mp4", "3gp", "ogg"}.intersection(container.split(",")): splits = container.split(",")
if not {"webm", "mp4", "3gp", "ogg"}.intersection(splits):
return "Container format is not in the approved list of WebM, MP4. " \ return "Container format is not in the approved list of WebM, MP4. " \
f"Actual: {container} [{scan_data['format']['format_long_name']}]" f"Actual: {container} [{scan_data['format']['format_long_name']}]"
if "matroska" in splits:
for stream in scan_data["streams"]:
if stream["codec_type"] == "video":
codec = stream["codec_name"]
if not {"vp8", "vp9", "av1"}.intersection(codec.split(",")):
return "WebM format requires VP8/9 or AV1 video. " \
f"Actual: {codec} [{stream['codec_long_name']}]"
elif stream["codec_type"] == "audio":
codec = stream["codec_name"]
if not {"vorbis", "opus"}.intersection(codec.split(",")):
return "WebM format requires Vorbis or Opus audio. " \
f"Actual: {codec} [{stream['codec_long_name']}]"
return "" return ""
@staticmethod @staticmethod
@ -289,20 +304,22 @@ class VideoFileAnalyzer:
# if we are vp8/vp9/av1 we want webm # if we are vp8/vp9/av1 we want webm
# use mp4 for anything else # use mp4 for anything else
if not video_encoder: # not re-encoding video if video_encoder: # not re-encoding video
for stream in scan_data["streams"]: if "theora" in video_encoder:
if stream["codec_type"] != "video": return "ogv"
continue if re.search(r"vp[89x]|av1", video_encoder.split(" ", 1)[0]):
codec = stream["codec_name"].split(",") return "webm"
if "theora" in codec: return "mp4"
return "ogv"
if {"vp8", "vp9", "av1"}.intersection(codec): for stream in scan_data["streams"]:
return "webm" if stream["codec_type"] != "video":
continue
codec = stream["codec_name"].split(",")
if "theora" in codec:
return "ogv"
if {"vp8", "vp9", "av1"}.intersection(codec):
return "webm"
if "theora" in video_encoder:
return "ogv"
elif re.search(r"vp[89x]|av1", video_encoder.split(" ", 1)[0]):
return "webm"
return "mp4" return "mp4"
async def _get_scan_data(self, validate, file_path): async def _get_scan_data(self, validate, file_path):