diff --git a/backend/package/yuxi/repositories/channels/__init__.py b/backend/package/yuxi/repositories/channels/__init__.py index 09167de5..d6753fe2 100644 --- a/backend/package/yuxi/repositories/channels/__init__.py +++ b/backend/package/yuxi/repositories/channels/__init__.py @@ -1,10 +1,10 @@ """渠道仓储层聚合导出。 本包提供： -- 11 个 ``Channel*Repository`` / ``ChannelConversationRepository`` / +- 10 个 ``Channel*Repository`` / ``ChannelConversationRepository`` / ``ChannelMessageRepository`` 实现类 （直接操作 ORM Model，原子 CRUD，不含业务语义） -- ``Repositories`` 聚合 dataclass（共享 ``db`` 会话的 11 个仓储实例） +- ``Repositories`` 聚合 dataclass（共享 ``db`` 会话的 10 个仓储实例） - ``create_repositories(db)`` factory 函数（由框架层在请求开始时调用） 依赖边界：只依赖 ``yuxi.storage.postgres.models_channels`` / @@ -30,7 +30,6 @@ from .channel_idempotency_repository import ChannelIdempotencyRepository from .channel_message_repository import ChannelMessageRepository from .channel_outbox_repository import ChannelOutboxRepository from .channel_pairing_repository import ChannelPairingRepository -from .channel_report_repository import ChannelReportRepository from .channel_session_repository import ChannelSessionRepository from .content_review_record_repository import ChannelContentReviewRecordRepository from .route_binding_repository import ChannelRouteBindingRepository @@ -46,7 +45,6 @@ __all__ = [ "ChannelMessageRepository", "ChannelOutboxRepository", "ChannelPairingRepository", - "ChannelReportRepository", "ChannelRouteBindingRepository", "ChannelSessionRepository", "Repositories", @@ -72,7 +70,6 @@ class Repositories: idempotency: ChannelIdempotencyRepository conversation: ChannelConversationRepository message: ChannelMessageRepository - report: ChannelReportRepository content_review_record: ChannelContentReviewRecordRepository route_binding: ChannelRouteBindingRepository @@ -94,7 +91,6 @@ def create_repositories(db: AsyncSession) -> Repositories: idempotency=ChannelIdempotencyRepository(db), conversation=ChannelConversationRepository(db), message=ChannelMessageRepository(db), - report=ChannelReportRepository(db), content_review_record=ChannelContentReviewRecordRepository(db), route_binding=ChannelRouteBindingRepository(db), ) diff --git a/backend/package/yuxi/repositories/channels/channel_account_repository.py b/backend/package/yuxi/repositories/channels/channel_account_repository.py index d91b3e09..6a8e52c2 100644 --- a/backend/package/yuxi/repositories/channels/channel_account_repository.py +++ b/backend/package/yuxi/repositories/channels/channel_account_repository.py @@ -39,14 +39,11 @@ class ChannelAccountRepository(BaseRepository): account_id=data["account_id"], display_name=data["display_name"], config=data.get("config") or {}, - capabilities=data.get("capabilities") or {}, enabled=data.get("enabled", True), status=data.get("status", "active"), plugin_status=data.get("plugin_status", "stopped"), - config_version=data.get("config_version", 1), last_error=data.get("last_error"), last_error_at=data.get("last_error_at"), - last_message_at=data.get("last_message_at"), service_user_uid=data.get("service_user_uid"), version=1, created_by=data.get("created_by"), @@ -156,15 +153,12 @@ class ChannelAccountRepository(BaseRepository): updatable_fields = ( "display_name", "config", - "capabilities", "enabled", "status", "plugin_status", - "config_version", "last_error", "last_error_at", "last_health_check_at", - "last_message_at", "updated_by", ) for field in updatable_fields: @@ -234,27 +228,6 @@ class ChannelAccountRepository(BaseRepository): await self.db.flush() return result.rowcount - async def update_last_message_at( - self, - record_id: int, - *, - commit: bool = True, - ) -> int: - """按主键更新最近消息时间（冗余字段，FR-35 健康检查）。""" - now = utc_now_naive() - stmt = ( - update(ChannelAccount) - .where(ChannelAccount.id == record_id) - .where(self._not_deleted()) - .values(last_message_at=now, updated_at=now) - ) - result = await self.db.execute(stmt) - if commit: - await self.db.commit() - else: - await self.db.flush() - return result.rowcount - async def update_last_health_check_at( self, record_id: int, diff --git a/backend/package/yuxi/repositories/channels/channel_idempotency_repository.py b/backend/package/yuxi/repositories/channels/channel_idempotency_repository.py index 48184812..bd95ecaa 100644 --- a/backend/package/yuxi/repositories/channels/channel_idempotency_repository.py +++ b/backend/package/yuxi/repositories/channels/channel_idempotency_repository.py @@ -41,8 +41,6 @@ class ChannelIdempotencyRepository(BaseRepository): record = ChannelIdempotency( idempotency_key=data["idempotency_key"], operation=data["operation"], - account_id=data.get("account_id"), - request_hash=data.get("request_hash"), status="in_progress", in_progress_started_at=now, response_body=data.get("response_body"), @@ -63,6 +61,7 @@ class ChannelIdempotencyRepository(BaseRepository): status: str, *, response_body: dict[str, Any] | None = None, + updated_by: str | None = None, commit: bool = True, ) -> int: """更新幂等记录状态（状态机流转，不加载 ORM 实例）。 @@ -75,6 +74,7 @@ class ChannelIdempotencyRepository(BaseRepository): record_id: 幂等记录主键。 status: 目标状态（completed / failed）。 response_body: 首次请求的响应体（仅 completed 时填充）。 + updated_by: 状态变更操作人（审计用）。 commit: True 时提交事务，False 时仅 flush。 Returns: @@ -87,6 +87,8 @@ class ChannelIdempotencyRepository(BaseRepository): } if response_body is not None: values["response_body"] = response_body + if updated_by is not None: + values["updated_by"] = updated_by stmt = update(ChannelIdempotency).where(ChannelIdempotency.id == record_id).values(**values) result = await self.db.execute(stmt) if commit: @@ -99,14 +101,6 @@ class ChannelIdempotencyRepository(BaseRepository): # 查询 # ------------------------------------------------------------------ - async def get_by_id(self, record_id: int, *, for_update: bool = False) -> ChannelIdempotency | None: - """根据主键 ID 获取幂等记录。""" - stmt = select(ChannelIdempotency).where(ChannelIdempotency.id == record_id) - if for_update: - stmt = stmt.with_for_update() - result = await self.db.execute(stmt) - return result.scalar_one_or_none() - async def get_by_key(self, idempotency_key: str, *, for_update: bool = False) -> ChannelIdempotency | None: """根据幂等键获取记录（重复请求回放）。""" stmt = select(ChannelIdempotency).where(ChannelIdempotency.idempotency_key == idempotency_key) @@ -115,28 +109,6 @@ class ChannelIdempotencyRepository(BaseRepository): result = await self.db.execute(stmt) return result.scalar_one_or_none() - async def list_in_progress_stale( - self, - before: dt.datetime, - *, - limit: int = 100, - ) -> list[ChannelIdempotency]: - """列出超时的 in_progress 记录（worker 崩溃回收）。 - - Args: - before: 截止时间，in_progress_started_at 早于此时间的记录将被列出。 - limit: 返回数量上限。 - """ - stmt = ( - select(ChannelIdempotency) - .where(ChannelIdempotency.status == "in_progress") - .where(ChannelIdempotency.in_progress_started_at < before) - .order_by(ChannelIdempotency.in_progress_started_at.asc()) - .limit(limit) - ) - result = await self.db.execute(stmt) - return list(result.scalars().all()) - # ------------------------------------------------------------------ # 清理（物理删除，不使用软删除） # ------------------------------------------------------------------ @@ -147,12 +119,11 @@ class ChannelIdempotencyRepository(BaseRepository): *, commit: bool = True, ) -> int: - """物理删除早于指定时间的幂等记录（定时清理任务）。 + """物理删除已过期的幂等记录（定时清理任务）。 - 幂等记录不使用软删除，过期后直接物理删除。 - 默认保留 24 小时（IDEMPOTENCY_DEFAULT_TTL_HOURS）。 + 按 ``expires_at`` 判断过期，幂等记录不使用软删除，过期后直接物理删除。 """ - stmt = sa_delete(ChannelIdempotency).where(ChannelIdempotency.created_at < before) + stmt = sa_delete(ChannelIdempotency).where(ChannelIdempotency.expires_at < before) result = await self.db.execute(stmt) if commit: await self.db.commit() diff --git a/backend/package/yuxi/repositories/channels/channel_outbox_repository.py b/backend/package/yuxi/repositories/channels/channel_outbox_repository.py index ea8c97b7..72f92f9d 100644 --- a/backend/package/yuxi/repositories/channels/channel_outbox_repository.py +++ b/backend/package/yuxi/repositories/channels/channel_outbox_repository.py @@ -257,8 +257,6 @@ class ChannelOutboxRepository(BaseRepository): next_retry_at: dt.datetime | None = None, last_retry_at: dt.datetime | None = None, sent_at: dt.datetime | None = None, - release_lock: bool = False, - worker_id: str | None = None, updated_by: str | None = None, commit: bool = True, ) -> int: @@ -275,8 +273,6 @@ class ChannelOutboxRepository(BaseRepository): next_retry_at: 下次重试时间（指数退避）。 last_retry_at: 上次重试时间（用于诊断退避进度与 SLA）。 sent_at: 首次成功投递时间（用于计算投递延迟）。 - release_lock: True 时清空 locked_by / locked_at（投递完成后释放租约）。 - worker_id: 指定 worker_id 时仅释放该 worker 持有的锁（防止误释放他人锁）。 updated_by: 操作人。 commit: True 时提交事务，False 时仅 flush。 @@ -301,12 +297,7 @@ class ChannelOutboxRepository(BaseRepository): values["last_retry_at"] = last_retry_at if sent_at is not None: values["sent_at"] = sent_at - if release_lock: - values["locked_by"] = None - values["locked_at"] = None stmt = update(ChannelOutboxEntry).where(ChannelOutboxEntry.id == record_id).where(self._not_deleted()) - if release_lock and worker_id is not None: - stmt = stmt.where(ChannelOutboxEntry.locked_by == worker_id) stmt = stmt.values(**values) result = await self.db.execute(stmt) if commit: @@ -315,74 +306,6 @@ class ChannelOutboxRepository(BaseRepository): await self.db.flush() return result.rowcount - async def acquire_lock( - self, - record_id: int, - worker_id: str, - *, - commit: bool = True, - ) -> int: - """原子抢占投递锁（at-least-once 语义）。 - - 仅当 ``locked_by`` 为空时才能抢占成功，保证同一时刻只有一个 worker - 处理该条目。 - - Args: - record_id: 发件箱条目主键。 - worker_id: 抢占方 worker 标识。 - commit: True 时提交事务，False 时仅 flush。 - - Returns: - 受影响行数（1 = 抢占成功，0 = 已被其他 worker 持有）。 - """ - now = utc_now_naive() - stmt = ( - update(ChannelOutboxEntry) - .where(ChannelOutboxEntry.id == record_id) - .where(ChannelOutboxEntry.locked_by.is_(None)) - .where(self._not_deleted()) - .values(locked_by=worker_id, locked_at=now, updated_at=now) - ) - result = await self.db.execute(stmt) - if commit: - await self.db.commit() - else: - await self.db.flush() - return result.rowcount - - async def release_lock( - self, - record_id: int, - *, - worker_id: str | None = None, - commit: bool = True, - ) -> int: - """释放投递锁（清空 locked_by / locked_at）。 - - Args: - record_id: 发件箱条目主键。 - worker_id: 指定 worker_id 时仅释放该 worker 持有的锁（防止误释放他人锁）。 - commit: True 时提交事务，False 时仅 flush。 - - Returns: - 受影响行数。 - """ - now = utc_now_naive() - stmt = ( - update(ChannelOutboxEntry) - .where(ChannelOutboxEntry.id == record_id) - .where(self._not_deleted()) - .values(locked_by=None, locked_at=None, updated_at=now) - ) - if worker_id is not None: - stmt = stmt.where(ChannelOutboxEntry.locked_by == worker_id) - result = await self.db.execute(stmt) - if commit: - await self.db.commit() - else: - await self.db.flush() - return result.rowcount - async def list_by_message_id(self, message_id: int) -> list[ChannelOutboxEntry]: """按消息 ID 查询投递状态（FR-22 状态回查）。""" stmt = ( diff --git a/backend/package/yuxi/repositories/channels/channel_report_repository.py b/backend/package/yuxi/repositories/channels/channel_report_repository.py deleted file mode 100644 index 7fa7dedf..00000000 --- a/backend/package/yuxi/repositories/channels/channel_report_repository.py +++ /dev/null @@ -1,294 +0,0 @@ -# yuxi/repositories/channels/channel_report_repository.py -"""渠道报告数据访问层。""" - -from __future__ import annotations - -from datetime import datetime -from typing import Any - -from sqlalchemy import delete, func, select - -from yuxi.repositories.channels.base import BaseRepository -from yuxi.storage.postgres.models_channels import ChannelReport - - -class ChannelReportRepository(BaseRepository): - """``channel_reports`` 表的异步数据访问封装。 - - 报告记录由 scheduler handler 异步生成后写入，状态机为 - ``pending → generating → ready / failed``，终态不可变更。 - """ - - model = ChannelReport - - # ------------------------------------------------------------------ - # 写操作 - # ------------------------------------------------------------------ - - async def create(self, data: dict[str, Any], *, commit: bool = True) -> ChannelReport: - """创建报告记录。 - - Args: - data: 报告字段字典，必须包含 report_id / task_id / report_type。 - commit: True 时提交事务，False 时仅 flush。 - """ - record = ChannelReport( - report_id=data["report_id"], - task_id=data["task_id"], - report_type=data["report_type"], - status=data.get("status", "pending"), - params=data.get("params") or {}, - content=data.get("content"), - error_message=data.get("error_message"), - created_by=data.get("created_by", "system"), - retried_from=data.get("retried_from"), - retried_at=data.get("retried_at"), - ) - self.db.add(record) - if commit: - await self.db.commit() - else: - await self.db.flush() - await self.db.refresh(record) - return record - - async def update_status( - self, - report_id: str, - *, - status: str | None = None, - content: dict[str, Any] | None = None, - error_message: str | None = None, - ready_at: datetime | None = None, - commit: bool = True, - ) -> ChannelReport | None: - """按 report_id 更新报告状态字段；记录不存在返回 None。 - - 仅更新提供的字段（局部更新），用于 ``updateReportStatus`` 用例。 - 与子包内其他 ``update`` 方法一致，无条件 ``refresh`` 以返回最新状态。 - """ - stmt = select(ChannelReport).where( - ChannelReport.report_id == report_id, - self._not_deleted(), - ) - result = await self.db.execute(stmt) - orm = result.scalar_one_or_none() - if orm is None: - return None - if status is not None: - orm.status = status - if content is not None: - orm.content = content - if error_message is not None: - orm.error_message = error_message - if ready_at is not None: - orm.ready_at = ready_at - if commit: - await self.db.commit() - else: - await self.db.flush() - await self.db.refresh(orm) - return orm - - async def update_fields( - self, - report_id: str, - data: dict[str, Any], - *, - commit: bool = True, - ) -> ChannelReport | None: - """按 report_id 完整更新报告的可变字段；记录不存在返回 None。 - - 用于重试场景标记原报告的 ``retried_at`` 字段。``report_id`` / - ``task_id`` 作为业务标识不更新。与子包内其他 ``update`` 方法一致， - 无条件 ``refresh`` 以返回最新状态。 - """ - stmt = select(ChannelReport).where( - ChannelReport.report_id == report_id, - self._not_deleted(), - ) - result = await self.db.execute(stmt) - orm = result.scalar_one_or_none() - if orm is None: - return None - updatable_fields = ( - "status", - "content", - "error_message", - "ready_at", - "retried_from", - "retried_at", - ) - for field in updatable_fields: - if field in data: - setattr(orm, field, data[field]) - if commit: - await self.db.commit() - else: - await self.db.flush() - await self.db.refresh(orm) - return orm - - # ------------------------------------------------------------------ - # 查询 - # ------------------------------------------------------------------ - - async def get_by_report_id(self, report_id: str, *, for_update: bool = False) -> ChannelReport | None: - """根据业务标识 report_id 获取报告（排除已软删除）。""" - stmt = select(ChannelReport).where( - ChannelReport.report_id == report_id, - self._not_deleted(), - ) - if for_update: - stmt = stmt.with_for_update() - result = await self.db.execute(stmt) - return result.scalar_one_or_none() - - async def get_by_task_id(self, task_id: str, *, for_update: bool = False) -> ChannelReport | None: - """根据 scheduler task_id 获取报告（排除已软删除）。""" - stmt = select(ChannelReport).where( - ChannelReport.task_id == task_id, - self._not_deleted(), - ) - if for_update: - stmt = stmt.with_for_update() - result = await self.db.execute(stmt) - return result.scalar_one_or_none() - - async def list( - self, - *, - status: str | None = None, - report_type: str | None = None, - start_time: datetime | None = None, - end_time: datetime | None = None, - page: int | None = None, - page_size: int | None = None, - limit: int | None = None, - offset: int | None = None, - ) -> list[ChannelReport]: - """列出报告，按 created_at 倒序，支持过滤与分页。 - - 分页参数与子包内其他仓储一致：优先使用 ``page`` / ``page_size``， - 其次使用 ``limit`` / ``offset``（由 ``_resolve_pagination`` 解析）。 - """ - stmt = self._build_query( - status=status, - report_type=report_type, - start_time=start_time, - end_time=end_time, - ) - stmt = stmt.order_by(ChannelReport.created_at.desc()) - effective_limit, effective_offset = self._resolve_pagination(page, page_size, limit, offset) - if effective_limit is not None: - stmt = stmt.limit(effective_limit).offset(effective_offset) - result = await self.db.execute(stmt) - return list(result.scalars().all()) - - async def count( - self, - *, - status: str | None = None, - report_type: str | None = None, - start_time: datetime | None = None, - end_time: datetime | None = None, - ) -> int: - """统计满足过滤条件的报告数量。""" - stmt = self._build_query( - status=status, - report_type=report_type, - start_time=start_time, - end_time=end_time, - select_count=True, - ) - result = await self.db.execute(stmt) - return int(result.scalar() or 0) - - async def list_stale_generating( - self, - before: datetime, - *, - limit: int = 100, - ) -> list[ChannelReport]: - """列出长时间停留在 generating 状态的陈旧报告。 - - Args: - before: 阈值，updated_at 早于此时间的 generating 报告将被列出。 - limit: 返回数量上限。 - """ - stmt = ( - select(ChannelReport) - .where( - ChannelReport.status == "generating", - ChannelReport.updated_at < before, - self._not_deleted(), - ) - .order_by(ChannelReport.updated_at.asc()) - .limit(limit) - ) - result = await self.db.execute(stmt) - return list(result.scalars().all()) - - # ------------------------------------------------------------------ - # 清理（物理删除，配合 scheduler 清理任务） - # ------------------------------------------------------------------ - - async def cleanup_old_reports( - self, - before: datetime, - *, - limit: int = 100, - commit: bool = True, - ) -> int: - """物理删除已完结（ready/failed）的陈旧报告，返回删除记录数。 - - Args: - before: 阈值，updated_at 早于此时间的终态报告将被删除。 - limit: 单次删除上限。 - commit: True 时提交事务，False 时仅 flush（用于组合事务）。 - """ - subq = ( - select(ChannelReport.id) - .where( - ChannelReport.status.in_(["ready", "failed"]), - ChannelReport.updated_at < before, - self._not_deleted(), - ) - .limit(limit) - ) - stmt = delete(ChannelReport).where(ChannelReport.id.in_(subq)) - result = await self.db.execute(stmt) - if commit: - await self.db.commit() - else: - await self.db.flush() - return result.rowcount - - # ------------------------------------------------------------------ - # 内部查询构建 - # ------------------------------------------------------------------ - - def _build_query( - self, - *, - status: str | None = None, - report_type: str | None = None, - start_time: datetime | None = None, - end_time: datetime | None = None, - select_count: bool = False, - ): - """构建查询语句，统一附加软删除过滤。""" - if select_count: - stmt = select(func.count(ChannelReport.id)).select_from(ChannelReport) - else: - stmt = select(ChannelReport) - stmt = stmt.where(self._not_deleted()) - if status is not None: - stmt = stmt.where(ChannelReport.status == status) - if report_type is not None: - stmt = stmt.where(ChannelReport.report_type == report_type) - if start_time is not None: - stmt = stmt.where(ChannelReport.created_at >= start_time) - if end_time is not None: - stmt = stmt.where(ChannelReport.created_at <= end_time) - return stmt diff --git a/backend/package/yuxi/repositories/channels/content_review_record_repository.py b/backend/package/yuxi/repositories/channels/content_review_record_repository.py index 00565914..c494eda5 100644 --- a/backend/package/yuxi/repositories/channels/content_review_record_repository.py +++ b/backend/package/yuxi/repositories/channels/content_review_record_repository.py @@ -3,10 +3,11 @@ from __future__ import annotations -from datetime import datetime +from datetime import datetime, timezone + from typing import Any -from sqlalchemy import case, func, select +from sqlalchemy import case, delete as sa_delete, func, select from yuxi.repositories.channels.base import BaseRepository from yuxi.storage.postgres.models_channels import ChannelContentReviewRecord @@ -55,6 +56,7 @@ class ChannelContentReviewRecordRepository(BaseRepository): reviewer=data["reviewer"], source=data["source"], trace_id=data.get("trace_id"), + decision_started_at=data.get("decision_started_at"), created_by=data.get("created_by"), updated_by=data.get("updated_by"), ) @@ -89,6 +91,8 @@ class ChannelContentReviewRecordRepository(BaseRepository): return None orm.verdict = verdict orm.reviewer = reviewer + orm.updated_by = reviewer + orm.decision_completed_at = datetime.now(timezone.utc).replace(tzinfo=None) if commit: await self.db.commit() else: @@ -96,6 +100,34 @@ class ChannelContentReviewRecordRepository(BaseRepository): await self.db.refresh(orm) return orm + async def delete_old_records( + self, + before: datetime, + *, + limit: int = 1000, + commit: bool = True, + ) -> int: + """物理删除早于指定时间的审核记录（retention）。 + + 与 ``ChannelAuditLog`` 一致采用物理删除（不使用软删除路径）， + 由 retention scheduler handler 周期触发，避免表无限增长。 + + Args: + before: 截止时间，早于此时间的记录将被删除。 + limit: 单次最多删除的记录数，避免长事务锁竞争。 + commit: True 时提交事务，False 时仅 flush。 + + Returns: + 被删除的记录数。 + """ + stmt = sa_delete(ChannelContentReviewRecord).where(ChannelContentReviewRecord.reviewed_at < before).limit(limit) + result = await self.db.execute(stmt) + if commit: + await self.db.commit() + else: + await self.db.flush() + return result.rowcount + # ------------------------------------------------------------------ # 查询 # ------------------------------------------------------------------ @@ -117,6 +149,8 @@ class ChannelContentReviewRecordRepository(BaseRepository): channel_type: str | None = None, account_id: str | None = None, verdict: str | None = None, + resource_type: str | None = None, + trace_id: str | None = None, start_time: datetime | None = None, end_time: datetime | None = None, page: int | None = None, @@ -133,6 +167,8 @@ class ChannelContentReviewRecordRepository(BaseRepository): channel_type=channel_type, account_id=account_id, verdict=verdict, + resource_type=resource_type, + trace_id=trace_id, start_time=start_time, end_time=end_time, ) @@ -149,6 +185,8 @@ class ChannelContentReviewRecordRepository(BaseRepository): channel_type: str | None = None, account_id: str | None = None, verdict: str | None = None, + resource_type: str | None = None, + trace_id: str | None = None, start_time: datetime | None = None, end_time: datetime | None = None, ) -> int: @@ -157,6 +195,8 @@ class ChannelContentReviewRecordRepository(BaseRepository): channel_type=channel_type, account_id=account_id, verdict=verdict, + resource_type=resource_type, + trace_id=trace_id, start_time=start_time, end_time=end_time, select_count=True, @@ -307,7 +347,13 @@ class ChannelContentReviewRecordRepository(BaseRepository): if end_time is not None: base_filters.append(ChannelContentReviewRecord.reviewed_at <= end_time) - # 全局聚合：总数 + pass/review/block + 人工介入计数 + # 全局聚合：总数 + pass/review/block + 人工介入计数 + 平均决策时长 + # avg_decision_seconds 仅对有 decision_completed_at 的记录计算 + # （首次审核即终态的记录无 completed_at，不计入） + decision_duration = func.extract( + "EPOCH", + ChannelContentReviewRecord.decision_completed_at - ChannelContentReviewRecord.decision_started_at, + ) agg_stmt = select( func.count(ChannelContentReviewRecord.id).label("total_reviews"), func.sum( @@ -334,6 +380,15 @@ class ChannelContentReviewRecordRepository(BaseRepository): else_=0, ) ).label("manual_count"), + func.avg( + case( + ( + ChannelContentReviewRecord.decision_completed_at.isnot(None), + decision_duration, + ), + else_=None, + ) + ).label("avg_decision_seconds"), ).where(*base_filters) agg_row = (await self.db.execute(agg_stmt)).one() total_reviews = int(agg_row.total_reviews or 0) @@ -341,6 +396,7 @@ class ChannelContentReviewRecordRepository(BaseRepository): review_count = int(agg_row.review_count or 0) block_count = int(agg_row.block_count or 0) manual_count = int(agg_row.manual_count or 0) + avg_decision_seconds = float(agg_row.avg_decision_seconds or 0.0) pass_rate = (pass_count / total_reviews * 100) if total_reviews > 0 else 0.0 block_rate = (block_count / total_reviews * 100) if total_reviews > 0 else 0.0 manual_intervention_rate = (manual_count / total_reviews * 100) if total_reviews > 0 else 0.0 @@ -400,6 +456,7 @@ class ChannelContentReviewRecordRepository(BaseRepository): "pass_rate": pass_rate, "block_rate": block_rate, "manual_intervention_rate": manual_intervention_rate, + "avg_decision_seconds": avg_decision_seconds, "by_category": by_category, "trend": trend, } @@ -414,15 +471,17 @@ class ChannelContentReviewRecordRepository(BaseRepository): channel_type: str | None = None, account_id: str | None = None, verdict: str | None = None, + resource_type: str | None = None, + trace_id: str | None = None, start_time: datetime | None = None, end_time: datetime | None = None, select_count: bool = False, ): """构建查询语句，统一附加软删除过滤。 - 支持 ``channel_type`` / ``account_id`` / ``verdict`` / - ``start_time`` / ``end_time`` 过滤，与 ``list`` / ``count`` 共享 - 同一组条件构造逻辑。 + 支持 ``channel_type`` / ``account_id`` / ``verdict`` / ``resource_type`` + / ``trace_id`` / ``start_time`` / ``end_time`` 过滤，与 ``list`` / + ``count`` 共享同一组条件构造逻辑。 """ if select_count: stmt = select(func.count(ChannelContentReviewRecord.id)).select_from(ChannelContentReviewRecord) @@ -435,6 +494,10 @@ class ChannelContentReviewRecordRepository(BaseRepository): stmt = stmt.where(ChannelContentReviewRecord.account_id == account_id) if verdict is not None: stmt = stmt.where(ChannelContentReviewRecord.verdict == verdict) + if resource_type is not None: + stmt = stmt.where(ChannelContentReviewRecord.resource_type == resource_type) + if trace_id is not None: + stmt = stmt.where(ChannelContentReviewRecord.trace_id == trace_id) if start_time is not None: stmt = stmt.where(ChannelContentReviewRecord.reviewed_at >= start_time) if end_time is not None: diff --git a/backend/package/yuxi/repositories/channels/route_binding_repository.py b/backend/package/yuxi/repositories/channels/route_binding_repository.py index b9cd0ba2..df4ff827 100644 --- a/backend/package/yuxi/repositories/channels/route_binding_repository.py +++ b/backend/package/yuxi/repositories/channels/route_binding_repository.py @@ -212,24 +212,3 @@ class ChannelRouteBindingRepository(BaseRepository): else: await self.db.flush() return result.rowcount > 0 - - # ------------------------------------------------------------------ - # 业务专用查询 - # ------------------------------------------------------------------ - - async def list_enabled_by_account( - self, - channel_type: str, - account_id: str, - ) -> list[ChannelRouteBinding]: - """加载某账户下所有启用且未删除的规则。""" - stmt = ( - select(ChannelRouteBinding) - .where(ChannelRouteBinding.channel_type == channel_type) - .where(ChannelRouteBinding.account_id == account_id) - .where(ChannelRouteBinding.enabled.is_(True)) - .where(self._not_deleted()) - .order_by(ChannelRouteBinding.created_at.desc()) - ) - result = await self.db.execute(stmt) - return list(result.scalars().all())