diff --git a/backend/package/yuxi/repositories/channels/__init__.py b/backend/package/yuxi/repositories/channels/__init__.py index afa2467e..4c5627de 100644 --- a/backend/package/yuxi/repositories/channels/__init__.py +++ b/backend/package/yuxi/repositories/channels/__init__.py @@ -1,9 +1,10 @@ """渠道仓储层聚合导出。 本包提供： -- 9 个 ``Channel*Repository`` / ``ChannelConversationRepository`` / - ``ChannelMessageRepository`` 实现类（直接操作 ORM Model，原子 CRUD，不含业务语义） -- ``Repositories`` 聚合 dataclass（共享 ``db`` 会话的 9 个仓储实例） +- 11 个 ``Channel*Repository`` / ``ChannelConversationRepository`` / + ``ChannelMessageRepository`` 实现类 + （直接操作 ORM Model，原子 CRUD，不含业务语义） +- ``Repositories`` 聚合 dataclass（共享 ``db`` 会话的 11 个仓储实例） - ``create_repositories(db)`` factory 函数（由框架层在请求开始时调用） 依赖边界：只依赖 ``yuxi.storage.postgres.models_channels`` / @@ -21,7 +22,7 @@ from dataclasses import dataclass from sqlalchemy.ext.asyncio import AsyncSession -from .base import BaseRepository +from .base import BaseRepository, RepositoryError from .channel_account_repository import ChannelAccountRepository from .channel_audit_log_repository import ChannelAuditLogRepository from .channel_conversation_repository import ChannelConversationRepository @@ -29,20 +30,25 @@ 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 .user_identity_repository import UserIdentityRepository __all__ = [ "BaseRepository", "ChannelAccountRepository", "ChannelAuditLogRepository", + "ChannelContentReviewRecordRepository", "ChannelConversationRepository", "ChannelIdempotencyRepository", "ChannelMessageRepository", "ChannelOutboxRepository", "ChannelPairingRepository", + "ChannelReportRepository", "ChannelSessionRepository", "Repositories", + "RepositoryError", "UserIdentityRepository", "create_repositories", ] @@ -64,6 +70,8 @@ class Repositories: idempotency: ChannelIdempotencyRepository conversation: ChannelConversationRepository message: ChannelMessageRepository + report: ChannelReportRepository + content_review_record: ChannelContentReviewRecordRepository def create_repositories(db: AsyncSession) -> Repositories: @@ -83,4 +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), ) diff --git a/backend/package/yuxi/repositories/channels/base.py b/backend/package/yuxi/repositories/channels/base.py index b90d57e2..5b2411a4 100644 --- a/backend/package/yuxi/repositories/channels/base.py +++ b/backend/package/yuxi/repositories/channels/base.py @@ -19,6 +19,11 @@ - ``hard_delete_by_id`` —— 按主键物理删除记录（仅删除 ``is_deleted=1`` 的记录） - ``hard_delete_before`` —— 物理删除 ``deleted_at`` 早于截止时间的软删除记录 - ``count_undeleted_by_fields`` —— 按字段条件统计未删除记录数（恢复前唯一约束预检） + +依赖边界：仅依赖 ``sqlalchemy`` / ``yuxi.utils``，**不依赖** +``yuxi.channels``（领域层 / 契约层 / 适配器层），避免基础设施层反向依赖 +领域层。仓储层抛出的 ``RepositoryError`` 由适配器层（``channels/adapters``） +捕获并翻译为 ``DependencyError``，确保原生异常不穿透到核心层。 """ from __future__ import annotations @@ -31,6 +36,19 @@ from sqlalchemy.ext.asyncio import AsyncSession from yuxi.utils.datetime_utils import utc_now_naive +class RepositoryError(Exception): + """仓储层基础设施异常。 + + 用于表达仓储内部编程错误（如字段名映射不存在）等不应穿透到核心层的 + 故障。适配器层应捕获 ``RepositoryError`` 并翻译为契约层 + ``DependencyError``，保持核心层不感知基础设施异常类型。 + + Note: + 本类刻意不继承 ``yuxi.channels.contract.errors.Error``，以避免 + 基础设施层反向依赖领域契约层。 + """ + + class BaseRepository: """渠道仓库基类。 @@ -197,15 +215,15 @@ class BaseRepository: async def hard_delete_before( self, - cutoff_at: datetime, + before: datetime, *, commit: bool = True, ) -> int: - """物理删除 deleted_at 早于 cutoff_at 的软删除记录。返回受影响行数。""" + """物理删除 deleted_at 早于 before 的软删除记录。返回受影响行数。""" stmt = ( sa_delete(self.model) .where(self.model.is_deleted == 1) - .where(self.model.deleted_at < cutoff_at) + .where(self.model.deleted_at < before) ) result = await self.db.execute(stmt) if commit: @@ -224,6 +242,9 @@ class BaseRepository: **field_filters: 字段名 → 值的映射，将转为 ``where(model.field == value)`` 条件。 字段名必须为 ORM 模型的实际列名（如 slug / system_id / env_key）。 + Raises: + RepositoryError: 字段名在 ORM 模型上不存在（编程错误，应由调用方修正）。 + Example: await repo.count_undeleted_by_fields(slug="old_crm") await repo.count_undeleted_by_fields(system_id=1, env_key="prod") @@ -232,7 +253,9 @@ class BaseRepository: for field_name, value in field_filters.items(): column = getattr(self.model, field_name, None) if column is None: - raise ValueError(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 bd1c8e83..b124a98d 100644 --- a/backend/package/yuxi/repositories/channels/channel_account_repository.py +++ b/backend/package/yuxi/repositories/channels/channel_account_repository.py @@ -49,6 +49,7 @@ class ChannelAccountRepository(BaseRepository): 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"), updated_by=data.get("updated_by"), 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 f17b4dd4..7d80288d 100644 --- a/backend/package/yuxi/repositories/channels/channel_audit_log_repository.py +++ b/backend/package/yuxi/repositories/channels/channel_audit_log_repository.py @@ -6,7 +6,7 @@ from __future__ import annotations import datetime as dt from typing import Any -from sqlalchemy import func, select +from sqlalchemy import delete as sa_delete, func, select from yuxi.repositories.channels.base import BaseRepository from yuxi.storage.postgres.models_channels import ChannelAuditLog @@ -70,19 +70,25 @@ class ChannelAuditLogRepository(BaseRepository): # 查询 # ------------------------------------------------------------------ - async def get_by_id(self, record_id: int) -> 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: + stmt = stmt.with_for_update() result = await self.db.execute(stmt) return result.scalar_one_or_none() async def get_by_audit_log_id( - self, audit_log_id: str + 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 ) + if for_update: + stmt = stmt.with_for_update() result = await self.db.execute(stmt) return result.scalar_one_or_none() @@ -94,8 +100,8 @@ class ChannelAuditLogRepository(BaseRepository): target_channel: str | None = None, target_account: str | None = None, trace_id: str | None = None, - start: dt.datetime | None = None, - end: dt.datetime | None = None, + start_time: dt.datetime | None = None, + end_time: dt.datetime | None = None, page: int | None = None, page_size: int | None = None, limit: int | None = None, @@ -108,8 +114,8 @@ class ChannelAuditLogRepository(BaseRepository): target_channel=target_channel, target_account=target_account, trace_id=trace_id, - start=start, - end=end, + start_time=start_time, + end_time=end_time, ) stmt = stmt.order_by(ChannelAuditLog.timestamp.desc()) effective_limit, effective_offset = self._resolve_pagination( @@ -128,8 +134,8 @@ class ChannelAuditLogRepository(BaseRepository): target_channel: str | None = None, target_account: str | None = None, trace_id: str | None = None, - start: dt.datetime | None = None, - end: dt.datetime | None = None, + start_time: dt.datetime | None = None, + end_time: dt.datetime | None = None, ) -> int: """统计满足过滤条件的审计日志数量。""" stmt = self._build_query( @@ -138,8 +144,8 @@ class ChannelAuditLogRepository(BaseRepository): target_channel=target_channel, target_account=target_account, trace_id=trace_id, - start=start, - end=end, + start_time=start_time, + end_time=end_time, select_count=True, ) result = await self.db.execute(stmt) @@ -181,25 +187,6 @@ class ChannelAuditLogRepository(BaseRepository): result = await self.db.execute(stmt) return [row[0] for row in result.all() if row[0]] - async def count_by_operation( - self, - *, - start: dt.datetime | None = None, - end: dt.datetime | None = None, - ) -> dict[str, int]: - """按操作类型分组统计日志数量。""" - stmt = ( - select(ChannelAuditLog.operation, func.count(ChannelAuditLog.id)) - .where(ChannelAuditLog.operation.isnot(None)) - .group_by(ChannelAuditLog.operation) - ) - if start is not None: - stmt = stmt.where(ChannelAuditLog.timestamp >= start) - if end is not None: - stmt = stmt.where(ChannelAuditLog.timestamp <= end) - result = await self.db.execute(stmt) - return {row[0]: int(row[1]) for row in result.all() if row[0]} - async def delete_old_logs( self, before: dt.datetime, @@ -210,8 +197,6 @@ class ChannelAuditLogRepository(BaseRepository): 审计日志为 append-only，不使用软删除，直接物理删除。 """ - from sqlalchemy import delete as sa_delete - stmt = sa_delete(ChannelAuditLog).where(ChannelAuditLog.timestamp < before) result = await self.db.execute(stmt) if commit: @@ -227,13 +212,13 @@ class ChannelAuditLogRepository(BaseRepository): operator: str | None = None, target_channel: str | None = None, target_account: str | None = None, - start: dt.datetime | None = None, - end: dt.datetime | None = None, + start_time: dt.datetime | None = None, + end_time: dt.datetime | None = None, ) -> dict[str, Any]: """按条件统计审计日志聚合（总数 / 按操作类型分组 / 按结果分组 / 时间范围）。 与 ``list`` / ``count`` 共享同一组过滤字段，返回聚合统计结果供 - 适配器构造 ``AuditLogStats`` DTO。统计查询不含分页参数，对全量 + 适配器构造 ``AuditLogStats`` DTO。统计查询不含分页参数，对全量 匹配数据聚合。 Args: @@ -241,8 +226,8 @@ class ChannelAuditLogRepository(BaseRepository): operator: 操作人过滤（可选）。 target_channel: 目标渠道过滤（可选）。 target_account: 目标账户过滤（可选）。 - start: 起始时间过滤（可选）。 - end: 结束时间过滤（可选）。 + start_time: 起始时间过滤（可选）。 + end_time: 结束时间过滤（可选）。 Returns: 含 ``total`` / ``by_operation`` / ``by_result`` / @@ -257,10 +242,10 @@ class ChannelAuditLogRepository(BaseRepository): conditions.append(ChannelAuditLog.target_channel == target_channel) if target_account is not None: conditions.append(ChannelAuditLog.target_account == target_account) - if start is not None: - conditions.append(ChannelAuditLog.timestamp >= start) - if end is not None: - conditions.append(ChannelAuditLog.timestamp <= end) + if start_time is not None: + conditions.append(ChannelAuditLog.timestamp >= start_time) + if end_time is not None: + conditions.append(ChannelAuditLog.timestamp <= end_time) total_stmt = select( func.count(ChannelAuditLog.id), @@ -315,11 +300,14 @@ class ChannelAuditLogRepository(BaseRepository): target_channel: str | None = None, target_account: str | None = None, trace_id: str | None = None, - start: dt.datetime | None = None, - end: dt.datetime | None = None, + start_time: dt.datetime | None = None, + end_time: dt.datetime | None = None, select_count: bool = False, ): - """构建查询语句。审计日志不附加软删除过滤（is_deleted 恒为 0）。""" + """构建查询语句。审计日志不附加软删除过滤（is_deleted 恒为 0）。 + + ``start_time`` / ``end_time`` 按 ``timestamp`` 时间范围过滤。 + """ if select_count: stmt = select(func.count(ChannelAuditLog.id)).select_from(ChannelAuditLog) else: @@ -334,8 +322,8 @@ class ChannelAuditLogRepository(BaseRepository): stmt = stmt.where(ChannelAuditLog.target_account == target_account) if trace_id is not None: stmt = stmt.where(ChannelAuditLog.trace_id == trace_id) - if start is not None: - stmt = stmt.where(ChannelAuditLog.timestamp >= start) - if end is not None: - stmt = stmt.where(ChannelAuditLog.timestamp <= end) + if start_time is not None: + stmt = stmt.where(ChannelAuditLog.timestamp >= start_time) + if end_time is not None: + stmt = stmt.where(ChannelAuditLog.timestamp <= end_time) return stmt diff --git a/backend/package/yuxi/repositories/channels/channel_idempotency_repository.py b/backend/package/yuxi/repositories/channels/channel_idempotency_repository.py index be5b4f6f..01b90fab 100644 --- a/backend/package/yuxi/repositories/channels/channel_idempotency_repository.py +++ b/backend/package/yuxi/repositories/channels/channel_idempotency_repository.py @@ -6,7 +6,7 @@ from __future__ import annotations import datetime as dt from typing import Any -from sqlalchemy import delete as sa_delete, func, select, update +from sqlalchemy import delete as sa_delete, select, update from yuxi.repositories.channels.base import BaseRepository from yuxi.storage.postgres.models_channels import ChannelIdempotency @@ -105,64 +105,28 @@ class ChannelIdempotencyRepository(BaseRepository): # 查询 # ------------------------------------------------------------------ - async def get_by_id(self, record_id: int) -> 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: + 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 + self, idempotency_key: str, *, for_update: bool = False ) -> ChannelIdempotency | None: """根据幂等键获取记录（重复请求回放）。""" stmt = select(ChannelIdempotency).where( ChannelIdempotency.idempotency_key == idempotency_key ) + if for_update: + stmt = stmt.with_for_update() result = await self.db.execute(stmt) return result.scalar_one_or_none() - async def list( - self, - *, - operation: str | None = None, - account_id: int | None = None, - page: int | None = None, - page_size: int | None = None, - limit: int | None = None, - offset: int | None = None, - ) -> list[ChannelIdempotency]: - """列出幂等记录，支持过滤与分页。""" - stmt = select(ChannelIdempotency) - if operation is not None: - stmt = stmt.where(ChannelIdempotency.operation == operation) - if account_id is not None: - stmt = stmt.where(ChannelIdempotency.account_id == account_id) - stmt = stmt.order_by(ChannelIdempotency.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, - *, - operation: str | None = None, - account_id: int | None = None, - ) -> int: - """统计幂等记录数量。""" - stmt = select(func.count(ChannelIdempotency.id)).select_from( - ChannelIdempotency - ) - if operation is not None: - stmt = stmt.where(ChannelIdempotency.operation == operation) - if account_id is not None: - stmt = stmt.where(ChannelIdempotency.account_id == account_id) - result = await self.db.execute(stmt) - return int(result.scalar() or 0) - async def list_in_progress_stale( self, before: dt.datetime, diff --git a/backend/package/yuxi/repositories/channels/channel_message_repository.py b/backend/package/yuxi/repositories/channels/channel_message_repository.py index 99b4a389..63678d72 100644 --- a/backend/package/yuxi/repositories/channels/channel_message_repository.py +++ b/backend/package/yuxi/repositories/channels/channel_message_repository.py @@ -37,7 +37,7 @@ class ChannelMessageRepository: channel_status: str | None = None, channel_msg_id: str | None = None, ref_channel_msg_id: str | None = None, - channel_status_history: list[dict] | None = None, + channel_status_history: tuple[dict, ...] | None = None, commit: bool = True, ) -> Message | None: """初始化 Message 的渠道扩展字段（FR-09）。 @@ -50,7 +50,7 @@ class ChannelMessageRepository: channel_status: 渠道侧初始状态（如 ``sent``）。 channel_msg_id: 渠道侧消息 ID。 ref_channel_msg_id: 引用的渠道消息 ID（编辑、回复场景）。 - channel_status_history: 初始状态历史数组。 + channel_status_history: 初始状态历史（不可变元组，内部转 list 落库）。 commit: True 时提交事务，False 时仅 flush。 Returns: diff --git a/backend/package/yuxi/repositories/channels/channel_outbox_repository.py b/backend/package/yuxi/repositories/channels/channel_outbox_repository.py index 9cfacb89..e4e23711 100644 --- a/backend/package/yuxi/repositories/channels/channel_outbox_repository.py +++ b/backend/package/yuxi/repositories/channels/channel_outbox_repository.py @@ -94,6 +94,8 @@ class ChannelOutboxRepository(BaseRepository): async def get_by_channel_msg_id( self, channel_msg_id: str, + *, + for_update: bool = False, ) -> ChannelOutboxEntry | None: """按渠道消息 ID 查询发件箱条目（FR-09 状态回写定位）。 @@ -102,6 +104,7 @@ class ChannelOutboxRepository(BaseRepository): Args: channel_msg_id: 渠道侧消息 ID。 + for_update: True 时加悲观锁。 Returns: 匹配的发件箱条目；不存在时返回 None。 @@ -113,6 +116,8 @@ class ChannelOutboxRepository(BaseRepository): .order_by(ChannelOutboxEntry.created_at.desc()) .limit(1) ) + if for_update: + stmt = stmt.with_for_update() result = await self.db.execute(stmt) return result.scalar_one_or_none() @@ -180,24 +185,24 @@ class ChannelOutboxRepository(BaseRepository): self, *, limit: int = 100, - before_at: dt.datetime | None = None, + before: dt.datetime | None = None, ) -> list[ChannelOutboxEntry]: """列出待发送的发件箱条目（FR-22 Worker 拉取）。 Args: limit: 返回数量上限。 - before_at: 截止时间，next_retry_at 早于此时间的 PENDING 条目将被列出。 - None 时不过滤 next_retry_at。 + before: 截止时间，next_retry_at 早于此时间的 PENDING 条目将被列出。 + None 时不过滤 next_retry_at。 """ stmt = ( select(ChannelOutboxEntry) .where(ChannelOutboxEntry.status == "pending") .where(self._not_deleted()) ) - if before_at is not None: + if before is not None: stmt = stmt.where( (ChannelOutboxEntry.next_retry_at.is_(None)) - | (ChannelOutboxEntry.next_retry_at <= before_at) + | (ChannelOutboxEntry.next_retry_at <= before) ) stmt = stmt.order_by(ChannelOutboxEntry.created_at.asc()).limit(limit) result = await self.db.execute(stmt) @@ -225,54 +230,6 @@ class ChannelOutboxRepository(BaseRepository): result = await self.db.execute(stmt) return list(result.scalars().all()) - async def list_retryable( - self, - *, - limit: int = 100, - before_at: dt.datetime | None = None, - ) -> list[ChannelOutboxEntry]: - """列出可重试的失败条目（FR-22 重试任务扫描）。 - - Args: - limit: 返回数量上限。 - before_at: 截止时间，next_retry_at 早于此时间的 FAILED 条目将被列出。 - """ - stmt = ( - select(ChannelOutboxEntry) - .where(ChannelOutboxEntry.status == "failed") - .where(self._not_deleted()) - ) - if before_at is not None: - stmt = stmt.where(ChannelOutboxEntry.next_retry_at <= before_at) - stmt = stmt.order_by(ChannelOutboxEntry.next_retry_at.asc()).limit(limit) - result = await self.db.execute(stmt) - return list(result.scalars().all()) - - async def list_expired( - self, - *, - limit: int = 100, - before_at: dt.datetime | None = None, - ) -> list[ChannelOutboxEntry]: - """列出已过期但仍为 PENDING/FAILED 的条目（标记 dead 任务扫描）。 - - Args: - limit: 返回数量上限。 - before_at: 截止时间，expires_at 早于此时间的条目将被列出。 - None 时使用当前时间。 - """ - cutoff = before_at or utc_now_naive() - stmt = ( - select(ChannelOutboxEntry) - .where(ChannelOutboxEntry.status.in_(("pending", "failed"))) - .where(ChannelOutboxEntry.expires_at < cutoff) - .where(self._not_deleted()) - .order_by(ChannelOutboxEntry.expires_at.asc()) - .limit(limit) - ) - result = await self.db.execute(stmt) - return list(result.scalars().all()) - async def update_status( self, record_id: int, @@ -356,8 +313,7 @@ class ChannelOutboxRepository(BaseRepository): """原子抢占投递锁（at-least-once 语义）。 仅当 ``locked_by`` 为空时才能抢占成功，保证同一时刻只有一个 worker - 处理该条目。worker 崩溃后由回收任务通过 ``list_stale_locked`` 发现 - 超时锁并重置后重新投递。 + 处理该条目。 Args: record_id: 发件箱条目主键。 @@ -415,29 +371,6 @@ class ChannelOutboxRepository(BaseRepository): await self.db.flush() return result.rowcount - async def list_stale_locked( - self, - lock_timeout: dt.datetime, - *, - limit: int = 100, - ) -> list[ChannelOutboxEntry]: - """列出锁已超时的条目（worker 崩溃回收）。 - - Args: - lock_timeout: 锁超时阈值，locked_at 早于此时间的条目将被列出。 - limit: 返回数量上限。 - """ - stmt = ( - select(ChannelOutboxEntry) - .where(ChannelOutboxEntry.locked_by.isnot(None)) - .where(ChannelOutboxEntry.locked_at < lock_timeout) - .where(self._not_deleted()) - .order_by(ChannelOutboxEntry.locked_at.asc()) - .limit(limit) - ) - result = await self.db.execute(stmt) - return list(result.scalars().all()) - async def list_by_message_id( self, message_id: int ) -> list[ChannelOutboxEntry]: @@ -471,15 +404,9 @@ class ChannelOutboxRepository(BaseRepository): status: str | None = None, account_id: int | None = None, message_id: int | None = None, - select_count: bool = False, ): """构建查询语句，统一附加软删除过滤。""" - if select_count: - stmt = select(func.count(ChannelOutboxEntry.id)).select_from( - ChannelOutboxEntry - ) - else: - stmt = select(ChannelOutboxEntry) + stmt = select(ChannelOutboxEntry) stmt = stmt.where(self._not_deleted()) if status is not None: stmt = stmt.where(ChannelOutboxEntry.status == status) diff --git a/backend/package/yuxi/repositories/channels/channel_pairing_repository.py b/backend/package/yuxi/repositories/channels/channel_pairing_repository.py index 61acf731..6571b749 100644 --- a/backend/package/yuxi/repositories/channels/channel_pairing_repository.py +++ b/backend/package/yuxi/repositories/channels/channel_pairing_repository.py @@ -89,6 +89,8 @@ class ChannelPairingRepository(BaseRepository): 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, page: int | None = None, page_size: int | None = None, limit: int | None = None, @@ -103,6 +105,8 @@ class ChannelPairingRepository(BaseRepository): 跨渠道查询场景，可能匹配多个渠道账户）。 peer_id: 对端用户 ID 过滤。 status: 状态过滤。 + start_time: 起始时间（按 ``requested_at`` 过滤，含）。 + end_time: 截止时间（按 ``requested_at`` 过滤，含）。 page: 页码（与 ``page_size`` 共同使用）。 page_size: 每页大小。 limit: 返回数量上限（与 ``offset`` 共同使用）。 @@ -113,6 +117,8 @@ class ChannelPairingRepository(BaseRepository): account_ids=account_ids, peer_id=peer_id, status=status, + start_time=start_time, + end_time=end_time, ) stmt = stmt.order_by(ChannelPairing.requested_at.desc()) effective_limit, effective_offset = self._resolve_pagination( @@ -194,18 +200,18 @@ class ChannelPairingRepository(BaseRepository): return list(result.scalars().all()) async def list_expired_pending( - self, before_at: datetime, *, limit: int = 100 + self, before: datetime, *, limit: int = 100 ) -> list[ChannelPairing]: """列出已过期但仍为 PENDING 的配对（过期清理任务）。 Args: - before_at: 截止时间，expires_at 早于此时间的 PENDING 记录将被列出。 + before: 截止时间，expires_at 早于此时间的 PENDING 记录将被列出。 limit: 返回数量上限。 """ stmt = ( select(ChannelPairing) .where(ChannelPairing.status == "pending") - .where(ChannelPairing.expires_at < before_at) + .where(ChannelPairing.expires_at < before) .where(self._not_deleted()) .order_by(ChannelPairing.expires_at.asc()) .limit(limit) @@ -215,7 +221,7 @@ class ChannelPairingRepository(BaseRepository): async def batch_expire_pending( self, - before_at: datetime, + before: datetime, *, commit: bool = True, ) -> int: @@ -228,7 +234,7 @@ class ChannelPairingRepository(BaseRepository): stmt = ( update(ChannelPairing) .where(ChannelPairing.status == "pending") - .where(ChannelPairing.expires_at < before_at) + .where(ChannelPairing.expires_at < before) .where(self._not_deleted()) .values(status="expired", expired_at=now, updated_at=now) ) @@ -260,13 +266,14 @@ class ChannelPairingRepository(BaseRepository): account_ids: list[int] | None = None, peer_id: str | None = None, status: str | None = None, - select_count: bool = False, + start_time: datetime | None = None, + end_time: datetime | None = None, ): - """构建查询语句，统一附加软删除过滤。""" - if select_count: - stmt = select(func.count(ChannelPairing.id)).select_from(ChannelPairing) - else: - stmt = select(ChannelPairing) + """构建查询语句，统一附加软删除过滤。 + + ``start_time`` / ``end_time`` 按 ``requested_at`` 时间范围过滤。 + """ + stmt = select(ChannelPairing) stmt = stmt.where(self._not_deleted()) if account_id is not None: stmt = stmt.where(ChannelPairing.account_id == account_id) @@ -276,4 +283,8 @@ class ChannelPairingRepository(BaseRepository): stmt = stmt.where(ChannelPairing.peer_id == peer_id) if status is not None: stmt = stmt.where(ChannelPairing.status == status) + if start_time is not None: + stmt = stmt.where(ChannelPairing.requested_at >= start_time) + if end_time is not None: + stmt = stmt.where(ChannelPairing.requested_at <= end_time) return stmt diff --git a/backend/package/yuxi/repositories/channels/channel_report_repository.py b/backend/package/yuxi/repositories/channels/channel_report_repository.py new file mode 100644 index 00000000..f0fda04b --- /dev/null +++ b/backend/package/yuxi/repositories/channels/channel_report_repository.py @@ -0,0 +1,302 @@ +# 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/channel_session_repository.py b/backend/package/yuxi/repositories/channels/channel_session_repository.py index 13754630..ae5784e9 100644 --- a/backend/package/yuxi/repositories/channels/channel_session_repository.py +++ b/backend/package/yuxi/repositories/channels/channel_session_repository.py @@ -3,16 +3,20 @@ from __future__ import annotations -from datetime import datetime +from datetime import datetime, timedelta from typing import Any -from sqlalchemy import func, select, update +from sqlalchemy import func, or_, select, update from yuxi.repositories.channels.base import BaseRepository -from yuxi.storage.postgres.models_channels import ChannelSession +from yuxi.storage.postgres.models_channels import ChannelAccount, ChannelSession from yuxi.utils.datetime_utils import utc_now_naive +# 僵尸会话阈值：最近消息时间早于该天数视为僵尸会话（仅未关闭） +_ZOMBIE_SESSION_DAYS = 7 + + class ChannelSessionRepository(BaseRepository): """``channel_sessions`` 表的异步数据访问封装。""" @@ -116,6 +120,13 @@ class ChannelSessionRepository(BaseRepository): unified_identity_id: str | None = None, conversation_id: int | None = None, is_temporary: bool | None = None, + start_time: datetime | None = None, + end_time: datetime | None = None, + owner_peer_id: str | None = None, + status: str | None = None, + last_message_after: datetime | None = None, + last_message_before: datetime | None = None, + abnormal: bool = False, page: int | None = None, page_size: int | None = None, limit: int | None = None, @@ -123,8 +134,9 @@ class ChannelSessionRepository(BaseRepository): ) -> list[ChannelSession]: """列出渠道会话，支持过滤与分页。 - ``peer_id`` 提供时执行模糊匹配（``LIKE '%peer_id%'``），供运维排查 - 按对端标识检索会话。 + ``peer_id`` 提供时执行大小写不敏感的模糊匹配（``ILIKE '%peer_id%'``）， + 转义 SQL LIKE 特殊字符，供运维排查按对端标识检索会话。 + ``start_time`` / ``end_time`` 提供时按会话创建时间范围过滤。 """ stmt = self._build_query( account_id=account_id, @@ -133,8 +145,15 @@ class ChannelSessionRepository(BaseRepository): unified_identity_id=unified_identity_id, conversation_id=conversation_id, is_temporary=is_temporary, + start_time=start_time, + end_time=end_time, + owner_peer_id=owner_peer_id, + status=status, + last_message_after=last_message_after, + last_message_before=last_message_before, + abnormal=abnormal, ) - stmt = stmt.order_by(ChannelSession.created_at.desc()) + stmt = stmt.order_by(ChannelSession.last_message_at.desc().nullslast()) effective_limit, effective_offset = self._resolve_pagination( page, page_size, limit, offset ) @@ -152,6 +171,13 @@ class ChannelSessionRepository(BaseRepository): unified_identity_id: str | None = None, conversation_id: int | None = None, is_temporary: bool | None = None, + start_time: datetime | None = None, + end_time: datetime | None = None, + owner_peer_id: str | None = None, + status: str | None = None, + last_message_after: datetime | None = None, + last_message_before: datetime | None = None, + abnormal: bool = False, ) -> int: """统计满足过滤条件的未软删除会话数。 @@ -164,6 +190,13 @@ class ChannelSessionRepository(BaseRepository): unified_identity_id=unified_identity_id, conversation_id=conversation_id, is_temporary=is_temporary, + start_time=start_time, + end_time=end_time, + owner_peer_id=owner_peer_id, + status=status, + last_message_after=last_message_after, + last_message_before=last_message_before, + abnormal=abnormal, select_count=True, ) result = await self.db.execute(stmt) @@ -232,21 +265,6 @@ class ChannelSessionRepository(BaseRepository): await self.db.flush() return result.rowcount - async def soft_delete_by_id( - self, - record_id: int, - *, - updated_by: str | None = None, - commit: bool = True, - ) -> int: - """软删除渠道会话（合并场景 FR-07）。 - - 复用基类 _soft_delete_by_id，设置 is_deleted=1, deleted_at=now。 - """ - return await self._soft_delete_by_id( - record_id, updated_by=updated_by, commit=commit - ) - async def update_last_message_at( self, record_id: int, @@ -307,26 +325,6 @@ class ChannelSessionRepository(BaseRepository): result = await self.db.execute(stmt) return list(result.scalars().all()) - async def list_temporary_inactive( - self, before_at: datetime, *, limit: int = 100 - ) -> list[ChannelSession]: - """列出超时的临时会话（FR-27 清理任务）。 - - Args: - before_at: 截止时间，last_message_at 早于此时间的临时会话将被列出。 - limit: 返回数量上限。 - """ - stmt = ( - select(ChannelSession) - .where(ChannelSession.is_temporary.is_(True)) - .where(self._not_deleted()) - .where(ChannelSession.last_message_at < before_at) - .order_by(ChannelSession.last_message_at.asc()) - .limit(limit) - ) - result = await self.db.execute(stmt) - return list(result.scalars().all()) - # ------------------------------------------------------------------ # 内部查询构建 # ------------------------------------------------------------------ @@ -340,11 +338,21 @@ class ChannelSessionRepository(BaseRepository): unified_identity_id: str | None = None, conversation_id: int | None = None, is_temporary: bool | None = None, + start_time: datetime | None = None, + end_time: datetime | None = None, + owner_peer_id: str | None = None, + status: str | None = None, + last_message_after: datetime | None = None, + last_message_before: datetime | None = None, + abnormal: bool = False, select_count: bool = False, ): """构建查询语句，统一附加软删除过滤。 - ``peer_id`` 提供时执行模糊匹配（``LIKE '%peer_id%'``）。 + ``peer_id`` 提供时执行大小写不敏感的模糊匹配（``ILIKE '%peer_id%'``）， + 并通过 ``_escape_like_pattern`` 转义 ``%`` / ``_`` / ``\\`` 特殊字符， + 与 ``ChannelAccountRepository`` 的关键词搜索约定一致。 + ``start_time`` / ``end_time`` 提供时按会话创建时间范围过滤。 """ if select_count: stmt = select(func.count(ChannelSession.id)).select_from(ChannelSession) @@ -356,11 +364,42 @@ class ChannelSessionRepository(BaseRepository): if channel_type is not None: stmt = stmt.where(ChannelSession.channel_type == channel_type) if peer_id is not None: - stmt = stmt.where(ChannelSession.peer_id.like(f"%{peer_id}%")) + pattern = f"%{self._escape_like_pattern(peer_id)}%" + stmt = stmt.where(ChannelSession.peer_id.ilike(pattern, escape="\\")) if unified_identity_id is not None: stmt = stmt.where(ChannelSession.unified_identity_id == unified_identity_id) if conversation_id is not None: stmt = stmt.where(ChannelSession.conversation_id == conversation_id) if is_temporary is not None: stmt = stmt.where(ChannelSession.is_temporary.is_(is_temporary)) + if start_time is not None: + stmt = stmt.where(ChannelSession.created_at >= start_time) + if end_time is not None: + stmt = stmt.where(ChannelSession.created_at <= end_time) + if owner_peer_id is not None: + stmt = stmt.where(ChannelSession.owner_peer_id == owner_peer_id) + if status is not None: + if status == "active": + stmt = stmt.where(ChannelSession.closed_at.is_(None)) + elif status == "closed": + stmt = stmt.where(ChannelSession.closed_at.is_not(None)) + if last_message_after is not None: + stmt = stmt.where(ChannelSession.last_message_at >= last_message_after) + 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 + ) + zombie_threshold = utc_now_naive() - timedelta(days=_ZOMBIE_SESSION_DAYS) + zombie_cond = or_( + ChannelSession.last_message_at.is_(None), + ChannelSession.last_message_at < zombie_threshold, + ) & ChannelSession.closed_at.is_(None) + account_unavailable_cond = ChannelAccount.status != "active" + ownerless_cond = or_( + ChannelSession.owner_peer_id.is_(None), + ChannelSession.owner_peer_id == "", + ) + stmt = stmt.where(or_(zombie_cond, account_unavailable_cond, ownerless_cond)) 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 new file mode 100644 index 00000000..3868c388 --- /dev/null +++ b/backend/package/yuxi/repositories/channels/content_review_record_repository.py @@ -0,0 +1,464 @@ +# yuxi/repositories/channels/content_review_record_repository.py +"""内容审核历史记录数据访问层。""" + +from __future__ import annotations + +from datetime import datetime +from typing import Any + +from sqlalchemy import case, func, select + +from yuxi.repositories.channels.base import BaseRepository +from yuxi.storage.postgres.models_channels import ChannelContentReviewRecord + + +class ChannelContentReviewRecordRepository(BaseRepository): + """``channel_content_review_records`` 表的异步数据访问封装。 + + 持久化审核结果与命中片段，独立于 ``channel_audit_logs``（操作记录） + 与 ``channel_outbox_entries``（投递记录）。 + + Note: + 本表的 ``account_id`` 列为 ``String(128)`` 业务标识（对齐 + ``channel_accounts.account_id``），与子包内其他仓储的 ``account_id`` + （``Integer`` ORM 外键 → ``channel_accounts.id``）语义不同。 + ``list`` / ``count`` / ``query_stats`` 的 ``account_id`` 参数 + 均为字符串业务标识。 + """ + + model = ChannelContentReviewRecord + + # ------------------------------------------------------------------ + # 写操作 + # ------------------------------------------------------------------ + + async def create( + self, data: dict[str, Any], *, commit: bool = True + ) -> ChannelContentReviewRecord: + """创建审核记录。 + + Args: + data: 审核记录字段字典，必须包含 review_id / channel_type / + account_id / resource_type / content_preview / verdict / + confidence / categories / detail / reviewed_at / reviewer / source。 + commit: True 时提交事务，False 时仅 flush。 + """ + record = ChannelContentReviewRecord( + review_id=data["review_id"], + channel_type=data["channel_type"], + account_id=data["account_id"], + resource_type=data["resource_type"], + content_preview=data["content_preview"], + verdict=data["verdict"], + confidence=data["confidence"], + categories=data.get("categories") or [], + detail=data.get("detail") or [], + reviewed_at=data["reviewed_at"], + reviewer=data["reviewer"], + source=data["source"], + trace_id=data.get("trace_id"), + created_by=data.get("created_by"), + updated_by=data.get("updated_by"), + ) + 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_verdict( + self, + review_id: str, + verdict: str, + reviewer: str, + *, + commit: bool = True, + ) -> ChannelContentReviewRecord | None: + """更新审核记录结论与审核人；记录不存在返回 None。 + + 用于人工覆盖审核结论（CR-DECISION-BATCH）。与子包内其他 ``update`` + 方法一致，无条件 ``refresh`` 以返回最新状态。 + """ + stmt = select(ChannelContentReviewRecord).where( + ChannelContentReviewRecord.review_id == review_id, + self._not_deleted(), + ) + result = await self.db.execute(stmt) + orm = result.scalar_one_or_none() + if orm is None: + return None + orm.verdict = verdict + orm.reviewer = reviewer + if commit: + await self.db.commit() + else: + await self.db.flush() + await self.db.refresh(orm) + return orm + + # ------------------------------------------------------------------ + # 查询 + # ------------------------------------------------------------------ + + 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, + 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, + *, + channel_type: str | None = None, + account_id: str | None = None, + verdict: 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[ChannelContentReviewRecord]: + """按过滤条件列出审核记录，按 reviewed_at 倒序，支持分页。 + + 分页参数与子包内其他仓储一致：优先使用 ``page`` / ``page_size``， + 其次使用 ``limit`` / ``offset``（由 ``_resolve_pagination`` 解析）。 + """ + stmt = self._build_query( + channel_type=channel_type, + account_id=account_id, + verdict=verdict, + start_time=start_time, + end_time=end_time, + ) + stmt = stmt.order_by(ChannelContentReviewRecord.reviewed_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, + *, + channel_type: str | None = None, + account_id: str | None = None, + verdict: str | None = None, + start_time: datetime | None = None, + end_time: datetime | None = None, + ) -> int: + """按过滤条件统计审核记录数量。""" + stmt = self._build_query( + channel_type=channel_type, + account_id=account_id, + verdict=verdict, + start_time=start_time, + end_time=end_time, + select_count=True, + ) + result = await self.db.execute(stmt) + return int(result.scalar() or 0) + + # ------------------------------------------------------------------ + # 聚合查询（ANL-CR / CR-STATS-01） + # ------------------------------------------------------------------ + + async def query_analytics( + self, + *, + start_time: datetime, + end_time: datetime, + channel_type: str | None = None, + granularity: str = "day", + ) -> dict[str, Any]: + """审核分析趋势聚合查询（ANL-CR）。 + + 时间范围为半开区间 ``[start_time, end_time)``，返回全局聚合 + （总数 + 拦截数 + 拦截率）、按命中分类切片、按时间粒度分桶的 + 审核/拦截双计数趋势。 + + Args: + start_time: 起始时间（含）。 + end_time: 结束时间（不含）。 + channel_type: 渠道类型过滤（可选）。 + granularity: 时间分桶粒度（hour/day/week）。 + + Returns: + 含 ``total_reviews`` / ``block_count`` / ``block_rate`` / + ``by_category`` / ``trend`` 五个键的字典。``trend`` 每项的 + ``timestamp`` 为 ``date_trunc`` 返回的 naive datetime（UTC）， + 与 ``query_stats`` 保持一致，由序列化层统一格式化。 + """ + base_filters = [ + ChannelContentReviewRecord.reviewed_at >= start_time, + ChannelContentReviewRecord.reviewed_at < end_time, + self._not_deleted(), + ] + if channel_type is not None: + base_filters.append(ChannelContentReviewRecord.channel_type == channel_type) + + # 全局聚合：总审核数 + 拦截数 + agg_stmt = select( + func.count(ChannelContentReviewRecord.id).label("total_reviews"), + func.sum( + case( + (ChannelContentReviewRecord.verdict == "block", 1), + else_=0, + ) + ).label("block_count"), + ).where(*base_filters) + 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 + ) + + # 按分类切片：展开 categories JSON 数组后分组计数 + category_expr = func.json_array_elements_text( + ChannelContentReviewRecord.categories + ).label("category") + category_stmt = ( + select( + category_expr, + func.count(ChannelContentReviewRecord.id).label("count"), + ) + .where(*base_filters) + .group_by(category_expr) + ) + 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() + ] + + # 时间序列：按 granularity 分桶，每桶含 reviewed / blocked 双计数 + bucket_expr = func.date_trunc( + granularity, ChannelContentReviewRecord.reviewed_at + ) + trend_stmt = ( + select( + bucket_expr.label("bucket"), + func.count(ChannelContentReviewRecord.id).label("reviewed"), + func.sum( + case( + (ChannelContentReviewRecord.verdict == "block", 1), + else_=0, + ) + ).label("blocked"), + ) + .where(*base_filters) + .group_by(bucket_expr) + .order_by(bucket_expr) + ) + trend_result = await self.db.execute(trend_stmt) + trend: list[dict[str, Any]] = [ + { + "timestamp": bucket, + "reviewed": int(reviewed or 0), + "blocked": int(blocked or 0), + } + for bucket, reviewed, blocked in trend_result.all() + ] + + return { + "total_reviews": total_reviews, + "block_count": block_count, + "block_rate": block_rate, + "by_category": by_category, + "trend": trend, + } + + async def query_stats( + self, + *, + channel_type: str | None = None, + account_id: str | None = None, + start_time: datetime | None = None, + end_time: datetime | None = None, + granularity: str = "day", + ) -> dict[str, Any]: + """审核统计聚合查询（CR-STATS-01）。 + + 时间范围为可选过滤（含端点），返回全局聚合（总数 + pass/review/block + 计数 + 通过率 + 拦截率 + 人工介入率）、按命中分类切片、按时间粒度 + 分桶的审核趋势（pass/block 双计数）。 + + Args: + channel_type: 渠道类型过滤（可选）。 + account_id: 渠道账户 ID 过滤（可选）。 + start_time: 起始时间（含，可选）。 + end_time: 结束时间（含，可选）。 + granularity: 时间分桶粒度（hour/day/week）。 + + Returns: + 含 ``total_reviews`` / ``pass_count`` / ``review_count`` / + ``block_count`` / ``pass_rate`` / ``block_rate`` / + ``manual_intervention_rate`` / ``by_category`` / ``trend`` 九个键的 + 字典。``trend`` 每项的 ``timestamp`` 为 ``date_trunc`` 返回的 + naive datetime（UTC），与 ``query_analytics`` 保持一致，由序列化层 + 统一格式化。 + """ + base_filters = [self._not_deleted()] + if channel_type is not None: + base_filters.append(ChannelContentReviewRecord.channel_type == channel_type) + if account_id is not None: + base_filters.append(ChannelContentReviewRecord.account_id == account_id) + if start_time is not None: + base_filters.append(ChannelContentReviewRecord.reviewed_at >= start_time) + if end_time is not None: + base_filters.append(ChannelContentReviewRecord.reviewed_at <= end_time) + + # 全局聚合：总数 + pass/review/block + 人工介入计数 + agg_stmt = select( + func.count(ChannelContentReviewRecord.id).label("total_reviews"), + func.sum( + case( + (ChannelContentReviewRecord.verdict == "pass", 1), + else_=0, + ) + ).label("pass_count"), + func.sum( + case( + (ChannelContentReviewRecord.verdict == "review", 1), + else_=0, + ) + ).label("review_count"), + func.sum( + case( + (ChannelContentReviewRecord.verdict == "block", 1), + else_=0, + ) + ).label("block_count"), + func.sum( + case( + (ChannelContentReviewRecord.source == "manual_preview", 1), + else_=0, + ) + ).label("manual_count"), + ).where(*base_filters) + agg_row = (await self.db.execute(agg_stmt)).one() + total_reviews = int(agg_row.total_reviews or 0) + pass_count = int(agg_row.pass_count or 0) + 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) + 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 + ) + + # 按分类切片：展开 categories JSON 数组后分组计数 + category_expr = func.json_array_elements_text( + ChannelContentReviewRecord.categories + ).label("category") + category_stmt = ( + select( + category_expr, + func.count(ChannelContentReviewRecord.id).label("count"), + ) + .where(*base_filters) + .group_by(category_expr) + ) + 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() + ] + + # 时间序列：按 granularity 分桶，每桶含 pass / block 双计数 + bucket_expr = func.date_trunc( + granularity, ChannelContentReviewRecord.reviewed_at + ) + trend_stmt = ( + select( + bucket_expr.label("bucket"), + func.sum( + case( + (ChannelContentReviewRecord.verdict == "pass", 1), + else_=0, + ) + ).label("pass_cnt"), + func.sum( + case( + (ChannelContentReviewRecord.verdict == "block", 1), + else_=0, + ) + ).label("block_cnt"), + ) + .where(*base_filters) + .group_by(bucket_expr) + .order_by(bucket_expr) + ) + trend_result = await self.db.execute(trend_stmt) + trend: list[dict[str, Any]] = [ + { + "timestamp": bucket, + "pass_count": int(pass_val or 0), + "block_count": int(block_val or 0), + } + for bucket, pass_val, block_val in trend_result.all() + ] + + return { + "total_reviews": total_reviews, + "pass_count": pass_count, + "review_count": review_count, + "block_count": block_count, + "pass_rate": pass_rate, + "block_rate": block_rate, + "manual_intervention_rate": manual_intervention_rate, + "by_category": by_category, + "trend": trend, + } + + # ------------------------------------------------------------------ + # 内部查询构建 + # ------------------------------------------------------------------ + + def _build_query( + self, + *, + channel_type: str | None = None, + account_id: str | None = None, + verdict: 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`` 共享 + 同一组条件构造逻辑。 + """ + if select_count: + stmt = select(func.count(ChannelContentReviewRecord.id)).select_from( + ChannelContentReviewRecord + ) + else: + stmt = select(ChannelContentReviewRecord) + stmt = stmt.where(self._not_deleted()) + if channel_type is not None: + stmt = stmt.where(ChannelContentReviewRecord.channel_type == channel_type) + if account_id is not None: + stmt = stmt.where(ChannelContentReviewRecord.account_id == account_id) + if verdict is not None: + stmt = stmt.where(ChannelContentReviewRecord.verdict == verdict) + if start_time is not None: + stmt = stmt.where(ChannelContentReviewRecord.reviewed_at >= start_time) + if end_time is not None: + stmt = stmt.where(ChannelContentReviewRecord.reviewed_at <= end_time) + return stmt diff --git a/backend/package/yuxi/repositories/channels/user_identity_repository.py b/backend/package/yuxi/repositories/channels/user_identity_repository.py index 7edff18e..f86d77a4 100644 --- a/backend/package/yuxi/repositories/channels/user_identity_repository.py +++ b/backend/package/yuxi/repositories/channels/user_identity_repository.py @@ -182,6 +182,59 @@ class UserIdentityRepository(BaseRepository): await self.db.flush() return result.rowcount + async def update_with_optimistic_lock( + self, + identity_id: str, + expected_version: int, + updates: dict[str, Any], + *, + commit: bool = True, + ) -> ChannelUserIdentity | None: + """乐观锁更新身份记录（P3 渐进式绑定）。 + + 通过 ``WHERE version = expected_version`` 实现乐观并发控制， + 版本不匹配（并发修改或记录不存在）时返回 ``None``，由调用方 + 翻译为 ``ConflictError``。 + + Args: + identity_id: 统一身份 ID。 + expected_version: 期望的当前版本号（聚合根 bindUser/unbindUser + 递增前的版本）。 + updates: 待更新字段（``user_id`` / ``identity_type`` / + ``updated_by``），其中 ``user_id`` 为 ``None`` 时显式置空 + （解绑场景）。 + commit: 是否提交事务，``False`` 时仅 ``flush`` 加入外层事务。 + + Returns: + 更新后的 ORM 实例；版本不匹配时返回 ``None``。 + """ + now = utc_now_naive() + values: dict[str, Any] = { + "version": expected_version + 1, + "updated_at": now, + } + if "user_id" in updates: + values["user_id"] = updates["user_id"] + if "identity_type" in updates: + values["identity_type"] = updates["identity_type"] + if updates.get("updated_by") is not None: + values["updated_by"] = updates["updated_by"] + stmt = ( + update(ChannelUserIdentity) + .where(ChannelUserIdentity.identity_id == identity_id) + .where(ChannelUserIdentity.version == expected_version) + .where(self._not_deleted()) + .values(**values) + ) + result = await self.db.execute(stmt) + if result.rowcount == 0: + return None + if commit: + await self.db.commit() + else: + await self.db.flush() + return await self.get_by_identity_id(identity_id) + # ------------------------------------------------------------------ # 业务专用查询（技术语义命名） # ------------------------------------------------------------------ @@ -251,15 +304,9 @@ class UserIdentityRepository(BaseRepository): identity_type: str | None = None, channel_type: str | None = None, confidence: str | None = None, - select_count: bool = False, ): """构建查询语句，统一附加软删除过滤。""" - if select_count: - stmt = select(func.count(ChannelUserIdentity.id)).select_from( - ChannelUserIdentity - ) - else: - stmt = select(ChannelUserIdentity) + stmt = select(ChannelUserIdentity) stmt = stmt.where(self._not_deleted()) if user_id is not None: stmt = stmt.where(ChannelUserIdentity.user_id == user_id)