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
100
101
102
103
104
105
106
107
108
109
|
from datetime import datetime
from typing import Dict
from szurubooru import model, rest, search
from szurubooru.func import (
auth,
comments,
posts,
scores,
serialization,
versions,
)
_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)
|
