feat(notice): 公告已读状态（未读数 / 标记已读）

backend/app/alembic/versions/2026_11_03_0900_a1b2c3d4e5f6.py

@@ -0,0 +1,57 @@
+"""公告已读记录表
+
+Revision ID: a1b2c3d4e5f6
+Revises: e082c18b2f01
+Create Date: 2026-11-03 09:00:00
+
+「谁读过哪条公告」是按用户产生的事实，单独一张 sys_notice_read：
+``(user_id, notice_id)`` 唯一，重复标记幂等；公告/用户删除时级联清理。
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "a1b2c3d4e5f6"
+down_revision: Union[str, None] = "e082c18b2f01"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "sys_notice_read",
+        sa.Column("user_id", sa.Integer(), nullable=False, comment="用户ID"),
+        sa.Column("notice_id", sa.Integer(), nullable=False, comment="公告ID"),
+        sa.Column("read_at", sa.DateTime(timezone=True), nullable=True, comment="已读时间"),
+        sa.Column("id", sa.Integer(), autoincrement=True, nullable=False, comment="主键ID"),
+        sa.Column("uuid", sa.String(length=64), nullable=False, comment="UUID全局唯一标识"),
+        sa.Column("is_deleted", sa.Boolean(), nullable=False, comment="是否已删除(0:未删除 1:已删除)"),
+        sa.Column("created_time", sa.DateTime(timezone=True), nullable=False, comment="创建时间"),
+        sa.Column("updated_time", sa.DateTime(timezone=True), nullable=False, comment="更新时间"),
+        sa.Column("deleted_time", sa.DateTime(timezone=True), nullable=True, comment="删除时间"),
+        sa.ForeignKeyConstraint(["notice_id"], ["sys_notice.id"], name=op.f("fk_sys_notice_read_notice_id_sys_notice"), ondelete="CASCADE"),
+        sa.ForeignKeyConstraint(["user_id"], ["sys_user.id"], name=op.f("fk_sys_notice_read_user_id_sys_user"), ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_sys_notice_read")),
+        sa.UniqueConstraint("user_id", "notice_id", name="uq_sys_notice_read_user_notice"),
+        comment="公告已读记录表",
+    )
+    with op.batch_alter_table("sys_notice_read", schema=None) as batch_op:
+        batch_op.create_index(batch_op.f("ix_sys_notice_read_created_time"), ["created_time"], unique=False)
+        batch_op.create_index(batch_op.f("ix_sys_notice_read_id"), ["id"], unique=False)
+        batch_op.create_index(batch_op.f("ix_sys_notice_read_is_deleted"), ["is_deleted"], unique=False)
+        batch_op.create_index(batch_op.f("ix_sys_notice_read_notice_id"), ["notice_id"], unique=False)
+        batch_op.create_index(batch_op.f("ix_sys_notice_read_user_id"), ["user_id"], unique=False)
+        batch_op.create_index(batch_op.f("ix_sys_notice_read_uuid"), ["uuid"], unique=True)
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("sys_notice_read", schema=None) as batch_op:
+        batch_op.drop_index(batch_op.f("ix_sys_notice_read_uuid"))
+        batch_op.drop_index(batch_op.f("ix_sys_notice_read_user_id"))
+        batch_op.drop_index(batch_op.f("ix_sys_notice_read_notice_id"))
+        batch_op.drop_index(batch_op.f("ix_sys_notice_read_is_deleted"))
+        batch_op.drop_index(batch_op.f("ix_sys_notice_read_id"))
+        batch_op.drop_index(batch_op.f("ix_sys_notice_read_created_time"))
+    op.drop_table("sys_notice_read")

backend/app/modules/system/notice/controller.py

@@ -13,7 +13,7 @@
 from app.core.router_class import OperationLogRoute
 from app.core.sse_manager import SSE_QUEUE_MAX_SIZE
 
-from .schema import NoticeCreateSchema, NoticeOutSchema, NoticeQueryParam, NoticeUpdateSchema
+from .schema import NoticeCreateSchema, NoticeOutSchema, NoticeQueryParam, NoticeReadSchema, NoticeUnreadSchema, NoticeUpdateSchema
 from .service import NoticeService
 from .stream import notice_stream_manager
 
@@ -96,6 +96,37 @@ async def get_notice_list_available_controller(
     return SuccessResponse(data=result_dict.items, msg="查询已启用公告列表成功")
 
 
+@NoticeRouter.get("/unread-count", summary="我的未读公告数", response_model=ResponseSchema[NoticeUnreadSchema])
+async def notice_unread_count_controller(
+    auth: Annotated[AuthSchema, Security(get_current_user)],
+    db: Annotated[AsyncSession, Depends(db_getter)],
+) -> JSONResponse:
+    """当前用户的未读公告数（客户端红点）；登录即可，不要求运营端权限码。"""
+    unread = await NoticeService(auth, db).unread_count(user_id=auth.user.id)
+    return SuccessResponse(data=NoticeUnreadSchema(unread=unread), msg="获取未读公告数成功")
+
+
+@NoticeRouter.get("/read-ids", summary="我读过的公告ID", response_model=ResponseSchema[list[int]])
+async def notice_read_ids_controller(
+    auth: Annotated[AuthSchema, Security(get_current_user)],
+    db: Annotated[AsyncSession, Depends(db_getter)],
+) -> JSONResponse:
+    """当前用户已读的公告 ID 列表（列表页标未读）。"""
+    ids = await NoticeService(auth, db).read_ids(user_id=auth.user.id)
+    return SuccessResponse(data=ids, msg="获取已读公告ID成功")
+
+
+@NoticeRouter.post("/read", summary="标记公告已读", response_model=ResponseSchema[NoticeUnreadSchema])
+async def notice_mark_read_controller(
+    auth: Annotated[AuthSchema, Security(get_current_user)],
+    db: Annotated[AsyncSession, Depends(db_getter)],
+    data: Annotated[NoticeReadSchema, Body(description="公告ID列表，缺省表示全部")],
+) -> JSONResponse:
+    """标记公告已读：传 ``notice_ids`` 标指定公告，不传则全部容器内启用公告标为已读。"""
+    unread = await NoticeService(auth, db).mark_read(user_id=auth.user.id, notice_ids=data.notice_ids)
+    return SuccessResponse(data=NoticeUnreadSchema(unread=unread), msg="标记已读成功")
+
+
 @NoticeRouter.get("/stream", summary="通知实时流(SSE)", response_class=EventSourceResponse)
 async def notice_stream_controller(
     auth: Annotated[AuthSchema, Security(get_current_user)],

backend/app/modules/system/notice/model.py

@@ -1,4 +1,6 @@
-from sqlalchemy import Integer, String, Text
+from datetime import datetime
+
+from sqlalchemy import DateTime, ForeignKey, Integer, String, Text, UniqueConstraint
 from sqlalchemy.orm import Mapped, mapped_column
 
 from app.core.base_model import ModelMixin, UserMixin
@@ -15,3 +17,25 @@ class NoticeModel(ModelMixin, UserMixin):
     notice_content: Mapped[str | None] = mapped_column(Text, nullable=True, comment="公告内容")
     status: Mapped[int] = mapped_column(Integer, default=0, nullable=False, comment="状态(0:草稿 1:已发布 2:已归档)")
     description: Mapped[str | None] = mapped_column(Text, default=None, nullable=True, comment="备注")
+
+
+class NoticeReadModel(ModelMixin):
+    """公告已读记录（用户 × 公告 唯一）
+
+    公告是广播内容，本身没有“收件人”；「谁读过」是按人产生的事实，所以单独一张表。
+    重复标记由 ``(user_id, notice_id)`` 唯一约束兜底，服务层先查后插，避免无谓报错。
+    """
+
+    __tablename__: str = "sys_notice_read"
+    __table_args__ = (
+        UniqueConstraint("user_id", "notice_id", name="uq_sys_notice_read_user_notice"),
+        {"comment": "公告已读记录表"},
+    )
+
+    user_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("sys_user.id", ondelete="CASCADE"), nullable=False, index=True, comment="用户ID"
+    )
+    notice_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("sys_notice.id", ondelete="CASCADE"), nullable=False, index=True, comment="公告ID"
+    )
+    read_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, comment="已读时间")

backend/app/modules/system/notice/schema.py

@@ -57,3 +57,15 @@ class NoticeQueryParam(BaseQueryParam, UserByQueryParam):
     notice_title: str | None = Field(None, description="公告标题", json_schema_extra={"q": "like"})
     notice_type: str | None = Field(None, description="公告类型", json_schema_extra={"q": "eq"})
     status: int | None = Field(None, ge=0, le=2, description="状态(0:草稿 1:已发布 2:已归档)", json_schema_extra={"q": "eq"})
+
+
+class NoticeReadSchema(BaseModel):
+    """标记公告已读（不传 notice_ids 表示全部启用公告）"""
+
+    notice_ids: list[int] | None = Field(default=None, description="公告ID列表，缺省表示全部")
+
+
+class NoticeUnreadSchema(BaseModel):
+    """未读公告数"""
+
+    unread: int = Field(default=0, description="未读公告数")

backend/app/modules/system/notice/service.py

@@ -1,11 +1,16 @@
+from datetime import UTC, datetime
+
+from sqlalchemy import false, func, select
 from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import selectinload
 
 from app.core.base_schema import AuthSchema, BatchSetAvailable, PageResultSchema
 from app.core.exceptions import CustomException
 from app.core.logger import logger
 from app.utils.common_util import search_to_dict
 
 from .crud import NoticeCRUD
+from .model import NoticeModel, NoticeReadModel
 from .schema import NoticeCreateSchema, NoticeOutSchema, NoticeQueryParam, NoticeUpdateSchema
 from .stream import notice_stream_manager
 
@@ -89,15 +94,66 @@ async def page(
         )
 
     async def available_page(self) -> PageResultSchema[NoticeOutSchema]:
-        """获取已启用的公告（首页展示用，最多 10 条）"""
-        return await NoticeCRUD(self.auth, self.db).page(
-            offset=0,
-            limit=10,
-            order_by=[{"id": "asc"}],
-            search={"status": ("eq", 0)},
-            out_schema=NoticeOutSchema,
+        """获取已启用的公告（首页展示用，最多 10 条）
+
+        **公开内容必须绕过 CRUD 的数据权限**：``CRUDBase`` 会给非超管叠加
+        ``created_id = 当前用户`` 的数据范围条件，公告由运营创建，普通用户据此查询会一条都看不到。
+        这里显式直查；``created_by/updated_by`` 需预加载，否则异步序列化触发 MissingGreenlet。
+        """
+        conditions = self._enabled_condition()
+        total = int((await self.db.scalar(select(func.count(NoticeModel.id)).where(*conditions))) or 0)
+        rows = (
+            await self.db.scalars(
+                select(NoticeModel)
+                .where(*conditions)
+                .order_by(NoticeModel.id.asc())
+                .limit(10)
+                .options(selectinload(NoticeModel.created_by), selectinload(NoticeModel.updated_by))
+            )
+        ).all()
+        return PageResultSchema(
+            page_no=1, page_size=10, total=total, has_next=False, items=[NoticeOutSchema.model_validate(r) for r in rows]
         )
 
+    @staticmethod
+    def _enabled_condition() -> tuple:
+        """对外可见的公告：启用(0) 且未删除。
+
+        ``available_page`` 走 ``NoticeCRUD.page``，会被基座的 created_id 数据范围再过滤一层，
+        所以这里做「未读数 / 全部已读」时统一用直查，保证口径与 C 端看到的一致。
+        """
+        return (NoticeModel.status == 0, NoticeModel.is_deleted == false())
+
+    async def unread_count(self, user_id: int) -> int:
+        """未读公告数 = 启用公告总数 − 我读过的（已读集合里若公告已停用则不参与相减）"""
+        read_ids = select(NoticeReadModel.notice_id).where(NoticeReadModel.user_id == user_id)
+        stmt = select(func.count(NoticeModel.id)).where(*self._enabled_condition(), NoticeModel.id.not_in(read_ids))
+        return int((await self.db.scalar(stmt)) or 0)
+
+    async def read_ids(self, user_id: int) -> list[int]:
+        """我已读的公告 ID（供列表页判断哪条未读）"""
+        rows = await self.db.scalars(select(NoticeReadModel.notice_id).where(NoticeReadModel.user_id == user_id))
+        return [int(x) for x in rows.all()]
+
+    async def mark_read(self, user_id: int, notice_ids: list[int] | None = None) -> int:
+        """标记已读（不传 notice_ids 表示全部启用公告），返回剩余未读数。
+
+        只记录仍然存在的启用公告；已存在则跳过，重复调用是幂等的。
+        """
+        stmt = select(NoticeModel.id).where(*self._enabled_condition())
+        if notice_ids:
+            stmt = stmt.where(NoticeModel.id.in_(notice_ids))
+        target_ids = [int(x) for x in (await self.db.scalars(stmt)).all()]
+        if target_ids:
+            existing = set(await self.read_ids(user_id))
+            now = datetime.now(UTC)
+            for notice_id in target_ids:
+                if notice_id in existing:
+                    continue
+                self.db.add(NoticeReadModel(user_id=user_id, notice_id=notice_id, read_at=now))
+            await self.db.flush()
+        return await self.unread_count(user_id)
+
     async def create(self, data: NoticeCreateSchema) -> NoticeOutSchema:
         """创建公告
 

backend/tests/conftest.py

@@ -23,10 +23,22 @@
 # 测试环境变量
 # ============================================================
 
+# 测试库类型：默认 sqlite（开箱即跑）；设置 TEST_DB_TYPE=postgres 可切到真实 PostgreSQL
+# ——SQLite 与 PG 在类型/约束/迁移上的行为并不一致，涉及迁移或数据库特有语义时请在 PG 上验。
+_TEST_DB_TYPE = os.getenv("TEST_DB_TYPE", "sqlite").lower()
 _TEST_DB_PATH = tempfile.NamedTemporaryFile(suffix=".db", delete=False).name
 
-os.environ["DATABASE_TYPE"] = "sqlite"
-os.environ["DATABASE_NAME"] = _TEST_DB_PATH
+if _TEST_DB_TYPE == "postgres":
+    os.environ["DATABASE_TYPE"] = "postgres"
+    os.environ["DATABASE_HOST"] = os.getenv("TEST_DB_HOST", "127.0.0.1")
+    os.environ["DATABASE_PORT"] = os.getenv("TEST_DB_PORT", "5432")
+    os.environ["DATABASE_USER"] = os.getenv("TEST_DB_USER", "fastapiadmin")
+    os.environ["DATABASE_PASSWORD"] = os.getenv("TEST_DB_PASSWORD", "")
+    os.environ["DATABASE_NAME"] = os.getenv("TEST_DB_NAME", "fastapiadmin_test")
+else:
+    os.environ["DATABASE_TYPE"] = "sqlite"
+    os.environ["DATABASE_NAME"] = _TEST_DB_PATH
+
 os.environ["POOL_SIZE"] = "1"
 os.environ["MAX_OVERFLOW"] = "1"
 os.environ["SECRET_KEY"] = "unit-test-secret-key-not-a-real-secret"  # Settings 必填项，固定注入保证测试确定性
@@ -39,8 +51,15 @@
 
 from app.config.setting import settings
 
-settings.DATABASE_TYPE = "sqlite"
-settings.DATABASE_NAME = _TEST_DB_PATH
+settings.DATABASE_TYPE = _TEST_DB_TYPE
+if _TEST_DB_TYPE == "postgres":
+    settings.DATABASE_HOST = os.environ["DATABASE_HOST"]
+    settings.DATABASE_PORT = int(os.environ["DATABASE_PORT"])
+    settings.DATABASE_USER = os.environ["DATABASE_USER"]
+    settings.DATABASE_PASSWORD = os.environ["DATABASE_PASSWORD"]
+    settings.DATABASE_NAME = os.environ["DATABASE_NAME"]
+else:
+    settings.DATABASE_NAME = _TEST_DB_PATH
 settings.POOL_SIZE = 1
 settings.MAX_OVERFLOW = 1
 settings.CAPTCHA_ENABLE = False  # 测试环境关闭验证码

backend/tests/test_notice_read_state.py

@@ -0,0 +1,73 @@
+"""公告已读状态：未读数、按条标记、全部标记、按用户隔离。
+
+口径：公告状态 0 = 启用（对外可见）、1 = 停用；未读数只统计启用公告，
+已读按用户存储，重复标记幂等。
+
+断言一律用**增量**而不是绝对值：测试库可能承接多次运行的数据（PostgreSQL 不像 sqlite
+每次都重建），`/notice/available` 也固定最多返回 10 条。
+"""
+
+import uuid
+
+from fastapi.testclient import TestClient
+
+
+def _create_notice(client: TestClient, headers: dict, title: str, status: int = 0) -> int:
+    unique_title = f"{title}-{uuid.uuid4().hex[:8]}"
+    resp = client.post(
+        "/system/notice/create",
+        json={"notice_title": unique_title, "notice_type": "1", "notice_content": f"{unique_title} 正文", "status": status},
+        headers=headers,
+    )
+    assert resp.json()["success"] is True, resp.text
+    return resp.json()["data"]["id"]
+
+
+def _login(client: TestClient, username: str = "user", password: str = "123456") -> dict:
+    resp = client.post("/system/auth/login", data={"username": username, "password": password})
+    assert resp.status_code == 200, resp.text
+    return {"Authorization": f"Bearer {resp.json()['data']['access_token']}"}
+
+
+def _unread(client: TestClient, headers: dict) -> int:
+    resp = client.get("/system/notice/unread-count", headers=headers)
+    assert resp.json()["success"] is True, resp.text
+    return resp.json()["data"]["unread"]
+
+
+class TestNoticeReadState:
+    def test_unread_count_and_mark_read(self, test_client: TestClient, auth_headers: dict) -> None:
+        user_h = _login(test_client)
+        before = _unread(test_client, user_h)
+
+        first = _create_notice(test_client, auth_headers, "已读测试-A")
+        _create_notice(test_client, auth_headers, "已读测试-B")
+        # 停用的不计入未读
+        _create_notice(test_client, auth_headers, "已读测试-停用", status=1)
+        assert _unread(test_client, user_h) == before + 2
+
+        # 按条标记：未读 -1，已读 ID 里能看到它
+        resp = test_client.post("/system/notice/read", json={"notice_ids": [first]}, headers=user_h)
+        assert resp.json()["data"]["unread"] == before + 1
+        assert first in test_client.get("/system/notice/read-ids", headers=user_h).json()["data"]
+
+        # 重复标记幂等
+        again = test_client.post("/system/notice/read", json={"notice_ids": [first]}, headers=user_h)
+        assert again.json()["data"]["unread"] == before + 1
+
+        # 全部标记 → 0
+        assert test_client.post("/system/notice/read", json={}, headers=user_h).json()["data"]["unread"] == 0
+
+    def test_read_state_is_per_user(self, test_client: TestClient, auth_headers: dict) -> None:
+        """已读按用户隔离：我标记不影响别人的未读数。"""
+        user_a = _login(test_client)
+        other = _login(test_client, "admin", "admin123")
+        notice_id = _create_notice(test_client, auth_headers, "已读测试-隔离")
+
+        other_before = _unread(test_client, other)
+        test_client.post("/system/notice/read", json={"notice_ids": [notice_id]}, headers=user_a)
+        assert _unread(test_client, other) == other_before
+
+    def test_mark_read_requires_login(self, test_client: TestClient) -> None:
+        assert test_client.get("/system/notice/unread-count").status_code == 401
+        assert test_client.post("/system/notice/read", json={}).status_code == 401

frontend/app/src/api/module_system/notice.ts

@@ -28,6 +28,19 @@ export const NoticeAPI = {
   batchStatus(data: BatchSetStatus): Promise<void> {
     return http.Patch(`${SYSTEM_BASE}/notice/status/batch`, data)
   },
+  /** 我的未读公告数（首页公告条红点） */
+  getUnreadCount(): Promise<{ unread: number }> {
+    return http.Get(`${SYSTEM_BASE}/notice/unread-count`)
+  },
+  /** 我已读的公告 id（列表标未读） */
+  getReadIds(): Promise<number[]> {
+    return http.Get(`${SYSTEM_BASE}/notice/read-ids`)
+  },
+  /** 标记已读（不传 noticeIds 表示全部），返回剩余未读数 */
+  markRead(noticeIds?: number[]): Promise<{ unread: number }> {
+    return http.Post(`${SYSTEM_BASE}/notice/read`, noticeIds ? { notice_ids: noticeIds } : {})
+  },
+
   getAvailable(): Promise<NoticeItem[]> {
     return http.Get(`${SYSTEM_BASE}/notice/available`)
   },

frontend/app/src/locales/langs/en.json

@@ -238,7 +238,10 @@
     "contentPlaceholder": "Please enter content",
     "type": "Notice Type",
     "notice": "Notice",
-    "announce": "Announcement"
+    "announce": "Announcement",
+    "unreadCount": "{count} unread",
+    "markAllRead": "Mark all read",
+    "markAllReadSuccess": "All marked as read"
   },
   "chat": {
     "title": "AI Assistant",