Address pull request comments
* Not married to the constants on mime, but atleast standardized them * Moved application/ogg to a constant to be consistent * Removed extraneous newlines * appeased pycodestyle * TODO: ideas for how to let a post have multiple source formats?
This commit is contained in:
parent
ac8d683581
commit
cb1f79ae98
4 changed files with 43 additions and 36 deletions
|
@ -72,12 +72,15 @@ def create_post(
|
||||||
create_snapshots_for_post(post, new_tags, None if anonymous else ctx.user)
|
create_snapshots_for_post(post, new_tags, None if anonymous else ctx.user)
|
||||||
alternate_format_posts = posts.generate_alternate_formats(post, content)
|
alternate_format_posts = posts.generate_alternate_formats(post, content)
|
||||||
for alternate_post, alternate_post_new_tags in alternate_format_posts:
|
for alternate_post, alternate_post_new_tags in alternate_format_posts:
|
||||||
create_snapshots_for_post(alternate_post, alternate_post_new_tags, None if anonymous else ctx.user)
|
create_snapshots_for_post(alternate_post,
|
||||||
|
alternate_post_new_tags,
|
||||||
|
None if anonymous else ctx.user)
|
||||||
ctx.session.commit()
|
ctx.session.commit()
|
||||||
return _serialize_post(ctx, post)
|
return _serialize_post(ctx, post)
|
||||||
|
|
||||||
|
|
||||||
def create_snapshots_for_post(post: model.Post, new_tags: List[model.Tag], user: Optional[model.User]):
|
def create_snapshots_for_post(post: model.Post, new_tags: List[model.Tag],
|
||||||
|
user: Optional[model.User]):
|
||||||
snapshots.create(post, user)
|
snapshots.create(post, user)
|
||||||
for tag in new_tags:
|
for tag in new_tags:
|
||||||
snapshots.create(tag, user)
|
snapshots.create(tag, user)
|
||||||
|
|
|
@ -108,9 +108,7 @@ class Image:
|
||||||
])
|
])
|
||||||
|
|
||||||
def to_mp4(self) -> bytes:
|
def to_mp4(self) -> bytes:
|
||||||
|
|
||||||
with util.create_temp_file_path(suffix='.dat') as mp4_temp_path:
|
with util.create_temp_file_path(suffix='.dat') as mp4_temp_path:
|
||||||
|
|
||||||
width = self.width
|
width = self.width
|
||||||
height = self.height
|
height = self.height
|
||||||
altered_dimensions = False
|
altered_dimensions = False
|
||||||
|
|
|
@ -1,66 +1,67 @@
|
||||||
import re
|
import re
|
||||||
from typing import Optional
|
from typing import Optional
|
||||||
|
|
||||||
APPLICATION_SWF = 'application/x-shockwave-flash'
|
_APPLICATION_OCTET_STREAM = 'application/octet-stream'
|
||||||
IMAGE_JPEG = 'image/jpeg'
|
_APPLICATION_OGG = 'application/ogg'
|
||||||
IMAGE_PNG = 'image/png'
|
_APPLICATION_SWF = 'application/x-shockwave-flash'
|
||||||
IMAGE_GIF = 'image/gif'
|
_IMAGE_GIF = 'image/gif'
|
||||||
VIDEO_WEBM = 'video/webm'
|
_IMAGE_JPEG = 'image/jpeg'
|
||||||
VIDEO_MP4 = 'video/mp4'
|
_IMAGE_PNG = 'image/png'
|
||||||
APPLICATION_OCTET_STREAM = 'application/octet-stream'
|
_VIDEO_MP4 = 'video/mp4'
|
||||||
|
_VIDEO_WEBM = 'video/webm'
|
||||||
|
|
||||||
|
|
||||||
def get_mime_type(content: bytes) -> str:
|
def get_mime_type(content: bytes) -> str:
|
||||||
if not content:
|
if not content:
|
||||||
return APPLICATION_OCTET_STREAM
|
return _APPLICATION_OCTET_STREAM
|
||||||
|
|
||||||
if content[0:3] in (b'CWS', b'FWS', b'ZWS'):
|
if content[0:3] in (b'CWS', b'FWS', b'ZWS'):
|
||||||
return APPLICATION_SWF
|
return _APPLICATION_SWF
|
||||||
|
|
||||||
if content[0:3] == b'\xFF\xD8\xFF':
|
if content[0:3] == b'\xFF\xD8\xFF':
|
||||||
return IMAGE_JPEG
|
return _IMAGE_JPEG
|
||||||
|
|
||||||
if content[0:6] == b'\x89PNG\x0D\x0A':
|
if content[0:6] == b'\x89PNG\x0D\x0A':
|
||||||
return IMAGE_PNG
|
return _IMAGE_PNG
|
||||||
|
|
||||||
if content[0:6] in (b'GIF87a', b'GIF89a'):
|
if content[0:6] in (b'GIF87a', b'GIF89a'):
|
||||||
return IMAGE_GIF
|
return _IMAGE_GIF
|
||||||
|
|
||||||
if content[0:4] == b'\x1A\x45\xDF\xA3':
|
if content[0:4] == b'\x1A\x45\xDF\xA3':
|
||||||
return VIDEO_WEBM
|
return _VIDEO_WEBM
|
||||||
|
|
||||||
if content[4:12] in (b'ftypisom', b'ftypmp42'):
|
if content[4:12] in (b'ftypisom', b'ftypmp42'):
|
||||||
return VIDEO_MP4
|
return _VIDEO_MP4
|
||||||
|
|
||||||
return APPLICATION_OCTET_STREAM
|
return _APPLICATION_OCTET_STREAM
|
||||||
|
|
||||||
|
|
||||||
def get_extension(mime_type: str) -> Optional[str]:
|
def get_extension(mime_type: str) -> Optional[str]:
|
||||||
extension_map = {
|
extension_map = {
|
||||||
APPLICATION_SWF: 'swf',
|
_APPLICATION_SWF: 'swf',
|
||||||
IMAGE_GIF: 'gif',
|
_IMAGE_GIF: 'gif',
|
||||||
IMAGE_JPEG: 'jpg',
|
_IMAGE_JPEG: 'jpg',
|
||||||
IMAGE_PNG: 'png',
|
_IMAGE_PNG: 'png',
|
||||||
VIDEO_MP4: 'mp4',
|
_VIDEO_MP4: 'mp4',
|
||||||
VIDEO_WEBM: 'webm',
|
_VIDEO_WEBM: 'webm',
|
||||||
APPLICATION_OCTET_STREAM: 'dat',
|
_APPLICATION_OCTET_STREAM: 'dat',
|
||||||
}
|
}
|
||||||
return extension_map.get((mime_type or '').strip().lower(), None)
|
return extension_map.get((mime_type or '').strip().lower(), None)
|
||||||
|
|
||||||
|
|
||||||
def is_flash(mime_type: str) -> bool:
|
def is_flash(mime_type: str) -> bool:
|
||||||
return mime_type.lower() == APPLICATION_SWF
|
return mime_type.lower() == _APPLICATION_SWF
|
||||||
|
|
||||||
|
|
||||||
def is_video(mime_type: str) -> bool:
|
def is_video(mime_type: str) -> bool:
|
||||||
return mime_type.lower() in ('application/ogg', VIDEO_MP4, VIDEO_WEBM)
|
return mime_type.lower() in (_APPLICATION_OGG, _VIDEO_MP4, _VIDEO_WEBM)
|
||||||
|
|
||||||
|
|
||||||
def is_image(mime_type: str) -> bool:
|
def is_image(mime_type: str) -> bool:
|
||||||
return mime_type.lower() in (IMAGE_JPEG, IMAGE_PNG, IMAGE_GIF)
|
return mime_type.lower() in (_IMAGE_JPEG, _IMAGE_PNG, _IMAGE_GIF)
|
||||||
|
|
||||||
|
|
||||||
def is_animated_gif(content: bytes) -> bool:
|
def is_animated_gif(content: bytes) -> bool:
|
||||||
pattern = b'\x21\xF9\x04[\x00-\xFF]{4}\x00[\x2C\x21]'
|
pattern = b'\x21\xF9\x04[\x00-\xFF]{4}\x00[\x2C\x21]'
|
||||||
return get_mime_type(content) == IMAGE_GIF \
|
return (get_mime_type(content) == _IMAGE_GIF
|
||||||
and len(re.findall(pattern, content)) > 1
|
and len(re.findall(pattern, content)) > 1)
|
||||||
|
|
|
@ -429,22 +429,26 @@ def _sync_post_content(post: model.Post) -> None:
|
||||||
generate_post_thumbnail(post)
|
generate_post_thumbnail(post)
|
||||||
|
|
||||||
|
|
||||||
def generate_alternate_formats(post: model.Post, content: bytes) -> List[Tuple[model.Post, List[model.Tag]]]:
|
def generate_alternate_formats(post: model.Post, content: bytes) \
|
||||||
|
-> List[Tuple[model.Post, List[model.Tag]]]:
|
||||||
assert post
|
assert post
|
||||||
assert content
|
assert content
|
||||||
new_posts = []
|
new_posts = []
|
||||||
if mime.is_animated_gif(content):
|
if mime.is_animated_gif(content):
|
||||||
tag_names = [tag_name.name for tag_name in [tag.names for tag in post.tags]]
|
tag_names = [tag_name.name for tag_name in
|
||||||
|
[tag.names for tag in post.tags]]
|
||||||
|
|
||||||
if config.config['convert']['gif']['to_mp4']:
|
if config.config['convert']['gif']['to_mp4']:
|
||||||
mp4_post, new_tags = create_post(images.Image(content).to_mp4(), tag_names, post.user)
|
mp4_post, new_tags = create_post(images.Image(content).to_mp4(),
|
||||||
|
tag_names, post.user)
|
||||||
update_post_flags(mp4_post, ['loop'])
|
update_post_flags(mp4_post, ['loop'])
|
||||||
update_post_safety(mp4_post, post.safety)
|
update_post_safety(mp4_post, post.safety)
|
||||||
update_post_source(mp4_post, post.source)
|
update_post_source(mp4_post, post.source)
|
||||||
new_posts += [(mp4_post, new_tags)]
|
new_posts += [(mp4_post, new_tags)]
|
||||||
|
|
||||||
if config.config['convert']['gif']['to_webm']:
|
if config.config['convert']['gif']['to_webm']:
|
||||||
webm_post, new_tags = create_post(images.Image(content).to_webm(), tag_names, post.user)
|
webm_post, new_tags = create_post(images.Image(content).to_webm(),
|
||||||
|
tag_names, post.user)
|
||||||
update_post_flags(webm_post, ['loop'])
|
update_post_flags(webm_post, ['loop'])
|
||||||
update_post_safety(webm_post, post.safety)
|
update_post_safety(webm_post, post.safety)
|
||||||
update_post_source(webm_post, post.source)
|
update_post_source(webm_post, post.source)
|
||||||
|
@ -455,7 +459,8 @@ def generate_alternate_formats(post: model.Post, content: bytes) -> List[Tuple[m
|
||||||
new_posts = list(filter(lambda i: i[0] is not None, new_posts))
|
new_posts = list(filter(lambda i: i[0] is not None, new_posts))
|
||||||
|
|
||||||
new_relations = [p[0].post_id for p in new_posts]
|
new_relations = [p[0].post_id for p in new_posts]
|
||||||
update_post_relations(post, new_relations) if len(new_relations) > 0 else None
|
if len(new_relations) > 0:
|
||||||
|
update_post_relations(post, new_relations)
|
||||||
|
|
||||||
return new_posts
|
return new_posts
|
||||||
|
|
||||||
|
|
Reference in a new issue