|
import logging
|
|
from typing import Annotated
|
|
from datetime import datetime, timezone
|
|
from fastapi import APIRouter, Request, HTTPException, Form
|
|
from fastapi.responses import RedirectResponse, HTMLResponse
|
|
from devplacepy.constants import TOPICS
|
|
from devplacepy.database import get_table, get_comment_counts_by_post_uids, load_comments, db, resolve_by_slug
|
|
from devplacepy.templating import templates
|
|
from devplacepy.utils import generate_uid, get_current_user, require_user, time_ago, make_combined_slug, create_mention_notifications
|
|
from devplacepy.seo import base_seo_context, site_url, website_schema, breadcrumb_schema, discussion_forum_posting, combine, truncate
|
|
from devplacepy.attachments import get_attachments, link_attachments, delete_target_attachments, save_inline_image, delete_inline_image
|
|
from devplacepy.models import PostForm, PostEditForm
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter()
|
|
|
|
|
|
@router.post("/create")
|
|
async def create_post(request: Request, data: Annotated[PostForm, Form()]):
|
|
user = require_user(request)
|
|
content = data.content.strip()
|
|
title = data.title.strip()
|
|
topic = data.topic
|
|
project_uid = data.project_uid
|
|
|
|
image_filename = None
|
|
form = await request.form()
|
|
image_file = form.get("image")
|
|
if image_file is not None and hasattr(image_file, "filename") and image_file.filename:
|
|
image_filename = save_inline_image(await image_file.read(), image_file.filename)
|
|
if image_filename:
|
|
content += f"\n\n"
|
|
|
|
posts = get_table("posts")
|
|
uid = generate_uid()
|
|
slug_text = title if title else content[:50]
|
|
post_slug = make_combined_slug(slug_text, uid)
|
|
posts.insert({
|
|
"uid": uid,
|
|
"user_uid": user["uid"],
|
|
"title": title or None,
|
|
"slug": post_slug,
|
|
"content": content,
|
|
"topic": topic,
|
|
"project_uid": project_uid or None,
|
|
"image": image_filename,
|
|
"stars": 0,
|
|
"created_at": datetime.now(timezone.utc).isoformat(),
|
|
})
|
|
|
|
badges = get_table("badges")
|
|
existing = badges.find_one(user_uid=user["uid"], badge_name="First Post")
|
|
if not existing:
|
|
badges.insert({
|
|
"uid": generate_uid(),
|
|
"user_uid": user["uid"],
|
|
"badge_name": "First Post",
|
|
"created_at": datetime.now(timezone.utc).isoformat(),
|
|
})
|
|
|
|
if data.attachment_uids:
|
|
link_attachments(data.attachment_uids, "post", uid)
|
|
|
|
create_mention_notifications(content, user["uid"], f"/posts/{post_slug}")
|
|
logger.info(f"Post {uid} created by {user['username']}")
|
|
return RedirectResponse(url=f"/posts/{post_slug}", status_code=302)
|
|
|
|
|
|
@router.get("/{post_slug}", response_class=HTMLResponse)
|
|
async def view_post(request: Request, post_slug: str):
|
|
user = get_current_user(request)
|
|
posts = get_table("posts")
|
|
post = resolve_by_slug(posts, post_slug)
|
|
if not post:
|
|
raise HTTPException(status_code=404, detail="Post not found")
|
|
|
|
users_table = get_table("users")
|
|
author = users_table.find_one(uid=post["user_uid"])
|
|
|
|
top_level = load_comments("post", post["uid"])
|
|
|
|
def count_all(items):
|
|
total = len(items)
|
|
for item in items:
|
|
total += count_all(item.get("children", []))
|
|
return total
|
|
comment_count = count_all(top_level)
|
|
star_count = post.get("stars", 0)
|
|
base = site_url(request)
|
|
seo_ctx = base_seo_context(
|
|
request,
|
|
title=post.get("title") or "Post",
|
|
description=truncate(post.get("content", ""), 160),
|
|
breadcrumbs=[
|
|
{"name": "Home", "url": "/feed"},
|
|
{"name": "Feed", "url": "/feed"},
|
|
{"name": post.get("title") or "Post", "url": f"/posts/{post['slug'] or post['uid']}"},
|
|
],
|
|
og_type="article",
|
|
schemas=[
|
|
website_schema(base),
|
|
discussion_forum_posting(post, author, comment_count, star_count, base),
|
|
],
|
|
)
|
|
|
|
related_posts = []
|
|
if "posts" in db.tables:
|
|
rows = list(db.query(
|
|
"SELECT p.*, u.username FROM posts p JOIN users u ON p.user_uid = u.uid WHERE p.topic = :t AND p.uid != :uid ORDER BY p.created_at DESC LIMIT 5",
|
|
t=post.get("topic", ""), uid=post["uid"],
|
|
))
|
|
for r in rows:
|
|
related_posts.append({
|
|
"post": r,
|
|
"author": {"username": r["username"]},
|
|
"time_ago": time_ago(r["created_at"]),
|
|
})
|
|
|
|
post_attachments = get_attachments("post", post["uid"])
|
|
|
|
return templates.TemplateResponse(request, "post.html", {
|
|
**seo_ctx,
|
|
"request": request,
|
|
"user": user,
|
|
"post": post,
|
|
"author": author,
|
|
"comments": top_level,
|
|
"time_ago": time_ago(post["created_at"]),
|
|
"comment_count": comment_count,
|
|
"related_posts": related_posts,
|
|
"topics": list(TOPICS),
|
|
"attachments": post_attachments,
|
|
})
|
|
|
|
|
|
@router.post("/edit/{post_slug}")
|
|
async def edit_post(request: Request, post_slug: str, data: Annotated[PostEditForm, Form()]):
|
|
user = require_user(request)
|
|
posts = get_table("posts")
|
|
post = resolve_by_slug(posts, post_slug)
|
|
if not post or post["user_uid"] != user["uid"]:
|
|
return RedirectResponse(url="/feed", status_code=302)
|
|
|
|
posts.update({
|
|
"uid": post["uid"],
|
|
"content": data.content.strip(),
|
|
"title": data.title.strip() or None,
|
|
"topic": data.topic,
|
|
}, ["uid"])
|
|
|
|
logger.info(f"Post {post['uid']} edited by {user['username']}")
|
|
return RedirectResponse(url=f"/posts/{post['slug'] or post['uid']}", status_code=302)
|
|
|
|
|
|
@router.post("/delete/{post_slug}")
|
|
async def delete_post(request: Request, post_slug: str):
|
|
user = require_user(request)
|
|
posts = get_table("posts")
|
|
post = resolve_by_slug(posts, post_slug)
|
|
if post and post["user_uid"] == user["uid"]:
|
|
delete_target_attachments("post", post["uid"])
|
|
get_table("comments").delete(post_uid=post["uid"])
|
|
get_table("votes").delete(target_uid=post["uid"])
|
|
delete_inline_image(post.get("image"))
|
|
posts.delete(id=post["id"])
|
|
logger.info(f"Post {post['uid']} deleted by {user['username']}")
|
|
return RedirectResponse(url="/feed", status_code=302)
|