-
Notifications
You must be signed in to change notification settings - Fork 3
/
user_comments.py
182 lines (158 loc) · 4.87 KB
/
user_comments.py
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
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
from __future__ import annotations
from fastapi import Depends
from fastapi import Form
from rgdps import logger
from rgdps.api import commands
from rgdps.api import responses
from rgdps.api.context import HTTPContext
from rgdps.api.dependencies import authenticate_dependency
from rgdps.common import gd_obj
from rgdps.common.validators import Base64String
from rgdps.constants.errors import ServiceError
from rgdps.constants.likes import LikeType
from rgdps.constants.users import UserPrivileges
from rgdps.models.user import User
from rgdps.usecases import likes
from rgdps.usecases import user_comments
PAGE_SIZE = 10
async def user_comments_get(
target_id: int = Form(..., alias="accountID"),
page: int = Form(..., alias="page"),
ctx: HTTPContext = Depends(),
):
result = await user_comments.get_user(
ctx,
target_id,
page,
PAGE_SIZE,
)
if isinstance(result, ServiceError):
logger.info(
"Failed to view user comments.",
extra={
"error": result.value,
"target_id": target_id,
"page": page,
},
)
return responses.fail()
response = "|".join(
gd_obj.dumps(gd_obj.create_user_comment(comment, result.user), sep="~")
for comment in (result.comment)
)
response += "#" + gd_obj.create_pagination_info(result.total, page, PAGE_SIZE)
logger.info(
"Successfully viewed user comments.",
extra={
"target_id": target_id,
"page": page,
},
)
return response
async def user_comments_post(
ctx: HTTPContext = Depends(),
user: User = Depends(
authenticate_dependency(
required_privileges=UserPrivileges.USER_CREATE_USER_COMMENTS,
),
),
content: Base64String = Form(..., alias="comment", max_length=256),
):
# Allow users to run commands on themselves.
if commands.is_command(content):
result_str = await commands.router.entrypoint(
command=commands.strip_prefix(content),
user=user,
base_ctx=ctx,
level_id=None,
target_user_id=user.id,
)
# Comment bans allow you to show text to the user instead of posting the comment.
return gd_obj.comment_ban_string(
0,
result_str,
)
result = await user_comments.create(ctx, user.id, content)
if isinstance(result, ServiceError):
logger.info(
"Failed to post user comment.",
extra={
"error": result.value,
"user_id": user.id,
},
)
return responses.fail()
logger.info(
"Successfully posted a user comment.",
extra={
"user_id": user.id,
"comment_id": result.id,
},
)
return responses.success()
# TODO: MOVE
async def like_target_post(
ctx: HTTPContext = Depends(),
user: User = Depends(
authenticate_dependency(required_privileges=UserPrivileges.COMMENTS_LIKE),
),
target_type: LikeType = Form(..., alias="type"),
target_id: int = Form(..., alias="itemID"),
is_positive: bool = Form(..., alias="like"),
):
# TODO: Move this logic to a single usecase.
result = None
if target_type is LikeType.USER_COMMENT:
result = await likes.like_user_comment(
ctx,
user.id,
target_id,
int(is_positive),
)
elif target_type is LikeType.LEVEL:
result = await likes.like_level(ctx, user.id, target_id, int(is_positive))
else:
raise NotImplementedError
if isinstance(result, ServiceError):
logger.info(
f"Failed to like/dislike target.",
extra={
"user_id": user.id,
"target_type": target_type.value,
"target_id": target_id,
"is_positive": is_positive,
"error": result.value,
},
)
return responses.fail()
logger.info(
"Successfully liked/disliked target.",
extra={
"like_id": result.id,
},
)
return responses.success()
async def user_comment_delete(
ctx: HTTPContext = Depends(),
user: User = Depends(authenticate_dependency()),
comment_id: int = Form(..., alias="commentID"),
):
result = await user_comments.delete(ctx, user.id, comment_id)
if isinstance(result, ServiceError):
logger.info(
"Failed to delete user comment.",
extra={
"user_id": user.id,
"comment_id": comment_id,
"error": result.value,
},
)
return responses.fail()
logger.info(
"Successfully deleted comment.",
extra={
"user_id": user.id,
"comment_id": comment_id,
},
)
return responses.success()