mirror of
https://codeberg.org/rimu/pyfedi
synced 2025-01-23 19:36:56 -08:00
94 lines
4.3 KiB
Python
94 lines
4.3 KiB
Python
from app import cache
|
|
from app.activitypub.signature import default_context, post_request_in_background
|
|
from app.community.util import send_to_remote_instance
|
|
from app.models import Post, User
|
|
from app.utils import gibberish, instance_banned, render_template, authorise_api_user, recently_upvoted_posts, recently_downvoted_posts
|
|
|
|
from flask import current_app, request
|
|
from flask_login import current_user
|
|
|
|
|
|
# would be in app/constants.py
|
|
SRC_WEB = 1
|
|
SRC_PUB = 2
|
|
SRC_API = 3
|
|
|
|
# function can be shared between WEB and API (only API calls it for now)
|
|
# post_vote in app/post/routes would just need to do 'return vote_for_post(post_id, vote_direction, SRC_WEB)'
|
|
|
|
def vote_for_post(post_id: int, vote_direction, src, auth=None):
|
|
if src == SRC_API and auth is not None:
|
|
post = Post.query.get(post_id)
|
|
if not post:
|
|
raise Exception('post_not_found')
|
|
try:
|
|
user = authorise_api_user(auth, return_type='model')
|
|
except:
|
|
raise
|
|
else:
|
|
post = Post.query.get_or_404(post_id)
|
|
user = current_user
|
|
|
|
undo = post.vote(user, vote_direction)
|
|
|
|
if not post.community.local_only:
|
|
if undo:
|
|
action_json = {
|
|
'actor': user.public_url(not(post.community.instance.votes_are_public() and user.vote_privately())),
|
|
'type': 'Undo',
|
|
'id': f"https://{current_app.config['SERVER_NAME']}/activities/undo/{gibberish(15)}",
|
|
'audience': post.community.public_url(),
|
|
'object': {
|
|
'actor': user.public_url(not(post.community.instance.votes_are_public() and user.vote_privately())),
|
|
'object': post.public_url(),
|
|
'type': undo,
|
|
'id': f"https://{current_app.config['SERVER_NAME']}/activities/{undo.lower()}/{gibberish(15)}",
|
|
'audience': post.community.public_url()
|
|
}
|
|
}
|
|
else:
|
|
action_type = 'Like' if vote_direction == 'upvote' else 'Dislike'
|
|
action_json = {
|
|
'actor': user.public_url(not(post.community.instance.votes_are_public() and user.vote_privately())),
|
|
'object': post.profile_id(),
|
|
'type': action_type,
|
|
'id': f"https://{current_app.config['SERVER_NAME']}/activities/{action_type.lower()}/{gibberish(15)}",
|
|
'audience': post.community.public_url()
|
|
}
|
|
if post.community.is_local():
|
|
announce = {
|
|
"id": f"https://{current_app.config['SERVER_NAME']}/activities/announce/{gibberish(15)}",
|
|
"type": 'Announce',
|
|
"to": [
|
|
"https://www.w3.org/ns/activitystreams#Public"
|
|
],
|
|
"actor": post.community.public_url(),
|
|
"cc": [
|
|
post.community.ap_followers_url
|
|
],
|
|
'@context': default_context(),
|
|
'object': action_json
|
|
}
|
|
for instance in post.community.following_instances():
|
|
if instance.inbox and not user.has_blocked_instance(instance.id) and not instance_banned(instance.domain):
|
|
send_to_remote_instance(instance.id, post.community.id, announce)
|
|
else:
|
|
post_request_in_background(post.community.ap_inbox_url, action_json, user.private_key,
|
|
user.public_url(not(post.community.instance.votes_are_public() and user.vote_privately())) + '#main-key')
|
|
|
|
|
|
if src == SRC_API:
|
|
return user.id
|
|
else:
|
|
recently_upvoted = []
|
|
recently_downvoted = []
|
|
if vote_direction == 'upvote' and undo is None:
|
|
recently_upvoted = [post_id]
|
|
elif vote_direction == 'downvote' and undo is None:
|
|
recently_downvoted = [post_id]
|
|
cache.delete_memoized(recently_upvoted_posts, user.id)
|
|
cache.delete_memoized(recently_downvoted_posts, user.id)
|
|
|
|
template = 'post/_post_voting_buttons.html' if request.args.get('style', '') == '' else 'post/_post_voting_buttons_masonry.html'
|
|
return render_template(template, post=post, community=post.community, recently_upvoted=recently_upvoted,
|
|
recently_downvoted=recently_downvoted)
|