754 lines
24 KiB
Python
754 lines
24 KiB
Python
"""
|
|
SQLite storage for the private community module.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections import Counter
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
import os
|
|
import re
|
|
import sqlite3
|
|
import time
|
|
import uuid
|
|
|
|
_ROOT = Path(__file__).resolve().parent.parent.parent
|
|
DATA_DIR = Path(os.getenv("COMMUNITY_DATA_DIR", str(_ROOT / "data")))
|
|
DB_PATH = DATA_DIR / "community.db"
|
|
UPLOADS_DIR = DATA_DIR / "community_uploads"
|
|
|
|
UPVOTE_REACTION = "UPVOTE"
|
|
|
|
|
|
def _ensure_dirs() -> None:
|
|
DATA_DIR.mkdir(parents=True, exist_ok=True)
|
|
UPLOADS_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
|
|
def _get_conn() -> sqlite3.Connection:
|
|
_ensure_dirs()
|
|
conn = sqlite3.connect(str(DB_PATH))
|
|
conn.row_factory = sqlite3.Row
|
|
conn.execute("PRAGMA foreign_keys = ON")
|
|
return conn
|
|
|
|
|
|
def _now_iso() -> str:
|
|
return time.strftime("%Y-%m-%dT%H:%M:%S.000Z", time.gmtime())
|
|
|
|
|
|
def _normalize_memo_id(memo_id: str) -> str:
|
|
return memo_id.replace("memos/", "") if memo_id.startswith("memos/") else memo_id
|
|
|
|
|
|
def _extract_tags(content: str) -> list[str]:
|
|
matches = re.findall(r"#[\w\u4e00-\u9fa5]+", content or "")
|
|
return list(dict.fromkeys(match.strip("#") for match in matches))
|
|
|
|
|
|
def _extract_attachments(content: str) -> list[dict]:
|
|
items = []
|
|
for match in re.finditer(r"!\[([^\]]*)\]\(([^)]+)\)", content or ""):
|
|
alt, url = match.group(1).strip(), match.group(2).strip()
|
|
if url:
|
|
items.append({"url": url, "alt": alt})
|
|
return items
|
|
|
|
|
|
def init_schema() -> None:
|
|
_ensure_dirs()
|
|
conn = _get_conn()
|
|
try:
|
|
conn.executescript(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS community_memos (
|
|
id TEXT PRIMARY KEY,
|
|
user_id TEXT NOT NULL,
|
|
content TEXT NOT NULL,
|
|
pinned INTEGER NOT NULL DEFAULT 0,
|
|
visibility TEXT NOT NULL DEFAULT 'PRIVATE',
|
|
state TEXT NOT NULL DEFAULT 'NORMAL',
|
|
created TEXT NOT NULL,
|
|
updated TEXT NOT NULL
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_memos_user ON community_memos(user_id);
|
|
CREATE INDEX IF NOT EXISTS idx_memos_state ON community_memos(state);
|
|
CREATE INDEX IF NOT EXISTS idx_memos_created ON community_memos(created);
|
|
CREATE INDEX IF NOT EXISTS idx_memos_pinned ON community_memos(pinned);
|
|
|
|
CREATE TABLE IF NOT EXISTS community_uploads (
|
|
id TEXT PRIMARY KEY,
|
|
user_id TEXT NOT NULL,
|
|
original_name TEXT NOT NULL,
|
|
filename TEXT NOT NULL UNIQUE,
|
|
content_type TEXT NOT NULL,
|
|
size INTEGER NOT NULL DEFAULT 0,
|
|
url TEXT NOT NULL,
|
|
created TEXT NOT NULL
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_uploads_user ON community_uploads(user_id);
|
|
CREATE INDEX IF NOT EXISTS idx_uploads_created ON community_uploads(created);
|
|
|
|
CREATE TABLE IF NOT EXISTS community_comments (
|
|
id TEXT PRIMARY KEY,
|
|
memo_id TEXT NOT NULL,
|
|
user_id TEXT NOT NULL,
|
|
content TEXT NOT NULL,
|
|
created TEXT NOT NULL,
|
|
updated TEXT NOT NULL,
|
|
FOREIGN KEY (memo_id) REFERENCES community_memos(id) ON DELETE CASCADE
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_comments_memo ON community_comments(memo_id, created);
|
|
CREATE INDEX IF NOT EXISTS idx_comments_user ON community_comments(user_id);
|
|
|
|
CREATE TABLE IF NOT EXISTS community_reactions (
|
|
id TEXT PRIMARY KEY,
|
|
memo_id TEXT NOT NULL,
|
|
user_id TEXT NOT NULL,
|
|
reaction_type TEXT NOT NULL,
|
|
created TEXT NOT NULL,
|
|
UNIQUE(memo_id, user_id, reaction_type),
|
|
FOREIGN KEY (memo_id) REFERENCES community_memos(id) ON DELETE CASCADE
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_reactions_memo ON community_reactions(memo_id);
|
|
CREATE INDEX IF NOT EXISTS idx_reactions_user ON community_reactions(user_id);
|
|
"""
|
|
)
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def _base_memo_from_row(row: sqlite3.Row) -> dict:
|
|
content = row["content"]
|
|
return {
|
|
"id": row["id"],
|
|
"user_id": row["user_id"],
|
|
"content": content,
|
|
"pinned": bool(row["pinned"]),
|
|
"visibility": row["visibility"] or "PRIVATE",
|
|
"state": row["state"] or "NORMAL",
|
|
"created": row["created"],
|
|
"updated": row["updated"],
|
|
"tags": _extract_tags(content),
|
|
"attachments": _extract_attachments(content),
|
|
}
|
|
|
|
|
|
def _comment_from_row(row: sqlite3.Row) -> dict:
|
|
return {
|
|
"id": row["id"],
|
|
"memo_id": row["memo_id"],
|
|
"user_id": row["user_id"],
|
|
"content": row["content"],
|
|
"created": row["created"],
|
|
"updated": row["updated"],
|
|
}
|
|
|
|
|
|
def _comment_count(conn: sqlite3.Connection, memo_id: str) -> int:
|
|
cursor = conn.execute("SELECT COUNT(*) FROM community_comments WHERE memo_id = ?", (memo_id,))
|
|
return int(cursor.fetchone()[0] or 0)
|
|
|
|
|
|
def _reaction_count(conn: sqlite3.Connection, memo_id: str) -> int:
|
|
cursor = conn.execute(
|
|
"SELECT COUNT(*) FROM community_reactions WHERE memo_id = ? AND reaction_type = ?",
|
|
(memo_id, UPVOTE_REACTION),
|
|
)
|
|
return int(cursor.fetchone()[0] or 0)
|
|
|
|
|
|
def _liked_by_user(conn: sqlite3.Connection, memo_id: str, viewer_id: str = "") -> bool:
|
|
if not viewer_id:
|
|
return False
|
|
cursor = conn.execute(
|
|
"""
|
|
SELECT 1
|
|
FROM community_reactions
|
|
WHERE memo_id = ? AND user_id = ? AND reaction_type = ?
|
|
LIMIT 1
|
|
""",
|
|
(memo_id, viewer_id, UPVOTE_REACTION),
|
|
)
|
|
return cursor.fetchone() is not None
|
|
|
|
|
|
def list_comments(memo_id: str) -> list[dict]:
|
|
uid = _normalize_memo_id(memo_id)
|
|
conn = _get_conn()
|
|
try:
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT id, memo_id, user_id, content, created, updated
|
|
FROM community_comments
|
|
WHERE memo_id = ?
|
|
ORDER BY created ASC
|
|
""",
|
|
(uid,),
|
|
).fetchall()
|
|
return [_comment_from_row(row) for row in rows]
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def _enrich_memo(conn: sqlite3.Connection, memo: dict, viewer_id: str = "", include_comments: bool = False) -> dict:
|
|
enriched = dict(memo)
|
|
enriched["comment_count"] = _comment_count(conn, memo["id"])
|
|
enriched["reaction_count"] = _reaction_count(conn, memo["id"])
|
|
enriched["liked_by_me"] = _liked_by_user(conn, memo["id"], viewer_id)
|
|
if include_comments:
|
|
enriched["comments"] = list_comments(memo["id"])
|
|
return enriched
|
|
|
|
|
|
def _to_memo_format(memo: dict) -> dict:
|
|
payload = {
|
|
"name": f"memos/{memo['id']}",
|
|
"id": memo["id"],
|
|
"creator": f"users/{memo['user_id']}",
|
|
"user_id": memo["user_id"],
|
|
"content": memo["content"],
|
|
"pinned": memo["pinned"],
|
|
"visibility": memo["visibility"],
|
|
"state": memo["state"],
|
|
"create_time": memo["created"],
|
|
"update_time": memo["updated"],
|
|
"display_time": memo["created"],
|
|
"tags": memo["tags"],
|
|
"attachments": memo["attachments"],
|
|
"comment_count": memo["comment_count"],
|
|
"reaction_count": memo["reaction_count"],
|
|
"liked_by_me": memo["liked_by_me"],
|
|
}
|
|
if "comments" in memo:
|
|
payload["comments"] = memo["comments"]
|
|
return payload
|
|
|
|
|
|
def _normalize_memo(memo: dict) -> dict:
|
|
payload = {
|
|
"id": memo["id"],
|
|
"user_id": memo["user_id"],
|
|
"content": memo["content"],
|
|
"pinned": memo["pinned"],
|
|
"visibility": memo["visibility"],
|
|
"state": memo["state"],
|
|
"created": memo["created"],
|
|
"updated": memo["updated"],
|
|
"tags": memo["tags"],
|
|
"attachments": memo["attachments"],
|
|
"comment_count": memo["comment_count"],
|
|
"reaction_count": memo["reaction_count"],
|
|
"liked_by_me": memo["liked_by_me"],
|
|
}
|
|
if "comments" in memo:
|
|
payload["comments"] = memo["comments"]
|
|
return payload
|
|
|
|
|
|
def _build_where_clause(state: str = "NORMAL", query: str = "") -> tuple[str, list[str]]:
|
|
clauses: list[str] = []
|
|
params: list[str] = []
|
|
|
|
normalized_state = (state or "NORMAL").upper()
|
|
if normalized_state != "ALL":
|
|
clauses.append("state = ?")
|
|
params.append(normalized_state)
|
|
|
|
if query:
|
|
lowered = query.lower()
|
|
clauses.append("LOWER(content) LIKE ?")
|
|
params.append(f"%{lowered}%")
|
|
|
|
if not clauses:
|
|
return "", params
|
|
return f"WHERE {' AND '.join(clauses)}", params
|
|
|
|
|
|
def list_memos(
|
|
page: int = 1,
|
|
per_page: int = 50,
|
|
state: str = "NORMAL",
|
|
order_by: str = "-pinned,-created",
|
|
memo_format: bool = False,
|
|
query: str = "",
|
|
viewer_id: str = "",
|
|
) -> dict:
|
|
conn = _get_conn()
|
|
try:
|
|
order_clause = "pinned DESC, created DESC"
|
|
if "created" in order_by and "pinned" not in order_by:
|
|
order_clause = "created DESC"
|
|
|
|
offset = (page - 1) * per_page
|
|
where_clause, params = _build_where_clause(state=state, query=query)
|
|
|
|
total = int(
|
|
conn.execute(f"SELECT COUNT(*) FROM community_memos {where_clause}", params).fetchone()[0] or 0
|
|
)
|
|
|
|
rows = conn.execute(
|
|
f"""
|
|
SELECT id, user_id, content, pinned, visibility, state, created, updated
|
|
FROM community_memos
|
|
{where_clause}
|
|
ORDER BY {order_clause}
|
|
LIMIT ? OFFSET ?
|
|
""",
|
|
[*params, per_page, offset],
|
|
).fetchall()
|
|
|
|
items = []
|
|
for row in rows:
|
|
enriched = _enrich_memo(conn, _base_memo_from_row(row), viewer_id=viewer_id)
|
|
items.append(_to_memo_format(enriched) if memo_format else _normalize_memo(enriched))
|
|
|
|
next_token = "" if len(items) < per_page else str(page + 1)
|
|
return {
|
|
"memos": items,
|
|
"items": items,
|
|
"totalItems": total,
|
|
"page": page,
|
|
"pageSize": per_page,
|
|
"next_page_token": next_token,
|
|
}
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def get_memo(memo_id: str, viewer_id: str = "", include_comments: bool = True) -> dict | None:
|
|
uid = _normalize_memo_id(memo_id)
|
|
conn = _get_conn()
|
|
try:
|
|
row = conn.execute(
|
|
"""
|
|
SELECT id, user_id, content, pinned, visibility, state, created, updated
|
|
FROM community_memos
|
|
WHERE id = ?
|
|
""",
|
|
(uid,),
|
|
).fetchone()
|
|
if not row:
|
|
return None
|
|
|
|
enriched = _enrich_memo(
|
|
conn,
|
|
_base_memo_from_row(row),
|
|
viewer_id=viewer_id,
|
|
include_comments=include_comments,
|
|
)
|
|
return _to_memo_format(enriched)
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def create_memo(
|
|
user_id: str,
|
|
content: str,
|
|
pinned: bool = False,
|
|
visibility: str = "PRIVATE",
|
|
state: str = "NORMAL",
|
|
) -> dict:
|
|
content = (content or "").strip()
|
|
if not content:
|
|
raise ValueError("content cannot be empty")
|
|
if len(content) > 10000:
|
|
raise ValueError("content is too long")
|
|
|
|
memo_id = uuid.uuid4().hex[:16]
|
|
now = _now_iso()
|
|
|
|
conn = _get_conn()
|
|
try:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO community_memos (id, user_id, content, pinned, visibility, state, created, updated)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(memo_id, user_id, content, 1 if pinned else 0, visibility, state, now, now),
|
|
)
|
|
conn.commit()
|
|
memo = _enrich_memo(
|
|
conn,
|
|
{
|
|
"id": memo_id,
|
|
"user_id": user_id,
|
|
"content": content,
|
|
"pinned": pinned,
|
|
"visibility": visibility,
|
|
"state": state,
|
|
"created": now,
|
|
"updated": now,
|
|
"tags": _extract_tags(content),
|
|
"attachments": _extract_attachments(content),
|
|
},
|
|
viewer_id=user_id,
|
|
)
|
|
return _to_memo_format(memo)
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def update_memo(
|
|
memo_id: str,
|
|
user_id: str,
|
|
pinned: Optional[bool] = None,
|
|
content: Optional[str] = None,
|
|
visibility: Optional[str] = None,
|
|
state: Optional[str] = None,
|
|
) -> dict | None:
|
|
uid = _normalize_memo_id(memo_id)
|
|
conn = _get_conn()
|
|
try:
|
|
row = conn.execute("SELECT * FROM community_memos WHERE id = ?", (uid,)).fetchone()
|
|
if not row:
|
|
return None
|
|
|
|
memo = _base_memo_from_row(row)
|
|
if memo["user_id"] != user_id:
|
|
return None
|
|
|
|
updates = []
|
|
params = []
|
|
|
|
if pinned is not None:
|
|
updates.append("pinned = ?")
|
|
params.append(1 if pinned else 0)
|
|
if content is not None:
|
|
content = content.strip()
|
|
if not content:
|
|
raise ValueError("content cannot be empty")
|
|
if len(content) > 10000:
|
|
raise ValueError("content is too long")
|
|
updates.append("content = ?")
|
|
params.append(content)
|
|
if visibility is not None:
|
|
updates.append("visibility = ?")
|
|
params.append(visibility)
|
|
if state is not None:
|
|
updates.append("state = ?")
|
|
params.append(state)
|
|
|
|
if not updates:
|
|
return _to_memo_format(_enrich_memo(conn, memo, viewer_id=user_id))
|
|
|
|
now = _now_iso()
|
|
updates.append("updated = ?")
|
|
params.append(now)
|
|
params.append(uid)
|
|
conn.execute(f"UPDATE community_memos SET {', '.join(updates)} WHERE id = ?", params)
|
|
conn.commit()
|
|
|
|
latest_content = content if content is not None else memo["content"]
|
|
memo.update(
|
|
{
|
|
"pinned": pinned if pinned is not None else memo["pinned"],
|
|
"content": latest_content,
|
|
"visibility": visibility if visibility is not None else memo["visibility"],
|
|
"state": state if state is not None else memo["state"],
|
|
"updated": now,
|
|
"tags": _extract_tags(latest_content),
|
|
"attachments": _extract_attachments(latest_content),
|
|
}
|
|
)
|
|
return _to_memo_format(_enrich_memo(conn, memo, viewer_id=user_id))
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def delete_memo(memo_id: str, user_id: str) -> bool:
|
|
uid = _normalize_memo_id(memo_id)
|
|
conn = _get_conn()
|
|
try:
|
|
row = conn.execute("SELECT user_id FROM community_memos WHERE id = ?", (uid,)).fetchone()
|
|
if not row or row["user_id"] != user_id:
|
|
return False
|
|
conn.execute("DELETE FROM community_memos WHERE id = ?", (uid,))
|
|
conn.commit()
|
|
return True
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def create_comment(memo_id: str, user_id: str, content: str) -> dict:
|
|
uid = _normalize_memo_id(memo_id)
|
|
content = (content or "").strip()
|
|
if not content:
|
|
raise ValueError("comment content cannot be empty")
|
|
if len(content) > 3000:
|
|
raise ValueError("comment content is too long")
|
|
|
|
conn = _get_conn()
|
|
try:
|
|
memo_exists = conn.execute("SELECT 1 FROM community_memos WHERE id = ?", (uid,)).fetchone()
|
|
if not memo_exists:
|
|
raise ValueError("memo does not exist")
|
|
|
|
comment_id = uuid.uuid4().hex[:16]
|
|
now = _now_iso()
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO community_comments (id, memo_id, user_id, content, created, updated)
|
|
VALUES (?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(comment_id, uid, user_id, content, now, now),
|
|
)
|
|
conn.commit()
|
|
return {
|
|
"id": comment_id,
|
|
"memo_id": uid,
|
|
"user_id": user_id,
|
|
"content": content,
|
|
"created": now,
|
|
"updated": now,
|
|
}
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def delete_comment(comment_id: str, user_id: str) -> bool:
|
|
conn = _get_conn()
|
|
try:
|
|
row = conn.execute(
|
|
"""
|
|
SELECT c.user_id AS comment_user_id, m.user_id AS memo_user_id
|
|
FROM community_comments c
|
|
JOIN community_memos m ON m.id = c.memo_id
|
|
WHERE c.id = ?
|
|
""",
|
|
(comment_id,),
|
|
).fetchone()
|
|
if not row:
|
|
return False
|
|
if row["comment_user_id"] != user_id and row["memo_user_id"] != user_id:
|
|
return False
|
|
conn.execute("DELETE FROM community_comments WHERE id = ?", (comment_id,))
|
|
conn.commit()
|
|
return True
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def toggle_reaction(memo_id: str, user_id: str, reaction_type: str = UPVOTE_REACTION) -> dict:
|
|
uid = _normalize_memo_id(memo_id)
|
|
reaction_type = (reaction_type or UPVOTE_REACTION).upper()
|
|
conn = _get_conn()
|
|
try:
|
|
memo_exists = conn.execute("SELECT 1 FROM community_memos WHERE id = ?", (uid,)).fetchone()
|
|
if not memo_exists:
|
|
raise ValueError("memo does not exist")
|
|
|
|
existing = conn.execute(
|
|
"""
|
|
SELECT id
|
|
FROM community_reactions
|
|
WHERE memo_id = ? AND user_id = ? AND reaction_type = ?
|
|
""",
|
|
(uid, user_id, reaction_type),
|
|
).fetchone()
|
|
|
|
liked = False
|
|
if existing:
|
|
conn.execute("DELETE FROM community_reactions WHERE id = ?", (existing["id"],))
|
|
else:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO community_reactions (id, memo_id, user_id, reaction_type, created)
|
|
VALUES (?, ?, ?, ?, ?)
|
|
""",
|
|
(uuid.uuid4().hex[:16], uid, user_id, reaction_type, _now_iso()),
|
|
)
|
|
liked = True
|
|
conn.commit()
|
|
|
|
return {
|
|
"memo_id": uid,
|
|
"reaction_type": reaction_type,
|
|
"liked": liked,
|
|
"reaction_count": _reaction_count(conn, uid),
|
|
}
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def save_upload(data: bytes, filename: str, content_type: str, user_id: str, base_url: str) -> dict:
|
|
_ensure_dirs()
|
|
ext = filename.rsplit(".", 1)[-1].lower() if "." in filename else "jpg"
|
|
if ext not in ("jpg", "jpeg", "png", "gif", "webp"):
|
|
ext = "jpg"
|
|
|
|
upload_id = uuid.uuid4().hex[:16]
|
|
stored_name = f"{int(time.time() * 1000)}_{upload_id[:8]}.{ext}"
|
|
path = UPLOADS_DIR / stored_name
|
|
path.write_bytes(data)
|
|
|
|
now = _now_iso()
|
|
url = f"{base_url.rstrip('/')}/community/files/{stored_name}"
|
|
|
|
conn = _get_conn()
|
|
try:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO community_uploads (id, user_id, original_name, filename, content_type, size, url, created)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(upload_id, user_id, filename, stored_name, content_type, len(data), url, now),
|
|
)
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
return {
|
|
"id": upload_id,
|
|
"url": url,
|
|
"markdown": f"",
|
|
"filename": stored_name,
|
|
"original_name": filename,
|
|
"content_type": content_type,
|
|
"size": len(data),
|
|
"created": now,
|
|
}
|
|
|
|
|
|
def list_uploads(page: int = 1, per_page: int = 24, user_id: str = "") -> dict:
|
|
offset = (page - 1) * per_page
|
|
conn = _get_conn()
|
|
try:
|
|
where_clause = ""
|
|
params: list[str] = []
|
|
if user_id:
|
|
where_clause = "WHERE user_id = ?"
|
|
params.append(user_id)
|
|
|
|
total = int(
|
|
conn.execute(f"SELECT COUNT(*) FROM community_uploads {where_clause}", params).fetchone()[0] or 0
|
|
)
|
|
rows = conn.execute(
|
|
f"""
|
|
SELECT id, user_id, original_name, filename, content_type, size, url, created
|
|
FROM community_uploads
|
|
{where_clause}
|
|
ORDER BY created DESC
|
|
LIMIT ? OFFSET ?
|
|
""",
|
|
[*params, per_page, offset],
|
|
).fetchall()
|
|
|
|
items = [
|
|
{
|
|
"id": row["id"],
|
|
"user_id": row["user_id"],
|
|
"filename": row["filename"],
|
|
"original_name": row["original_name"],
|
|
"content_type": row["content_type"],
|
|
"size": row["size"],
|
|
"url": row["url"],
|
|
"created": row["created"],
|
|
}
|
|
for row in rows
|
|
]
|
|
|
|
next_token = "" if len(items) < per_page else str(page + 1)
|
|
return {
|
|
"items": items,
|
|
"totalItems": total,
|
|
"page": page,
|
|
"pageSize": per_page,
|
|
"next_page_token": next_token,
|
|
}
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def get_stats(member_count: int | None = None) -> dict:
|
|
conn = _get_conn()
|
|
try:
|
|
totals = conn.execute(
|
|
"""
|
|
SELECT
|
|
COUNT(*) AS total_memos,
|
|
SUM(CASE WHEN state = 'NORMAL' THEN 1 ELSE 0 END) AS normal_memos,
|
|
SUM(CASE WHEN state = 'ARCHIVED' THEN 1 ELSE 0 END) AS archived_memos,
|
|
SUM(CASE WHEN pinned = 1 THEN 1 ELSE 0 END) AS pinned_memos,
|
|
COUNT(DISTINCT user_id) AS contributors
|
|
FROM community_memos
|
|
"""
|
|
).fetchone()
|
|
|
|
total_uploads = int(conn.execute("SELECT COUNT(*) FROM community_uploads").fetchone()[0] or 0)
|
|
total_comments = int(conn.execute("SELECT COUNT(*) FROM community_comments").fetchone()[0] or 0)
|
|
total_reactions = int(conn.execute("SELECT COUNT(*) FROM community_reactions").fetchone()[0] or 0)
|
|
|
|
today_prefix = _now_iso()[:10]
|
|
today_memos = int(
|
|
conn.execute("SELECT COUNT(*) FROM community_memos WHERE created LIKE ?", (f"{today_prefix}%",)).fetchone()[0]
|
|
or 0
|
|
)
|
|
|
|
top_authors_rows = conn.execute(
|
|
"""
|
|
SELECT user_id, COUNT(*) AS memo_count, MAX(created) AS latest_created
|
|
FROM community_memos
|
|
WHERE state = 'NORMAL'
|
|
GROUP BY user_id
|
|
ORDER BY memo_count DESC, latest_created DESC
|
|
LIMIT 6
|
|
"""
|
|
).fetchall()
|
|
top_authors = [
|
|
{
|
|
"user_id": row["user_id"],
|
|
"memo_count": row["memo_count"],
|
|
"latest_created": row["latest_created"],
|
|
}
|
|
for row in top_authors_rows
|
|
]
|
|
|
|
tag_counter: Counter[str] = Counter()
|
|
for row in conn.execute("SELECT content FROM community_memos WHERE state = 'NORMAL'").fetchall():
|
|
tag_counter.update(_extract_tags(row["content"]))
|
|
|
|
top_tags = [{"tag": tag, "count": count} for tag, count in tag_counter.most_common(10)]
|
|
|
|
latest_rows = conn.execute(
|
|
"""
|
|
SELECT id, user_id, content, pinned, visibility, state, created, updated
|
|
FROM community_memos
|
|
WHERE state = 'NORMAL'
|
|
ORDER BY pinned DESC, created DESC
|
|
LIMIT 3
|
|
"""
|
|
).fetchall()
|
|
latest_memos = [_normalize_memo(_enrich_memo(conn, _base_memo_from_row(row))) for row in latest_rows]
|
|
|
|
return {
|
|
"stats": {
|
|
"members": member_count,
|
|
"total_memos": int(totals["total_memos"] or 0),
|
|
"normal_memos": int(totals["normal_memos"] or 0),
|
|
"archived_memos": int(totals["archived_memos"] or 0),
|
|
"pinned_memos": int(totals["pinned_memos"] or 0),
|
|
"contributors": int(totals["contributors"] or 0),
|
|
"attachments": total_uploads,
|
|
"comments": total_comments,
|
|
"reactions": total_reactions,
|
|
"today_memos": today_memos,
|
|
},
|
|
"top_tags": top_tags,
|
|
"top_authors": top_authors,
|
|
"latest_memos": latest_memos,
|
|
}
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def get_upload_path(filename: str) -> Path | None:
|
|
if not filename or ".." in filename or "/" in filename or "\\" in filename:
|
|
return None
|
|
path = UPLOADS_DIR / filename
|
|
if not path.exists() or not path.is_file():
|
|
return None
|
|
return path
|