mirror of
https://github.com/rr-/szurubooru.git
synced 2025-07-17 08:26:24 +00:00
4c78cf8c47
This will remove the dependency on the Elasticsearch database. The search query is passed currently as raw SQL. Proper implementation using SQLAlchemy will need custom ORM classed to be made. Additional config parameter "allow_broken_uploads" has been added.
271 lines
10 KiB
Python
271 lines
10 KiB
Python
from typing import Optional, Dict, List
|
|
from datetime import datetime
|
|
from szurubooru import db, model, errors, rest, search
|
|
from szurubooru.func import (
|
|
auth, tags, posts, snapshots, favorites, scores, serialization, versions)
|
|
|
|
|
|
_search_executor_config = search.configs.PostSearchConfig()
|
|
_search_executor = search.Executor(_search_executor_config)
|
|
|
|
|
|
def _get_post_id(params: Dict[str, str]) -> int:
|
|
try:
|
|
return int(params['post_id'])
|
|
except TypeError:
|
|
raise posts.InvalidPostIdError(
|
|
'Invalid post ID: %r.' % params['post_id'])
|
|
|
|
|
|
def _get_post(params: Dict[str, str]) -> model.Post:
|
|
return posts.get_post_by_id(_get_post_id(params))
|
|
|
|
|
|
def _serialize_post(
|
|
ctx: rest.Context, post: Optional[model.Post]) -> rest.Response:
|
|
return posts.serialize_post(
|
|
post,
|
|
ctx.user,
|
|
options=serialization.get_serialization_options(ctx))
|
|
|
|
|
|
@rest.routes.get('/posts/?')
|
|
def get_posts(
|
|
ctx: rest.Context, _params: Dict[str, str] = {}) -> rest.Response:
|
|
auth.verify_privilege(ctx.user, 'posts:list')
|
|
_search_executor_config.user = ctx.user
|
|
return _search_executor.execute_and_serialize(
|
|
ctx, lambda post: _serialize_post(ctx, post))
|
|
|
|
|
|
@rest.routes.post('/posts/?')
|
|
def create_post(
|
|
ctx: rest.Context, _params: Dict[str, str] = {}) -> rest.Response:
|
|
anonymous = ctx.get_param_as_bool('anonymous', default=False)
|
|
if anonymous:
|
|
auth.verify_privilege(ctx.user, 'posts:create:anonymous')
|
|
else:
|
|
auth.verify_privilege(ctx.user, 'posts:create:identified')
|
|
content = ctx.get_file('content')
|
|
tag_names = ctx.get_param_as_string_list('tags', default=[])
|
|
safety = ctx.get_param_as_string('safety')
|
|
source = ctx.get_param_as_string('source', default='')
|
|
if ctx.has_param('contentUrl') and not source:
|
|
source = ctx.get_param_as_string('contentUrl', default='')
|
|
relations = ctx.get_param_as_int_list('relations', default=[])
|
|
notes = ctx.get_param_as_list('notes', default=[])
|
|
flags = ctx.get_param_as_string_list('flags', default=[])
|
|
|
|
post, new_tags = posts.create_post(
|
|
content, tag_names, None if anonymous else ctx.user)
|
|
if len(new_tags):
|
|
auth.verify_privilege(ctx.user, 'tags:create')
|
|
posts.update_post_safety(post, safety)
|
|
posts.update_post_source(post, source)
|
|
posts.update_post_relations(post, relations)
|
|
posts.update_post_notes(post, notes)
|
|
posts.update_post_flags(post, flags)
|
|
posts.test_sound(post, content)
|
|
if ctx.has_file('thumbnail'):
|
|
posts.update_post_thumbnail(post, ctx.get_file('thumbnail'))
|
|
ctx.session.add(post)
|
|
ctx.session.flush()
|
|
create_snapshots_for_post(post, new_tags, None if anonymous else ctx.user)
|
|
alternate_format_posts = posts.generate_alternate_formats(post, content)
|
|
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)
|
|
ctx.session.commit()
|
|
return _serialize_post(ctx, post)
|
|
|
|
|
|
def create_snapshots_for_post(
|
|
post: model.Post,
|
|
new_tags: List[model.Tag],
|
|
user: Optional[model.User]):
|
|
snapshots.create(post, user)
|
|
for tag in new_tags:
|
|
snapshots.create(tag, user)
|
|
|
|
|
|
@rest.routes.get('/post/(?P<post_id>[^/]+)/?')
|
|
def get_post(ctx: rest.Context, params: Dict[str, str]) -> rest.Response:
|
|
auth.verify_privilege(ctx.user, 'posts:view')
|
|
post = _get_post(params)
|
|
return _serialize_post(ctx, post)
|
|
|
|
|
|
@rest.routes.put('/post/(?P<post_id>[^/]+)/?')
|
|
def update_post(ctx: rest.Context, params: Dict[str, str]) -> rest.Response:
|
|
post = _get_post(params)
|
|
versions.verify_version(post, ctx)
|
|
versions.bump_version(post)
|
|
if ctx.has_file('content'):
|
|
auth.verify_privilege(ctx.user, 'posts:edit:content')
|
|
posts.update_post_content(post, ctx.get_file('content'))
|
|
if ctx.has_param('tags'):
|
|
auth.verify_privilege(ctx.user, 'posts:edit:tags')
|
|
new_tags = posts.update_post_tags(
|
|
post, ctx.get_param_as_string_list('tags'))
|
|
if len(new_tags):
|
|
auth.verify_privilege(ctx.user, 'tags:create')
|
|
db.session.flush()
|
|
for tag in new_tags:
|
|
snapshots.create(tag, ctx.user)
|
|
if ctx.has_param('safety'):
|
|
auth.verify_privilege(ctx.user, 'posts:edit:safety')
|
|
posts.update_post_safety(post, ctx.get_param_as_string('safety'))
|
|
if ctx.has_param('source'):
|
|
auth.verify_privilege(ctx.user, 'posts:edit:source')
|
|
posts.update_post_source(post, ctx.get_param_as_string('source'))
|
|
elif ctx.has_param('contentUrl'):
|
|
posts.update_post_source(post, ctx.get_param_as_string('contentUrl'))
|
|
if ctx.has_param('relations'):
|
|
auth.verify_privilege(ctx.user, 'posts:edit:relations')
|
|
posts.update_post_relations(
|
|
post, ctx.get_param_as_int_list('relations'))
|
|
if ctx.has_param('notes'):
|
|
auth.verify_privilege(ctx.user, 'posts:edit:notes')
|
|
posts.update_post_notes(post, ctx.get_param_as_list('notes'))
|
|
if ctx.has_param('flags'):
|
|
auth.verify_privilege(ctx.user, 'posts:edit:flags')
|
|
posts.update_post_flags(post, ctx.get_param_as_string_list('flags'))
|
|
if ctx.has_file('thumbnail'):
|
|
auth.verify_privilege(ctx.user, 'posts:edit:thumbnail')
|
|
posts.update_post_thumbnail(post, ctx.get_file('thumbnail'))
|
|
post.last_edit_time = datetime.utcnow()
|
|
ctx.session.flush()
|
|
snapshots.modify(post, ctx.user)
|
|
ctx.session.commit()
|
|
return _serialize_post(ctx, post)
|
|
|
|
|
|
@rest.routes.delete('/post/(?P<post_id>[^/]+)/?')
|
|
def delete_post(ctx: rest.Context, params: Dict[str, str]) -> rest.Response:
|
|
auth.verify_privilege(ctx.user, 'posts:delete')
|
|
post = _get_post(params)
|
|
versions.verify_version(post, ctx)
|
|
snapshots.delete(post, ctx.user)
|
|
posts.delete(post)
|
|
ctx.session.commit()
|
|
return {}
|
|
|
|
|
|
@rest.routes.post('/post-merge/?')
|
|
def merge_posts(
|
|
ctx: rest.Context, _params: Dict[str, str] = {}) -> rest.Response:
|
|
source_post_id = ctx.get_param_as_int('remove')
|
|
target_post_id = ctx.get_param_as_int('mergeTo')
|
|
source_post = posts.get_post_by_id(source_post_id)
|
|
target_post = posts.get_post_by_id(target_post_id)
|
|
replace_content = ctx.get_param_as_bool('replaceContent')
|
|
versions.verify_version(source_post, ctx, 'removeVersion')
|
|
versions.verify_version(target_post, ctx, 'mergeToVersion')
|
|
versions.bump_version(target_post)
|
|
auth.verify_privilege(ctx.user, 'posts:merge')
|
|
posts.merge_posts(source_post, target_post, replace_content)
|
|
snapshots.merge(source_post, target_post, ctx.user)
|
|
ctx.session.commit()
|
|
return _serialize_post(ctx, target_post)
|
|
|
|
|
|
@rest.routes.get('/featured-post/?')
|
|
def get_featured_post(
|
|
ctx: rest.Context, _params: Dict[str, str] = {}) -> rest.Response:
|
|
auth.verify_privilege(ctx.user, 'posts:view:featured')
|
|
post = posts.try_get_featured_post()
|
|
return _serialize_post(ctx, post)
|
|
|
|
|
|
@rest.routes.post('/featured-post/?')
|
|
def set_featured_post(
|
|
ctx: rest.Context, _params: Dict[str, str] = {}) -> rest.Response:
|
|
auth.verify_privilege(ctx.user, 'posts:feature')
|
|
post_id = ctx.get_param_as_int('id')
|
|
post = posts.get_post_by_id(post_id)
|
|
featured_post = posts.try_get_featured_post()
|
|
if featured_post and featured_post.post_id == post.post_id:
|
|
raise posts.PostAlreadyFeaturedError(
|
|
'Post %r is already featured.' % post_id)
|
|
posts.feature_post(post, ctx.user)
|
|
snapshots.modify(post, ctx.user)
|
|
ctx.session.commit()
|
|
return _serialize_post(ctx, post)
|
|
|
|
|
|
@rest.routes.put('/post/(?P<post_id>[^/]+)/score/?')
|
|
def set_post_score(ctx: rest.Context, params: Dict[str, str]) -> rest.Response:
|
|
auth.verify_privilege(ctx.user, 'posts:score')
|
|
post = _get_post(params)
|
|
score = ctx.get_param_as_int('score')
|
|
scores.set_score(post, ctx.user, score)
|
|
ctx.session.commit()
|
|
return _serialize_post(ctx, post)
|
|
|
|
|
|
@rest.routes.delete('/post/(?P<post_id>[^/]+)/score/?')
|
|
def delete_post_score(
|
|
ctx: rest.Context, params: Dict[str, str]) -> rest.Response:
|
|
auth.verify_privilege(ctx.user, 'posts:score')
|
|
post = _get_post(params)
|
|
scores.delete_score(post, ctx.user)
|
|
ctx.session.commit()
|
|
return _serialize_post(ctx, post)
|
|
|
|
|
|
@rest.routes.post('/post/(?P<post_id>[^/]+)/favorite/?')
|
|
def add_post_to_favorites(
|
|
ctx: rest.Context, params: Dict[str, str]) -> rest.Response:
|
|
auth.verify_privilege(ctx.user, 'posts:favorite')
|
|
post = _get_post(params)
|
|
favorites.set_favorite(post, ctx.user)
|
|
ctx.session.commit()
|
|
return _serialize_post(ctx, post)
|
|
|
|
|
|
@rest.routes.delete('/post/(?P<post_id>[^/]+)/favorite/?')
|
|
def delete_post_from_favorites(
|
|
ctx: rest.Context, params: Dict[str, str]) -> rest.Response:
|
|
auth.verify_privilege(ctx.user, 'posts:favorite')
|
|
post = _get_post(params)
|
|
favorites.unset_favorite(post, ctx.user)
|
|
ctx.session.commit()
|
|
return _serialize_post(ctx, post)
|
|
|
|
|
|
@rest.routes.get('/post/(?P<post_id>[^/]+)/around/?')
|
|
def get_posts_around(
|
|
ctx: rest.Context, params: Dict[str, str]) -> rest.Response:
|
|
auth.verify_privilege(ctx.user, 'posts:list')
|
|
_search_executor_config.user = ctx.user
|
|
post_id = _get_post_id(params)
|
|
return _search_executor.get_around_and_serialize(
|
|
ctx, post_id, lambda post: _serialize_post(ctx, post))
|
|
|
|
|
|
@rest.routes.post('/posts/reverse-search/?')
|
|
def get_posts_by_image(
|
|
ctx: rest.Context, _params: Dict[str, str] = {}) -> rest.Response:
|
|
auth.verify_privilege(ctx.user, 'posts:reverse_search')
|
|
content = ctx.get_file('content')
|
|
|
|
try:
|
|
lookalikes = posts.search_by_image(content)
|
|
except (errors.ThirdPartyError, errors.ProcessingError):
|
|
lookalikes = []
|
|
|
|
return {
|
|
'exactPost':
|
|
_serialize_post(ctx, posts.search_by_image_exact(content)),
|
|
'similarPosts':
|
|
[
|
|
{
|
|
'distance': distance,
|
|
'post': _serialize_post(ctx, post),
|
|
}
|
|
for distance, post in lookalikes
|
|
],
|
|
}
|