""" 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)