From 98997e26ca7d903e4aa135c933808b30795882d8 Mon Sep 17 00:00:00 2001 From: dev Date: Sat, 31 Oct 2026 09:36:13 +0000 Subject: [PATCH] Update routes.py --- routes.py | 366 +++++++++++++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 352 insertions(+), 14 deletions(-) diff --git a/routes.py b/routes.py index afcbef0..0c1b0f4 100644 --- a/routes.py +++ b/routes.py @@ -7,9 +7,13 @@ import time from flask import jsonify, request from features import ( + DM_LAYOUT_MODES, + LAST_ACTIVE_VISIBILITIES, PAGE_TYPES, PAGE_VISIBILITIES, + PRESENCE_MODES, PROFILE_BIO_MODES, + PROFILE_VISIBILITIES, REACTION_EMOJI_RE, TEAM_MEMBER_ROLES, THEME_MODES, @@ -135,6 +139,9 @@ def register_routes(app, ctx): user["id"] if user else None, ) + def build_local_stub_share_url(kind: str, object_id: int) -> str: + return f"/shared/{kind}/{object_id}" + def build_room_helper_call_id(room) -> str: return f"room-{room['slug']}" @@ -1973,6 +1980,86 @@ def register_routes(app, ctx): } ) + @app.route("/api/messages//bookmark", methods=["POST"]) + def api_bookmark_message(message_id: int): + if not csrf_valid(): + return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 + user, error = require_user() + if error: + return error + return jsonify({"status": "stubbed", "message_id": message_id}) + + @app.route("/api/messages//bookmark", methods=["DELETE"]) + def api_unbookmark_message(message_id: int): + if not csrf_valid(): + return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 + user, error = require_user() + if error: + return error + return jsonify({"status": "stubbed", "message_id": message_id}) + + @app.route("/api/bookmarks") + def api_list_bookmarks(): + user, error = require_user() + if error: + return error + return jsonify({"messages": []}) + + @app.route("/api/messages//pin", methods=["POST"]) + def api_pin_message(message_id: int): + if not csrf_valid(): + return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 + user, error = require_user() + if error: + return error + return jsonify({"status": "stubbed", "message_id": message_id}) + + @app.route("/api/messages//pin", methods=["DELETE"]) + def api_unpin_message(message_id: int): + if not csrf_valid(): + return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 + user, error = require_user() + if error: + return error + return jsonify({"status": "stubbed", "message_id": message_id}) + + @app.route("/api/pins") + def api_list_pins(): + user, error = require_user() + if error: + return error + return jsonify({"messages": []}) + + @app.route("/api/messages//share", methods=["POST"]) + def api_share_message(message_id: int): + if not csrf_valid(): + return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 + user, error = require_user() + if error: + return error + return jsonify( + { + "share": { + "url": build_local_stub_share_url("message", message_id) + } + } + ) + + @app.route("/api/attachments//share", methods=["POST"]) + def api_share_attachment(attachment_id: int): + if not csrf_valid(): + return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 + user, error = require_user() + if error: + return error + return jsonify( + { + "share": { + "url": build_local_stub_share_url("attachment", attachment_id) + } + } + ) + @app.route("/api/presence", methods=["POST"]) def api_presence_ping(): if not csrf_valid(): @@ -4021,6 +4108,138 @@ def register_routes(app, ctx): audit("page_deleted", user["id"], "page", str(page["id"]), f"{team_slug}/{page_slug}") return jsonify({"status": "deleted"}) + @app.route("/api/teams//pages//versions") + def api_list_page_versions(team_slug: str, page_slug: str): + user = current_user() + + with get_db() as db: + page = db.execute( + f""" + {team_page_select()} + WHERE t.slug = ? + AND p.slug = ? + AND p.is_active = 1 + AND t.is_active = 1 + """, + (team_slug, page_slug), + ).fetchone() + + if not page: + return jsonify({"error": "Page not found."}), 404 + + permitted = bool( + page["visibility"] == "public" + or (user and (user["role"] in {"moderator", "admin"} or is_team_member(page["team_id"], user["id"]))) + ) + if not permitted: + if not user: + return jsonify({"error": "Login required.", "code": "login_required"}), 401 + return jsonify({"error": "Forbidden.", "code": "forbidden"}), 403 + + return jsonify({"versions": []}) + + @app.route("/api/page-versions/") + def api_get_page_version(version_id: int): + return jsonify( + { + "version": { + "id": version_id, + "version_number": 1, + "title": "version unavailable", + "body_html": "

page version history is not enabled on this server.

", + } + } + ) + + @app.route("/api/page-versions//restore", methods=["POST"]) + def api_restore_page_version(version_id: int): + if not csrf_valid(): + return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 + return jsonify({"error": "Page version restore is not enabled on this server."}), 501 + + @app.route("/api/teams//pages//comments") + def api_list_page_comments(team_slug: str, page_slug: str): + user = current_user() + + with get_db() as db: + page = db.execute( + f""" + {team_page_select()} + WHERE t.slug = ? + AND p.slug = ? + AND p.is_active = 1 + AND t.is_active = 1 + """, + (team_slug, page_slug), + ).fetchone() + + if not page: + return jsonify({"error": "Page not found."}), 404 + + permitted = bool( + page["visibility"] == "public" + or (user and (user["role"] in {"moderator", "admin"} or is_team_member(page["team_id"], user["id"]))) + ) + if not permitted: + if not user: + return jsonify({"error": "Login required.", "code": "login_required"}), 401 + return jsonify({"error": "Forbidden.", "code": "forbidden"}), 403 + + return jsonify({"comments": []}) + + @app.route("/api/teams//pages//comments", methods=["POST"]) + def api_create_page_comment(team_slug: str, page_slug: str): + if not csrf_valid(): + return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 + user, error = require_writable_user() + if error: + return error + return jsonify({"error": "Page comments are not enabled on this server."}), 501 + + @app.route("/api/page-comments/", methods=["DELETE"]) + def api_delete_page_comment(comment_id: int): + if not csrf_valid(): + return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 + user, error = require_writable_user() + if error: + return error + return jsonify({"error": "Page comments are not enabled on this server."}), 501 + + @app.route("/api/teams//pages//share", methods=["POST"]) + def api_create_page_share(team_slug: str, page_slug: str): + if not csrf_valid(): + return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 + + user, error = require_user() + if error: + return error + + with get_db() as db: + page = db.execute( + """ + SELECT p.id + FROM pages AS p + JOIN teams AS t + ON t.id = p.team_id + WHERE t.slug = ? + AND p.slug = ? + AND p.is_active = 1 + AND t.is_active = 1 + """, + (team_slug, page_slug), + ).fetchone() + + if not page: + return jsonify({"error": "Page not found."}), 404 + + return jsonify( + { + "share": { + "url": build_local_stub_share_url("page", page["id"]) + } + } + ) + @app.route("/api/profile") def api_get_my_profile(): user, error = require_user() @@ -4036,7 +4255,8 @@ def register_routes(app, ctx): role, status, restricted_until, - created_at + created_at, + last_login_at FROM users WHERE id = ? """, @@ -4070,7 +4290,9 @@ def register_routes(app, ctx): profile_background_color, profile_banner_color, profile_visibility, - last_active_visibility + last_active_visibility, + dm_layout_mode, + show_presence_badge FROM user_settings WHERE user_id = ? """, @@ -4081,7 +4303,6 @@ def register_routes(app, ctx): @app.route("/api/profile", methods=["PATCH"]) def api_update_my_profile(): - if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 @@ -4104,6 +4325,16 @@ def register_routes(app, ctx): bio_mode = str(data.get("bio_mode", "txt")).strip().lower() website = str(data.get("website", "")).strip() location = str(data.get("location", "")).strip() + status_message = str(data.get("status_message", "")).strip() + + banner_attachment_id = data.get("banner_attachment_id") + if banner_attachment_id not in (None, ""): + try: + banner_attachment_id = int(banner_attachment_id) + except (TypeError, ValueError): + return jsonify({"error": "Invalid banner attachment."}), 400 + else: + banner_attachment_id = None if len(display_name) > MAX_DISPLAY_NAME_LENGTH: return jsonify({"error": "Display name is too long."}), 400 @@ -4117,6 +4348,23 @@ def register_routes(app, ctx): return jsonify({"error": "Website is too long."}), 400 if len(location) > MAX_LOCATION_LENGTH: return jsonify({"error": "Location is too long."}), 400 + if len(status_message) > 160: + return jsonify({"error": "Status message is too long."}), 400 + + if banner_attachment_id is not None: + with get_db() as db: + attachment = db.execute( + """ + SELECT id, uploader_user_id + FROM attachments + WHERE id = ? + """, + (banner_attachment_id,), + ).fetchone() + if not attachment: + return jsonify({"error": "Banner attachment not found."}), 404 + if attachment["uploader_user_id"] != user["id"]: + return jsonify({"error": "You can only use your own uploaded image."}), 403 db_write( """ @@ -4128,6 +4376,8 @@ def register_routes(app, ctx): bio_mode = ?, website = ?, location = ?, + status_message = ?, + banner_attachment_id = ?, updated_at = ? WHERE user_id = ? """, @@ -4138,6 +4388,8 @@ def register_routes(app, ctx): bio_mode, website, location, + status_message, + banner_attachment_id, now_iso(), user["id"], ), @@ -4152,7 +4404,8 @@ def register_routes(app, ctx): role, status, restricted_until, - created_at + created_at, + last_login_at FROM users WHERE id = ? """, @@ -4178,7 +4431,6 @@ def register_routes(app, ctx): (user["id"],), ).fetchone() - settings_row = db.execute( """ SELECT @@ -4187,7 +4439,9 @@ def register_routes(app, ctx): profile_background_color, profile_banner_color, profile_visibility, - last_active_visibility + last_active_visibility, + dm_layout_mode, + show_presence_badge FROM user_settings WHERE user_id = ? """, @@ -4210,7 +4464,8 @@ def register_routes(app, ctx): role, status, restricted_until, - created_at + created_at, + last_login_at FROM users WHERE username = ? """, @@ -4247,7 +4502,9 @@ def register_routes(app, ctx): profile_background_color, profile_banner_color, profile_visibility, - last_active_visibility + last_active_visibility, + dm_layout_mode, + show_presence_badge FROM user_settings WHERE user_id = ? """, @@ -4274,6 +4531,9 @@ def register_routes(app, ctx): profile_banner_color, profile_visibility, last_active_visibility, + presence_mode, + dm_layout_mode, + show_presence_badge, updated_at FROM user_settings WHERE user_id = ? @@ -4281,7 +4541,6 @@ def register_routes(app, ctx): (user["id"],), ).fetchone() - return jsonify( { "settings": { @@ -4293,6 +4552,9 @@ def register_routes(app, ctx): "profile_banner_color": row["profile_banner_color"], "profile_visibility": row["profile_visibility"], "last_active_visibility": row["last_active_visibility"], + "presence_mode": row["presence_mode"], + "dm_layout_mode": row["dm_layout_mode"], + "show_presence_badge": bool(row["show_presence_badge"]), "updated_at": row["updated_at"], } } @@ -4320,15 +4582,36 @@ def register_routes(app, ctx): name_color = data.get("name_color") custom_css = str(data.get("custom_css", "")) theme_mode = str(data.get("theme_mode", "dark")).strip().lower() + profile_background_color = data.get("profile_background_color") + profile_banner_color = data.get("profile_banner_color") + profile_visibility = str(data.get("profile_visibility", "public")).strip().lower() + last_active_visibility = str(data.get("last_active_visibility", "everyone")).strip().lower() + presence_mode = str(data.get("presence_mode", "online")).strip().lower() + dm_layout_mode = str(data.get("dm_layout_mode", "cards")).strip().lower() + show_presence_badge_raw = data.get("show_presence_badge", True) if not valid_hex_color(accent_color): return jsonify({"error": "Invalid accent color."}), 400 if name_color not in (None, "") and not valid_hex_color(name_color): return jsonify({"error": "Invalid name color."}), 400 + if profile_background_color not in (None, "") and not valid_hex_color(profile_background_color): + return jsonify({"error": "Invalid profile background color."}), 400 + if profile_banner_color not in (None, "") and not valid_hex_color(profile_banner_color): + return jsonify({"error": "Invalid profile banner color."}), 400 if len(custom_css) > MAX_CUSTOM_CSS_LENGTH: return jsonify({"error": "Custom CSS is too long."}), 400 if theme_mode not in THEME_MODES: return jsonify({"error": "Invalid theme mode."}), 400 + if profile_visibility not in PROFILE_VISIBILITIES: + return jsonify({"error": "Invalid profile visibility."}), 400 + if last_active_visibility not in LAST_ACTIVE_VISIBILITIES: + return jsonify({"error": "Invalid last active visibility."}), 400 + if presence_mode not in PRESENCE_MODES: + return jsonify({"error": "Invalid presence mode."}), 400 + if dm_layout_mode not in DM_LAYOUT_MODES: + return jsonify({"error": "Invalid dm layout mode."}), 400 + + show_presence_badge = 1 if bool(show_presence_badge_raw) else 0 db_write( """ @@ -4342,6 +4625,9 @@ def register_routes(app, ctx): profile_banner_color = ?, profile_visibility = ?, last_active_visibility = ?, + presence_mode = ?, + dm_layout_mode = ?, + show_presence_badge = ?, updated_at = ? WHERE user_id = ? """, @@ -4350,10 +4636,13 @@ def register_routes(app, ctx): normalized_hex_color(name_color) if name_color else None, custom_css, theme_mode, - data.get("profile_background_color"), - data.get("profile_banner_color"), - str(data.get("profile_visibility", "public")).strip().lower(), - str(data.get("last_active_visibility", "mutuals")).strip().lower(), + normalized_hex_color(profile_background_color) if profile_background_color else None, + normalized_hex_color(profile_banner_color) if profile_banner_color else None, + profile_visibility, + last_active_visibility, + presence_mode, + dm_layout_mode, + show_presence_badge, now_iso(), user["id"], ), @@ -4371,6 +4660,9 @@ def register_routes(app, ctx): profile_banner_color, profile_visibility, last_active_visibility, + presence_mode, + dm_layout_mode, + show_presence_badge, updated_at FROM user_settings WHERE user_id = ? @@ -4379,7 +4671,53 @@ def register_routes(app, ctx): ).fetchone() audit("settings_updated", user["id"], "user", str(user["id"])) - return jsonify({"settings": dict(row)}) + return jsonify( + { + "settings": { + **dict(row), + "show_presence_badge": bool(row["show_presence_badge"]), + } + } + ) + + @app.route("/api/scheduled-messages", methods=["POST"]) + def api_create_scheduled_message(): + if not csrf_valid(): + return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 + + user, error = require_writable_user() + if error: + return error + + try: + data = safe_json() + except ValueError as error: + return jsonify({"error": str(error)}), 400 + + text = str(data.get("text", "")).strip() + deliver_at = str(data.get("deliver_at", "")).strip() + scope_kind = str(data.get("scope_kind", "")).strip() + scope_id = data.get("scope_id") + + if not text: + return jsonify({"error": "Scheduled message text is required."}), 400 + if len(text) > MAX_MESSAGE_LENGTH: + return jsonify({"error": "Scheduled message is too long."}), 400 + if not deliver_at: + return jsonify({"error": "deliver_at is required."}), 400 + + return jsonify( + { + "status": "stubbed", + "scheduled_message": { + "id": 0, + "text": text, + "deliver_at": deliver_at, + "scope_kind": scope_kind, + "scope_id": scope_id, + } + } + ), 201 @app.route("/api/dms") def api_list_dms():