802 lines
25 KiB
Python
802 lines
25 KiB
Python
"""Community data — PocketBase backed (discussions, meetups, gigs, notifications)."""
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
import secrets
|
|
from datetime import datetime, timezone
|
|
from typing import Any
|
|
|
|
from app.services.pb_client import pb, pb_quote
|
|
from app.services.pb_repo import (
|
|
find_by_legacy,
|
|
pb_date,
|
|
q,
|
|
rid,
|
|
safe_create,
|
|
safe_first,
|
|
safe_list,
|
|
safe_update,
|
|
use_pb,
|
|
)
|
|
|
|
LIKE_ACTIONS = frozenset({"like", "right", "superlike"})
|
|
|
|
|
|
def _now_iso() -> str:
|
|
return datetime.now(timezone.utc).strftime("%Y-%m-%d")
|
|
|
|
|
|
def _slugify(text: str) -> str:
|
|
s = re.sub(r"[^a-z0-9\u4e00-\u9fff]+", "-", text.lower()).strip("-")
|
|
return (s[:48] or secrets.token_hex(4))
|
|
|
|
|
|
# ── Meetups ──────────────────────────────────────────────────────────────
|
|
|
|
def _meetup_from_pb(r: dict) -> dict:
|
|
return {
|
|
"id": rid(r),
|
|
"pb_id": r["id"],
|
|
"title": r.get("title", ""),
|
|
"city": r.get("city", ""),
|
|
"destination_slug": r.get("destinationSlug", ""),
|
|
"emoji": r.get("emoji", "🎉"),
|
|
"date": pb_date(r.get("date")),
|
|
"time": r.get("time", ""),
|
|
"venue": r.get("venue", ""),
|
|
"description": r.get("description", ""),
|
|
"mode": r.get("mode", "offline"),
|
|
"access_level": r.get("accessLevel", "public"),
|
|
"mirotalkRoom": r.get("mirotalkRoom", ""),
|
|
"loungeChannel": r.get("loungeChannel", ""),
|
|
"meetingUrl": r.get("meetingUrl", ""),
|
|
"rsvp_count": int(r.get("rsvpCount") or 0),
|
|
"max_attendees": int(r.get("maxAttendees") or 30),
|
|
"organizer": r.get("organizer", ""),
|
|
"organizer_id": r.get("organizerId", ""),
|
|
"tags": r.get("tags") or [],
|
|
"is_upcoming": bool(r.get("isUpcoming", True)),
|
|
}
|
|
|
|
|
|
def _rsvp_count(meetup_pb_id: str) -> int:
|
|
rows = safe_list("meetup_rsvps", filter=f"meetupId={q(meetup_pb_id)}")
|
|
return len(rows)
|
|
|
|
|
|
def list_meetups(upcoming: bool = True) -> list[dict]:
|
|
if not use_pb():
|
|
return _json_list_meetups(upcoming)
|
|
rows = safe_list("meetups")
|
|
rows.sort(key=lambda r: r.get("created", ""), reverse=True)
|
|
items = []
|
|
for r in rows:
|
|
m = _meetup_from_pb(r)
|
|
if upcoming and not m.get("is_upcoming", True):
|
|
continue
|
|
m["rsvp_count"] = max(m["rsvp_count"], _rsvp_count(r["id"]))
|
|
items.append(m)
|
|
return items
|
|
|
|
|
|
def get_meetup(meetup_id: str) -> dict | None:
|
|
if not use_pb():
|
|
return _json_get_meetup(meetup_id)
|
|
r = find_by_legacy("meetups", meetup_id)
|
|
if not r:
|
|
return None
|
|
m = _meetup_from_pb(r)
|
|
m["rsvp_count"] = max(m["rsvp_count"], _rsvp_count(r["id"]))
|
|
return m
|
|
|
|
|
|
def create_meetup(user_id: str, user_name: str, payload: dict) -> dict:
|
|
if not use_pb():
|
|
return _json_create_meetup(user_id, user_name, payload)
|
|
mid = _slugify(payload.get("title", "meetup")) + "-" + secrets.token_hex(3)
|
|
mode = payload.get("mode", "offline")
|
|
room = payload.get("mirotalkRoom") or ""
|
|
if mode in ("online", "hybrid") and not room:
|
|
room = f"nomadro-{_slugify(payload.get('city', 'online'))}-{secrets.token_hex(2)}"
|
|
lounge = payload.get("loungeChannel") or (f"#{room}" if room else "")
|
|
date_val = payload.get("date", _now_iso())
|
|
if date_val and "T" not in str(date_val):
|
|
date_val = f"{date_val} 00:00:00.000Z"
|
|
row = safe_create(
|
|
"meetups",
|
|
{
|
|
"legacyId": mid,
|
|
"title": payload["title"],
|
|
"city": payload.get("city", "线上"),
|
|
"destinationSlug": payload.get("destination_slug", ""),
|
|
"emoji": payload.get("emoji", "🎉"),
|
|
"date": date_val,
|
|
"time": payload.get("time", "19:00"),
|
|
"venue": payload.get("venue", "待定"),
|
|
"description": payload.get("description", ""),
|
|
"mode": mode,
|
|
"accessLevel": payload.get("access_level", "public"),
|
|
"mirotalkRoom": room,
|
|
"loungeChannel": lounge,
|
|
"rsvpCount": 0,
|
|
"maxAttendees": int(payload.get("max_attendees", 30)),
|
|
"organizer": user_name,
|
|
"organizerId": user_id,
|
|
"tags": payload.get("tags", []),
|
|
"isUpcoming": True,
|
|
"status": "published",
|
|
},
|
|
)
|
|
return _meetup_from_pb(row) if row else {}
|
|
|
|
|
|
def rsvp_meetup(meetup_id: str, user_id: str | None = None) -> dict:
|
|
if not use_pb():
|
|
return _json_rsvp_meetup(meetup_id, user_id)
|
|
m = get_meetup(meetup_id)
|
|
if not m:
|
|
return {"ok": False, "error": "not_found"}
|
|
uid = user_id or f"guest-{secrets.token_hex(4)}"
|
|
pb_id = m["pb_id"]
|
|
existing = safe_first("meetup_rsvps", filter=f"meetupId={q(pb_id)} && userId={q(uid)}")
|
|
count = _rsvp_count(pb_id)
|
|
if existing:
|
|
return {"ok": True, "message": "已报名", "rsvp_count": count}
|
|
if count >= m["max_attendees"]:
|
|
return {"ok": False, "error": "full"}
|
|
safe_create("meetup_rsvps", {"meetupId": pb_id, "userId": uid, "status": "going"})
|
|
count += 1
|
|
safe_update("meetups", pb_id, {"rsvpCount": count})
|
|
return {"ok": True, "message": f"已报名「{m['title']}」", "rsvp_count": count}
|
|
|
|
|
|
def cancel_rsvp(meetup_id: str, user_id: str) -> dict:
|
|
if not use_pb():
|
|
return _json_cancel_rsvp(meetup_id, user_id)
|
|
m = get_meetup(meetup_id)
|
|
if not m:
|
|
return {"ok": False, "error": "not_found"}
|
|
rows = safe_list("meetup_rsvps", filter=f"meetupId={q(m['pb_id'])} && userId={q(user_id)}")
|
|
for r in rows:
|
|
try:
|
|
pb.delete_record("meetup_rsvps", r["id"])
|
|
except Exception:
|
|
pass
|
|
count = _rsvp_count(m["pb_id"])
|
|
safe_update("meetups", m["pb_id"], {"rsvpCount": count})
|
|
return {"ok": True, "rsvp_count": count}
|
|
|
|
|
|
def user_rsvp_ids(user_id: str) -> set[str]:
|
|
if not use_pb():
|
|
return _json_user_rsvp_ids(user_id)
|
|
rows = safe_list("meetup_rsvps", filter=f"userId={q(user_id)}")
|
|
ids: set[str] = set()
|
|
for r in rows:
|
|
meetup = safe_first("meetups", filter=f"id={q(r.get('meetupId', ''))}")
|
|
if meetup:
|
|
ids.add(rid(meetup))
|
|
return ids
|
|
|
|
|
|
# ── Discussions ────────────────────────────────────────────────────────
|
|
|
|
def _discussion_from_pb(r: dict, *, like_extra: int = 0, reply_count: int | None = None) -> dict:
|
|
base_likes = int(r.get("likes") or 0)
|
|
replies = reply_count if reply_count is not None else int(r.get("replies") or 0)
|
|
return {
|
|
"id": rid(r),
|
|
"pb_id": r["id"],
|
|
"title": r.get("title", ""),
|
|
"excerpt": r.get("excerpt", ""),
|
|
"author": r.get("author", ""),
|
|
"author_id": r.get("authorUserId", ""),
|
|
"author_emoji": r.get("authorEmoji", "🧑💻"),
|
|
"category": r.get("category", "社区"),
|
|
"reply_count": replies,
|
|
"like_count": base_likes + like_extra,
|
|
"view_count": int(r.get("views") or 0),
|
|
"is_pinned": bool(r.get("pinned")),
|
|
"created_at": pb_date(r.get("created")),
|
|
"tags": r.get("tags") or [],
|
|
}
|
|
|
|
|
|
def _discussion_likes_count(discussion_pb_id: str) -> int:
|
|
return len(safe_list("discussion_likes", filter=f"discussionId={q(discussion_pb_id)}"))
|
|
|
|
|
|
def list_discussions(category: str | None = None) -> list[dict]:
|
|
if not use_pb():
|
|
return _json_list_discussions(category)
|
|
rows = safe_list("discussions")
|
|
rows.sort(key=lambda r: r.get("created", ""), reverse=True)
|
|
items = []
|
|
for r in rows:
|
|
d = _discussion_from_pb(r, like_extra=_discussion_likes_count(r["id"]))
|
|
if category and d["category"] != category:
|
|
continue
|
|
items.append(d)
|
|
pinned = sorted([d for d in items if d.get("is_pinned")], key=lambda x: x["created_at"], reverse=True)
|
|
rest = sorted([d for d in items if not d.get("is_pinned")], key=lambda x: x["created_at"], reverse=True)
|
|
return pinned + rest
|
|
|
|
|
|
def get_discussion(discussion_id: str, increment_view: bool = False) -> dict | None:
|
|
if not use_pb():
|
|
return _json_get_discussion(discussion_id, increment_view)
|
|
r = find_by_legacy("discussions", discussion_id)
|
|
if not r:
|
|
return None
|
|
if increment_view:
|
|
views = int(r.get("views") or 0) + 1
|
|
safe_update("discussions", r["id"], {"views": views})
|
|
r["views"] = views
|
|
replies_raw = safe_list("discussion_replies", filter=f"discussionId={q(r['id'])}")
|
|
replies_raw.sort(key=lambda rep: rep.get("created", ""))
|
|
replies = [
|
|
{
|
|
"id": rep["id"],
|
|
"author": rep.get("author", ""),
|
|
"author_id": rep.get("userId", ""),
|
|
"author_emoji": rep.get("authorEmoji", "🧑💻"),
|
|
"content": rep.get("body", ""),
|
|
"created_at": pb_date(rep.get("created")),
|
|
"like_count": 0,
|
|
}
|
|
for rep in replies_raw
|
|
]
|
|
d = _discussion_from_pb(
|
|
r,
|
|
like_extra=_discussion_likes_count(r["id"]),
|
|
reply_count=len(replies),
|
|
)
|
|
return {**d, "replies": replies}
|
|
|
|
|
|
def create_discussion(user_id: str, user_name: str, payload: dict) -> dict:
|
|
if not use_pb():
|
|
return _json_create_discussion(user_id, user_name, payload)
|
|
did = _slugify(payload["title"]) + "-" + secrets.token_hex(3)
|
|
row = safe_create(
|
|
"discussions",
|
|
{
|
|
"legacyId": did,
|
|
"title": payload["title"],
|
|
"excerpt": (payload.get("excerpt") or payload.get("content") or "")[:1000],
|
|
"author": user_name,
|
|
"authorUserId": user_id,
|
|
"authorEmoji": payload.get("author_emoji", "🧑💻"),
|
|
"category": payload.get("category", "社区"),
|
|
"tags": payload.get("tags", []),
|
|
"replies": 0,
|
|
"views": 0,
|
|
"likes": 0,
|
|
"pinned": False,
|
|
},
|
|
)
|
|
return _discussion_from_pb(row) if row else {}
|
|
|
|
|
|
def add_reply(discussion_id: str, user_id: str, user_name: str, content: str, emoji: str = "🧑💻") -> dict | None:
|
|
if not use_pb():
|
|
return _json_add_reply(discussion_id, user_id, user_name, content, emoji)
|
|
r = find_by_legacy("discussions", discussion_id)
|
|
if not r:
|
|
return None
|
|
rep = safe_create(
|
|
"discussion_replies",
|
|
{
|
|
"discussionId": r["id"],
|
|
"userId": user_id,
|
|
"author": user_name,
|
|
"authorEmoji": emoji,
|
|
"body": content,
|
|
},
|
|
)
|
|
if not rep:
|
|
return None
|
|
safe_update("discussions", r["id"], {"replies": int(r.get("replies") or 0) + 1})
|
|
return {
|
|
"id": rep["id"],
|
|
"author": user_name,
|
|
"author_id": user_id,
|
|
"author_emoji": emoji,
|
|
"content": content,
|
|
"created_at": _now_iso(),
|
|
"like_count": 0,
|
|
}
|
|
|
|
|
|
def toggle_discussion_like(discussion_id: str, user_id: str) -> dict:
|
|
if not use_pb():
|
|
return _json_toggle_discussion_like(discussion_id, user_id)
|
|
r = find_by_legacy("discussions", discussion_id)
|
|
if not r:
|
|
return {"liked": False, "like_count": 0}
|
|
existing = safe_first(
|
|
"discussion_likes",
|
|
filter=f"discussionId={q(r['id'])} && userId={q(user_id)}",
|
|
)
|
|
liked = False
|
|
if existing:
|
|
try:
|
|
pb.delete_record("discussion_likes", existing["id"])
|
|
except Exception:
|
|
pass
|
|
else:
|
|
safe_create("discussion_likes", {"discussionId": r["id"], "userId": user_id})
|
|
liked = True
|
|
count = _discussion_likes_count(r["id"])
|
|
return {"liked": liked, "like_count": count}
|
|
|
|
|
|
def toggle_pin(discussion_id: str, user_id: str) -> dict | None:
|
|
if not use_pb():
|
|
return _json_toggle_pin(discussion_id, user_id)
|
|
r = find_by_legacy("discussions", discussion_id)
|
|
if not r or r.get("authorUserId") != user_id:
|
|
return None
|
|
pinned = not bool(r.get("pinned"))
|
|
updated = safe_update("discussions", r["id"], {"pinned": pinned})
|
|
return _discussion_from_pb(updated or r, like_extra=_discussion_likes_count(r["id"])) if updated or r else None
|
|
|
|
|
|
def search_discussions(qs: str) -> list[dict]:
|
|
s = qs.lower()
|
|
return [d for d in list_discussions() if s in d["title"].lower() or s in d.get("excerpt", "").lower()][:20]
|
|
|
|
|
|
# ── Gigs ───────────────────────────────────────────────────────────────
|
|
|
|
def _gig_from_pb(r: dict) -> dict:
|
|
budget = r.get("budget")
|
|
return {
|
|
"id": rid(r),
|
|
"pb_id": r["id"],
|
|
"title": r.get("title", ""),
|
|
"description": r.get("description", ""),
|
|
"budget": f"¥{budget}" if isinstance(budget, (int, float)) and budget else str(budget or "面议"),
|
|
"deadline": r.get("deadline", ""),
|
|
"tags": r.get("tags") or [],
|
|
"poster": r.get("posterName", ""),
|
|
"poster_id": r.get("posterId", ""),
|
|
"status": r.get("status", "open"),
|
|
"category": r.get("category", ""),
|
|
"location": r.get("location", ""),
|
|
}
|
|
|
|
|
|
def list_gigs() -> list[dict]:
|
|
if not use_pb():
|
|
return _json_list_gigs()
|
|
return [_gig_from_pb(r) for r in safe_list("gigs") if r.get("status", "open") == "open"]
|
|
|
|
|
|
def create_gig(user_id: str, user_name: str, payload: dict) -> dict:
|
|
if not use_pb():
|
|
return _json_create_gig(user_id, user_name, payload)
|
|
gid = _slugify(payload["title"]) + "-" + secrets.token_hex(2)
|
|
budget_raw = payload.get("budget", "面议")
|
|
budget = int(budget_raw) if str(budget_raw).isdigit() else 0
|
|
row = safe_create(
|
|
"gigs",
|
|
{
|
|
"legacyId": gid,
|
|
"title": payload["title"],
|
|
"description": payload.get("description", ""),
|
|
"budget": budget,
|
|
"location": payload.get("location", ""),
|
|
"category": payload.get("category", ""),
|
|
"posterId": user_id,
|
|
"posterName": user_name,
|
|
"status": "open",
|
|
},
|
|
)
|
|
return _gig_from_pb(row) if row else {}
|
|
|
|
|
|
def apply_gig(gig_id: str, user_id: str, user_name: str, message: str) -> dict:
|
|
if not use_pb():
|
|
return _json_apply_gig(gig_id, user_id, user_name, message)
|
|
g = find_by_legacy("gigs", gig_id)
|
|
if not g:
|
|
return {"ok": False, "error": "not_found"}
|
|
dup = safe_first(
|
|
"gig_applications",
|
|
filter=f"gigId={q(g['id'])} && userId={q(user_id)}",
|
|
)
|
|
if dup:
|
|
return {"ok": False, "error": "duplicate"}
|
|
safe_create(
|
|
"gig_applications",
|
|
{"gigId": g["id"], "userId": user_id, "applicant": user_name, "message": message, "status": "pending"},
|
|
)
|
|
return {
|
|
"ok": True,
|
|
"message": "申请已提交",
|
|
"poster_id": g.get("posterId", ""),
|
|
"title": g.get("title", ""),
|
|
}
|
|
|
|
|
|
# ── Notifications ──────────────────────────────────────────────────────
|
|
|
|
def _notif_from_pb(r: dict) -> dict:
|
|
return {
|
|
"id": r["id"],
|
|
"title": r.get("title", ""),
|
|
"body": r.get("body", ""),
|
|
"link": r.get("actionUrl", ""),
|
|
"category": r.get("category", "general"),
|
|
"read": bool(r.get("read")),
|
|
"archived": bool(r.get("archived")),
|
|
"pinned": bool(r.get("pinned")),
|
|
"created_at": pb_date(r.get("created")),
|
|
}
|
|
|
|
|
|
def add_notification(user_id: str, title: str, body: str, link: str = "", category: str = "general") -> None:
|
|
if not use_pb():
|
|
return _json_add_notification(user_id, title, body, link, category)
|
|
safe_create(
|
|
"notifications",
|
|
{
|
|
"title": title,
|
|
"body": body,
|
|
"actionUrl": link,
|
|
"category": category,
|
|
"targetUserId": user_id,
|
|
"status": "published",
|
|
"read": False,
|
|
"archived": False,
|
|
"pinned": False,
|
|
},
|
|
)
|
|
|
|
|
|
def list_notifications_filtered(user_id: str, include_archived: bool = False) -> list[dict]:
|
|
if not use_pb():
|
|
return _json_list_notifications_filtered(user_id, include_archived)
|
|
filt = f"targetUserId={q(user_id)}"
|
|
if not include_archived:
|
|
filt += " && archived=false"
|
|
rows = safe_list("notifications", filter=filt)
|
|
rows.sort(key=lambda r: r.get("created", ""), reverse=True)
|
|
items = [_notif_from_pb(r) for r in rows]
|
|
pinned = [n for n in items if n.get("pinned")]
|
|
rest = [n for n in items if not n.get("pinned")]
|
|
return pinned + rest
|
|
|
|
|
|
def list_notifications(user_id: str) -> list[dict]:
|
|
return list_notifications_filtered(user_id)
|
|
|
|
|
|
def notification_unread_count(user_id: str) -> int:
|
|
if not use_pb():
|
|
return _json_notification_unread_count(user_id)
|
|
rows = safe_list(
|
|
"notifications",
|
|
filter=f"targetUserId={q(user_id)} && read=false && archived=false",
|
|
)
|
|
return len(rows)
|
|
|
|
|
|
def update_notification(user_id: str, notif_id: str, action: str) -> dict | None:
|
|
if not use_pb():
|
|
return _json_update_notification(user_id, notif_id, action)
|
|
r = safe_first("notifications", filter=f"id={q(notif_id)} && targetUserId={q(user_id)}")
|
|
if not r:
|
|
return None
|
|
patch: dict[str, Any] = {}
|
|
if action == "read":
|
|
patch["read"] = True
|
|
elif action == "unread":
|
|
patch["read"] = False
|
|
elif action == "archive":
|
|
patch["archived"] = True
|
|
elif action == "restore":
|
|
patch["archived"] = False
|
|
elif action == "pin":
|
|
patch["pinned"] = True
|
|
elif action == "unpin":
|
|
patch["pinned"] = False
|
|
updated = safe_update("notifications", r["id"], patch)
|
|
return _notif_from_pb(updated or r)
|
|
|
|
|
|
def mark_notifications_read(user_id: str) -> None:
|
|
if not use_pb():
|
|
return _json_mark_notifications_read(user_id)
|
|
for r in safe_list("notifications", filter=f"targetUserId={q(user_id)} && read=false"):
|
|
safe_update("notifications", r["id"], {"read": True})
|
|
|
|
|
|
def add_feedback(user_id: str | None, user_name: str, content: str, category: str = "general") -> dict:
|
|
if not use_pb():
|
|
return _json_add_feedback(user_id, user_name, content, category)
|
|
row = safe_create(
|
|
"feedback",
|
|
{
|
|
"type": category,
|
|
"email": "",
|
|
"title": user_name,
|
|
"content": content[:2000],
|
|
"status": "open",
|
|
},
|
|
)
|
|
return {"id": row["id"] if row else secrets.token_hex(6), "user_name": user_name, "content": content}
|
|
|
|
|
|
def meetup_social_suggestions(meetup_id: str, limit: int = 6) -> list[dict]:
|
|
m = get_meetup(meetup_id)
|
|
if not m:
|
|
return []
|
|
city = m.get("city", "")
|
|
from app.data.social_profiles import CANDIDATE_PROFILES
|
|
out = []
|
|
for p in CANDIDATE_PROFILES:
|
|
if city and city not in (p.get("location") or ""):
|
|
continue
|
|
out.append(p)
|
|
if len(out) >= limit:
|
|
break
|
|
return out[:limit] if out else CANDIDATE_PROFILES[:limit]
|
|
|
|
|
|
def stats_overview() -> dict:
|
|
if not use_pb():
|
|
return _json_stats_overview()
|
|
return {
|
|
"meetups": len(safe_list("meetups")),
|
|
"discussions": len(safe_list("discussions")),
|
|
"gigs": len(safe_list("gigs")),
|
|
"members_active": len(safe_list("meetup_rsvps")) + len(safe_list("gig_applications")),
|
|
}
|
|
|
|
|
|
# ── JSON fallback (dev without PocketBase) ───────────────────────────
|
|
|
|
import json
|
|
from copy import deepcopy
|
|
from pathlib import Path
|
|
|
|
from app.data import community_data
|
|
|
|
_JSON_PATH = Path(__file__).resolve().parents[1] / "data" / "community_store.json"
|
|
_j: dict[str, Any] = {}
|
|
|
|
|
|
def _jload() -> None:
|
|
global _j
|
|
if _j:
|
|
return
|
|
if _JSON_PATH.exists():
|
|
try:
|
|
_j.update(json.loads(_JSON_PATH.read_text(encoding="utf-8")))
|
|
except (OSError, json.JSONDecodeError):
|
|
pass
|
|
needs_seed = not _j.get("seeded") or not (
|
|
(_j.get("meetups") or []) and (_j.get("discussions") or []) and (_j.get("gigs") or [])
|
|
)
|
|
if needs_seed:
|
|
_j.update({
|
|
"meetups": deepcopy(community_data.MEETUPS),
|
|
"discussions": deepcopy(community_data.DISCUSSIONS),
|
|
"replies": deepcopy(community_data.DISCUSSION_REPLIES),
|
|
"gigs": deepcopy(getattr(community_data, "GIGS", [])),
|
|
"discussion_likes": _j.get("discussion_likes") or {},
|
|
"rsvps": _j.get("rsvps") or {},
|
|
"gig_apps": _j.get("gig_apps") or [],
|
|
"notifications": _j.get("notifications") or {},
|
|
"feedback": _j.get("feedback") or [],
|
|
"views": _j.get("views") or {},
|
|
"seeded": True,
|
|
})
|
|
_jsave()
|
|
|
|
|
|
def _jsave() -> None:
|
|
_JSON_PATH.parent.mkdir(parents=True, exist_ok=True)
|
|
_JSON_PATH.write_text(json.dumps(_j, ensure_ascii=False), encoding="utf-8")
|
|
|
|
|
|
def _json_list_meetups(upcoming: bool) -> list[dict]:
|
|
_jload()
|
|
items = list(_j.get("meetups") or [])
|
|
if upcoming:
|
|
items = [m for m in items if m.get("is_upcoming", True)]
|
|
return items
|
|
|
|
|
|
def _json_get_meetup(mid: str) -> dict | None:
|
|
_jload()
|
|
return next((m for m in _j.get("meetups") or [] if m["id"] == mid), None)
|
|
|
|
|
|
def _json_create_meetup(user_id: str, user_name: str, payload: dict) -> dict:
|
|
_jload()
|
|
mid = _slugify(payload.get("title", "meetup")) + "-" + secrets.token_hex(3)
|
|
meetup = {"id": mid, "title": payload["title"], "city": payload.get("city", "线上"), "organizer": user_name, "organizer_id": user_id, "is_upcoming": True, "rsvp_count": 0, "max_attendees": 30, **payload}
|
|
_j.setdefault("meetups", []).insert(0, meetup)
|
|
_jsave()
|
|
return meetup
|
|
|
|
|
|
def _json_rsvp_meetup(meetup_id: str, user_id: str | None) -> dict:
|
|
_jload()
|
|
m = _json_get_meetup(meetup_id)
|
|
if not m:
|
|
return {"ok": False, "error": "not_found"}
|
|
uid = user_id or f"guest-{secrets.token_hex(4)}"
|
|
rsvps = _j.setdefault("rsvps", {})
|
|
rset = set(rsvps.get(meetup_id, []))
|
|
if uid in rset:
|
|
return {"ok": True, "message": "已报名", "rsvp_count": len(rset)}
|
|
rset.add(uid)
|
|
rsvps[meetup_id] = list(rset)
|
|
_jsave()
|
|
return {"ok": True, "message": "已报名", "rsvp_count": len(rset)}
|
|
|
|
|
|
def _json_cancel_rsvp(meetup_id: str, user_id: str) -> dict:
|
|
_jload()
|
|
rsvps = _j.setdefault("rsvps", {})
|
|
rset = set(rsvps.get(meetup_id, []))
|
|
rset.discard(user_id)
|
|
rsvps[meetup_id] = list(rset)
|
|
_jsave()
|
|
return {"ok": True, "rsvp_count": len(rset)}
|
|
|
|
|
|
def _json_user_rsvp_ids(user_id: str) -> set[str]:
|
|
_jload()
|
|
return {mid for mid, uids in (_j.get("rsvps") or {}).items() if user_id in uids}
|
|
|
|
|
|
def _json_list_discussions(category: str | None) -> list[dict]:
|
|
_jload()
|
|
items = list(_j.get("discussions") or [])
|
|
if category:
|
|
items = [d for d in items if d.get("category") == category]
|
|
return items
|
|
|
|
|
|
def _json_get_discussion(did: str, inc: bool) -> dict | None:
|
|
_jload()
|
|
for d in _j.get("discussions") or []:
|
|
if d["id"] == did:
|
|
if inc:
|
|
views = _j.setdefault("views", {})
|
|
views[did] = views.get(did, 0) + 1
|
|
_jsave()
|
|
replies = _j.get("replies", {}).get(did, [])
|
|
return {**d, "replies": replies}
|
|
return None
|
|
|
|
|
|
def _json_create_discussion(user_id: str, user_name: str, payload: dict) -> dict:
|
|
_jload()
|
|
did = _slugify(payload["title"]) + "-" + secrets.token_hex(3)
|
|
d = {"id": did, "title": payload["title"], "author": user_name, "author_id": user_id, "created_at": _now_iso(), **payload}
|
|
_j.setdefault("discussions", []).insert(0, d)
|
|
_j.setdefault("replies", {})[did] = []
|
|
_jsave()
|
|
return d
|
|
|
|
|
|
def _json_add_reply(did: str, user_id: str, user_name: str, content: str, emoji: str) -> dict | None:
|
|
_jload()
|
|
if not any(d["id"] == did for d in _j.get("discussions") or []):
|
|
return None
|
|
reply = {"id": secrets.token_hex(6), "author": user_name, "author_id": user_id, "content": content, "created_at": _now_iso()}
|
|
_j.setdefault("replies", {}).setdefault(did, []).append(reply)
|
|
_jsave()
|
|
return reply
|
|
|
|
|
|
def _json_toggle_discussion_like(did: str, user_id: str) -> dict:
|
|
_jload()
|
|
likes = _j.setdefault("discussion_likes", {}).setdefault(did, [])
|
|
if user_id in likes:
|
|
likes.remove(user_id)
|
|
liked = False
|
|
else:
|
|
likes.append(user_id)
|
|
liked = True
|
|
_jsave()
|
|
return {"liked": liked, "like_count": len(likes)}
|
|
|
|
|
|
def _json_toggle_pin(did: str, user_id: str) -> dict | None:
|
|
_jload()
|
|
for d in _j.get("discussions") or []:
|
|
if d["id"] == did and d.get("author_id") == user_id:
|
|
d["is_pinned"] = not d.get("is_pinned", False)
|
|
_jsave()
|
|
return d
|
|
return None
|
|
|
|
|
|
def _json_list_gigs() -> list[dict]:
|
|
_jload()
|
|
return [g for g in _j.get("gigs") or [] if g.get("status", "open") == "open"]
|
|
|
|
|
|
def _json_create_gig(user_id: str, user_name: str, payload: dict) -> dict:
|
|
_jload()
|
|
gid = _slugify(payload["title"]) + "-" + secrets.token_hex(2)
|
|
gig = {"id": gid, "poster": user_name, "poster_id": user_id, "status": "open", **payload}
|
|
_j.setdefault("gigs", []).insert(0, gig)
|
|
_jsave()
|
|
return gig
|
|
|
|
|
|
def _json_apply_gig(gig_id: str, user_id: str, user_name: str, message: str) -> dict:
|
|
_jload()
|
|
gig = next((g for g in (_j.get("gigs") or []) if g.get("id") == gig_id), {}) or {}
|
|
_j.setdefault("gig_apps", []).append({"gig_id": gig_id, "user_id": user_id, "user_name": user_name, "message": message})
|
|
_jsave()
|
|
return {
|
|
"ok": True,
|
|
"message": "申请已提交",
|
|
"poster_id": gig.get("poster_id", ""),
|
|
"title": gig.get("title", ""),
|
|
}
|
|
|
|
|
|
def _json_add_notification(user_id: str, title: str, body: str, link: str, category: str) -> None:
|
|
_jload()
|
|
items = _j.setdefault("notifications", {}).setdefault(user_id, [])
|
|
items.insert(0, {"id": secrets.token_hex(6), "title": title, "body": body, "link": link, "category": category, "read": False, "archived": False, "pinned": False, "created_at": _now_iso()})
|
|
_jsave()
|
|
|
|
|
|
def _json_list_notifications_filtered(user_id: str, include_archived: bool) -> list[dict]:
|
|
_jload()
|
|
items = _j.get("notifications", {}).get(user_id, [])
|
|
if not include_archived:
|
|
items = [n for n in items if not n.get("archived")]
|
|
return items
|
|
|
|
|
|
def _json_notification_unread_count(user_id: str) -> int:
|
|
return sum(1 for n in _json_list_notifications_filtered(user_id, False) if not n.get("read"))
|
|
|
|
|
|
def _json_update_notification(user_id: str, notif_id: str, action: str) -> dict | None:
|
|
_jload()
|
|
for n in _j.get("notifications", {}).get(user_id, []):
|
|
if n["id"] == notif_id:
|
|
if action == "read":
|
|
n["read"] = True
|
|
_jsave()
|
|
return n
|
|
return None
|
|
|
|
|
|
def _json_mark_notifications_read(user_id: str) -> None:
|
|
_jload()
|
|
for n in _j.get("notifications", {}).get(user_id, []):
|
|
n["read"] = True
|
|
_jsave()
|
|
|
|
|
|
def _json_add_feedback(user_id: str | None, user_name: str, content: str, category: str) -> dict:
|
|
_jload()
|
|
item = {"id": secrets.token_hex(6), "user_name": user_name, "content": content, "category": category}
|
|
_j.setdefault("feedback", []).append(item)
|
|
_jsave()
|
|
return item
|
|
|
|
|
|
def _json_stats_overview() -> dict:
|
|
_jload()
|
|
return {
|
|
"meetups": len(_j.get("meetups") or []),
|
|
"discussions": len(_j.get("discussions") or []),
|
|
"gigs": len(_j.get("gigs") or []),
|
|
"members_active": len(_j.get("rsvps") or {}),
|
|
}
|