117 lines
2.9 KiB
Python
117 lines
2.9 KiB
Python
"""
|
|
Private community service layer.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
|
|
from ..config import BASE_URL
|
|
from ..db import community_sqlite as sqlite_db
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
try:
|
|
sqlite_db.init_schema()
|
|
logger.info("community: SQLite initialized at %s", sqlite_db.DB_PATH)
|
|
except Exception as exc:
|
|
logger.warning("community SQLite init failed: %s", exc)
|
|
|
|
|
|
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:
|
|
return sqlite_db.list_memos(
|
|
page=page,
|
|
per_page=per_page,
|
|
state=state,
|
|
order_by=order_by,
|
|
memo_format=memo_format,
|
|
query=query,
|
|
viewer_id=viewer_id,
|
|
)
|
|
|
|
|
|
def get_memo(memo_id: str, viewer_id: str = "", include_comments: bool = True) -> dict | None:
|
|
return sqlite_db.get_memo(memo_id, viewer_id=viewer_id, include_comments=include_comments)
|
|
|
|
|
|
def create_memo(
|
|
user_id: str,
|
|
content: str,
|
|
pinned: bool = False,
|
|
visibility: str = "PRIVATE",
|
|
state: str = "NORMAL",
|
|
) -> dict:
|
|
return sqlite_db.create_memo(
|
|
user_id=user_id,
|
|
content=content,
|
|
pinned=pinned,
|
|
visibility=visibility,
|
|
state=state,
|
|
)
|
|
|
|
|
|
def update_memo(
|
|
memo_id: str,
|
|
user_id: str,
|
|
pinned: bool | None = None,
|
|
content: str | None = None,
|
|
visibility: str | None = None,
|
|
state: str | None = None,
|
|
) -> dict | None:
|
|
return sqlite_db.update_memo(
|
|
memo_id,
|
|
user_id,
|
|
pinned=pinned,
|
|
content=content,
|
|
visibility=visibility,
|
|
state=state,
|
|
)
|
|
|
|
|
|
def delete_memo(memo_id: str, user_id: str) -> bool:
|
|
return sqlite_db.delete_memo(memo_id, user_id)
|
|
|
|
|
|
def create_comment(memo_id: str, user_id: str, content: str) -> dict:
|
|
return sqlite_db.create_comment(memo_id, user_id, content)
|
|
|
|
|
|
def list_comments(memo_id: str) -> list[dict]:
|
|
return sqlite_db.list_comments(memo_id)
|
|
|
|
|
|
def delete_comment(comment_id: str, user_id: str) -> bool:
|
|
return sqlite_db.delete_comment(comment_id, user_id)
|
|
|
|
|
|
def toggle_reaction(memo_id: str, user_id: str, reaction_type: str = sqlite_db.UPVOTE_REACTION) -> dict:
|
|
return sqlite_db.toggle_reaction(memo_id, user_id, reaction_type)
|
|
|
|
|
|
def upload_image(
|
|
data: bytes,
|
|
filename: str,
|
|
content_type: str,
|
|
user_id: str,
|
|
base_url: str = "",
|
|
) -> dict:
|
|
base = base_url or (os.getenv("BASE_URL") or BASE_URL or "http://localhost:8000")
|
|
return sqlite_db.save_upload(data, filename, content_type, user_id, base)
|
|
|
|
|
|
def list_uploads(page: int = 1, per_page: int = 24, user_id: str = "") -> dict:
|
|
return sqlite_db.list_uploads(page=page, per_page=per_page, user_id=user_id)
|
|
|
|
|
|
def get_stats(member_count: int | None = None) -> dict:
|
|
return sqlite_db.get_stats(member_count=member_count)
|