diff --git a/backend/package/yuxi/repositories/channels/__init__.py b/backend/package/yuxi/repositories/channels/__init__.py index 4c5627de..09167de5 100644 --- a/backend/package/yuxi/repositories/channels/__init__.py +++ b/backend/package/yuxi/repositories/channels/__init__.py @@ -33,6 +33,7 @@ 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 from .user_identity_repository import UserIdentityRepository __all__ = [ @@ -46,6 +47,7 @@ __all__ = [ "ChannelOutboxRepository", "ChannelPairingRepository", "ChannelReportRepository", + "ChannelRouteBindingRepository", "ChannelSessionRepository", "Repositories", "RepositoryError", @@ -72,6 +74,7 @@ class Repositories: message: ChannelMessageRepository report: ChannelReportRepository content_review_record: ChannelContentReviewRecordRepository + route_binding: ChannelRouteBindingRepository def create_repositories(db: AsyncSession) -> Repositories: @@ -93,4 +96,5 @@ def create_repositories(db: AsyncSession) -> Repositories: message=ChannelMessageRepository(db), report=ChannelReportRepository(db), content_review_record=ChannelContentReviewRecordRepository(db), + route_binding=ChannelRouteBindingRepository(db), ) diff --git a/backend/package/yuxi/repositories/channels/base.py b/backend/package/yuxi/repositories/channels/base.py index 5b2411a4..a69f578f 100644 --- a/backend/package/yuxi/repositories/channels/base.py +++ b/backend/package/yuxi/repositories/channels/base.py @@ -100,9 +100,7 @@ class BaseRepository: 当 ``commit=False`` 时仅执行 ``flush()`` 使变更进入会话但不提交, 便于上层在事务中组合多个写操作。 """ - return await self._soft_delete_by_id( - record_id, updated_by=updated_by, commit=commit - ) + return await self._soft_delete_by_id(record_id, updated_by=updated_by, commit=commit) async def _soft_delete_by_id( self, @@ -135,11 +133,7 @@ class BaseRepository: 用于恢复前的记录查询与唯一约束预检,避免 list_deleted(limit=1) 的误匹配。 """ - stmt = ( - select(self.model) - .where(self.model.id == record_id) - .where(self.model.is_deleted == 1) - ) + stmt = select(self.model).where(self.model.id == record_id).where(self.model.is_deleted == 1) result = await self.db.execute(stmt) return result.scalars().first() @@ -201,11 +195,7 @@ class BaseRepository: commit: bool = True, ) -> int: """按主键物理删除记录(仅删除 is_deleted=1 的记录)。返回受影响行数。""" - stmt = ( - sa_delete(self.model) - .where(self.model.id == record_id) - .where(self.model.is_deleted == 1) - ) + stmt = sa_delete(self.model).where(self.model.id == record_id).where(self.model.is_deleted == 1) result = await self.db.execute(stmt) if commit: await self.db.commit() @@ -220,11 +210,7 @@ class BaseRepository: commit: bool = True, ) -> int: """物理删除 deleted_at 早于 before 的软删除记录。返回受影响行数。""" - stmt = ( - sa_delete(self.model) - .where(self.model.is_deleted == 1) - .where(self.model.deleted_at < before) - ) + stmt = sa_delete(self.model).where(self.model.is_deleted == 1).where(self.model.deleted_at < before) result = await self.db.execute(stmt) if commit: await self.db.commit() @@ -253,9 +239,7 @@ class BaseRepository: for field_name, value in field_filters.items(): column = getattr(self.model, field_name, None) if column is None: - raise RepositoryError( - f"模型 {self.model.__name__} 无字段 {field_name}" - ) + raise RepositoryError(f"模型 {self.model.__name__} 无字段 {field_name}") stmt = stmt.where(column == value) result = await self.db.execute(stmt) return int(result.scalar() or 0) diff --git a/backend/package/yuxi/repositories/channels/channel_account_repository.py b/backend/package/yuxi/repositories/channels/channel_account_repository.py index b124a98d..d91b3e09 100644 --- a/backend/package/yuxi/repositories/channels/channel_account_repository.py +++ b/backend/package/yuxi/repositories/channels/channel_account_repository.py @@ -22,9 +22,7 @@ class ChannelAccountRepository(BaseRepository): # 默认增删改查 # ------------------------------------------------------------------ - async def create( - self, data: dict[str, Any], *, commit: bool = True - ) -> ChannelAccount: + async def create(self, data: dict[str, Any], *, commit: bool = True) -> ChannelAccount: """创建渠道账户。 Args: @@ -64,9 +62,7 @@ class ChannelAccountRepository(BaseRepository): await self.db.refresh(account) return account - async def get_by_id( - self, record_id: int, *, for_update: bool = False - ) -> ChannelAccount | None: + async def get_by_id(self, record_id: int, *, for_update: bool = False) -> ChannelAccount | None: """根据主键 ID 获取渠道账户(排除已软删除)。""" stmt = select(ChannelAccount).where( ChannelAccount.id == record_id, @@ -116,9 +112,7 @@ class ChannelAccountRepository(BaseRepository): keyword=keyword, ) stmt = stmt.order_by(ChannelAccount.created_at.desc()) - effective_limit, effective_offset = self._resolve_pagination( - page, page_size, limit, offset - ) + 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) @@ -209,12 +203,7 @@ class ChannelAccountRepository(BaseRepository): values["last_error"] = last_error if last_error_at is not None: values["last_error_at"] = last_error_at - stmt = ( - update(ChannelAccount) - .where(ChannelAccount.id == record_id) - .where(self._not_deleted()) - .values(**values) - ) + stmt = update(ChannelAccount).where(ChannelAccount.id == record_id).where(self._not_deleted()).values(**values) result = await self.db.execute(stmt) if commit: await self.db.commit() diff --git a/backend/package/yuxi/repositories/channels/channel_audit_log_repository.py b/backend/package/yuxi/repositories/channels/channel_audit_log_repository.py index 7d80288d..3fe62b83 100644 --- a/backend/package/yuxi/repositories/channels/channel_audit_log_repository.py +++ b/backend/package/yuxi/repositories/channels/channel_audit_log_repository.py @@ -24,9 +24,7 @@ class ChannelAuditLogRepository(BaseRepository): # 写操作(仅 create,append-only) # ------------------------------------------------------------------ - async def create( - self, data: dict[str, Any], *, commit: bool = True - ) -> ChannelAuditLog: + async def create(self, data: dict[str, Any], *, commit: bool = True) -> ChannelAuditLog: """创建审计日志记录(append-only)。 Args: @@ -70,9 +68,7 @@ class ChannelAuditLogRepository(BaseRepository): # 查询 # ------------------------------------------------------------------ - async def get_by_id( - self, record_id: int, *, for_update: bool = False - ) -> ChannelAuditLog | None: + async def get_by_id(self, record_id: int, *, for_update: bool = False) -> ChannelAuditLog | None: """根据主键 ID 获取审计日志。""" stmt = select(ChannelAuditLog).where(ChannelAuditLog.id == record_id) if for_update: @@ -80,13 +76,9 @@ class ChannelAuditLogRepository(BaseRepository): result = await self.db.execute(stmt) return result.scalar_one_or_none() - async def get_by_audit_log_id( - self, audit_log_id: str, *, for_update: bool = False - ) -> ChannelAuditLog | None: + async def get_by_audit_log_id(self, audit_log_id: str, *, for_update: bool = False) -> ChannelAuditLog | None: """根据业务标识 audit_log_id 获取审计日志。""" - stmt = select(ChannelAuditLog).where( - ChannelAuditLog.audit_log_id == audit_log_id - ) + stmt = select(ChannelAuditLog).where(ChannelAuditLog.audit_log_id == audit_log_id) if for_update: stmt = stmt.with_for_update() result = await self.db.execute(stmt) @@ -118,9 +110,7 @@ class ChannelAuditLogRepository(BaseRepository): end_time=end_time, ) stmt = stmt.order_by(ChannelAuditLog.timestamp.desc()) - effective_limit, effective_offset = self._resolve_pagination( - page, page_size, limit, offset - ) + 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) @@ -265,9 +255,7 @@ class ChannelAuditLogRepository(BaseRepository): for cond in conditions: op_stmt = op_stmt.where(cond) result = await self.db.execute(op_stmt) - by_operation: dict[str, int] = { - row[0]: int(row[1]) for row in result.all() if row[0] is not None - } + by_operation: dict[str, int] = {row[0]: int(row[1]) for row in result.all() if row[0] is not None} res_stmt = select( ChannelAuditLog.result, @@ -276,9 +264,7 @@ class ChannelAuditLogRepository(BaseRepository): for cond in conditions: res_stmt = res_stmt.where(cond) result = await self.db.execute(res_stmt) - by_result: dict[str, int] = { - row[0]: int(row[1]) for row in result.all() if row[0] is not None - } + by_result: dict[str, int] = {row[0]: int(row[1]) for row in result.all() if row[0] is not None} return { "total": total, diff --git a/backend/package/yuxi/repositories/channels/channel_idempotency_repository.py b/backend/package/yuxi/repositories/channels/channel_idempotency_repository.py index 01b90fab..48184812 100644 --- a/backend/package/yuxi/repositories/channels/channel_idempotency_repository.py +++ b/backend/package/yuxi/repositories/channels/channel_idempotency_repository.py @@ -26,9 +26,7 @@ class ChannelIdempotencyRepository(BaseRepository): # 写操作 # ------------------------------------------------------------------ - async def create( - self, data: dict[str, Any], *, commit: bool = True - ) -> ChannelIdempotency: + async def create(self, data: dict[str, Any], *, commit: bool = True) -> ChannelIdempotency: """创建幂等记录(status='in_progress',持有处理权)。 利用 idempotency_key UNIQUE 约束实现首次写入抢占: @@ -89,11 +87,7 @@ class ChannelIdempotencyRepository(BaseRepository): } if response_body is not None: values["response_body"] = response_body - stmt = ( - update(ChannelIdempotency) - .where(ChannelIdempotency.id == record_id) - .values(**values) - ) + stmt = update(ChannelIdempotency).where(ChannelIdempotency.id == record_id).values(**values) result = await self.db.execute(stmt) if commit: await self.db.commit() @@ -105,9 +99,7 @@ class ChannelIdempotencyRepository(BaseRepository): # 查询 # ------------------------------------------------------------------ - async def get_by_id( - self, record_id: int, *, for_update: bool = False - ) -> ChannelIdempotency | None: + 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: @@ -115,13 +107,9 @@ class ChannelIdempotencyRepository(BaseRepository): 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: + async def get_by_key(self, idempotency_key: str, *, for_update: bool = False) -> ChannelIdempotency | None: """根据幂等键获取记录(重复请求回放)。""" - stmt = select(ChannelIdempotency).where( - ChannelIdempotency.idempotency_key == idempotency_key - ) + stmt = select(ChannelIdempotency).where(ChannelIdempotency.idempotency_key == idempotency_key) if for_update: stmt = stmt.with_for_update() result = await self.db.execute(stmt) @@ -164,9 +152,7 @@ class ChannelIdempotencyRepository(BaseRepository): 幂等记录不使用软删除,过期后直接物理删除。 默认保留 24 小时(IDEMPOTENCY_DEFAULT_TTL_HOURS)。 """ - stmt = sa_delete(ChannelIdempotency).where( - ChannelIdempotency.created_at < before - ) + stmt = sa_delete(ChannelIdempotency).where(ChannelIdempotency.created_at < before) result = await self.db.execute(stmt) if commit: await self.db.commit() @@ -181,9 +167,7 @@ class ChannelIdempotencyRepository(BaseRepository): commit: bool = True, ) -> int: """按幂等键物理删除记录(手动清理)。""" - stmt = sa_delete(ChannelIdempotency).where( - ChannelIdempotency.idempotency_key == idempotency_key - ) + stmt = sa_delete(ChannelIdempotency).where(ChannelIdempotency.idempotency_key == idempotency_key) 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 e4e23711..dffbaa67 100644 --- a/backend/package/yuxi/repositories/channels/channel_outbox_repository.py +++ b/backend/package/yuxi/repositories/channels/channel_outbox_repository.py @@ -21,9 +21,7 @@ class ChannelOutboxRepository(BaseRepository): # 默认增删改查 # ------------------------------------------------------------------ - async def create( - self, data: dict[str, Any], *, commit: bool = True - ) -> ChannelOutboxEntry: + async def create(self, data: dict[str, Any], *, commit: bool = True) -> ChannelOutboxEntry: """创建发件箱条目。 Note: @@ -62,9 +60,7 @@ class ChannelOutboxRepository(BaseRepository): await self.db.refresh(entry) return entry - async def get_by_id( - self, record_id: int, *, for_update: bool = False - ) -> ChannelOutboxEntry | None: + async def get_by_id(self, record_id: int, *, for_update: bool = False) -> ChannelOutboxEntry | None: """根据主键 ID 获取发件箱条目(排除已软删除)。""" stmt = select(ChannelOutboxEntry).where( ChannelOutboxEntry.id == record_id, @@ -139,9 +135,7 @@ class ChannelOutboxRepository(BaseRepository): message_id=message_id, ) stmt = stmt.order_by(ChannelOutboxEntry.created_at.desc()) - effective_limit, effective_offset = self._resolve_pagination( - page, page_size, limit, offset - ) + 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) @@ -194,15 +188,10 @@ class ChannelOutboxRepository(BaseRepository): before: 截止时间,next_retry_at 早于此时间的 PENDING 条目将被列出。 None 时不过滤 next_retry_at。 """ - stmt = ( - select(ChannelOutboxEntry) - .where(ChannelOutboxEntry.status == "pending") - .where(self._not_deleted()) - ) + stmt = select(ChannelOutboxEntry).where(ChannelOutboxEntry.status == "pending").where(self._not_deleted()) if before is not None: stmt = stmt.where( - (ChannelOutboxEntry.next_retry_at.is_(None)) - | (ChannelOutboxEntry.next_retry_at <= before) + (ChannelOutboxEntry.next_retry_at.is_(None)) | (ChannelOutboxEntry.next_retry_at <= before) ) stmt = stmt.order_by(ChannelOutboxEntry.created_at.asc()).limit(limit) result = await self.db.execute(stmt) @@ -288,11 +277,7 @@ class ChannelOutboxRepository(BaseRepository): if release_lock: values["locked_by"] = None values["locked_at"] = None - stmt = ( - update(ChannelOutboxEntry) - .where(ChannelOutboxEntry.id == record_id) - .where(self._not_deleted()) - ) + 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) @@ -371,9 +356,7 @@ class ChannelOutboxRepository(BaseRepository): await self.db.flush() return result.rowcount - async def list_by_message_id( - self, message_id: int - ) -> list[ChannelOutboxEntry]: + async def list_by_message_id(self, message_id: int) -> list[ChannelOutboxEntry]: """按消息 ID 查询投递状态(FR-22 状态回查)。""" stmt = ( select(ChannelOutboxEntry) diff --git a/backend/package/yuxi/repositories/channels/channel_pairing_repository.py b/backend/package/yuxi/repositories/channels/channel_pairing_repository.py index 6571b749..386da887 100644 --- a/backend/package/yuxi/repositories/channels/channel_pairing_repository.py +++ b/backend/package/yuxi/repositories/channels/channel_pairing_repository.py @@ -22,9 +22,7 @@ class ChannelPairingRepository(BaseRepository): # 默认增删改查 # ------------------------------------------------------------------ - async def create( - self, data: dict[str, Any], *, commit: bool = True - ) -> ChannelPairing: + async def create(self, data: dict[str, Any], *, commit: bool = True) -> ChannelPairing: """创建配对审批记录。""" data = dict(data) now = utc_now_naive() @@ -56,9 +54,7 @@ class ChannelPairingRepository(BaseRepository): await self.db.refresh(pairing) return pairing - async def get_by_id( - self, record_id: int, *, for_update: bool = False - ) -> ChannelPairing | None: + async def get_by_id(self, record_id: int, *, for_update: bool = False) -> ChannelPairing | None: """根据主键 ID 获取配对记录(排除已软删除)。""" stmt = select(ChannelPairing).where( ChannelPairing.id == record_id, @@ -69,9 +65,7 @@ class ChannelPairingRepository(BaseRepository): result = await self.db.execute(stmt) return result.scalar_one_or_none() - async def get_by_pairing_id( - self, pairing_id: str, *, for_update: bool = False - ) -> ChannelPairing | None: + async def get_by_pairing_id(self, pairing_id: str, *, for_update: bool = False) -> ChannelPairing | None: """根据业务标识 pairing_id 获取配对记录(排除已软删除)。""" stmt = select(ChannelPairing).where( ChannelPairing.pairing_id == pairing_id, @@ -121,9 +115,7 @@ class ChannelPairingRepository(BaseRepository): end_time=end_time, ) stmt = stmt.order_by(ChannelPairing.requested_at.desc()) - effective_limit, effective_offset = self._resolve_pagination( - page, page_size, limit, offset - ) + 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) @@ -199,9 +191,7 @@ class ChannelPairingRepository(BaseRepository): result = await self.db.execute(stmt) return list(result.scalars().all()) - async def list_expired_pending( - self, before: datetime, *, limit: int = 100 - ) -> list[ChannelPairing]: + async def list_expired_pending(self, before: datetime, *, limit: int = 100) -> list[ChannelPairing]: """列出已过期但仍为 PENDING 的配对(过期清理任务)。 Args: @@ -245,6 +235,32 @@ class ChannelPairingRepository(BaseRepository): await self.db.flush() return result.rowcount + async def count( + self, + *, + account_id: int | None = None, + account_ids: list[int] | None = None, + peer_id: str | None = None, + status: str | None = None, + start_time: datetime | None = None, + end_time: datetime | None = None, + ) -> int: + """按过滤条件统计配对记录数(与 ``list`` 过滤语义一致,仅返回计数)。 + + 供 ``countPairings`` 等轻量计数场景使用,不分页。 + """ + stmt = self._build_query( + account_id=account_id, + account_ids=account_ids, + peer_id=peer_id, + status=status, + start_time=start_time, + end_time=end_time, + ) + count_stmt = select(func.count()).select_from(stmt.subquery()) + result = await self.db.execute(count_stmt) + return int(result.scalar() or 0) + async def count_by_status(self) -> dict[str, int]: """按状态分组统计配对数量。""" stmt = ( diff --git a/backend/package/yuxi/repositories/channels/channel_report_repository.py b/backend/package/yuxi/repositories/channels/channel_report_repository.py index f0fda04b..7fa7dedf 100644 --- a/backend/package/yuxi/repositories/channels/channel_report_repository.py +++ b/backend/package/yuxi/repositories/channels/channel_report_repository.py @@ -25,9 +25,7 @@ class ChannelReportRepository(BaseRepository): # 写操作 # ------------------------------------------------------------------ - async def create( - self, data: dict[str, Any], *, commit: bool = True - ) -> ChannelReport: + async def create(self, data: dict[str, Any], *, commit: bool = True) -> ChannelReport: """创建报告记录。 Args: @@ -135,9 +133,7 @@ class ChannelReportRepository(BaseRepository): # 查询 # ------------------------------------------------------------------ - async def get_by_report_id( - self, report_id: str, *, for_update: bool = False - ) -> ChannelReport | None: + 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, @@ -148,9 +144,7 @@ class ChannelReportRepository(BaseRepository): 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: + 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, @@ -185,9 +179,7 @@ class ChannelReportRepository(BaseRepository): end_time=end_time, ) stmt = stmt.order_by(ChannelReport.created_at.desc()) - effective_limit, effective_offset = self._resolve_pagination( - page, page_size, limit, offset - ) + 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) diff --git a/backend/package/yuxi/repositories/channels/channel_session_repository.py b/backend/package/yuxi/repositories/channels/channel_session_repository.py index ae5784e9..cbccd430 100644 --- a/backend/package/yuxi/repositories/channels/channel_session_repository.py +++ b/backend/package/yuxi/repositories/channels/channel_session_repository.py @@ -26,9 +26,7 @@ class ChannelSessionRepository(BaseRepository): # 默认增删改查 # ------------------------------------------------------------------ - async def create( - self, data: dict[str, Any], *, commit: bool = True - ) -> ChannelSession: + async def create(self, data: dict[str, Any], *, commit: bool = True) -> ChannelSession: """创建渠道会话。""" data = dict(data) now = utc_now_naive() @@ -58,9 +56,7 @@ class ChannelSessionRepository(BaseRepository): await self.db.refresh(session) return session - async def get_by_id( - self, record_id: int, *, for_update: bool = False - ) -> ChannelSession | None: + async def get_by_id(self, record_id: int, *, for_update: bool = False) -> ChannelSession | None: """根据主键 ID 获取渠道会话(排除已软删除)。""" stmt = select(ChannelSession).where( ChannelSession.id == record_id, @@ -71,9 +67,7 @@ class ChannelSessionRepository(BaseRepository): result = await self.db.execute(stmt) return result.scalar_one_or_none() - async def get_by_session_id( - self, session_id: str, *, for_update: bool = False - ) -> ChannelSession | None: + async def get_by_session_id(self, session_id: str, *, for_update: bool = False) -> ChannelSession | None: """根据业务标识 session_id 获取渠道会话(排除已软删除)。""" stmt = select(ChannelSession).where( ChannelSession.session_id == session_id, @@ -154,9 +148,7 @@ class ChannelSessionRepository(BaseRepository): abnormal=abnormal, ) stmt = stmt.order_by(ChannelSession.last_message_at.desc().nullslast()) - effective_limit, effective_offset = self._resolve_pagination( - page, page_size, limit, offset - ) + 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) @@ -252,12 +244,7 @@ class ChannelSessionRepository(BaseRepository): } if route_match_source is not None: values["route_match_source"] = route_match_source - stmt = ( - update(ChannelSession) - .where(ChannelSession.id == record_id) - .where(self._not_deleted()) - .values(**values) - ) + stmt = update(ChannelSession).where(ChannelSession.id == record_id).where(self._not_deleted()).values(**values) result = await self.db.execute(stmt) if commit: await self.db.commit() @@ -312,9 +299,7 @@ class ChannelSessionRepository(BaseRepository): result = await self.db.execute(stmt) return list(result.scalars().all()) - async def list_by_conversation( - self, conversation_id: int - ) -> list[ChannelSession]: + async def list_by_conversation(self, conversation_id: int) -> list[ChannelSession]: """按内部会话 ID 反查渠道会话(FR-06 跨渠道关联)。""" stmt = ( select(ChannelSession) @@ -388,9 +373,7 @@ class ChannelSessionRepository(BaseRepository): if last_message_before is not None: stmt = stmt.where(ChannelSession.last_message_at <= last_message_before) if abnormal: - stmt = stmt.join( - ChannelAccount, ChannelAccount.id == ChannelSession.account_id - ) + stmt = stmt.join(ChannelAccount, ChannelAccount.id == ChannelSession.account_id) zombie_threshold = utc_now_naive() - timedelta(days=_ZOMBIE_SESSION_DAYS) zombie_cond = or_( ChannelSession.last_message_at.is_(None), 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 3868c388..00565914 100644 --- a/backend/package/yuxi/repositories/channels/content_review_record_repository.py +++ b/backend/package/yuxi/repositories/channels/content_review_record_repository.py @@ -32,9 +32,7 @@ class ChannelContentReviewRecordRepository(BaseRepository): # 写操作 # ------------------------------------------------------------------ - async def create( - self, data: dict[str, Any], *, commit: bool = True - ) -> ChannelContentReviewRecord: + async def create(self, data: dict[str, Any], *, commit: bool = True) -> ChannelContentReviewRecord: """创建审核记录。 Args: @@ -102,9 +100,7 @@ class ChannelContentReviewRecordRepository(BaseRepository): # 查询 # ------------------------------------------------------------------ - async def get_by_review_id( - self, review_id: str, *, for_update: bool = False - ) -> ChannelContentReviewRecord | None: + async def get_by_review_id(self, review_id: str, *, for_update: bool = False) -> ChannelContentReviewRecord | None: """根据业务标识 review_id 获取审核记录(排除已软删除)。""" stmt = select(ChannelContentReviewRecord).where( ChannelContentReviewRecord.review_id == review_id, @@ -141,9 +137,7 @@ class ChannelContentReviewRecordRepository(BaseRepository): end_time=end_time, ) stmt = stmt.order_by(ChannelContentReviewRecord.reviewed_at.desc()) - effective_limit, effective_offset = self._resolve_pagination( - page, page_size, limit, offset - ) + 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) @@ -221,14 +215,10 @@ class ChannelContentReviewRecordRepository(BaseRepository): agg_row = (await self.db.execute(agg_stmt)).one() total_reviews = int(agg_row.total_reviews or 0) block_count = int(agg_row.block_count or 0) - block_rate = ( - (block_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 # 按分类切片:展开 categories JSON 数组后分组计数 - category_expr = func.json_array_elements_text( - ChannelContentReviewRecord.categories - ).label("category") + category_expr = func.json_array_elements_text(ChannelContentReviewRecord.categories).label("category") category_stmt = ( select( category_expr, @@ -239,14 +229,11 @@ class ChannelContentReviewRecordRepository(BaseRepository): ) category_result = await self.db.execute(category_stmt) by_category: list[dict[str, Any]] = [ - {"category": row.category, "count": int(row.count or 0)} - for row in category_result.all() + {"category": row.category, "count": int(row.count or 0)} for row in category_result.all() ] # 时间序列:按 granularity 分桶,每桶含 reviewed / blocked 双计数 - bucket_expr = func.date_trunc( - granularity, ChannelContentReviewRecord.reviewed_at - ) + bucket_expr = func.date_trunc(granularity, ChannelContentReviewRecord.reviewed_at) trend_stmt = ( select( bucket_expr.label("bucket"), @@ -356,14 +343,10 @@ class ChannelContentReviewRecordRepository(BaseRepository): manual_count = int(agg_row.manual_count or 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 - ) + manual_intervention_rate = (manual_count / total_reviews * 100) if total_reviews > 0 else 0.0 # 按分类切片:展开 categories JSON 数组后分组计数 - category_expr = func.json_array_elements_text( - ChannelContentReviewRecord.categories - ).label("category") + category_expr = func.json_array_elements_text(ChannelContentReviewRecord.categories).label("category") category_stmt = ( select( category_expr, @@ -374,14 +357,11 @@ class ChannelContentReviewRecordRepository(BaseRepository): ) category_result = await self.db.execute(category_stmt) by_category: list[dict[str, Any]] = [ - {"category": row.category, "count": int(row.count or 0)} - for row in category_result.all() + {"category": row.category, "count": int(row.count or 0)} for row in category_result.all() ] # 时间序列:按 granularity 分桶,每桶含 pass / block 双计数 - bucket_expr = func.date_trunc( - granularity, ChannelContentReviewRecord.reviewed_at - ) + bucket_expr = func.date_trunc(granularity, ChannelContentReviewRecord.reviewed_at) trend_stmt = ( select( bucket_expr.label("bucket"), @@ -445,9 +425,7 @@ class ChannelContentReviewRecordRepository(BaseRepository): 同一组条件构造逻辑。 """ if select_count: - stmt = select(func.count(ChannelContentReviewRecord.id)).select_from( - ChannelContentReviewRecord - ) + stmt = select(func.count(ChannelContentReviewRecord.id)).select_from(ChannelContentReviewRecord) else: stmt = select(ChannelContentReviewRecord) stmt = stmt.where(self._not_deleted()) diff --git a/backend/package/yuxi/repositories/channels/route_binding_repository.py b/backend/package/yuxi/repositories/channels/route_binding_repository.py new file mode 100644 index 00000000..b9cd0ba2 --- /dev/null +++ b/backend/package/yuxi/repositories/channels/route_binding_repository.py @@ -0,0 +1,235 @@ +"""渠道路由绑定规则仓储。 + +提供 ``channel_route_bindings`` 表的异步数据访问封装,遵循现有 11 个 +repository 模式:直接使用 SQLAlchemy AsyncSession,不依赖抽象端口; +``commit=True`` 单方法提交,``commit=False`` 仅 flush(由上层 tx 控制)。 +""" + +from __future__ import annotations + +from typing import Any + +from sqlalchemy import case, func, select, update + +from yuxi.channels.contract.dtos.route import RouteBindingFilter +from yuxi.repositories.channels.base import BaseRepository +from yuxi.storage.postgres.models_channels import ChannelRouteBinding +from yuxi.utils.datetime_utils import utc_now_naive + +__all__ = ["ChannelRouteBindingRepository"] + +#: 内置匹配层级优先级(与 factory._DEFAULT_MATCH_TIERS / mappers._TIER_PRIORITY_MAP 对齐) +#: 用于仓储层按运行时命中顺序排序;仅展示用,不参与路由决策。 +_TIER_PRIORITY_MAP: dict[str, int] = { + "session_key": 800, + "identity_id": 700, + "peer_id": 600, + "chat_type": 500, + "channel_session": 400, + "channel_type": 300, + "account": 200, + "default": 100, +} + + +def _priority_order_clause(): + """构造 match_source → priority 的 CASE 排序表达式。""" + return case( + *_build_when_pairs(), + else_=0, + ) + + +def _build_when_pairs(): + """生成 SQLAlchemy case 的 when 子句列表。""" + return [(ChannelRouteBinding.match_source == tier, priority) for tier, priority in _TIER_PRIORITY_MAP.items()] + + +class ChannelRouteBindingRepository(BaseRepository): + """``channel_route_bindings`` 表的异步数据访问封装。""" + + model = ChannelRouteBinding + + # ------------------------------------------------------------------ + # 默认增删改查 + # ------------------------------------------------------------------ + + async def create(self, data: dict[str, Any], *, commit: bool = True) -> ChannelRouteBinding: + """创建路由绑定规则。 + + Args: + data: 字段字典,必须包含 binding_id / channel_type / account_id / + match_source / agent_binding。 + commit: True 时提交事务,False 时仅 flush。 + + Returns: + 创建后的 ChannelRouteBinding ORM 实例。 + """ + data = dict(data) + now = utc_now_naive() + binding = ChannelRouteBinding( + binding_id=data["binding_id"], + channel_type=data["channel_type"], + account_id=data["account_id"], + match_source=data["match_source"], + match_value=data.get("match_value"), + agent_binding=data["agent_binding"], + enabled=data.get("enabled", True), + description=data.get("description"), + version=1, + created_by=data.get("created_by"), + updated_by=data.get("updated_by"), + created_at=now, + updated_at=now, + ) + self.db.add(binding) + if commit: + await self.db.commit() + else: + await self.db.flush() + await self.db.refresh(binding) + return binding + + async def get_by_binding_id( + self, + binding_id: str, + *, + for_update: bool = False, + ) -> ChannelRouteBinding | None: + """按 binding_id 查询(排除已软删除,可选行级锁)。""" + stmt = select(ChannelRouteBinding).where( + ChannelRouteBinding.binding_id == binding_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 count( + self, + *, + filter: RouteBindingFilter, + ) -> int: + """按过滤条件统计总数(始终排除已软删除)。""" + stmt = select(func.count(ChannelRouteBinding.id)) + stmt = self._apply_filter_to_stmt(stmt, filter) + result = await self.db.execute(stmt) + return result.scalar() or 0 + + def _apply_filter_to_stmt(self, stmt: select, filter: RouteBindingFilter) -> select: + """把过滤条件应用到计数或列表查询语句。""" + if filter.channel_type is not None: + stmt = stmt.where(ChannelRouteBinding.channel_type == filter.channel_type) + if filter.account_id is not None: + stmt = stmt.where(ChannelRouteBinding.account_id == filter.account_id) + if filter.match_source is not None: + stmt = stmt.where(ChannelRouteBinding.match_source == filter.match_source) + if filter.enabled is not None: + stmt = stmt.where(ChannelRouteBinding.enabled.is_(filter.enabled)) + if filter.agent_binding is not None: + stmt = stmt.where(ChannelRouteBinding.agent_binding == filter.agent_binding) + return stmt + + async def list( + self, + *, + filter: RouteBindingFilter, + limit: int = 1000, + offset: int = 0, + ) -> list[ChannelRouteBinding]: + """按过滤条件列表查询(始终排除已软删除)。 + + 默认按匹配层级优先级降序、同层级内按创建时间降序排列, + 使列表顺序与运行时命中顺序一致。 + """ + stmt = select(ChannelRouteBinding).where(self._not_deleted()) + stmt = self._apply_filter_to_stmt(stmt, filter) + stmt = stmt.order_by(_priority_order_clause().desc(), ChannelRouteBinding.created_at.desc()) + stmt = stmt.limit(limit).offset(offset) + result = await self.db.execute(stmt) + return list(result.scalars().all()) + + async def update( + self, + record: ChannelRouteBinding, + data: dict[str, Any], + *, + commit: bool = True, + ) -> ChannelRouteBinding: + """更新路由绑定规则(局部更新,仅更新 data 中提供的字段)。 + + Args: + record: ORM 实例(通常由 get_by_binding_id 获取,支持悲观锁)。 + data: 待更新字段字典。 + commit: True 时提交事务,False 时仅 flush。 + """ + updatable_fields = ( + "match_value", + "agent_binding", + "enabled", + "description", + "updated_by", + ) + for field in updatable_fields: + if field in data and data[field] is not None: + setattr(record, field, data[field]) + record.updated_at = utc_now_naive() + if commit: + await self.db.commit() + else: + await self.db.flush() + await self.db.refresh(record) + return record + + async def soft_delete( + self, + binding_id: str, + *, + operator: str, + commit: bool = True, + ) -> bool: + """软删除(标记 is_deleted=1)。 + + Returns: + True 若命中并删除,False 若记录不存在或已软删除。 + """ + now = utc_now_naive() + stmt = ( + update(ChannelRouteBinding) + .where(ChannelRouteBinding.binding_id == binding_id) + .where(ChannelRouteBinding.is_deleted == 0) + .values( + is_deleted=1, + deleted_at=now, + updated_at=now, + updated_by=operator, + ) + ) + result = await self.db.execute(stmt) + if commit: + await self.db.commit() + 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()) diff --git a/backend/package/yuxi/repositories/channels/user_identity_repository.py b/backend/package/yuxi/repositories/channels/user_identity_repository.py index f86d77a4..c7cf9d6b 100644 --- a/backend/package/yuxi/repositories/channels/user_identity_repository.py +++ b/backend/package/yuxi/repositories/channels/user_identity_repository.py @@ -21,9 +21,7 @@ class UserIdentityRepository(BaseRepository): # 默认增删改查 # ------------------------------------------------------------------ - async def create( - self, data: dict[str, Any], *, commit: bool = True - ) -> ChannelUserIdentity: + async def create(self, data: dict[str, Any], *, commit: bool = True) -> ChannelUserIdentity: """创建统一身份记录。""" data = dict(data) now = utc_now_naive() @@ -52,9 +50,7 @@ class UserIdentityRepository(BaseRepository): await self.db.refresh(identity) return identity - async def get_by_id( - self, record_id: int, *, for_update: bool = False - ) -> ChannelUserIdentity | None: + async def get_by_id(self, record_id: int, *, for_update: bool = False) -> ChannelUserIdentity | None: """根据主键 ID 获取身份记录(排除已软删除)。""" stmt = select(ChannelUserIdentity).where( ChannelUserIdentity.id == record_id, @@ -65,9 +61,7 @@ class UserIdentityRepository(BaseRepository): result = await self.db.execute(stmt) return result.scalar_one_or_none() - async def get_by_identity_id( - self, identity_id: str, *, for_update: bool = False - ) -> ChannelUserIdentity | None: + async def get_by_identity_id(self, identity_id: str, *, for_update: bool = False) -> ChannelUserIdentity | None: """根据业务标识 identity_id 获取身份记录(排除已软删除)。""" stmt = select(ChannelUserIdentity).where( ChannelUserIdentity.identity_id == identity_id, @@ -116,9 +110,7 @@ class UserIdentityRepository(BaseRepository): confidence=confidence, ) stmt = stmt.order_by(ChannelUserIdentity.created_at.desc()) - effective_limit, effective_offset = self._resolve_pagination( - page, page_size, limit, offset - ) + 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) @@ -271,9 +263,7 @@ class UserIdentityRepository(BaseRepository): result = await self.db.execute(stmt) return list(result.scalars().all()) - async def list_by_merged_from( - self, identity_id: str - ) -> list[ChannelUserIdentity]: + async def list_by_merged_from(self, identity_id: str) -> list[ChannelUserIdentity]: """查询合并来源包含指定 identity_id 的记录(FR-07 合并回滚)。""" stmt = ( select(ChannelUserIdentity)