|
import logging
|
|
from typing import Annotated
|
|
from datetime import datetime, timezone
|
|
from fastapi import APIRouter, Request, Form
|
|
from fastapi.responses import RedirectResponse
|
|
from devplacepy.database import get_table, update_target_stars, get_target_owner_uid
|
|
from devplacepy.utils import generate_uid, require_user, create_notification
|
|
from devplacepy.models import VoteForm
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter()
|
|
|
|
NOTIFY_ON_VOTE: set[str] = {"post", "comment", "gist", "project"}
|
|
|
|
|
|
@router.post("/{target_type}/{target_uid}")
|
|
async def vote(request: Request, target_type: str, target_uid: str, data: Annotated[VoteForm, Form()]):
|
|
user = require_user(request)
|
|
value = data.value
|
|
|
|
votes = get_table("votes")
|
|
existing = votes.find_one(user_uid=user["uid"], target_uid=target_uid, target_type=target_type)
|
|
|
|
if existing:
|
|
if int(existing["value"]) == value:
|
|
votes.delete(id=existing["id"])
|
|
else:
|
|
votes.update({"id": existing["id"], "value": value}, ["id"])
|
|
else:
|
|
votes.insert({
|
|
"uid": generate_uid(),
|
|
"user_uid": user["uid"],
|
|
"target_uid": target_uid,
|
|
"target_type": target_type,
|
|
"value": value,
|
|
"created_at": datetime.now(timezone.utc).isoformat(),
|
|
})
|
|
|
|
up_count = votes.count(target_uid=target_uid, value=1)
|
|
down_count = votes.count(target_uid=target_uid, value=-1)
|
|
net = up_count - down_count
|
|
|
|
update_target_stars(target_type, target_uid, net)
|
|
|
|
if value == 1 and target_type in NOTIFY_ON_VOTE:
|
|
owner_uid = get_target_owner_uid(target_type, target_uid)
|
|
if owner_uid and owner_uid != user["uid"]:
|
|
create_notification(owner_uid, "vote", f"{user['username']} ++'d your {target_type}", user["uid"])
|
|
|
|
referer = request.headers.get("Referer", "/feed")
|
|
return RedirectResponse(url=referer, status_code=302)
|