refactor(repositories): 统一格式化仓库方法签名并新增路由绑定仓储

1.  统一所有仓库类的方法换行格式，将多行参数折叠为单行
2.  新增ChannelRouteBindingRepository仓储实现，用于渠道路由绑定规则的数据访问封装

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

backend/package/yuxi/repositories/channels/base.py

@@ -100,9 +100,7 @@ class BaseRepository:
         当 ``commit=False`` 时仅执行 ``flush()`` 使变更进入会话但不提交，
         便于上层在事务中组合多个写操作。
         """
-        return await self._soft_delete_by_id(
+        return await self._soft_delete_by_id(record_id, updated_by=updated_by, commit=commit)
-            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 = (
+        stmt = select(self.model).where(self.model.id == record_id).where(self.model.is_deleted == 1)
-            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 = (
+        stmt = sa_delete(self.model).where(self.model.id == record_id).where(self.model.is_deleted == 1)
-            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 = (
+        stmt = sa_delete(self.model).where(self.model.is_deleted == 1).where(self.model.deleted_at < before)
-            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(
+                raise RepositoryError(f"模型 {self.model.__name__} 无字段 {field_name}")
-                    f"模型 {self.model.__name__} 无字段 {field_name}"
-                )
             stmt = stmt.where(column == value)
         result = await self.db.execute(stmt)
         return int(result.scalar() or 0)

backend/package/yuxi/repositories/channels/channel_account_repository.py

@@ -22,9 +22,7 @@ class ChannelAccountRepository(BaseRepository):
     # 默认增删改查
     # ------------------------------------------------------------------
 
-    async def create(
+    async def create(self, data: dict[str, Any], *, commit: bool = True) -> ChannelAccount:
-        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(
+    async def get_by_id(self, record_id: int, *, for_update: bool = False) -> ChannelAccount | None:
-        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(
+        effective_limit, effective_offset = self._resolve_pagination(page, page_size, limit, offset)
-            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 = (
+        stmt = update(ChannelAccount).where(ChannelAccount.id == record_id).where(self._not_deleted()).values(**values)
-            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()

backend/package/yuxi/repositories/channels/channel_audit_log_repository.py

@@ -24,9 +24,7 @@ class ChannelAuditLogRepository(BaseRepository):
     # 写操作（仅 create，append-only）
     # ------------------------------------------------------------------
 
-    async def create(
+    async def create(self, data: dict[str, Any], *, commit: bool = True) -> ChannelAuditLog:
-        self, data: dict[str, Any], *, commit: bool = True
-    ) -> ChannelAuditLog:
         """创建审计日志记录（append-only）。
 
         Args:
@@ -70,9 +68,7 @@ class ChannelAuditLogRepository(BaseRepository):
     # 查询
     # ------------------------------------------------------------------
 
-    async def get_by_id(
+    async def get_by_id(self, record_id: int, *, for_update: bool = False) -> ChannelAuditLog | None:
-        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(
+    async def get_by_audit_log_id(self, audit_log_id: str, *, for_update: bool = False) -> ChannelAuditLog | None:
-        self, audit_log_id: str, *, for_update: bool = False
-    ) -> ChannelAuditLog | None:
         """根据业务标识 audit_log_id 获取审计日志。"""
-        stmt = select(ChannelAuditLog).where(
+        stmt = select(ChannelAuditLog).where(ChannelAuditLog.audit_log_id == audit_log_id)
-            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(
+        effective_limit, effective_offset = self._resolve_pagination(page, page_size, limit, offset)
-            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] = {
+        by_operation: dict[str, int] = {row[0]: int(row[1]) for row in result.all() if row[0] is not None}
-            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] = {
+        by_result: dict[str, int] = {row[0]: int(row[1]) for row in result.all() if row[0] is not None}
-            row[0]: int(row[1]) for row in result.all() if row[0] is not None
-        }
 
         return {
             "total": total,

backend/package/yuxi/repositories/channels/channel_idempotency_repository.py

@@ -26,9 +26,7 @@ class ChannelIdempotencyRepository(BaseRepository):
     # 写操作
     # ------------------------------------------------------------------
 
-    async def create(
+    async def create(self, data: dict[str, Any], *, commit: bool = True) -> ChannelIdempotency:
-        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 = (
+        stmt = update(ChannelIdempotency).where(ChannelIdempotency.id == record_id).values(**values)
-            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(
+    async def get_by_id(self, record_id: int, *, for_update: bool = False) -> ChannelIdempotency | None:
-        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(
+    async def get_by_key(self, idempotency_key: str, *, for_update: bool = False) -> ChannelIdempotency | None:
-        self, idempotency_key: str, *, for_update: bool = False
-    ) -> ChannelIdempotency | None:
         """根据幂等键获取记录（重复请求回放）。"""
-        stmt = select(ChannelIdempotency).where(
+        stmt = select(ChannelIdempotency).where(ChannelIdempotency.idempotency_key == idempotency_key)
-            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(
+        stmt = sa_delete(ChannelIdempotency).where(ChannelIdempotency.created_at < before)
-            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(
+        stmt = sa_delete(ChannelIdempotency).where(ChannelIdempotency.idempotency_key == idempotency_key)
-            ChannelIdempotency.idempotency_key == idempotency_key
-        )
         result = await self.db.execute(stmt)
         if commit:
             await self.db.commit()

backend/package/yuxi/repositories/channels/channel_outbox_repository.py

@@ -21,9 +21,7 @@ class ChannelOutboxRepository(BaseRepository):
     # 默认增删改查
     # ------------------------------------------------------------------
 
-    async def create(
+    async def create(self, data: dict[str, Any], *, commit: bool = True) -> ChannelOutboxEntry:
-        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(
+    async def get_by_id(self, record_id: int, *, for_update: bool = False) -> ChannelOutboxEntry | None:
-        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(
+        effective_limit, effective_offset = self._resolve_pagination(page, page_size, limit, offset)
-            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 = (
+        stmt = select(ChannelOutboxEntry).where(ChannelOutboxEntry.status == "pending").where(self._not_deleted())
-            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.is_(None)) | (ChannelOutboxEntry.next_retry_at <= before)
-                | (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 = (
+        stmt = update(ChannelOutboxEntry).where(ChannelOutboxEntry.id == record_id).where(self._not_deleted())
-            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(
+    async def list_by_message_id(self, message_id: int) -> list[ChannelOutboxEntry]:
-        self, message_id: int
-    ) -> list[ChannelOutboxEntry]:
         """按消息 ID 查询投递状态（FR-22 状态回查）。"""
         stmt = (
             select(ChannelOutboxEntry)

backend/package/yuxi/repositories/channels/channel_pairing_repository.py

@@ -22,9 +22,7 @@ class ChannelPairingRepository(BaseRepository):
     # 默认增删改查
     # ------------------------------------------------------------------
 
-    async def create(
+    async def create(self, data: dict[str, Any], *, commit: bool = True) -> ChannelPairing:
-        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(
+    async def get_by_id(self, record_id: int, *, for_update: bool = False) -> ChannelPairing | None:
-        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(
+    async def get_by_pairing_id(self, pairing_id: str, *, for_update: bool = False) -> ChannelPairing | None:
-        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(
+        effective_limit, effective_offset = self._resolve_pagination(page, page_size, limit, offset)
-            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(
+    async def list_expired_pending(self, before: datetime, *, limit: int = 100) -> list[ChannelPairing]:
-        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 = (

backend/package/yuxi/repositories/channels/channel_report_repository.py

@@ -25,9 +25,7 @@ class ChannelReportRepository(BaseRepository):
     # 写操作
     # ------------------------------------------------------------------
 
-    async def create(
+    async def create(self, data: dict[str, Any], *, commit: bool = True) -> ChannelReport:
-        self, data: dict[str, Any], *, commit: bool = True
-    ) -> ChannelReport:
         """创建报告记录。
 
         Args:
@@ -135,9 +133,7 @@ class ChannelReportRepository(BaseRepository):
     # 查询
     # ------------------------------------------------------------------
 
-    async def get_by_report_id(
+    async def get_by_report_id(self, report_id: str, *, for_update: bool = False) -> ChannelReport | None:
-        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(
+    async def get_by_task_id(self, task_id: str, *, for_update: bool = False) -> ChannelReport | None:
-        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(
+        effective_limit, effective_offset = self._resolve_pagination(page, page_size, limit, offset)
-            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)

backend/package/yuxi/repositories/channels/channel_session_repository.py

@@ -26,9 +26,7 @@ class ChannelSessionRepository(BaseRepository):
     # 默认增删改查
     # ------------------------------------------------------------------
 
-    async def create(
+    async def create(self, data: dict[str, Any], *, commit: bool = True) -> ChannelSession:
-        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(
+    async def get_by_id(self, record_id: int, *, for_update: bool = False) -> ChannelSession | None:
-        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(
+    async def get_by_session_id(self, session_id: str, *, for_update: bool = False) -> ChannelSession | None:
-        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(
+        effective_limit, effective_offset = self._resolve_pagination(page, page_size, limit, offset)
-            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 = (
+        stmt = update(ChannelSession).where(ChannelSession.id == record_id).where(self._not_deleted()).values(**values)
-            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(
+    async def list_by_conversation(self, conversation_id: int) -> list[ChannelSession]:
-        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(
+            stmt = stmt.join(ChannelAccount, ChannelAccount.id == ChannelSession.account_id)
-                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),

backend/package/yuxi/repositories/channels/content_review_record_repository.py

@@ -32,9 +32,7 @@ class ChannelContentReviewRecordRepository(BaseRepository):
     # 写操作
     # ------------------------------------------------------------------
 
-    async def create(
+    async def create(self, data: dict[str, Any], *, commit: bool = True) -> ChannelContentReviewRecord:
-        self, data: dict[str, Any], *, commit: bool = True
-    ) -> ChannelContentReviewRecord:
         """创建审核记录。
 
         Args:
@@ -102,9 +100,7 @@ class ChannelContentReviewRecordRepository(BaseRepository):
     # 查询
     # ------------------------------------------------------------------
 
-    async def get_by_review_id(
+    async def get_by_review_id(self, review_id: str, *, for_update: bool = False) -> ChannelContentReviewRecord | None:
-        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(
+        effective_limit, effective_offset = self._resolve_pagination(page, page_size, limit, offset)
-            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_rate = (block_count / total_reviews * 100) if total_reviews > 0 else 0.0
-            (block_count / total_reviews * 100) if total_reviews > 0 else 0.0
-        )
 
         # 按分类切片：展开 categories JSON 数组后分组计数
-        category_expr = func.json_array_elements_text(
+        category_expr = func.json_array_elements_text(ChannelContentReviewRecord.categories).label("category")
-            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)}
+            {"category": row.category, "count": int(row.count or 0)} for row in category_result.all()
-            for row in category_result.all()
         ]
 
         # 时间序列：按 granularity 分桶，每桶含 reviewed / blocked 双计数
-        bucket_expr = func.date_trunc(
+        bucket_expr = func.date_trunc(granularity, ChannelContentReviewRecord.reviewed_at)
-            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_intervention_rate = (manual_count / total_reviews * 100) if total_reviews > 0 else 0.0
-            (manual_count / total_reviews * 100) if total_reviews > 0 else 0.0
-        )
 
         # 按分类切片：展开 categories JSON 数组后分组计数
-        category_expr = func.json_array_elements_text(
+        category_expr = func.json_array_elements_text(ChannelContentReviewRecord.categories).label("category")
-            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)}
+            {"category": row.category, "count": int(row.count or 0)} for row in category_result.all()
-            for row in category_result.all()
         ]
 
         # 时间序列：按 granularity 分桶，每桶含 pass / block 双计数
-        bucket_expr = func.date_trunc(
+        bucket_expr = func.date_trunc(granularity, ChannelContentReviewRecord.reviewed_at)
-            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(
+            stmt = select(func.count(ChannelContentReviewRecord.id)).select_from(ChannelContentReviewRecord)
-                ChannelContentReviewRecord
-            )
         else:
             stmt = select(ChannelContentReviewRecord)
         stmt = stmt.where(self._not_deleted())

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

backend/package/yuxi/repositories/channels/user_identity_repository.py

@@ -21,9 +21,7 @@ class UserIdentityRepository(BaseRepository):
     # 默认增删改查
     # ------------------------------------------------------------------
 
-    async def create(
+    async def create(self, data: dict[str, Any], *, commit: bool = True) -> ChannelUserIdentity:
-        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(
+    async def get_by_id(self, record_id: int, *, for_update: bool = False) -> ChannelUserIdentity | None:
-        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(
+    async def get_by_identity_id(self, identity_id: str, *, for_update: bool = False) -> ChannelUserIdentity | None:
-        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(
+        effective_limit, effective_offset = self._resolve_pagination(page, page_size, limit, offset)
-            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(
+    async def list_by_merged_from(self, identity_id: str) -> list[ChannelUserIdentity]:
-        self, identity_id: str
-    ) -> list[ChannelUserIdentity]:
         """查询合并来源包含指定 identity_id 的记录（FR-07 合并回滚）。"""
         stmt = (
             select(ChannelUserIdentity)