blindxfishandClaude Fable 5
72e088c160
DevPlace CI / test (pull_request) Has been cancelled
The project detail page becomes a full project showcase built entirely
from existing platform mechanisms. One encompassing dark card wraps the
page; inner panels (tab bar, sidebar cards, devlog entries, comments)
sit one elevation lighter. The hero opens with a cover banner and an
optional logo tile, both plain attachment references
(cover_attachment_uid/logo_attachment_uid) uploaded through the
standard dp-upload attachment widget and linked via the existing
link_attachments choke point - the route validates each uid belongs to
the actor and is an image, and an empty value on edit keeps the current
one. The title block, type/platform chips and author row overlay the
banner behind a scrim with a dark text shadow, next to an owner-set
Visit Website CTA; website_url and repo_url are normalized in models
and render with rel noopener nofollow.
An anchor tab bar (Overview, Devlog, Screenshots when present,
Comments, Files) navigates the page. The main column keeps About, the
devlog timeline (with devlog_count and an owner Post update button
opening the shared composer preset to the devlog topic + project - the
form now lives once in _post_composer_form.html, included by feed.html
and project_detail.html), a Screenshots gallery built from image
attachments minus the cover/logo (thumbnails, lightbox, 12 rendered),
and the comment thread; the sidebar holds Links, Stats and the Author
card. Owners add gallery images from the More menu via
POST /projects/{slug}/screenshots (owner-only, audit
project.screenshots.add, Devii action project_add_screenshots, docs id
projects-screenshots). comment_count/devlog_count ride
ProjectDetailOut, the new fields ride ProjectOut, and the create/edit
faces (modals, Devii actions, API docs) carry them. The project
comment/files e2e tests scope their locators per the documented
dual-control idiom, and new unit/api/e2e tests cover URL normalization,
the counts, the hero attachment guard, the screenshots flow and the
preset composer.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
880 lines
28 KiB
Python
880 lines
28 KiB
Python
# retoor <retoor@molodetz.nl>
|
|
import logging
|
|
from typing import Any
|
|
from datetime import datetime, timezone
|
|
from fastapi.responses import RedirectResponse
|
|
from devplacepy.attachments import (
|
|
soft_delete_attachments_for,
|
|
get_attachments,
|
|
link_attachments,
|
|
)
|
|
from devplacepy.database import (
|
|
get_table,
|
|
resolve_by_slug,
|
|
get_users_by_uids,
|
|
get_vote_counts,
|
|
get_comment_counts_by_post_uids,
|
|
paginate,
|
|
STAR_TARGETS,
|
|
get_user_votes,
|
|
get_reactions_by_targets,
|
|
get_user_bookmarks,
|
|
get_blocked_uids,
|
|
get_poll_for_post,
|
|
update_target_stars,
|
|
clear_user_stars,
|
|
clear_user_post_count,
|
|
get_target_owner_uid,
|
|
resolve_object_url,
|
|
soft_delete,
|
|
soft_delete_in,
|
|
soft_delete_engagement,
|
|
soft_delete_fork_relations,
|
|
load_comments,
|
|
band_allows_mature,
|
|
band_allows_restricted,
|
|
get_maturity,
|
|
get_maturity_by_targets,
|
|
get_int_setting,
|
|
_now_iso,
|
|
db,
|
|
)
|
|
from devplacepy.utils import (
|
|
time_ago,
|
|
generate_uid,
|
|
make_combined_slug,
|
|
award_rewards,
|
|
track_action,
|
|
create_notification,
|
|
create_mention_notifications,
|
|
is_admin,
|
|
is_primary_admin,
|
|
XP_COMMENT,
|
|
XP_UPVOTE,
|
|
)
|
|
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.seo_meta import schedule_seo_meta_for_table
|
|
from devplacepy.services.moderation.screening import (
|
|
record as record_screening,
|
|
refuse_if_blocked,
|
|
screen_fields,
|
|
)
|
|
|
|
CREATE_METADATA_KEYS = ("project_type", "is_private", "language", "topic", "status")
|
|
|
|
BOOKMARKABLE_TYPES = {"post", "gist", "project", "news", "quiz"}
|
|
REACTABLE_TYPES = {"post", "comment", "gist", "project", "quiz"}
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def get_project_by_uid(project_uid: str | None) -> dict | None:
|
|
if not project_uid:
|
|
return None
|
|
project = get_table("projects").find_one(uid=project_uid)
|
|
if not project:
|
|
return None
|
|
slug = project.get("slug") or project["uid"]
|
|
return {
|
|
"uid": project["uid"],
|
|
"name": project.get("title") or project.get("name", ""),
|
|
"slug": slug,
|
|
"url": f"/projects/{slug}",
|
|
}
|
|
|
|
|
|
def is_owner(item: dict | None, user: dict | None) -> bool:
|
|
return bool(item and user and item["user_uid"] == user["uid"])
|
|
|
|
|
|
def mature_hidden_by_default() -> bool:
|
|
return get_int_setting("moderation_mature_default_hidden", 1) != 0
|
|
|
|
|
|
def maturity_hidden(level: str | None, user: dict | None) -> bool:
|
|
if not level or level == "general":
|
|
return False
|
|
if not mature_hidden_by_default():
|
|
return False
|
|
if not user:
|
|
return True
|
|
band = user.get("age_band") or "adult"
|
|
allowed = (
|
|
band_allows_restricted(band) if level == "restricted" else band_allows_mature(band)
|
|
)
|
|
return not (allowed and bool(user.get("mature_opt_in")))
|
|
|
|
|
|
def is_suspended(user: dict | None) -> bool:
|
|
from devplacepy.database import suspension_active
|
|
|
|
return suspension_active(user)
|
|
|
|
|
|
def _owner_is_admin(project: dict) -> bool:
|
|
owner_uid = project.get("user_uid")
|
|
owner = get_users_by_uids([owner_uid]).get(owner_uid) if owner_uid else None
|
|
return is_admin(owner)
|
|
|
|
|
|
def can_view_project(project: dict | None, user: dict | None) -> bool:
|
|
if not project:
|
|
return False
|
|
if not project.get("is_private"):
|
|
return True
|
|
if is_owner(project, user):
|
|
return True
|
|
if not is_admin(user):
|
|
return False
|
|
return not _owner_is_admin(project)
|
|
|
|
|
|
def owns_instance(
|
|
instance: dict | None, project: dict | None, user: dict | None
|
|
) -> bool:
|
|
if not instance or not user:
|
|
return False
|
|
uid = user.get("uid")
|
|
if not uid:
|
|
return False
|
|
if instance.get("created_by") == uid:
|
|
return True
|
|
return bool(project and project.get("user_uid") == uid)
|
|
|
|
|
|
def can_view_project_containers(project: dict | None, user: dict | None) -> bool:
|
|
if not project or not is_admin(user):
|
|
return False
|
|
if is_primary_admin(user) or is_owner(project, user):
|
|
return True
|
|
return not project.get("is_private")
|
|
|
|
|
|
def can_view_instance(
|
|
instance: dict | None, project: dict | None, user: dict | None
|
|
) -> bool:
|
|
if not instance or not is_admin(user):
|
|
return False
|
|
if is_primary_admin(user) or owns_instance(instance, project, user):
|
|
return True
|
|
return bool(project) and not project.get("is_private")
|
|
|
|
|
|
def can_manage_instance(
|
|
instance: dict | None, project: dict | None, user: dict | None
|
|
) -> bool:
|
|
if not instance or not is_admin(user):
|
|
return False
|
|
return is_primary_admin(user) or owns_instance(instance, project, user)
|
|
|
|
|
|
def workspaces_enabled() -> bool:
|
|
from devplacepy.database import get_setting
|
|
|
|
return get_setting("workspace_enabled", "0") == "1"
|
|
|
|
|
|
def can_open_workspace(project: dict | None, user: dict | None) -> bool:
|
|
if not project or not user or not user.get("uid"):
|
|
return False
|
|
if not workspaces_enabled():
|
|
return False
|
|
return is_owner(project, user) or is_admin(user)
|
|
|
|
|
|
def owns_workspace(instance: dict | None, user: dict | None) -> bool:
|
|
if not instance or not user:
|
|
return False
|
|
uid = user.get("uid")
|
|
if not uid:
|
|
return False
|
|
return instance.get("workspace_owner_uid") == uid
|
|
|
|
|
|
def can_manage_workspace(
|
|
instance: dict | None, project: dict | None, user: dict | None
|
|
) -> bool:
|
|
if not instance or not user:
|
|
return False
|
|
if owns_workspace(instance, user):
|
|
return True
|
|
return can_manage_instance(instance, project, user)
|
|
|
|
|
|
def can_manage_tunnel(
|
|
instance: dict | None, project: dict | None, user: dict | None
|
|
) -> bool:
|
|
return can_manage_workspace(instance, project, user)
|
|
|
|
|
|
def canonical_redirect(
|
|
area: str, item: dict, requested: str
|
|
) -> RedirectResponse | None:
|
|
canonical = item.get("slug") or item["uid"]
|
|
if requested == canonical:
|
|
return None
|
|
return RedirectResponse(url=f"/{area}/{canonical}", status_code=301)
|
|
|
|
|
|
def first_image_url(item: dict, attachments: list | None) -> str | None:
|
|
inline = item.get("image")
|
|
if inline:
|
|
return f"/static/uploads/{inline}"
|
|
for attachment in attachments or []:
|
|
if attachment.get("is_image"):
|
|
return attachment["url"]
|
|
return None
|
|
|
|
|
|
def create_content_item(
|
|
table_name: str,
|
|
target_type: str,
|
|
user: dict,
|
|
fields: dict,
|
|
slug_source: str,
|
|
xp: int,
|
|
badge: str,
|
|
mention_text: str,
|
|
attachment_uids: list | None,
|
|
request=None,
|
|
) -> tuple[str, str]:
|
|
screening = screen_fields(table_name, fields)
|
|
refuse_if_blocked(screening)
|
|
uid = generate_uid()
|
|
slug = make_combined_slug(slug_source, uid)
|
|
get_table(table_name).insert(
|
|
{
|
|
"uid": uid,
|
|
"user_uid": user["uid"],
|
|
"slug": slug,
|
|
"stars": 0,
|
|
"created_at": datetime.now(timezone.utc).isoformat(),
|
|
"deleted_at": None,
|
|
"deleted_by": None,
|
|
**fields,
|
|
}
|
|
)
|
|
if table_name == "posts":
|
|
clear_user_post_count(user["uid"])
|
|
if table_name == "projects":
|
|
from devplacepy.templating import clear_user_projects_cache
|
|
|
|
clear_user_projects_cache(user["uid"])
|
|
award_rewards(user["uid"], xp, badge)
|
|
if attachment_uids:
|
|
link_attachments(attachment_uids, target_type, uid)
|
|
create_mention_notifications(mention_text, user["uid"], f"/{table_name}/{slug}")
|
|
logger.info(f"{target_type} {uid} created by {user['username']}")
|
|
label = fields.get("title") or slug
|
|
links = [audit.target(target_type, uid, label)]
|
|
if fields.get("project_uid"):
|
|
links.append(audit.project(fields["project_uid"]))
|
|
metadata = {
|
|
key: fields[key] for key in CREATE_METADATA_KEYS if fields.get(key) is not None
|
|
}
|
|
if attachment_uids:
|
|
metadata["attachment_count"] = len(attachment_uids)
|
|
audit.record(
|
|
request,
|
|
f"{target_type}.create",
|
|
user=user,
|
|
target_type=target_type,
|
|
target_uid=uid,
|
|
target_label=label,
|
|
summary=f"{user['username']} created {target_type} {label}",
|
|
metadata=metadata or None,
|
|
links=links,
|
|
)
|
|
record_screening(
|
|
screening,
|
|
target_type=target_type,
|
|
target_uid=uid,
|
|
actor_uid=user["uid"],
|
|
request=request,
|
|
)
|
|
schedule_correction(user, table_name, uid, request)
|
|
schedule_modification(user, table_name, uid, request)
|
|
schedule_seo_meta_for_table(table_name, uid)
|
|
return uid, slug
|
|
|
|
|
|
VOTE_NOTIFY_TYPES = {"post", "comment", "gist", "project", "quiz"}
|
|
|
|
|
|
def apply_vote(request, user: dict, target_type: str, target_uid: str, value: int) -> dict:
|
|
votes = get_table("votes")
|
|
existing = votes.find_one(
|
|
user_uid=user["uid"], target_uid=target_uid, target_type=target_type
|
|
)
|
|
old_value = int(existing["value"]) if existing else 0
|
|
did_upvote = False
|
|
new_value = value
|
|
if existing:
|
|
if existing.get("deleted_at"):
|
|
votes.update(
|
|
{
|
|
"id": existing["id"],
|
|
"value": value,
|
|
"deleted_at": None,
|
|
"deleted_by": None,
|
|
},
|
|
["id"],
|
|
)
|
|
did_upvote = value == 1
|
|
elif int(existing["value"]) == value:
|
|
votes.update(
|
|
{"id": existing["id"], "deleted_at": _now_iso(), "deleted_by": user["uid"]},
|
|
["id"],
|
|
)
|
|
new_value = 0
|
|
else:
|
|
votes.update({"id": existing["id"], "value": value}, ["id"])
|
|
did_upvote = value == 1
|
|
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(),
|
|
"deleted_at": None,
|
|
"deleted_by": None,
|
|
}
|
|
)
|
|
did_upvote = value == 1
|
|
|
|
up_count = votes.count(
|
|
target_uid=target_uid, target_type=target_type, value=1, deleted_at=None
|
|
)
|
|
down_count = votes.count(
|
|
target_uid=target_uid, target_type=target_type, value=-1, deleted_at=None
|
|
)
|
|
net = up_count - down_count
|
|
update_target_stars(target_type, target_uid, net)
|
|
|
|
owner_uid = get_target_owner_uid(target_type, target_uid)
|
|
if owner_uid:
|
|
clear_user_stars(owner_uid)
|
|
direction = "clear" if new_value == 0 else ("up" if new_value == 1 else "down")
|
|
vote_links = [audit.target(target_type, target_uid)]
|
|
if owner_uid and owner_uid != user["uid"]:
|
|
vote_links.append(audit.author(owner_uid))
|
|
audit.record(
|
|
request,
|
|
f"vote.{target_type}.{direction}",
|
|
user=user,
|
|
target_type=target_type,
|
|
target_uid=target_uid,
|
|
old_value=old_value,
|
|
new_value=new_value,
|
|
metadata={"value_old": old_value, "value_new": new_value, "net": net},
|
|
summary=f"{user['username']} {direction} vote on {target_type} {target_uid}",
|
|
links=vote_links,
|
|
)
|
|
|
|
if did_upvote and target_type in VOTE_NOTIFY_TYPES:
|
|
if owner_uid and owner_uid != user["uid"]:
|
|
target_url = resolve_object_url(target_type, target_uid)
|
|
create_notification(
|
|
owner_uid,
|
|
"vote",
|
|
f"{user['username']} ++'d your {target_type}",
|
|
user["uid"],
|
|
target_url,
|
|
)
|
|
award_rewards(owner_uid, XP_UPVOTE)
|
|
|
|
if did_upvote:
|
|
track_action(user["uid"], "vote")
|
|
|
|
current = votes.find_one(
|
|
user_uid=user["uid"],
|
|
target_uid=target_uid,
|
|
target_type=target_type,
|
|
deleted_at=None,
|
|
)
|
|
current_value = int(current["value"]) if current else 0
|
|
return {"net": net, "up": up_count, "down": down_count, "value": current_value}
|
|
|
|
|
|
def create_comment_record(
|
|
request,
|
|
user: dict,
|
|
target_type: str,
|
|
target_uid: str,
|
|
content: str,
|
|
parent_uid: str | None = None,
|
|
attachment_uids: list | None = None,
|
|
) -> tuple[str, str]:
|
|
screening = screen_fields("comments", {"content": content})
|
|
refuse_if_blocked(screening)
|
|
comment_uid = generate_uid()
|
|
redirect_url = resolve_object_url(target_type, target_uid)
|
|
insert = {
|
|
"uid": comment_uid,
|
|
"target_uid": target_uid,
|
|
"target_type": target_type,
|
|
"user_uid": user["uid"],
|
|
"content": content,
|
|
"parent_uid": parent_uid or None,
|
|
"created_at": datetime.now(timezone.utc).isoformat(),
|
|
"deleted_at": None,
|
|
"deleted_by": None,
|
|
}
|
|
if target_type == "post":
|
|
insert["post_uid"] = target_uid
|
|
get_table("comments").insert(insert)
|
|
|
|
if attachment_uids:
|
|
link_attachments(attachment_uids, "comment", comment_uid)
|
|
|
|
award_rewards(user["uid"], XP_COMMENT, "First Comment")
|
|
|
|
comment_url = f"{redirect_url}#comment-{comment_uid}"
|
|
|
|
if target_type == "post":
|
|
if parent_uid:
|
|
parent = get_table("comments").find_one(uid=parent_uid, deleted_at=None)
|
|
if parent and parent["user_uid"] != user["uid"]:
|
|
create_notification(
|
|
parent["user_uid"],
|
|
"reply",
|
|
f"{user['username']} replied to your comment",
|
|
user["uid"],
|
|
comment_url,
|
|
)
|
|
else:
|
|
posts = get_table("posts")
|
|
post = posts.find_one(uid=target_uid)
|
|
if not post:
|
|
post = posts.find_one(slug=target_uid)
|
|
if post and post["user_uid"] != user["uid"]:
|
|
create_notification(
|
|
post["user_uid"],
|
|
"comment",
|
|
f"{user['username']} commented on your post",
|
|
user["uid"],
|
|
comment_url,
|
|
)
|
|
|
|
create_mention_notifications(content, user["uid"], comment_url)
|
|
record_screening(
|
|
screening,
|
|
target_type="comment",
|
|
target_uid=comment_uid,
|
|
actor_uid=user["uid"],
|
|
request=request,
|
|
)
|
|
schedule_correction(user, "comments", comment_uid, request)
|
|
schedule_modification(user, "comments", comment_uid, request)
|
|
logger.info(f"Comment by {user['username']} on {target_type} {target_uid}")
|
|
comment_links = [
|
|
audit.target("comment", comment_uid),
|
|
audit.parent(target_type, target_uid),
|
|
]
|
|
if parent_uid:
|
|
comment_links.append(audit.link("parent_comment", "comment", parent_uid))
|
|
audit.record(
|
|
request,
|
|
f"comment.create.{target_type}",
|
|
user=user,
|
|
target_type="comment",
|
|
target_uid=comment_uid,
|
|
summary=f"{user['username']} commented on {target_type} {target_uid}: {content}",
|
|
links=comment_links,
|
|
)
|
|
return comment_uid, comment_url
|
|
|
|
|
|
def edit_comment_record(request, user: dict, comment: dict, content: str) -> str:
|
|
target_type = comment.get("target_type", "post")
|
|
target_uid = comment.get("target_uid") or comment.get("post_uid", "")
|
|
screening = screen_fields("comments", {"content": content})
|
|
refuse_if_blocked(screening)
|
|
updated_at = datetime.now(timezone.utc).isoformat()
|
|
get_table("comments").update(
|
|
{"uid": comment["uid"], "content": content, "updated_at": updated_at}, ["uid"]
|
|
)
|
|
record_screening(
|
|
screening,
|
|
target_type="comment",
|
|
target_uid=comment["uid"],
|
|
actor_uid=user["uid"],
|
|
request=request,
|
|
)
|
|
schedule_correction(user, "comments", comment["uid"], request)
|
|
schedule_modification(user, "comments", comment["uid"], request)
|
|
logger.info(f"Comment {comment['uid']} edited by {user['username']}")
|
|
audit.record(
|
|
request,
|
|
"comment.edit",
|
|
user=user,
|
|
target_type="comment",
|
|
target_uid=comment["uid"],
|
|
summary=f"{user['username']} edited a comment under {target_type} {target_uid}",
|
|
links=[
|
|
audit.target("comment", comment["uid"]),
|
|
audit.parent(target_type, target_uid),
|
|
],
|
|
)
|
|
return updated_at
|
|
|
|
|
|
def delete_comment_record(request, user: dict, comment: dict) -> tuple[str, str]:
|
|
target_type = comment.get("target_type", "post")
|
|
target_uid = comment.get("target_uid") or comment.get("post_uid", "")
|
|
actor = user["uid"]
|
|
stamp = _now_iso()
|
|
soft_delete_attachments_for("comment", [comment["uid"]], actor)
|
|
soft_delete(
|
|
"votes", actor, stamp=stamp, target_uid=comment["uid"], target_type="comment"
|
|
)
|
|
soft_delete_engagement("comment", [comment["uid"]], actor)
|
|
soft_delete("comments", actor, stamp=stamp, uid=comment["uid"])
|
|
logger.info(f"Comment {comment['uid']} soft-deleted by {user['username']}")
|
|
audit.record(
|
|
request,
|
|
"comment.delete",
|
|
user=user,
|
|
target_type="comment",
|
|
target_uid=comment["uid"],
|
|
summary=f"{user['username']} deleted a comment under {target_type} {target_uid}",
|
|
links=[
|
|
audit.target("comment", comment["uid"]),
|
|
audit.parent(target_type, target_uid),
|
|
],
|
|
)
|
|
return target_type, target_uid
|
|
|
|
|
|
def set_bookmark(
|
|
request, user: dict, target_type: str, target_uid: str, saved: bool
|
|
) -> bool:
|
|
bookmarks = get_table("bookmarks")
|
|
existing = bookmarks.find_one(
|
|
user_uid=user["uid"], target_uid=target_uid, target_type=target_type
|
|
)
|
|
changed = False
|
|
if saved:
|
|
if existing and existing.get("deleted_at"):
|
|
bookmarks.update(
|
|
{"id": existing["id"], "deleted_at": None, "deleted_by": None}, ["id"]
|
|
)
|
|
changed = True
|
|
elif not existing:
|
|
bookmarks.insert(
|
|
{
|
|
"uid": generate_uid(),
|
|
"user_uid": user["uid"],
|
|
"target_uid": target_uid,
|
|
"target_type": target_type,
|
|
"created_at": datetime.now(timezone.utc).isoformat(),
|
|
"deleted_at": None,
|
|
"deleted_by": None,
|
|
}
|
|
)
|
|
changed = True
|
|
else:
|
|
if existing and not existing.get("deleted_at"):
|
|
bookmarks.update(
|
|
{
|
|
"id": existing["id"],
|
|
"deleted_at": _now_iso(),
|
|
"deleted_by": user["uid"],
|
|
},
|
|
["id"],
|
|
)
|
|
changed = True
|
|
if changed:
|
|
audit.record(
|
|
request,
|
|
"bookmark.add" if saved else "bookmark.remove",
|
|
user=user,
|
|
target_type=target_type,
|
|
target_uid=target_uid,
|
|
summary=f"{user['username']} {'bookmarked' if saved else 'removed bookmark from'} {target_type} {target_uid}",
|
|
links=[audit.target(target_type, target_uid)],
|
|
)
|
|
if saved:
|
|
track_action(user["uid"], "bookmark")
|
|
return saved
|
|
|
|
|
|
def detail_context(
|
|
request,
|
|
user: dict | None,
|
|
detail: dict,
|
|
key: str,
|
|
seo_ctx: dict,
|
|
extra: dict | None = None,
|
|
) -> dict:
|
|
context = {
|
|
**seo_ctx,
|
|
"request": request,
|
|
"user": user,
|
|
key: detail["item"],
|
|
"author": detail["author"],
|
|
"is_owner": detail["is_owner"],
|
|
"star_count": detail["star_count"],
|
|
"my_vote": detail["my_vote"],
|
|
"time_ago": detail["time_ago"],
|
|
"comments": detail["comments"],
|
|
"attachments": detail["attachments"],
|
|
"reactions": detail.get("reactions", {"counts": {}, "mine": []}),
|
|
"bookmarked": detail.get("bookmarked", False),
|
|
"poll": detail.get("poll"),
|
|
"project_link": detail.get("project_link"),
|
|
"maturity": detail.get("maturity", "general"),
|
|
}
|
|
if extra:
|
|
context.update(extra)
|
|
return context
|
|
|
|
|
|
def edit_content_item(
|
|
request,
|
|
table_name: str,
|
|
user: dict,
|
|
slug: str,
|
|
update_fields: dict,
|
|
redirect_fail: str,
|
|
target_type: str | None = None,
|
|
):
|
|
from devplacepy.responses import action_result, wants_json, json_error
|
|
|
|
kind = target_type or table_name.rstrip("s")
|
|
table = get_table(table_name)
|
|
item = resolve_by_slug(table, slug)
|
|
if not is_owner(item, user):
|
|
audit.record(
|
|
request,
|
|
f"{kind}.edit",
|
|
user=user,
|
|
result="denied",
|
|
target_type=kind,
|
|
target_uid=item["uid"] if item else slug,
|
|
target_label=item.get("title") if item else slug,
|
|
summary=f"{user['username']} denied editing {kind} {slug}",
|
|
)
|
|
if wants_json(request):
|
|
return json_error(403, "Not allowed")
|
|
return RedirectResponse(url=redirect_fail, status_code=302)
|
|
screening = screen_fields(table_name, update_fields)
|
|
refuse_if_blocked(screening)
|
|
update_fields = {
|
|
**update_fields,
|
|
"updated_at": datetime.now(timezone.utc).isoformat(),
|
|
}
|
|
table.update({"uid": item["uid"], **update_fields}, ["uid"])
|
|
record_screening(
|
|
screening,
|
|
target_type=kind,
|
|
target_uid=item["uid"],
|
|
actor_uid=user["uid"],
|
|
request=request,
|
|
)
|
|
schedule_correction(user, table_name, item["uid"], request)
|
|
schedule_modification(user, table_name, item["uid"], request)
|
|
schedule_seo_meta_for_table(table_name, item["uid"], regenerate=True)
|
|
logger.info(f"{table_name} {item['uid']} edited by {user['username']}")
|
|
label = update_fields.get("title") or item.get("title") or item["uid"]
|
|
audit.record(
|
|
request,
|
|
f"{kind}.edit",
|
|
user=user,
|
|
target_type=kind,
|
|
target_uid=item["uid"],
|
|
target_label=label,
|
|
summary=f"{user['username']} edited {kind} {label}",
|
|
metadata={"changed_fields": sorted(update_fields.keys())},
|
|
links=[audit.target(kind, item["uid"], label)],
|
|
)
|
|
url = f"/{table_name}/{item['slug'] or item['uid']}"
|
|
return action_result(
|
|
request, url, data={"uid": item["uid"], "slug": item.get("slug"), "url": url}
|
|
)
|
|
|
|
|
|
def delete_content_item(
|
|
request,
|
|
table_name: str,
|
|
target_type: str,
|
|
user: dict,
|
|
slug: str,
|
|
redirect_url: str,
|
|
inline_image_field: str | None = None,
|
|
):
|
|
from devplacepy.responses import action_result, wants_json, json_error
|
|
|
|
table = get_table(table_name)
|
|
item = resolve_by_slug(table, slug)
|
|
if not item or not (is_owner(item, user) or is_admin(user)):
|
|
audit.record(
|
|
request,
|
|
f"{target_type}.delete",
|
|
user=user,
|
|
result="denied",
|
|
target_type=target_type,
|
|
target_uid=item["uid"] if item else slug,
|
|
target_label=item.get("title") if item else slug,
|
|
summary=f"{user['username']} denied deleting {target_type} {slug}",
|
|
)
|
|
if wants_json(request):
|
|
return json_error(403, "Not allowed")
|
|
return RedirectResponse(url=redirect_url, status_code=302)
|
|
item_label = item.get("title") or item["uid"]
|
|
actor = user["uid"]
|
|
stamp = _now_iso()
|
|
soft_delete_attachments_for(target_type, [item["uid"]], actor)
|
|
comment_uids = []
|
|
if "comments" in db.tables:
|
|
comments = get_table("comments")
|
|
comment_uids = [
|
|
comment["uid"]
|
|
for comment in comments.find(target_uid=item["uid"], deleted_at=None)
|
|
]
|
|
soft_delete_attachments_for("comment", comment_uids, actor)
|
|
soft_delete("comments", actor, stamp=stamp, target_uid=item["uid"])
|
|
if "votes" in db.tables:
|
|
soft_delete("votes", actor, stamp=stamp, target_uid=item["uid"])
|
|
soft_delete_in(
|
|
"votes", "target_uid", comment_uids, actor, stamp=stamp, target_type="comment"
|
|
)
|
|
soft_delete_engagement(target_type, [item["uid"]], actor)
|
|
if comment_uids:
|
|
soft_delete_engagement("comment", comment_uids, actor)
|
|
if target_type == "post":
|
|
clear_user_post_count(item["user_uid"])
|
|
if target_type == "quiz":
|
|
from devplacepy.services.quiz.store import cascade_questions, clear_cache
|
|
|
|
cascade_questions(item["uid"], actor, stamp)
|
|
clear_cache()
|
|
if target_type == "project":
|
|
from devplacepy.project_files import soft_delete_all_project_files
|
|
from devplacepy.templating import clear_user_projects_cache
|
|
|
|
soft_delete_all_project_files(item["uid"], actor)
|
|
soft_delete_fork_relations(item["uid"], actor)
|
|
clear_user_projects_cache(item["user_uid"])
|
|
soft_delete(table_name, actor, stamp=stamp, uid=item["uid"])
|
|
logger.info(f"{table_name} {item['uid']} soft-deleted by {user['username']}")
|
|
audit.record(
|
|
request,
|
|
f"{target_type}.delete",
|
|
user=user,
|
|
target_type=target_type,
|
|
target_uid=item["uid"],
|
|
target_label=item_label,
|
|
summary=f"{user['username']} deleted {target_type} {item_label}",
|
|
metadata={"comment_count": len(comment_uids)},
|
|
links=[audit.target(target_type, item["uid"], item_label)],
|
|
)
|
|
return action_result(request, redirect_url)
|
|
|
|
|
|
def load_detail(
|
|
table_name: str, target_type: str, slug: str, user: dict | None
|
|
) -> dict | None:
|
|
item = resolve_by_slug(get_table(table_name), slug)
|
|
if not item:
|
|
return None
|
|
if user and item["user_uid"] in get_blocked_uids(user["uid"]):
|
|
return None
|
|
author = get_users_by_uids([item["user_uid"]]).get(item["user_uid"])
|
|
if target_type in STAR_TARGETS:
|
|
star_count = item.get("stars") or 0
|
|
else:
|
|
ups, downs = get_vote_counts([item["uid"]])
|
|
star_count = ups.get(item["uid"], 0) - downs.get(item["uid"], 0)
|
|
reactions = (
|
|
get_reactions_by_targets(target_type, [item["uid"]], user).get(
|
|
item["uid"], {"counts": {}, "mine": []}
|
|
)
|
|
if target_type in REACTABLE_TYPES
|
|
else {"counts": {}, "mine": []}
|
|
)
|
|
bookmarked = (
|
|
bool(user)
|
|
and target_type in BOOKMARKABLE_TYPES
|
|
and item["uid"] in get_user_bookmarks(user["uid"], target_type, [item["uid"]])
|
|
)
|
|
return {
|
|
"item": item,
|
|
"author": author,
|
|
"is_owner": bool(user and user["uid"] == item["user_uid"]),
|
|
"star_count": star_count,
|
|
"my_vote": get_user_votes(user["uid"], [item["uid"]]).get(item["uid"], 0)
|
|
if user
|
|
else 0,
|
|
"comments": load_comments(target_type, item["uid"], user),
|
|
"attachments": get_attachments(target_type, item["uid"]),
|
|
"time_ago": time_ago(item["created_at"]),
|
|
"reactions": reactions,
|
|
"bookmarked": bookmarked,
|
|
"poll": get_poll_for_post(item["uid"], user) if target_type == "post" else None,
|
|
"project_link": get_project_by_uid(item.get("project_uid")) if target_type == "post" else None,
|
|
"maturity": get_maturity(target_type, item["uid"])["level"],
|
|
}
|
|
|
|
|
|
def enrich_items(
|
|
items: list,
|
|
key: str,
|
|
authors: dict,
|
|
extra_maps: dict[str, Any] | None = None,
|
|
ts_field: str = "created_at",
|
|
user: dict | None = None,
|
|
) -> list:
|
|
extra_maps = extra_maps or {}
|
|
user_votes = (
|
|
get_user_votes(user["uid"], [item["uid"] for item in items]) if user else {}
|
|
)
|
|
maturity = get_maturity_by_targets(key, [item["uid"] for item in items])
|
|
enriched = []
|
|
for item in items:
|
|
entry = {
|
|
key: item,
|
|
"author": authors.get(item["user_uid"]),
|
|
"time_ago": time_ago(item[ts_field]),
|
|
"my_vote": user_votes.get(item["uid"], 0),
|
|
"maturity": maturity.get(item["uid"], {}).get("level", "general"),
|
|
}
|
|
for name, source in extra_maps.items():
|
|
entry[name] = (
|
|
source(item) if callable(source) else source.get(item["uid"], 0)
|
|
)
|
|
if key == "post" and item.get("project_uid"):
|
|
entry["project_link"] = get_project_by_uid(item["project_uid"])
|
|
enriched.append(entry)
|
|
return enriched
|
|
|
|
|
|
def count_project_devlog(project_uid: str) -> int:
|
|
return get_table("posts").count(project_uid=project_uid, deleted_at=None)
|
|
|
|
|
|
def get_project_devlog(
|
|
project_uid: str, before: str | None = None, viewer: dict | None = None
|
|
) -> tuple[list, str | None]:
|
|
posts, next_cursor = paginate(
|
|
get_table("posts"),
|
|
before=before,
|
|
viewer_uid=viewer["uid"] if viewer else None,
|
|
project_uid=project_uid,
|
|
)
|
|
if not posts:
|
|
return [], None
|
|
|
|
authors = get_users_by_uids([post["user_uid"] for post in posts])
|
|
counts = get_comment_counts_by_post_uids([post["uid"] for post in posts])
|
|
|
|
enriched = enrich_items(
|
|
posts, "post", authors, {"comment_count": counts}, user=viewer
|
|
)
|
|
return enriched, next_cursor
|