1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
|
from typing import Dict
from datetime import datetime
from szurubooru import search, rest, model
from szurubooru.func import (
auth, comments, posts, scores, versions, serialization)
_search_executor = search.Executor(search.configs.CommentSearchConfig())
def _get_comment(params: Dict[str, str]) -> model.Comment:
try:
comment_id = int(params['comment_id'])
except TypeError:
raise comments.InvalidCommentIdError(
'Invalid comment ID: %r.' % params['comment_id'])
return comments.get_comment_by_id(comment_id)
def _serialize(
ctx: rest.Context, comment: model.Comment) -> rest.Response:
return comments.serialize_comment(
comment,
ctx.user,
options=serialization.get_serialization_options(ctx))
@rest.routes.get('/comments/?')
def get_comments(
ctx: rest.Context, _params: Dict[str, str]={}) -> rest.Response:
auth.verify_privilege(ctx.user, 'comments:list')
return _search_executor.execute_and_serialize(
ctx, lambda comment: _serialize(ctx, comment))
@rest.routes.post('/comments/?')
def create_comment(
ctx: rest.Context, _params: Dict[str, str]={}) -> rest.Response:
auth.verify_privilege(ctx.user, 'comments:create')
text = ctx.get_param_as_string('text')
post_id = ctx.get_param_as_int('postId')
post = posts.get_post_by_id(post_id)
comment = comments.create_comment(ctx.user, post, text)
ctx.session.add(comment)
ctx.session.commit()
return _serialize(ctx, comment)
@rest.routes.get('/comment/(?P<comment_id>[^/]+)/?')
def get_comment(ctx: rest.Context, params: Dict[str, str]) -> rest.Response:
auth.verify_privilege(ctx.user, 'comments:view')
comment = _get_comment(params)
return _serialize(ctx, comment)
@rest.routes.put('/comment/(?P<comment_id>[^/]+)/?')
def update_comment(ctx: rest.Context, params: Dict[str, str]) -> rest.Response:
comment = _get_comment(params)
versions.verify_version(comment, ctx)
versions.bump_version(comment)
infix = 'own' if ctx.user.user_id == comment.user_id else 'any'
text = ctx.get_param_as_string('text')
auth.verify_privilege(ctx.user, 'comments:edit:%s' % infix)
comments.update_comment_text(comment, text)
comment.last_edit_time = datetime.utcnow()
ctx.session.commit()
return _serialize(ctx, comment)
@rest.routes.delete('/comment/(?P<comment_id>[^/]+)/?')
def delete_comment(ctx: rest.Context, params: Dict[str, str]) -> rest.Response:
comment = _get_comment(params)
versions.verify_version(comment, ctx)
infix = 'own' if ctx.user.user_id == comment.user_id else 'any'
auth.verify_privilege(ctx.user, 'comments:delete:%s' % infix)
ctx.session.delete(comment)
ctx.session.commit()
return {}
@rest.routes.put('/comment/(?P<comment_id>[^/]+)/score/?')
def set_comment_score(
ctx: rest.Context, params: Dict[str, str]) -> rest.Response:
auth.verify_privilege(ctx.user, 'comments:score')
score = ctx.get_param_as_int('score')
comment = _get_comment(params)
scores.set_score(comment, ctx.user, score)
ctx.session.commit()
return _serialize(ctx, comment)
@rest.routes.delete('/comment/(?P<comment_id>[^/]+)/score/?')
def delete_comment_score(
ctx: rest.Context, params: Dict[str, str]) -> rest.Response:
auth.verify_privilege(ctx.user, 'comments:score')
comment = _get_comment(params)
scores.delete_score(comment, ctx.user)
ctx.session.commit()
return _serialize(ctx, comment)
|
