|
import logging
|
|
import aiofiles
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from fastapi import APIRouter, Request, HTTPException
|
|
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.config import STATIC_DIR
|
|
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
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter()
|
|
|
|
|
|
@router.post("/create")
|
|
async def create_post(request: Request):
|
|
user = require_user(request)
|
|
form = await request.form()
|
|
content = form.get("content", "").strip()
|
|
title = form.get("title", "").strip()
|
|
topic = form.get("topic", "random")
|
|
project_uid = form.get("project_uid", "")
|
|
|
|
errors = []
|
|
if not content:
|
|
errors.append("Content is required")
|
|
if content and len(content) < 10:
|
|
errors.append("Content must be at least 10 characters")
|
|
if len(content) > 2000:
|
|
errors.append("Content too long (max 2000 characters)")
|
|
if topic not in TOPICS:
|
|
topic = "random"
|
|
|
|
if errors:
|
|
return RedirectResponse(url="/feed", status_code=302)
|
|
|
|
image_filename = None
|
|
image_file = form.get("image")
|
|
if image_file and hasattr(image_file, "filename") and image_file.filename:
|
|
try:
|
|
content_bytes = await image_file.read()
|
|
if len(content_bytes) > 5 * 1024 * 1024:
|
|
logger.warning(f"Image too large: {image_file.filename}")
|
|
else:
|
|
ext = Path(image_file.filename).suffix.lower()
|
|
allowed = {".png", ".jpg", ".jpeg", ".gif", ".webp", ".svg"}
|
|
if ext not in allowed:
|
|
logger.warning(f"Unsupported image type: {ext}")
|
|
else:
|
|
upload_dir = STATIC_DIR / "uploads"
|
|
upload_dir.mkdir(parents=True, exist_ok=True)
|
|
image_filename = f"{generate_uid()}{ext}"
|
|
file_path = upload_dir / image_filename
|
|
async with aiofiles.open(str(file_path), "wb") as f:
|
|
await f.write(content_bytes)
|
|
logger.info(f"Image saved: {image_filename}")
|
|
content += f"\n\n"
|
|
except Exception as e:
|
|
logger.warning(f"Image upload failed: {e}")
|
|
|
|
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(),
|
|
})
|
|
|
|
attachment_uids = form.getlist("attachment_uids")
|
|
if attachment_uids:
|
|
link_attachments(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):
|
|
user = require_user(request)
|
|
form = await request.form()
|
|
content = form.get("content", "").strip()
|
|
title = form.get("title", "").strip()
|
|
topic = form.get("topic", "random")
|
|
|
|
errors = []
|
|
if not content:
|
|
errors.append("Content is required")
|
|
if content and len(content) < 10:
|
|
errors.append("Content must be at least 10 characters")
|
|
if len(content) > 2000:
|
|
errors.append("Content too long (max 2000 characters)")
|
|
if topic not in TOPICS:
|
|
topic = "random"
|
|
|
|
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)
|
|
|
|
if errors:
|
|
return RedirectResponse(url=f"/posts/{post['slug'] or post['uid']}", status_code=302)
|
|
|
|
posts.update({
|
|
"uid": post["uid"],
|
|
"content": content,
|
|
"title": title or None,
|
|
"topic": 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"])
|
|
image = post.get("image")
|
|
if image:
|
|
try:
|
|
(STATIC_DIR / "uploads" / image).unlink(missing_ok=True)
|
|
except Exception as e:
|
|
logger.warning(f"Failed to delete image {image}: {e}")
|
|
posts.delete(id=post["id"])
|
|
logger.info(f"Post {post['uid']} deleted by {user['username']}")
|
|
return RedirectResponse(url="/feed", status_code=302)
|