refactor(channel-repo): 清理废弃仓储与冗余方法

1. 删除未使用的 ChannelReportRepository 及其所有引用
2. 移除 route_binding_repository 中的 list_enabled_by_account 方法
3. 精简 channel_account_repository 中的冗余字段与 update_last_message_at 方法
4. 简化 channel_outbox_repository 的锁相关与 update 参数
5. 精简 channel_idempotency_repository 的冗余查询方法与字段
6. 优化 content_review_record_repository 的查询与清理逻辑
This commit is contained in:
Kris 2026-07-08 03:57:28 +08:00
parent c7eb196a7e
commit 6f730dca06
7 changed files with 78 additions and 467 deletions

View File

@ -1,10 +1,10 @@
"""渠道仓储层聚合导出。 """渠道仓储层聚合导出。
本包提供 本包提供
- 11 ``Channel*Repository`` / ``ChannelConversationRepository`` / - 10 ``Channel*Repository`` / ``ChannelConversationRepository`` /
``ChannelMessageRepository`` 实现类 ``ChannelMessageRepository`` 实现类
直接操作 ORM Model原子 CRUD不含业务语义 直接操作 ORM Model原子 CRUD不含业务语义
- ``Repositories`` 聚合 dataclass共享 ``db`` 会话的 11 个仓储实例 - ``Repositories`` 聚合 dataclass共享 ``db`` 会话的 10 个仓储实例
- ``create_repositories(db)`` factory 函数由框架层在请求开始时调用 - ``create_repositories(db)`` factory 函数由框架层在请求开始时调用
依赖边界只依赖 ``yuxi.storage.postgres.models_channels`` / 依赖边界只依赖 ``yuxi.storage.postgres.models_channels`` /
@ -30,7 +30,6 @@ from .channel_idempotency_repository import ChannelIdempotencyRepository
from .channel_message_repository import ChannelMessageRepository from .channel_message_repository import ChannelMessageRepository
from .channel_outbox_repository import ChannelOutboxRepository from .channel_outbox_repository import ChannelOutboxRepository
from .channel_pairing_repository import ChannelPairingRepository from .channel_pairing_repository import ChannelPairingRepository
from .channel_report_repository import ChannelReportRepository
from .channel_session_repository import ChannelSessionRepository from .channel_session_repository import ChannelSessionRepository
from .content_review_record_repository import ChannelContentReviewRecordRepository from .content_review_record_repository import ChannelContentReviewRecordRepository
from .route_binding_repository import ChannelRouteBindingRepository from .route_binding_repository import ChannelRouteBindingRepository
@ -46,7 +45,6 @@ __all__ = [
"ChannelMessageRepository", "ChannelMessageRepository",
"ChannelOutboxRepository", "ChannelOutboxRepository",
"ChannelPairingRepository", "ChannelPairingRepository",
"ChannelReportRepository",
"ChannelRouteBindingRepository", "ChannelRouteBindingRepository",
"ChannelSessionRepository", "ChannelSessionRepository",
"Repositories", "Repositories",
@ -72,7 +70,6 @@ class Repositories:
idempotency: ChannelIdempotencyRepository idempotency: ChannelIdempotencyRepository
conversation: ChannelConversationRepository conversation: ChannelConversationRepository
message: ChannelMessageRepository message: ChannelMessageRepository
report: ChannelReportRepository
content_review_record: ChannelContentReviewRecordRepository content_review_record: ChannelContentReviewRecordRepository
route_binding: ChannelRouteBindingRepository route_binding: ChannelRouteBindingRepository
@ -94,7 +91,6 @@ def create_repositories(db: AsyncSession) -> Repositories:
idempotency=ChannelIdempotencyRepository(db), idempotency=ChannelIdempotencyRepository(db),
conversation=ChannelConversationRepository(db), conversation=ChannelConversationRepository(db),
message=ChannelMessageRepository(db), message=ChannelMessageRepository(db),
report=ChannelReportRepository(db),
content_review_record=ChannelContentReviewRecordRepository(db), content_review_record=ChannelContentReviewRecordRepository(db),
route_binding=ChannelRouteBindingRepository(db), route_binding=ChannelRouteBindingRepository(db),
) )

View File

@ -39,14 +39,11 @@ class ChannelAccountRepository(BaseRepository):
account_id=data["account_id"], account_id=data["account_id"],
display_name=data["display_name"], display_name=data["display_name"],
config=data.get("config") or {}, config=data.get("config") or {},
capabilities=data.get("capabilities") or {},
enabled=data.get("enabled", True), enabled=data.get("enabled", True),
status=data.get("status", "active"), status=data.get("status", "active"),
plugin_status=data.get("plugin_status", "stopped"), plugin_status=data.get("plugin_status", "stopped"),
config_version=data.get("config_version", 1),
last_error=data.get("last_error"), last_error=data.get("last_error"),
last_error_at=data.get("last_error_at"), last_error_at=data.get("last_error_at"),
last_message_at=data.get("last_message_at"),
service_user_uid=data.get("service_user_uid"), service_user_uid=data.get("service_user_uid"),
version=1, version=1,
created_by=data.get("created_by"), created_by=data.get("created_by"),
@ -156,15 +153,12 @@ class ChannelAccountRepository(BaseRepository):
updatable_fields = ( updatable_fields = (
"display_name", "display_name",
"config", "config",
"capabilities",
"enabled", "enabled",
"status", "status",
"plugin_status", "plugin_status",
"config_version",
"last_error", "last_error",
"last_error_at", "last_error_at",
"last_health_check_at", "last_health_check_at",
"last_message_at",
"updated_by", "updated_by",
) )
for field in updatable_fields: for field in updatable_fields:
@ -234,27 +228,6 @@ class ChannelAccountRepository(BaseRepository):
await self.db.flush() await self.db.flush()
return result.rowcount return result.rowcount
async def update_last_message_at(
self,
record_id: int,
*,
commit: bool = True,
) -> int:
"""按主键更新最近消息时间冗余字段FR-35 健康检查)。"""
now = utc_now_naive()
stmt = (
update(ChannelAccount)
.where(ChannelAccount.id == record_id)
.where(self._not_deleted())
.values(last_message_at=now, updated_at=now)
)
result = await self.db.execute(stmt)
if commit:
await self.db.commit()
else:
await self.db.flush()
return result.rowcount
async def update_last_health_check_at( async def update_last_health_check_at(
self, self,
record_id: int, record_id: int,

View File

@ -41,8 +41,6 @@ class ChannelIdempotencyRepository(BaseRepository):
record = ChannelIdempotency( record = ChannelIdempotency(
idempotency_key=data["idempotency_key"], idempotency_key=data["idempotency_key"],
operation=data["operation"], operation=data["operation"],
account_id=data.get("account_id"),
request_hash=data.get("request_hash"),
status="in_progress", status="in_progress",
in_progress_started_at=now, in_progress_started_at=now,
response_body=data.get("response_body"), response_body=data.get("response_body"),
@ -63,6 +61,7 @@ class ChannelIdempotencyRepository(BaseRepository):
status: str, status: str,
*, *,
response_body: dict[str, Any] | None = None, response_body: dict[str, Any] | None = None,
updated_by: str | None = None,
commit: bool = True, commit: bool = True,
) -> int: ) -> int:
"""更新幂等记录状态(状态机流转,不加载 ORM 实例)。 """更新幂等记录状态(状态机流转,不加载 ORM 实例)。
@ -75,6 +74,7 @@ class ChannelIdempotencyRepository(BaseRepository):
record_id: 幂等记录主键 record_id: 幂等记录主键
status: 目标状态completed / failed status: 目标状态completed / failed
response_body: 首次请求的响应体 completed 时填充 response_body: 首次请求的响应体 completed 时填充
updated_by: 状态变更操作人审计用
commit: True 时提交事务False 时仅 flush commit: True 时提交事务False 时仅 flush
Returns: Returns:
@ -87,6 +87,8 @@ class ChannelIdempotencyRepository(BaseRepository):
} }
if response_body is not None: if response_body is not None:
values["response_body"] = response_body values["response_body"] = response_body
if updated_by is not None:
values["updated_by"] = updated_by
stmt = update(ChannelIdempotency).where(ChannelIdempotency.id == record_id).values(**values) stmt = update(ChannelIdempotency).where(ChannelIdempotency.id == record_id).values(**values)
result = await self.db.execute(stmt) result = await self.db.execute(stmt)
if commit: if commit:
@ -99,14 +101,6 @@ class ChannelIdempotencyRepository(BaseRepository):
# 查询 # 查询
# ------------------------------------------------------------------ # ------------------------------------------------------------------
async def get_by_id(self, record_id: int, *, for_update: bool = False) -> ChannelIdempotency | None:
"""根据主键 ID 获取幂等记录。"""
stmt = select(ChannelIdempotency).where(ChannelIdempotency.id == record_id)
if for_update:
stmt = stmt.with_for_update()
result = await self.db.execute(stmt)
return result.scalar_one_or_none()
async def get_by_key(self, idempotency_key: str, *, for_update: bool = False) -> ChannelIdempotency | None: async def get_by_key(self, idempotency_key: str, *, for_update: bool = False) -> ChannelIdempotency | None:
"""根据幂等键获取记录(重复请求回放)。""" """根据幂等键获取记录(重复请求回放)。"""
stmt = select(ChannelIdempotency).where(ChannelIdempotency.idempotency_key == idempotency_key) stmt = select(ChannelIdempotency).where(ChannelIdempotency.idempotency_key == idempotency_key)
@ -115,28 +109,6 @@ class ChannelIdempotencyRepository(BaseRepository):
result = await self.db.execute(stmt) result = await self.db.execute(stmt)
return result.scalar_one_or_none() return result.scalar_one_or_none()
async def list_in_progress_stale(
self,
before: dt.datetime,
*,
limit: int = 100,
) -> list[ChannelIdempotency]:
"""列出超时的 in_progress 记录worker 崩溃回收)。
Args:
before: 截止时间in_progress_started_at 早于此时间的记录将被列出
limit: 返回数量上限
"""
stmt = (
select(ChannelIdempotency)
.where(ChannelIdempotency.status == "in_progress")
.where(ChannelIdempotency.in_progress_started_at < before)
.order_by(ChannelIdempotency.in_progress_started_at.asc())
.limit(limit)
)
result = await self.db.execute(stmt)
return list(result.scalars().all())
# ------------------------------------------------------------------ # ------------------------------------------------------------------
# 清理(物理删除,不使用软删除) # 清理(物理删除,不使用软删除)
# ------------------------------------------------------------------ # ------------------------------------------------------------------
@ -147,12 +119,11 @@ class ChannelIdempotencyRepository(BaseRepository):
*, *,
commit: bool = True, commit: bool = True,
) -> int: ) -> int:
"""物理删除早于指定时间的幂等记录(定时清理任务)。 """物理删除已过期的幂等记录(定时清理任务)。
幂等记录不使用软删除过期后直接物理删除 ``expires_at`` 判断过期幂等记录不使用软删除过期后直接物理删除
默认保留 24 小时IDEMPOTENCY_DEFAULT_TTL_HOURS
""" """
stmt = sa_delete(ChannelIdempotency).where(ChannelIdempotency.created_at < before) stmt = sa_delete(ChannelIdempotency).where(ChannelIdempotency.expires_at < before)
result = await self.db.execute(stmt) result = await self.db.execute(stmt)
if commit: if commit:
await self.db.commit() await self.db.commit()

View File

@ -257,8 +257,6 @@ class ChannelOutboxRepository(BaseRepository):
next_retry_at: dt.datetime | None = None, next_retry_at: dt.datetime | None = None,
last_retry_at: dt.datetime | None = None, last_retry_at: dt.datetime | None = None,
sent_at: dt.datetime | None = None, sent_at: dt.datetime | None = None,
release_lock: bool = False,
worker_id: str | None = None,
updated_by: str | None = None, updated_by: str | None = None,
commit: bool = True, commit: bool = True,
) -> int: ) -> int:
@ -275,8 +273,6 @@ class ChannelOutboxRepository(BaseRepository):
next_retry_at: 下次重试时间指数退避 next_retry_at: 下次重试时间指数退避
last_retry_at: 上次重试时间用于诊断退避进度与 SLA last_retry_at: 上次重试时间用于诊断退避进度与 SLA
sent_at: 首次成功投递时间用于计算投递延迟 sent_at: 首次成功投递时间用于计算投递延迟
release_lock: True 时清空 locked_by / locked_at投递完成后释放租约
worker_id: 指定 worker_id 时仅释放该 worker 持有的锁防止误释放他人锁
updated_by: 操作人 updated_by: 操作人
commit: True 时提交事务False 时仅 flush commit: True 时提交事务False 时仅 flush
@ -301,12 +297,7 @@ class ChannelOutboxRepository(BaseRepository):
values["last_retry_at"] = last_retry_at values["last_retry_at"] = last_retry_at
if sent_at is not None: if sent_at is not None:
values["sent_at"] = sent_at values["sent_at"] = sent_at
if release_lock:
values["locked_by"] = None
values["locked_at"] = None
stmt = update(ChannelOutboxEntry).where(ChannelOutboxEntry.id == record_id).where(self._not_deleted()) stmt = update(ChannelOutboxEntry).where(ChannelOutboxEntry.id == record_id).where(self._not_deleted())
if release_lock and worker_id is not None:
stmt = stmt.where(ChannelOutboxEntry.locked_by == worker_id)
stmt = stmt.values(**values) stmt = stmt.values(**values)
result = await self.db.execute(stmt) result = await self.db.execute(stmt)
if commit: if commit:
@ -315,74 +306,6 @@ class ChannelOutboxRepository(BaseRepository):
await self.db.flush() await self.db.flush()
return result.rowcount return result.rowcount
async def acquire_lock(
self,
record_id: int,
worker_id: str,
*,
commit: bool = True,
) -> int:
"""原子抢占投递锁at-least-once 语义)。
仅当 ``locked_by`` 为空时才能抢占成功保证同一时刻只有一个 worker
处理该条目
Args:
record_id: 发件箱条目主键
worker_id: 抢占方 worker 标识
commit: True 时提交事务False 时仅 flush
Returns:
受影响行数1 = 抢占成功0 = 已被其他 worker 持有
"""
now = utc_now_naive()
stmt = (
update(ChannelOutboxEntry)
.where(ChannelOutboxEntry.id == record_id)
.where(ChannelOutboxEntry.locked_by.is_(None))
.where(self._not_deleted())
.values(locked_by=worker_id, locked_at=now, updated_at=now)
)
result = await self.db.execute(stmt)
if commit:
await self.db.commit()
else:
await self.db.flush()
return result.rowcount
async def release_lock(
self,
record_id: int,
*,
worker_id: str | None = None,
commit: bool = True,
) -> int:
"""释放投递锁(清空 locked_by / locked_at
Args:
record_id: 发件箱条目主键
worker_id: 指定 worker_id 时仅释放该 worker 持有的锁防止误释放他人锁
commit: True 时提交事务False 时仅 flush
Returns:
受影响行数
"""
now = utc_now_naive()
stmt = (
update(ChannelOutboxEntry)
.where(ChannelOutboxEntry.id == record_id)
.where(self._not_deleted())
.values(locked_by=None, locked_at=None, updated_at=now)
)
if worker_id is not None:
stmt = stmt.where(ChannelOutboxEntry.locked_by == worker_id)
result = await self.db.execute(stmt)
if commit:
await self.db.commit()
else:
await self.db.flush()
return result.rowcount
async def list_by_message_id(self, message_id: int) -> list[ChannelOutboxEntry]: async def list_by_message_id(self, message_id: int) -> list[ChannelOutboxEntry]:
"""按消息 ID 查询投递状态FR-22 状态回查)。""" """按消息 ID 查询投递状态FR-22 状态回查)。"""
stmt = ( stmt = (

View File

@ -1,294 +0,0 @@
# yuxi/repositories/channels/channel_report_repository.py
"""渠道报告数据访问层。"""
from __future__ import annotations
from datetime import datetime
from typing import Any
from sqlalchemy import delete, func, select
from yuxi.repositories.channels.base import BaseRepository
from yuxi.storage.postgres.models_channels import ChannelReport
class ChannelReportRepository(BaseRepository):
"""``channel_reports`` 表的异步数据访问封装。
报告记录由 scheduler handler 异步生成后写入状态机为
``pending generating ready / failed``终态不可变更
"""
model = ChannelReport
# ------------------------------------------------------------------
# 写操作
# ------------------------------------------------------------------
async def create(self, data: dict[str, Any], *, commit: bool = True) -> ChannelReport:
"""创建报告记录。
Args:
data: 报告字段字典必须包含 report_id / task_id / report_type
commit: True 时提交事务False 时仅 flush
"""
record = ChannelReport(
report_id=data["report_id"],
task_id=data["task_id"],
report_type=data["report_type"],
status=data.get("status", "pending"),
params=data.get("params") or {},
content=data.get("content"),
error_message=data.get("error_message"),
created_by=data.get("created_by", "system"),
retried_from=data.get("retried_from"),
retried_at=data.get("retried_at"),
)
self.db.add(record)
if commit:
await self.db.commit()
else:
await self.db.flush()
await self.db.refresh(record)
return record
async def update_status(
self,
report_id: str,
*,
status: str | None = None,
content: dict[str, Any] | None = None,
error_message: str | None = None,
ready_at: datetime | None = None,
commit: bool = True,
) -> ChannelReport | None:
"""按 report_id 更新报告状态字段;记录不存在返回 None。
仅更新提供的字段局部更新用于 ``updateReportStatus`` 用例
与子包内其他 ``update`` 方法一致无条件 ``refresh`` 以返回最新状态
"""
stmt = select(ChannelReport).where(
ChannelReport.report_id == report_id,
self._not_deleted(),
)
result = await self.db.execute(stmt)
orm = result.scalar_one_or_none()
if orm is None:
return None
if status is not None:
orm.status = status
if content is not None:
orm.content = content
if error_message is not None:
orm.error_message = error_message
if ready_at is not None:
orm.ready_at = ready_at
if commit:
await self.db.commit()
else:
await self.db.flush()
await self.db.refresh(orm)
return orm
async def update_fields(
self,
report_id: str,
data: dict[str, Any],
*,
commit: bool = True,
) -> ChannelReport | None:
"""按 report_id 完整更新报告的可变字段;记录不存在返回 None。
用于重试场景标记原报告的 ``retried_at`` 字段``report_id`` /
``task_id`` 作为业务标识不更新与子包内其他 ``update`` 方法一致
无条件 ``refresh`` 以返回最新状态
"""
stmt = select(ChannelReport).where(
ChannelReport.report_id == report_id,
self._not_deleted(),
)
result = await self.db.execute(stmt)
orm = result.scalar_one_or_none()
if orm is None:
return None
updatable_fields = (
"status",
"content",
"error_message",
"ready_at",
"retried_from",
"retried_at",
)
for field in updatable_fields:
if field in data:
setattr(orm, field, data[field])
if commit:
await self.db.commit()
else:
await self.db.flush()
await self.db.refresh(orm)
return orm
# ------------------------------------------------------------------
# 查询
# ------------------------------------------------------------------
async def get_by_report_id(self, report_id: str, *, for_update: bool = False) -> ChannelReport | None:
"""根据业务标识 report_id 获取报告(排除已软删除)。"""
stmt = select(ChannelReport).where(
ChannelReport.report_id == report_id,
self._not_deleted(),
)
if for_update:
stmt = stmt.with_for_update()
result = await self.db.execute(stmt)
return result.scalar_one_or_none()
async def get_by_task_id(self, task_id: str, *, for_update: bool = False) -> ChannelReport | None:
"""根据 scheduler task_id 获取报告(排除已软删除)。"""
stmt = select(ChannelReport).where(
ChannelReport.task_id == task_id,
self._not_deleted(),
)
if for_update:
stmt = stmt.with_for_update()
result = await self.db.execute(stmt)
return result.scalar_one_or_none()
async def list(
self,
*,
status: str | None = None,
report_type: str | None = None,
start_time: datetime | None = None,
end_time: datetime | None = None,
page: int | None = None,
page_size: int | None = None,
limit: int | None = None,
offset: int | None = None,
) -> list[ChannelReport]:
"""列出报告,按 created_at 倒序,支持过滤与分页。
分页参数与子包内其他仓储一致优先使用 ``page`` / ``page_size``
其次使用 ``limit`` / ``offset`` ``_resolve_pagination`` 解析
"""
stmt = self._build_query(
status=status,
report_type=report_type,
start_time=start_time,
end_time=end_time,
)
stmt = stmt.order_by(ChannelReport.created_at.desc())
effective_limit, effective_offset = self._resolve_pagination(page, page_size, limit, offset)
if effective_limit is not None:
stmt = stmt.limit(effective_limit).offset(effective_offset)
result = await self.db.execute(stmt)
return list(result.scalars().all())
async def count(
self,
*,
status: str | None = None,
report_type: str | None = None,
start_time: datetime | None = None,
end_time: datetime | None = None,
) -> int:
"""统计满足过滤条件的报告数量。"""
stmt = self._build_query(
status=status,
report_type=report_type,
start_time=start_time,
end_time=end_time,
select_count=True,
)
result = await self.db.execute(stmt)
return int(result.scalar() or 0)
async def list_stale_generating(
self,
before: datetime,
*,
limit: int = 100,
) -> list[ChannelReport]:
"""列出长时间停留在 generating 状态的陈旧报告。
Args:
before: 阈值updated_at 早于此时间的 generating 报告将被列出
limit: 返回数量上限
"""
stmt = (
select(ChannelReport)
.where(
ChannelReport.status == "generating",
ChannelReport.updated_at < before,
self._not_deleted(),
)
.order_by(ChannelReport.updated_at.asc())
.limit(limit)
)
result = await self.db.execute(stmt)
return list(result.scalars().all())
# ------------------------------------------------------------------
# 清理(物理删除,配合 scheduler 清理任务)
# ------------------------------------------------------------------
async def cleanup_old_reports(
self,
before: datetime,
*,
limit: int = 100,
commit: bool = True,
) -> int:
"""物理删除已完结ready/failed的陈旧报告返回删除记录数。
Args:
before: 阈值updated_at 早于此时间的终态报告将被删除
limit: 单次删除上限
commit: True 时提交事务False 时仅 flush用于组合事务
"""
subq = (
select(ChannelReport.id)
.where(
ChannelReport.status.in_(["ready", "failed"]),
ChannelReport.updated_at < before,
self._not_deleted(),
)
.limit(limit)
)
stmt = delete(ChannelReport).where(ChannelReport.id.in_(subq))
result = await self.db.execute(stmt)
if commit:
await self.db.commit()
else:
await self.db.flush()
return result.rowcount
# ------------------------------------------------------------------
# 内部查询构建
# ------------------------------------------------------------------
def _build_query(
self,
*,
status: str | None = None,
report_type: str | None = None,
start_time: datetime | None = None,
end_time: datetime | None = None,
select_count: bool = False,
):
"""构建查询语句,统一附加软删除过滤。"""
if select_count:
stmt = select(func.count(ChannelReport.id)).select_from(ChannelReport)
else:
stmt = select(ChannelReport)
stmt = stmt.where(self._not_deleted())
if status is not None:
stmt = stmt.where(ChannelReport.status == status)
if report_type is not None:
stmt = stmt.where(ChannelReport.report_type == report_type)
if start_time is not None:
stmt = stmt.where(ChannelReport.created_at >= start_time)
if end_time is not None:
stmt = stmt.where(ChannelReport.created_at <= end_time)
return stmt

View File

@ -3,10 +3,11 @@
from __future__ import annotations from __future__ import annotations
from datetime import datetime from datetime import datetime, timezone
from typing import Any from typing import Any
from sqlalchemy import case, func, select from sqlalchemy import case, delete as sa_delete, func, select
from yuxi.repositories.channels.base import BaseRepository from yuxi.repositories.channels.base import BaseRepository
from yuxi.storage.postgres.models_channels import ChannelContentReviewRecord from yuxi.storage.postgres.models_channels import ChannelContentReviewRecord
@ -55,6 +56,7 @@ class ChannelContentReviewRecordRepository(BaseRepository):
reviewer=data["reviewer"], reviewer=data["reviewer"],
source=data["source"], source=data["source"],
trace_id=data.get("trace_id"), trace_id=data.get("trace_id"),
decision_started_at=data.get("decision_started_at"),
created_by=data.get("created_by"), created_by=data.get("created_by"),
updated_by=data.get("updated_by"), updated_by=data.get("updated_by"),
) )
@ -89,6 +91,8 @@ class ChannelContentReviewRecordRepository(BaseRepository):
return None return None
orm.verdict = verdict orm.verdict = verdict
orm.reviewer = reviewer orm.reviewer = reviewer
orm.updated_by = reviewer
orm.decision_completed_at = datetime.now(timezone.utc).replace(tzinfo=None)
if commit: if commit:
await self.db.commit() await self.db.commit()
else: else:
@ -96,6 +100,34 @@ class ChannelContentReviewRecordRepository(BaseRepository):
await self.db.refresh(orm) await self.db.refresh(orm)
return orm return orm
async def delete_old_records(
self,
before: datetime,
*,
limit: int = 1000,
commit: bool = True,
) -> int:
"""物理删除早于指定时间的审核记录retention
``ChannelAuditLog`` 一致采用物理删除不使用软删除路径
retention scheduler handler 周期触发避免表无限增长
Args:
before: 截止时间早于此时间的记录将被删除
limit: 单次最多删除的记录数避免长事务锁竞争
commit: True 时提交事务False 时仅 flush
Returns:
被删除的记录数
"""
stmt = sa_delete(ChannelContentReviewRecord).where(ChannelContentReviewRecord.reviewed_at < before).limit(limit)
result = await self.db.execute(stmt)
if commit:
await self.db.commit()
else:
await self.db.flush()
return result.rowcount
# ------------------------------------------------------------------ # ------------------------------------------------------------------
# 查询 # 查询
# ------------------------------------------------------------------ # ------------------------------------------------------------------
@ -117,6 +149,8 @@ class ChannelContentReviewRecordRepository(BaseRepository):
channel_type: str | None = None, channel_type: str | None = None,
account_id: str | None = None, account_id: str | None = None,
verdict: str | None = None, verdict: str | None = None,
resource_type: str | None = None,
trace_id: str | None = None,
start_time: datetime | None = None, start_time: datetime | None = None,
end_time: datetime | None = None, end_time: datetime | None = None,
page: int | None = None, page: int | None = None,
@ -133,6 +167,8 @@ class ChannelContentReviewRecordRepository(BaseRepository):
channel_type=channel_type, channel_type=channel_type,
account_id=account_id, account_id=account_id,
verdict=verdict, verdict=verdict,
resource_type=resource_type,
trace_id=trace_id,
start_time=start_time, start_time=start_time,
end_time=end_time, end_time=end_time,
) )
@ -149,6 +185,8 @@ class ChannelContentReviewRecordRepository(BaseRepository):
channel_type: str | None = None, channel_type: str | None = None,
account_id: str | None = None, account_id: str | None = None,
verdict: str | None = None, verdict: str | None = None,
resource_type: str | None = None,
trace_id: str | None = None,
start_time: datetime | None = None, start_time: datetime | None = None,
end_time: datetime | None = None, end_time: datetime | None = None,
) -> int: ) -> int:
@ -157,6 +195,8 @@ class ChannelContentReviewRecordRepository(BaseRepository):
channel_type=channel_type, channel_type=channel_type,
account_id=account_id, account_id=account_id,
verdict=verdict, verdict=verdict,
resource_type=resource_type,
trace_id=trace_id,
start_time=start_time, start_time=start_time,
end_time=end_time, end_time=end_time,
select_count=True, select_count=True,
@ -307,7 +347,13 @@ class ChannelContentReviewRecordRepository(BaseRepository):
if end_time is not None: if end_time is not None:
base_filters.append(ChannelContentReviewRecord.reviewed_at <= end_time) base_filters.append(ChannelContentReviewRecord.reviewed_at <= end_time)
# 全局聚合:总数 + pass/review/block + 人工介入计数 # 全局聚合:总数 + pass/review/block + 人工介入计数 + 平均决策时长
# avg_decision_seconds 仅对有 decision_completed_at 的记录计算
# (首次审核即终态的记录无 completed_at不计入
decision_duration = func.extract(
"EPOCH",
ChannelContentReviewRecord.decision_completed_at - ChannelContentReviewRecord.decision_started_at,
)
agg_stmt = select( agg_stmt = select(
func.count(ChannelContentReviewRecord.id).label("total_reviews"), func.count(ChannelContentReviewRecord.id).label("total_reviews"),
func.sum( func.sum(
@ -334,6 +380,15 @@ class ChannelContentReviewRecordRepository(BaseRepository):
else_=0, else_=0,
) )
).label("manual_count"), ).label("manual_count"),
func.avg(
case(
(
ChannelContentReviewRecord.decision_completed_at.isnot(None),
decision_duration,
),
else_=None,
)
).label("avg_decision_seconds"),
).where(*base_filters) ).where(*base_filters)
agg_row = (await self.db.execute(agg_stmt)).one() agg_row = (await self.db.execute(agg_stmt)).one()
total_reviews = int(agg_row.total_reviews or 0) total_reviews = int(agg_row.total_reviews or 0)
@ -341,6 +396,7 @@ class ChannelContentReviewRecordRepository(BaseRepository):
review_count = int(agg_row.review_count or 0) review_count = int(agg_row.review_count or 0)
block_count = int(agg_row.block_count or 0) block_count = int(agg_row.block_count or 0)
manual_count = int(agg_row.manual_count or 0) manual_count = int(agg_row.manual_count or 0)
avg_decision_seconds = float(agg_row.avg_decision_seconds or 0.0)
pass_rate = (pass_count / total_reviews * 100) if total_reviews > 0 else 0.0 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 block_rate = (block_count / total_reviews * 100) if total_reviews > 0 else 0.0
manual_intervention_rate = (manual_count / total_reviews * 100) if total_reviews > 0 else 0.0 manual_intervention_rate = (manual_count / total_reviews * 100) if total_reviews > 0 else 0.0
@ -400,6 +456,7 @@ class ChannelContentReviewRecordRepository(BaseRepository):
"pass_rate": pass_rate, "pass_rate": pass_rate,
"block_rate": block_rate, "block_rate": block_rate,
"manual_intervention_rate": manual_intervention_rate, "manual_intervention_rate": manual_intervention_rate,
"avg_decision_seconds": avg_decision_seconds,
"by_category": by_category, "by_category": by_category,
"trend": trend, "trend": trend,
} }
@ -414,15 +471,17 @@ class ChannelContentReviewRecordRepository(BaseRepository):
channel_type: str | None = None, channel_type: str | None = None,
account_id: str | None = None, account_id: str | None = None,
verdict: str | None = None, verdict: str | None = None,
resource_type: str | None = None,
trace_id: str | None = None,
start_time: datetime | None = None, start_time: datetime | None = None,
end_time: datetime | None = None, end_time: datetime | None = None,
select_count: bool = False, select_count: bool = False,
): ):
"""构建查询语句,统一附加软删除过滤。 """构建查询语句,统一附加软删除过滤。
支持 ``channel_type`` / ``account_id`` / ``verdict`` / 支持 ``channel_type`` / ``account_id`` / ``verdict`` / ``resource_type``
``start_time`` / ``end_time`` 过滤 ``list`` / ``count`` 共享 / ``trace_id`` / ``start_time`` / ``end_time`` 过滤 ``list`` /
同一组条件构造逻辑 ``count`` 共享同一组条件构造逻辑
""" """
if select_count: if select_count:
stmt = select(func.count(ChannelContentReviewRecord.id)).select_from(ChannelContentReviewRecord) stmt = select(func.count(ChannelContentReviewRecord.id)).select_from(ChannelContentReviewRecord)
@ -435,6 +494,10 @@ class ChannelContentReviewRecordRepository(BaseRepository):
stmt = stmt.where(ChannelContentReviewRecord.account_id == account_id) stmt = stmt.where(ChannelContentReviewRecord.account_id == account_id)
if verdict is not None: if verdict is not None:
stmt = stmt.where(ChannelContentReviewRecord.verdict == verdict) stmt = stmt.where(ChannelContentReviewRecord.verdict == verdict)
if resource_type is not None:
stmt = stmt.where(ChannelContentReviewRecord.resource_type == resource_type)
if trace_id is not None:
stmt = stmt.where(ChannelContentReviewRecord.trace_id == trace_id)
if start_time is not None: if start_time is not None:
stmt = stmt.where(ChannelContentReviewRecord.reviewed_at >= start_time) stmt = stmt.where(ChannelContentReviewRecord.reviewed_at >= start_time)
if end_time is not None: if end_time is not None:

View File

@ -212,24 +212,3 @@ class ChannelRouteBindingRepository(BaseRepository):
else: else:
await self.db.flush() await self.db.flush()
return result.rowcount > 0 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())