diff --git a/README.md b/README.md index d24ba02..95964d3 100644 --- a/README.md +++ b/README.md @@ -118,6 +118,11 @@ contactSnapshotGroupDelayMillis: 200 profileBatchTargetMessages: 120 # 全量推进群画像时，至少累计多少条尚未处理的有效群消息才启动 profileBulkGroupMinPendingMessages: 20 +# 每天按服务器本地时间自动追平近期群画像；默认关闭 +profileDailyGroupUpdateEnabled: false +profileDailyGroupUpdateTime: '04:30' +# 最早待处理消息距今不超过该天数的群才允许自动推进 +profileDailyGroupUpdateMaxPendingAgeDays: 7 profileBatchMaxEpisodes: 16 profileEpisodeGapMinutes: 60 profileContextBeforeMessages: 30 @@ -246,6 +251,13 @@ searchHistoryMaxRecords: 5000 已不存在的操作目标会被跳过，不会重新请求模型。显式传入群号时仍只分析指定群。全量模式仅发送启动和最终汇总， 逐群进度与结果写入日志，避免回执刷屏。 +启用 `profileDailyGroupUpdateEnabled` 后，插件会在服务器本地时间每天按 `profileDailyGroupUpdateTime`（默认 `04:30`） +自动推进群画像。没有群游标时从该群最早历史开始判断，因此新安装后积累的近期历史可直接进入每日维护；已经追平后 +长期沉寂的群不会因游标时间较旧而被误判。准入依据是游标之后最早一条未撤回群消息，只有它距今不超过 +`profileDailyGroupUpdateMaxPendingAgeDays`（默认 7 天）才自动处理，所以旧积压不会被后来的一两条新消息掩盖。 +每日任务不使用 20 条的人工全量启动门槛，会把合格群推进到本轮固定历史快照；内容不足门槛的参与者不会调用模型。 +它与人工全量共用现有的每群运行锁：仍在推进的群会跳过，已经失败并释放锁的群可以从已保存水位继续处理。 + 下一次正常群聊会自动携带触发者和最近发言者的认识。现有好感度、Bot 代号、标签和主观印象会与证据驱动的 长期画像按同一个人合并渲染，并明确给出长期画像条目数；私聊也会携带对方的可靠画像摘要和条目数。 模型需要完整细节时可主动调用 `queryUserProfile`，按 QQ 号、群名片、昵称或好友备注读取完整画像。两套画像数据 diff --git a/src/main/kotlin/JChatGPT.kt b/src/main/kotlin/JChatGPT.kt index 1cc4c7c..d544973 100644 --- a/src/main/kotlin/JChatGPT.kt +++ b/src/main/kotlin/JChatGPT.kt @@ -34,6 +34,7 @@ import top.jie65535.mirai.data.SkillStore import top.jie65535.mirai.data.TokenUsageStore import top.jie65535.mirai.llm.LargeLanguageModels import top.jie65535.mirai.profile.ProfileAutoMaintenance +import top.jie65535.mirai.profile.ProfileDailyMaintenance import top.jie65535.mirai.profile.UserProfileStore import kotlin.random.Random @@ -77,6 +78,7 @@ object JChatGPT : KotlinPlugin( .onFailure { logger.error("初始化用户画像数据库失败，画像分析将暂时禁用", it) } LargeLanguageModels.reload() + ProfileDailyMaintenance.reload() PluginCommands.register() keyword = PluginConfig.callKeyword.takeIf(String::isNotEmpty)?.let(::Regex) @@ -104,6 +106,7 @@ object JChatGPT : KotlinPlugin( ConversationEngine.clear() ConversationContext.clearAll() ProfileAutoMaintenance.clear() + ProfileDailyMaintenance.clear() ContactSnapshotRefresher.clear() UserProfileStore.close() ContactSnapshotStore.close() diff --git a/src/main/kotlin/command/PluginCommands.kt b/src/main/kotlin/command/PluginCommands.kt index bc37e3a..7196388 100644 --- a/src/main/kotlin/command/PluginCommands.kt +++ b/src/main/kotlin/command/PluginCommands.kt @@ -26,6 +26,7 @@ import top.jie65535.mirai.profile.ProfileAnalysisReport import top.jie65535.mirai.profile.ProfileAutoMaintenance import top.jie65535.mirai.profile.ProfileCategory import top.jie65535.mirai.profile.ProfileCompactionReport +import top.jie65535.mirai.profile.ProfileDailyMaintenance import top.jie65535.mirai.profile.ProfilePersistentText import top.jie65535.mirai.profile.UserProfileAnalysisService import top.jie65535.mirai.profile.UserProfileSnapshot @@ -46,6 +47,7 @@ object PluginCommands : CompositeCommand( PluginConfig.reload() PluginData.reload() LargeLanguageModels.reload() + ProfileDailyMaintenance.reload() if (!PluginConfig.profileEnabled || !PluginConfig.profileAutoUpdateEnabled) { ProfileAutoMaintenance.clear() } diff --git a/src/main/kotlin/config/PluginConfig.kt b/src/main/kotlin/config/PluginConfig.kt index eb131eb..f2793c3 100644 --- a/src/main/kotlin/config/PluginConfig.kt +++ b/src/main/kotlin/config/PluginConfig.kt @@ -99,6 +99,15 @@ object PluginConfig : AutoSavePluginConfig("Config") { @ValueDescription("全量推进群画像时，群内至少有多少条尚未处理的消息才启动；显式指定群号不受限制") val profileBulkGroupMinPendingMessages: Int by value(20) + @ValueDescription("是否每天定时推进已接近历史水位线的群画像") + val profileDailyGroupUpdateEnabled: Boolean by value(false) + + @ValueDescription("每日群画像推进时间，使用服务器本地时区，格式 HH:mm") + val profileDailyGroupUpdateTime: String by value("04:30") + + @ValueDescription("每日群画像只自动推进最早待处理消息距今不超过多少天的群，必须为正数") + val profileDailyGroupUpdateMaxPendingAgeDays: Int by value(7) + @ValueDescription("每个画像分析批次最多包含多少个离散对话片段；同一秒的消息仍会一起处理") val profileBatchMaxEpisodes: Int by value(16) diff --git a/src/main/kotlin/profile/ProfileDailyMaintenance.kt b/src/main/kotlin/profile/ProfileDailyMaintenance.kt new file mode 100644 index 0000000..9a1425c --- /dev/null +++ b/src/main/kotlin/profile/ProfileDailyMaintenance.kt @@ -0,0 +1,202 @@ +package top.jie65535.mirai.profile + +import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.Job +import kotlinx.coroutines.async +import kotlinx.coroutines.awaitAll +import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.currentCoroutineContext +import kotlinx.coroutines.delay +import kotlinx.coroutines.isActive +import kotlinx.coroutines.launch +import top.jie65535.mirai.JChatGPT +import top.jie65535.mirai.config.PluginConfig +import top.jie65535.mirai.llm.LargeLanguageModels +import java.time.Duration +import java.time.Instant +import java.time.LocalTime +import java.time.ZoneId +import java.time.ZonedDateTime +import java.time.format.DateTimeFormatter +import java.time.format.DateTimeFormatterBuilder +import java.time.format.ResolverStyle +import java.time.temporal.ChronoField +import java.util.Locale + +object ProfileDailyMaintenance { + private val lock = Any() + private var schedulerJob: Job? = null + + fun reload() { + synchronized(lock) { + schedulerJob?.cancel() + schedulerJob = null + if (!PluginConfig.profileEnabled || !PluginConfig.profileDailyGroupUpdateEnabled) return@synchronized + if (!UserProfileStore.isAvailable) { + JChatGPT.logger.warning("用户画像数据库不可用，每日群画像推进未启动") + return@synchronized + } + + val scheduledTime = parseProfileDailyUpdateTime(PluginConfig.profileDailyGroupUpdateTime) + if (scheduledTime == null) { + JChatGPT.logger.warning( + "每日群画像推进时间无效: '${PluginConfig.profileDailyGroupUpdateTime}'，请使用 HH:mm 格式" + ) + return@synchronized + } + val maxPendingAgeDays = PluginConfig.profileDailyGroupUpdateMaxPendingAgeDays + if (maxPendingAgeDays <= 0) { + JChatGPT.logger.warning("每日群画像最大积压天数必须为正数，当前值: $maxPendingAgeDays") + return@synchronized + } + + val zoneId = ZoneId.systemDefault() + schedulerJob = JChatGPT.launch { + scheduleLoop(scheduledTime, zoneId, maxPendingAgeDays) + } + } + } + + fun clear() { + synchronized(lock) { + schedulerJob?.cancel() + schedulerJob = null + } + } + + private suspend fun scheduleLoop( + scheduledTime: LocalTime, + zoneId: ZoneId, + maxPendingAgeDays: Int, + ) { + while (currentCoroutineContext().isActive) { + val now = ZonedDateTime.now(zoneId) + val nextRun = nextProfileDailyUpdateAt(now, scheduledTime) + val delayMillis = Duration.between(now.toInstant(), nextRun.toInstant()).toMillis().coerceAtLeast(1L) + JChatGPT.logger.info( + "每日群画像推进已计划: next=$nextRun maxPendingAgeDays=$maxPendingAgeDays" + ) + delay(delayMillis) + try { + runOnce(maxPendingAgeDays) + } catch (cause: CancellationException) { + throw cause + } catch (cause: Exception) { + JChatGPT.logger.error("每日群画像推进失败，将在下一计划时间重试", cause) + } + } + } + + private suspend fun runOnce(maxPendingAgeDays: Int) { + if (!PluginConfig.profileEnabled || !PluginConfig.profileDailyGroupUpdateEnabled) return + if (!UserProfileStore.isAvailable) { + JChatGPT.logger.warning("用户画像数据库不可用，跳过本次每日群画像推进") + return + } + if (LargeLanguageModels.profile == null) { + JChatGPT.logger.warning("画像分析模型未配置，跳过本次每日群画像推进") + return + } + + val oldestAllowedTime = oldestAllowedProfilePendingTime(Instant.now().epochSecond, maxPendingAgeDays) + val groupIds = UserProfileAnalysisService.listRecentPendingHistoryGroupIds(oldestAllowedTime) + if (groupIds.isEmpty()) { + JChatGPT.logger.info("每日群画像推进无需运行：没有符合 $maxPendingAgeDays 天积压门槛的群") + return + } + + JChatGPT.logger.info("每日群画像推进开始: groups=${groupIds.size} maxPendingAgeDays=$maxPendingAgeDays") + val runToken = UserProfileAnalysisService.newRunToken() + val outcomes = coroutineScope { + groupIds.map { groupId -> + async { analyzeGroup(groupId, runToken) } + }.awaitAll() + } + + val reports = outcomes.mapNotNull(DailyGroupOutcome::report) + JChatGPT.logger.info( + "每日群画像推进完成: selected=${groupIds.size} " + + "success=${outcomes.count { it.status == DailyGroupStatus.SUCCESS }} " + + "alreadyRunning=${outcomes.count { it.status == DailyGroupStatus.ALREADY_RUNNING }} " + + "missing=${outcomes.count { it.status == DailyGroupStatus.MISSING_HISTORY }} " + + "stopped=${outcomes.count { it.status == DailyGroupStatus.STOPPED }} " + + "failed=${outcomes.count { it.status == DailyGroupStatus.FAILED }} " + + "batches=${reports.sumOf(GroupProfileAnalysisReport::processedBatches)} " + + "messages=${reports.sumOf(GroupProfileAnalysisReport::processedMessages)} " + + "operations=${reports.sumOf(GroupProfileAnalysisReport::appliedOperations)} " + + "tokens=${reports.sumOf { it.usage.promptTokens }}/${reports.sumOf { it.usage.completionTokens }} " + + "cached=${reports.sumOf { it.usage.cachedTokens }}" + ) + } + + private suspend fun analyzeGroup( + groupId: Long, + runToken: ProfileAnalysisRunToken, + ): DailyGroupOutcome { + return try { + val report = UserProfileAnalysisService.analyzeGroup( + groupId = groupId, + maxBatches = Int.MAX_VALUE, + runToken = runToken, + ) { progress -> + JChatGPT.logger.info( + "PROFILE_DAILY_BATCH group=$groupId batch=${progress.batchIndex} " + + "range=${progress.startTime}-${progress.endTime} " + + "messages=${progress.messageCount} users=${progress.analyzedUsers} " + + "operations=${progress.appliedOperations} skipped=${progress.skippedOperations} " + + "tokens=${progress.usage.promptTokens}/${progress.usage.completionTokens} " + + "cached=${progress.usage.cachedTokens}" + ) + } + val status = when { + report.alreadyRunning -> DailyGroupStatus.ALREADY_RUNNING + report.botId == null -> DailyGroupStatus.MISSING_HISTORY + report.stopped -> DailyGroupStatus.STOPPED + else -> DailyGroupStatus.SUCCESS + } + DailyGroupOutcome(status, report.takeUnless { report.alreadyRunning || report.botId == null }) + } catch (cause: CancellationException) { + throw cause + } catch (cause: Exception) { + JChatGPT.logger.error("群 $groupId 每日画像推进失败", cause) + DailyGroupOutcome(DailyGroupStatus.FAILED) + } + } + + private data class DailyGroupOutcome( + val status: DailyGroupStatus, + val report: GroupProfileAnalysisReport? = null, + ) + + private enum class DailyGroupStatus { + SUCCESS, + ALREADY_RUNNING, + MISSING_HISTORY, + STOPPED, + FAILED, + } +} + +private val PROFILE_DAILY_TIME_FORMATTER: DateTimeFormatter = DateTimeFormatterBuilder() + .parseStrict() + .appendValue(ChronoField.HOUR_OF_DAY, 2) + .appendLiteral(':') + .appendValue(ChronoField.MINUTE_OF_HOUR, 2) + .toFormatter(Locale.ROOT) + .withResolverStyle(ResolverStyle.STRICT) + +internal fun parseProfileDailyUpdateTime(value: String): LocalTime? = runCatching { + LocalTime.parse(value.trim(), PROFILE_DAILY_TIME_FORMATTER) +}.getOrNull() + +internal fun nextProfileDailyUpdateAt(now: ZonedDateTime, scheduledTime: LocalTime): ZonedDateTime { + val today = now.toLocalDate().atTime(scheduledTime).atZone(now.zone) + return if (today.isAfter(now)) today else now.toLocalDate().plusDays(1).atTime(scheduledTime).atZone(now.zone) +} + +internal fun oldestAllowedProfilePendingTime(nowEpochSecond: Long, maxPendingAgeDays: Int): Int { + require(maxPendingAgeDays > 0) { "maxPendingAgeDays must be positive" } + return (nowEpochSecond - maxPendingAgeDays.toLong() * 24L * 60L * 60L) + .coerceIn(0L, Int.MAX_VALUE.toLong()) + .toInt() +} diff --git a/src/main/kotlin/profile/ProfileHistoryReader.kt b/src/main/kotlin/profile/ProfileHistoryReader.kt index 06b13eb..772d12b 100644 --- a/src/main/kotlin/profile/ProfileHistoryReader.kt +++ b/src/main/kotlin/profile/ProfileHistoryReader.kt @@ -181,6 +181,37 @@ class ProfileHistoryReader(private val databaseFile: File) { } } + fun filterGroupRangesByOldestPendingMessageTime( + ranges: List, + oldestAllowedTime: Int, + ): List { + if (ranges.isEmpty()) return emptyList() + require(oldestAllowedTime >= 0) { "oldestAllowedTime must not be negative" } + return openReadConnection().use { connection -> + connection.prepareStatement( + """ + SELECT MIN(time) AS oldest_time + FROM message_record + WHERE bot_id = ? AND target_id = ? AND kind = ? AND recalled = 0 + AND time >= ? AND time < ? + """.trimIndent() + ).use { statement -> + ranges.filter { range -> + statement.setLong(1, range.botId) + statement.setLong(2, range.groupId) + statement.setInt(3, MessageSourceKind.GROUP.ordinal) + statement.setInt(4, range.startTime) + statement.setInt(5, range.endTime) + statement.executeQuery().use { results -> + check(results.next()) { "读取群 ${range.groupId} 最早待处理消息失败" } + val oldestTime = results.getInt("oldest_time") + results.wasNull() || oldestTime >= oldestAllowedTime + } + } + } + } + } + fun loadNextBatch( userId: Long, startTime: Int, diff --git a/src/main/kotlin/profile/UserProfileAnalysisService.kt b/src/main/kotlin/profile/UserProfileAnalysisService.kt index da2c945..45301a4 100644 --- a/src/main/kotlin/profile/UserProfileAnalysisService.kt +++ b/src/main/kotlin/profile/UserProfileAnalysisService.kt @@ -54,6 +54,25 @@ object UserProfileAnalysisService { } } + suspend fun listRecentPendingHistoryGroupIds(oldestAllowedMessageTime: Int): List { + require(oldestAllowedMessageTime >= 0) { "oldestAllowedMessageTime must not be negative" } + check(PluginConfig.profileEnabled) { "历史用户画像分析未启用" } + check(UserProfileStore.isAvailable) { "用户画像数据库不可用" } + return withContext(Dispatchers.IO) { + val reader = ProfileHistoryReader(resolveHistoryFile()) + val historyBounds = reader.listGroupTimeBounds() + val cursors = UserProfileStore.loadGroupCursors() + .associateBy { cursor -> cursor.botId to cursor.groupId } + val pendingRanges = historyBounds.mapNotNull { bounds -> + pendingGroupAnalysisRange(bounds, cursors[bounds.botId to bounds.groupId]) + } + reader.filterGroupRangesByOldestPendingMessageTime( + ranges = pendingRanges, + oldestAllowedTime = oldestAllowedMessageTime, + ).map(ProfileHistoryReader.GroupTimeBounds::groupId) + } + } + suspend fun analyze( userId: Long, maxBatches: Int, diff --git a/src/test/kotlin/profile/ProfileDailyMaintenanceTest.kt b/src/test/kotlin/profile/ProfileDailyMaintenanceTest.kt new file mode 100644 index 0000000..de4aebc --- /dev/null +++ b/src/test/kotlin/profile/ProfileDailyMaintenanceTest.kt @@ -0,0 +1,46 @@ +package top.jie65535.mirai.profile + +import java.time.LocalTime +import java.time.ZoneId +import java.time.ZonedDateTime +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertNull + +class ProfileDailyMaintenanceTest { + @Test + fun parsesStrictDailyUpdateTime() { + assertEquals(LocalTime.of(4, 30), parseProfileDailyUpdateTime("04:30")) + assertEquals(LocalTime.of(23, 59), parseProfileDailyUpdateTime(" 23:59 ")) + assertNull(parseProfileDailyUpdateTime("4:30")) + assertNull(parseProfileDailyUpdateTime("24:00")) + assertNull(parseProfileDailyUpdateTime("04:60")) + } + + @Test + fun schedulesTodayOrTomorrowInServerTimeZone() { + val zone = ZoneId.of("Asia/Shanghai") + val scheduledTime = LocalTime.of(4, 30) + + assertEquals( + ZonedDateTime.of(2026, 8, 5, 4, 30, 0, 0, zone), + nextProfileDailyUpdateAt( + ZonedDateTime.of(2026, 8, 5, 3, 0, 0, 0, zone), + scheduledTime, + ), + ) + assertEquals( + ZonedDateTime.of(2026, 8, 6, 4, 30, 0, 0, zone), + nextProfileDailyUpdateAt( + ZonedDateTime.of(2026, 8, 5, 4, 30, 0, 0, zone), + scheduledTime, + ), + ) + } + + @Test + fun calculatesOldestAllowedPendingMessageTime() { + assertEquals(395_200, oldestAllowedProfilePendingTime(1_000_000, 7)) + assertEquals(0, oldestAllowedProfilePendingTime(100, 7)) + } +} diff --git a/src/test/kotlin/profile/ProfileHistoryReaderTest.kt b/src/test/kotlin/profile/ProfileHistoryReaderTest.kt index 8e3871d..c60e167 100644 --- a/src/test/kotlin/profile/ProfileHistoryReaderTest.kt +++ b/src/test/kotlin/profile/ProfileHistoryReaderTest.kt @@ -62,6 +62,8 @@ class ProfileHistoryReaderTest { insert(1, 20, 110, kind = MessageSourceKind.FRIEND) insert(2, 30, 100) insert(2, 30, 110) + insert(1, 40, 100) + insert(1, 40, 1_000) } } @@ -80,6 +82,13 @@ class ProfileHistoryReaderTest { listOf(10L, 20L), reader.filterGroupRangesByMinimumMessageCount(ranges, 1).map { it.groupId }, ) + assertEquals( + listOf(10L, 30L), + reader.filterGroupRangesByOldestPendingMessageTime( + ranges = ranges + ProfileHistoryReader.GroupTimeBounds(1, 40, 90, 1_001), + oldestAllowedTime = 115, + ).map { it.groupId }, + ) } finally { directory.toFile().deleteRecursively() }