# retoor <retoor@molodetz.nl>
import logging
import secrets
from datetime import datetime, timezone
from fastapi import APIRouter, Request
from devplacepy.database import get_table
from devplacepy.utils import XP_POST
from devplacepy.content import (
create_content_item,
delete_content_item,
apply_vote,
create_comment_record,
set_bookmark,
is_owner,
is_admin,
)
from devplacepy.services.audit import record as audit
from devplacepy.services.correction import schedule_correction
from devplacepy.services.ai_modifier import schedule_modification
from devplacepy.services.devrant.params import merge_params
from devplacepy.services.devrant.tokens import resolve_user
from devplacepy.services.devrant.ids import as_int, post_by_id
from devplacepy.services.devrant.feed import list_rants, search_rants, load_rant_detail
from devplacepy.services.devrant.serializers import encode_tags
from devplacepy.routers.devrant._shared import dr_ok, dr_error, unauthorized
logger = logging.getLogger(__name__)
router = APIRouter()
DEFAULT_LIMIT = 20
MAX_LIMIT = 50
def _parse_tags(raw: object) -> list:
if not raw:
return []
if isinstance(raw, list):
return [str(tag).strip() for tag in raw if str(tag).strip()]
return [tag.strip() for tag in str(raw).split(",") if tag.strip()]
@router.get("/devrant/rants")
async def rant_feed(request: Request):
params = await merge_params(request)
viewer = resolve_user(params)
sort = params.get("sort") or "recent"
limit = min(MAX_LIMIT, max(1, as_int(params.get("limit"), DEFAULT_LIMIT)))
skip = max(0, as_int(params.get("skip"), 0))
rants = list_rants(sort, limit, skip, viewer)
num_notifs = 0
if viewer:
num_notifs = get_table("notifications").count(
user_uid=viewer["uid"], read=False
)
return dr_ok(
rants=rants,
settings={"notif_state": -1, "notif_token": ""},
set=secrets.token_hex(6),
wrw=0,
dpp=0,
num_notifs=num_notifs,
unread={"total": num_notifs},
news=None,
)
@router.get("/devrant/search")
async def search(request: Request):
params = await merge_params(request)
viewer = resolve_user(params)
term = (params.get("term") or "").strip()
return dr_ok(results=search_rants(term, viewer) if term else [])
@router.post("/devrant/rants")
async def create_rant(request: Request):
params = await merge_params(request)
user = resolve_user(params)
if not user:
return unauthorized()
text = (params.get("rant") or "").strip()
if len(text) < 1:
return dr_error("Your rant is too short.")
if len(text) > 125000:
return dr_error("Your rant is too long.")
tags = _parse_tags(params.get("tags"))
project_uid = params.get("project_uid") or None
uid, slug = create_content_item(
"posts",
"post",
user,
{
"title": None,
"content": text,
"topic": "rant",
"project_uid": project_uid,
"image": None,
"tags": encode_tags(tags),
},
text[:50],
XP_POST,
"First Post",
text,
None,
request,
)
post = get_table("posts").find_one(uid=uid)
return dr_ok(rant_id=int(post["id"]))
@router.get("/devrant/rants/{rant_id}")
async def get_rant(request: Request, rant_id: str):
params = await merge_params(request)
viewer = resolve_user(params)
post = post_by_id(rant_id)
if not post:
return dr_error("This rant does not exist.")
detail = load_rant_detail(post, viewer)
return dr_ok(rant=detail["rant"], comments=detail["comments"], subscribed=0)
@router.post("/devrant/rants/{rant_id}")
async def edit_rant(request: Request, rant_id: str):
params = await merge_params(request)
user = resolve_user(params)
if not user:
return unauthorized()
post = post_by_id(rant_id)
if not post:
return dr_error("This rant does not exist.", fail_reason="not_found")
if not is_owner(post, user):
return dr_error("You can only edit your own rants.", fail_reason="not_owner")
text = (params.get("rant") or "").strip()
if len(text) < 1 or len(text) > 125000:
return dr_error("Invalid rant length.", fail_reason="length")
tags = _parse_tags(params.get("tags"))
get_table("posts").update(
{
"uid": post["uid"],
"content": text,
"tags": encode_tags(tags),
"updated_at": datetime.now(timezone.utc).isoformat(),
},
["uid"],
)
schedule_correction(user, "posts", post["uid"], request)
schedule_modification(user, "posts", post["uid"], request)
audit.record(
request,
"post.edit",
user=user,
target_type="post",
target_uid=post["uid"],
origin="devrant",
summary=f"{user['username']} edited rant {post['id']}",
links=[audit.target("post", post["uid"])],
)
return dr_ok()
@router.delete("/devrant/rants/{rant_id}")
async def delete_rant(request: Request, rant_id: str):
params = await merge_params(request)
user = resolve_user(params)
if not user:
return unauthorized()
post = post_by_id(rant_id)
if not post:
return dr_error("This rant does not exist.")
if not (is_owner(post, user) or is_admin(user)):
return dr_error("You can only delete your own rants.")
delete_content_item(request, "posts", "post", user, post["slug"], "/feed")
return dr_ok()
@router.post("/devrant/rants/{rant_id}/vote")
async def vote_rant(request: Request, rant_id: str):
params = await merge_params(request)
user = resolve_user(params)
if not user:
return unauthorized()
post = post_by_id(rant_id)
if not post:
return dr_error("This rant does not exist.")
value = as_int(params.get("vote"))
if value not in (-1, 0, 1):
return dr_error("Invalid vote.")
apply_vote(request, user, "post", post["uid"], value)
fresh = get_table("posts").find_one(uid=post["uid"])
detail = load_rant_detail(fresh, user)
return dr_ok(rant=detail["rant"])
@router.post("/devrant/rants/{rant_id}/favorite")
async def favorite_rant(request: Request, rant_id: str):
return await _set_favorite(request, rant_id, True)
@router.post("/devrant/rants/{rant_id}/unfavorite")
async def unfavorite_rant(request: Request, rant_id: str):
return await _set_favorite(request, rant_id, False)
async def _set_favorite(request: Request, rant_id: str, saved: bool):
params = await merge_params(request)
user = resolve_user(params)
if not user:
return unauthorized()
post = post_by_id(rant_id)
if not post:
return dr_error("This rant does not exist.")
set_bookmark(request, user, "post", post["uid"], saved)
return dr_ok()
@router.post("/devrant/rants/{rant_id}/comments")
async def comment_rant(request: Request, rant_id: str):
params = await merge_params(request)
user = resolve_user(params)
if not user:
return unauthorized()
post = post_by_id(rant_id)
if not post:
return dr_error("This rant does not exist.")
text = (params.get("comment") or "").strip()
if len(text) < 1:
return dr_error("Your comment is too short.")
if len(text) > 1000:
return dr_error("Your comment is too long.")
create_comment_record(request, user, "post", post["uid"], text)
return dr_ok()
Reference in New Issue View Git Blame Copy Permalink