profile: gate manual group analysis requests

This commit is contained in:
2026-08-08 13:12:24 +08:00
parent dc76f4831e
commit 9ab4432450
2 changed files with 46 additions and 10 deletions
+27 -10
View File
@@ -29,6 +29,8 @@ import top.jie65535.mirai.profile.ProfileAutoMaintenance
import top.jie65535.mirai.profile.ProfileCategory import top.jie65535.mirai.profile.ProfileCategory
import top.jie65535.mirai.profile.ProfileCompactionReport import top.jie65535.mirai.profile.ProfileCompactionReport
import top.jie65535.mirai.profile.ProfileDailyMaintenance import top.jie65535.mirai.profile.ProfileDailyMaintenance
import top.jie65535.mirai.profile.ProfileDailyRequestController
import top.jie65535.mirai.profile.ProfileDailyRunStoppedException
import top.jie65535.mirai.profile.ProfilePersistentText import top.jie65535.mirai.profile.ProfilePersistentText
import top.jie65535.mirai.profile.UserProfileAnalysisService import top.jie65535.mirai.profile.UserProfileAnalysisService
import top.jie65535.mirai.profile.UserProfileSnapshot import top.jie65535.mirai.profile.UserProfileSnapshot
@@ -115,25 +117,35 @@ object PluginCommands : CompositeCommand(
return return
} }
val runToken = UserProfileAnalysisService.newRunToken() val runToken = UserProfileAnalysisService.newRunToken()
val requestController = ProfileDailyRequestController(
maxConcurrentRequests = PluginConfig.profileMaxConcurrentRequests,
maxAttempts = PluginConfig.profileRetryMax.coerceIn(0, 3) + 1,
// A manual full run uses one shared endpoint. If its first task
// exhausts retries, probing more groups only repeats the same
// configuration failure and floods the log.
maxConsecutiveExhaustedTasks = 1,
)
sendMessage( sendMessage(
"已启动 ${parsedGroupIds.size} 个群的画像分析,每群最多推进 $batches 个批次。" + "已启动 ${parsedGroupIds.size} 个群的画像分析,每群最多推进 $batches 个批次。" +
if (analyzeAllGroups) { " 请求并发上限 ${normalizeMaxConcurrentRequests(PluginConfig.profileMaxConcurrentRequests)}" +
" 全量启动门槛 $minimumMessages 条待处理消息" + if (analyzeAllGroups) " 全量启动门槛 $minimumMessages 条待处理消息" else ""
"画像请求并发上限 " +
"${normalizeMaxConcurrentRequests(PluginConfig.profileMaxConcurrentRequests)}"
} else {
""
}
) )
val completedGroups = AtomicInteger() val completedGroups = AtomicInteger()
val successfulGroups = AtomicInteger() val successfulGroups = AtomicInteger()
val alreadyRunningGroups = AtomicInteger() val alreadyRunningGroups = AtomicInteger()
val missingGroups = AtomicInteger() val missingGroups = AtomicInteger()
val failedGroups = AtomicInteger() val failedGroups = AtomicInteger()
val stoppedGroups = AtomicInteger()
val failureLogged = java.util.concurrent.atomic.AtomicBoolean()
parsedGroupIds.forEach { groupId -> parsedGroupIds.forEach { groupId ->
JChatGPT.launch { JChatGPT.launch {
try { try {
val report = UserProfileAnalysisService.analyzeGroup(groupId, batches, runToken) { progress -> val report = UserProfileAnalysisService.analyzeGroupControlled(
groupId = groupId,
maxBatches = batches,
runToken = runToken,
requestController = requestController,
) { progress ->
JChatGPT.logger.info( JChatGPT.logger.info(
"PROFILE_GROUP_BATCH group=$groupId batch=${progress.batchIndex}/$batches " + "PROFILE_GROUP_BATCH group=$groupId batch=${progress.batchIndex}/$batches " +
"range=${progress.startTime}-${progress.endTime} " + "range=${progress.startTime}-${progress.endTime} " +
@@ -174,9 +186,13 @@ object PluginCommands : CompositeCommand(
} }
} catch (cause: CancellationException) { } catch (cause: CancellationException) {
throw cause throw cause
} catch (cause: ProfileDailyRunStoppedException) {
stoppedGroups.incrementAndGet()
} catch (cause: Exception) { } catch (cause: Exception) {
failedGroups.incrementAndGet() failedGroups.incrementAndGet()
JChatGPT.logger.error("$groupId 批量画像分析失败", cause) if (failureLogged.compareAndSet(false, true)) {
JChatGPT.logger.error("$groupId 批量画像分析失败,本轮仅输出一次代表性异常", cause)
}
if (!analyzeAllGroups) { if (!analyzeAllGroups) {
sendMessage("$groupId 批量画像分析失败:${cause.message ?: cause::class.simpleName}") sendMessage("$groupId 批量画像分析失败:${cause.message ?: cause::class.simpleName}")
} }
@@ -185,7 +201,8 @@ object PluginCommands : CompositeCommand(
sendMessage( sendMessage(
"全量群画像分析完成:成功 ${successfulGroups.get()}" + "全量群画像分析完成:成功 ${successfulGroups.get()}" +
"已在运行 ${alreadyRunningGroups.get()}" + "已在运行 ${alreadyRunningGroups.get()}" +
"无历史 ${missingGroups.get()},失败 ${failedGroups.get()}" "无历史 ${missingGroups.get()},失败 ${failedGroups.get()}" +
"停止 ${stoppedGroups.get()}"
) )
} }
} }
@@ -196,6 +196,25 @@ class ProfileDailyRequestControllerTest {
assertEquals(1, stats.recoveredIncidents) assertEquals(1, stats.recoveredIncidents)
} }
@Test
fun canStopTheRunAfterTheFirstTaskExhaustsItsAttempts() = runBlocking {
val controller = ProfileDailyRequestController(
maxConcurrentRequests = 128,
maxAttempts = 3,
maxConsecutiveExhaustedTasks = 1,
)
repeat(3) { attemptIndex -> fail(controller, key(1), attemptIndex + 1) }
val stats = controller.snapshot()
assertTrue(stats.stopped)
assertEquals(1, stats.exhaustedTasks)
assertFailsWith<ProfileDailyRunStoppedException> {
controller.execute(key(2), attempt = 1) { }
}
Unit
}
@Test @Test
fun cancellationWhileWaitingDoesNotLeaveAQueuedRequest() = runBlocking { fun cancellationWhileWaitingDoesNotLeaveAQueuedRequest() = runBlocking {
val controller = controller(maxConcurrency = 1) val controller = controller(maxConcurrency = 1)