nezumi/astrbot_plugin_qq_group_daily_analysis
1
0
Fork 0
mirror of https://github.com/Nezumi-2711/astrbot_plugin_qq_group_daily_analysis.git synced 2026-09-22 13:38:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore: 注释

Browse Source
This commit is contained in:
SXP-Simon
2026-02-08 15:06:33 +08:00
parent cbc98f8cfd
commit 39e5d6169d
29 changed files with 703 additions and 623 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/application/__init__.py
+1 -1
View File
@@ -1,4 +1,4 @@
# Application Layer - Orchestration and Use Cases
# 应用层 - 编排和用例
from .analysis_orchestrator import AnalysisOrchestrator
from .message_converter import MessageConverter
from .scheduling_service import SchedulingService
src/domain/entities/__init__.py
+1 -1
View File
@@ -1,4 +1,4 @@
# Domain Entities
# 领域实体
from .analysis_task import AnalysisTask, TaskStatus
from .analysis_result import GroupAnalysisResult
src/domain/entities/analysis_result.py
+12 -12
View File
@@ -1,5 +1,5 @@
"""
Group Analysis Result Entity
群聊分析结果实体
"""
from dataclasses import dataclass, field
@@ -10,7 +10,7 @@ import time
@dataclass
class SummaryTopic:
"""Topic summary"""
"""话题摘要"""
topic: str
contributors: List[str]
detail: str
@@ -18,7 +18,7 @@ class SummaryTopic:
@dataclass
class UserTitle:
"""User title/portrait"""
"""用户称号/画像"""
name: str
user_id: str
title: str
@@ -30,7 +30,7 @@ class UserTitle:
@dataclass
class GoldenQuote:
"""Golden quote"""
"""金句"""
content: str
sender: str
reason: str
@@ -40,7 +40,7 @@ class GoldenQuote:
@dataclass
class TokenUsage:
"""Token usage statistics"""
"""令牌使用统计"""
prompt_tokens: int = 0
completion_tokens: int = 0
total_tokens: int = 0
@@ -48,7 +48,7 @@ class TokenUsage:
@dataclass
class EmojiStatistics:
"""Emoji statistics"""
"""表情统计"""
face_count: int = 0
mface_count: int = 0
bface_count: int = 0
@@ -69,7 +69,7 @@ class EmojiStatistics:
@dataclass
class ActivityVisualization:
"""Activity visualization data"""
"""活动可视化数据"""
hourly_activity: dict = field(default_factory=dict)
daily_activity: dict = field(default_factory=dict)
user_activity_ranking: list = field(default_factory=list)
@@ -79,7 +79,7 @@ class ActivityVisualization:
@dataclass
class GroupStatistics:
"""Group statistics"""
"""群组统计"""
message_count: int = 0
total_characters: int = 0
participant_count: int = 0
@@ -91,25 +91,25 @@ class GroupStatistics:
@dataclass
class GroupAnalysisResult:
"""Group analysis result entity"""
"""群聊分析结果实体"""
id: str = field(default_factory=lambda: uuid.uuid4().hex[:8])
group_id: str = ""
group_name: str = ""
trace_id: str = ""
platform: str = ""
# Analysis results
# 分析结果
message_count: int = 0
statistics: GroupStatistics = field(default_factory=GroupStatistics)
topics: List[SummaryTopic] = field(default_factory=list)
user_titles: List[UserTitle] = field(default_factory=list)
golden_quotes: List[GoldenQuote] = field(default_factory=list)
# Metadata
# 元数据
token_usage: TokenUsage = field(default_factory=TokenUsage)
analysis_date: str = ""
created_at: float = field(default_factory=time.time)
def has_content(self) -> bool:
"""Check if result has any analysis content"""
"""检查结果是否有分析内容"""
return bool(self.topics or self.user_titles or self.golden_quotes)
src/domain/entities/analysis_task.py
+8 -8
View File
@@ -1,5 +1,5 @@
"""
Analysis Task Entity - Aggregate Root
分析任务实体 - 聚合根
"""
from dataclasses import dataclass, field
@@ -23,7 +23,7 @@ class TaskStatus(Enum):
@dataclass
class AnalysisTask:
"""Analysis task entity - Aggregate root"""
"""分析任务实体 - 聚合根"""
id: str = field(default_factory=lambda: uuid.uuid4().hex[:8])
group_id: str = ""
platform_name: str = ""
@@ -37,34 +37,34 @@ class AnalysisTask:
error_message: Optional[str] = None
def start(self, can_analyze: bool) -> bool:
"""Start task, validate platform capability"""
"""启动任务，验证平台能力"""
if not can_analyze:
self.status = TaskStatus.UNSUPPORTED_PLATFORM
self.error_message = f"Platform {self.platform_name} does not support analysis"
self.error_message = f"平台 {self.platform_name} 不支持分析"
return False
self.status = TaskStatus.FETCHING_MESSAGES
self.started_at = time.time()
return True
def advance_to(self, status: TaskStatus):
"""Advance to next status"""
"""推进到下一个状态"""
self.status = status
def complete(self, result_id: str):
"""Mark task as completed"""
"""标记任务为已完成"""
self.status = TaskStatus.COMPLETED
self.result_id = result_id
self.completed_at = time.time()
def fail(self, error: str):
"""Mark task as failed"""
"""标记任务为失败"""
self.status = TaskStatus.FAILED
self.error_message = error
self.completed_at = time.time()
@property
def duration(self) -> Optional[float]:
"""Get task duration in seconds"""
"""获取任务持续时间（秒）"""
if self.started_at and self.completed_at:
return self.completed_at - self.started_at
return None
src/domain/exceptions.py
+58 -59
View File
@@ -1,14 +1,13 @@
"""
Domain Exceptions - Custom exceptions for the domain layer
领域异常 - 领域层自定义异常
This module contains all domain-specific exceptions used throughout
the plugin. These exceptions are platform-agnostic and represent
business logic errors.
该模块包含插件中使用的所有领域特定异常。
这些异常是平台无关的，表示业务逻辑错误。
"""
class DomainException(Exception):
"""Base exception for all domain errors."""
"""所有领域错误的基础异常。"""
def __init__(self, message: str, code: str = "DOMAIN_ERROR"):
self.message = message
@@ -17,62 +16,62 @@ class DomainException(Exception):
# ============================================================================
# Analysis Exceptions
# 分析异常
# ============================================================================
class AnalysisException(DomainException):
"""Base exception for analysis-related errors."""
"""分析相关错误的基础异常。"""
def __init__(self, message: str, code: str = "ANALYSIS_ERROR"):
super().__init__(message, code)
class InsufficientDataException(AnalysisException):
"""Raised when there is not enough data to perform analysis."""
"""当数据不足以进行分析时抛出。"""
def __init__(self, message: str = "Insufficient data for analysis"):
def __init__(self, message: str = "数据不足，无法进行分析"):
super().__init__(message, "INSUFFICIENT_DATA")
class AnalysisTimeoutException(AnalysisException):
"""Raised when analysis takes too long."""
"""当分析超时时抛出。"""
def __init__(self, message: str = "Analysis timed out"):
def __init__(self, message: str = "分析超时"):
super().__init__(message, "ANALYSIS_TIMEOUT")
class LLMException(AnalysisException):
"""Raised when LLM API call fails."""
"""当 LLM API 调用失败时抛出。"""
def __init__(self, message: str = "LLM API call failed", provider: str = ""):
def __init__(self, message: str = "LLM API 调用失败", provider: str = ""):
self.provider = provider
super().__init__(f"{message} (provider: {provider})" if provider else message, "LLM_ERROR")
super().__init__(f"{message} (提供商: {provider})" if provider else message, "LLM_ERROR")
class LLMRateLimitException(LLMException):
"""Raised when LLM API rate limit is exceeded."""
"""当 LLM API 速率限制超出时抛出。"""
def __init__(self, message: str = "LLM rate limit exceeded", provider: str = ""):
def __init__(self, message: str = "LLM 速率限制超出", provider: str = ""):
super().__init__(message, provider)
self.code = "LLM_RATE_LIMIT"
class LLMQuotaExceededException(LLMException):
"""Raised when LLM API quota is exceeded."""
"""当 LLM API 配额超出时抛出。"""
def __init__(self, message: str = "LLM quota exceeded", provider: str = ""):
def __init__(self, message: str = "LLM 配额超出", provider: str = ""):
super().__init__(message, provider)
self.code = "LLM_QUOTA_EXCEEDED"
# ============================================================================
# Platform Exceptions
# 平台异常
# ============================================================================
class PlatformException(DomainException):
"""Base exception for platform-related errors."""
"""平台相关错误的基础异常。"""
def __init__(self, message: str, platform: str = "", code: str = "PLATFORM_ERROR"):
self.platform = platform
@@ -80,133 +79,133 @@ class PlatformException(DomainException):
class PlatformNotSupportedException(PlatformException):
"""Raised when a platform is not supported."""
"""当平台不被支持时抛出。"""
def __init__(self, platform: str):
super().__init__(f"Platform '{platform}' is not supported", platform, "PLATFORM_NOT_SUPPORTED")
super().__init__(f"平台 '{platform}' 不被支持", platform, "PLATFORM_NOT_SUPPORTED")
class PlatformConnectionException(PlatformException):
"""Raised when connection to platform fails."""
"""当连接平台失败时抛出。"""
def __init__(self, message: str = "Failed to connect to platform", platform: str = ""):
def __init__(self, message: str = "连接平台失败", platform: str = ""):
super().__init__(message, platform, "PLATFORM_CONNECTION_ERROR")
class PlatformAPIException(PlatformException):
"""Raised when platform API call fails."""
"""当平台 API 调用失败时抛出。"""
def __init__(self, message: str = "Platform API call failed", platform: str = ""):
def __init__(self, message: str = "平台 API 调用失败", platform: str = ""):
super().__init__(message, platform, "PLATFORM_API_ERROR")
class MessageFetchException(PlatformException):
"""Raised when fetching messages fails."""
"""当获取消息失败时抛出。"""
def __init__(self, message: str = "Failed to fetch messages", platform: str = "", group_id: str = ""):
def __init__(self, message: str = "获取消息失败", platform: str = "", group_id: str = ""):
self.group_id = group_id
super().__init__(f"{message} (group: {group_id})" if group_id else message, platform, "MESSAGE_FETCH_ERROR")
super().__init__(f"{message} (群组: {group_id})" if group_id else message, platform, "MESSAGE_FETCH_ERROR")
class MessageSendException(PlatformException):
"""Raised when sending a message fails."""
"""当发送消息失败时抛出。"""
def __init__(self, message: str = "Failed to send message", platform: str = "", group_id: str = ""):
def __init__(self, message: str = "发送消息失败", platform: str = "", group_id: str = ""):
self.group_id = group_id
super().__init__(f"{message} (group: {group_id})" if group_id else message, platform, "MESSAGE_SEND_ERROR")
super().__init__(f"{message} (群组: {group_id})" if group_id else message, platform, "MESSAGE_SEND_ERROR")
# ============================================================================
# Configuration Exceptions
# 配置异常
# ============================================================================
class ConfigurationException(DomainException):
"""Base exception for configuration-related errors."""
"""配置相关错误的基础异常。"""
def __init__(self, message: str, code: str = "CONFIG_ERROR"):
super().__init__(message, code)
class InvalidConfigurationException(ConfigurationException):
"""Raised when configuration is invalid."""
"""当配置无效时抛出。"""
def __init__(self, message: str = "Invalid configuration", key: str = ""):
def __init__(self, message: str = "无效的配置", key: str = ""):
self.key = key
super().__init__(f"{message}: {key}" if key else message, "INVALID_CONFIG")
class MissingConfigurationException(ConfigurationException):
"""Raised when required configuration is missing."""
"""当缺少必需配置时抛出。"""
def __init__(self, key: str):
self.key = key
super().__init__(f"Missing required configuration: {key}", "MISSING_CONFIG")
super().__init__(f"缺少必需配置: {key}", "MISSING_CONFIG")
# ============================================================================
# Repository Exceptions
# 仓储异常
# ============================================================================
class RepositoryException(DomainException):
"""Base exception for repository-related errors."""
"""仓储相关错误的基础异常。"""
def __init__(self, message: str, code: str = "REPOSITORY_ERROR"):
super().__init__(message, code)
class DataNotFoundException(RepositoryException):
"""Raised when requested data is not found."""
"""当请求的数据未找到时抛出。"""
def __init__(self, message: str = "Data not found", entity_type: str = "", entity_id: str = ""):
def __init__(self, message: str = "数据未找到", entity_type: str = "", entity_id: str = ""):
self.entity_type = entity_type
self.entity_id = entity_id
super().__init__(f"{entity_type} not found: {entity_id}" if entity_type else message, "DATA_NOT_FOUND")
super().__init__(f"{entity_type} 未找到: {entity_id}" if entity_type else message, "DATA_NOT_FOUND")
class DataPersistenceException(RepositoryException):
"""Raised when data persistence fails."""
"""当数据持久化失败时抛出。"""
def __init__(self, message: str = "Failed to persist data"):
def __init__(self, message: str = "数据持久化失败"):
super().__init__(message, "DATA_PERSISTENCE_ERROR")
# ============================================================================
# Scheduling Exceptions
# 调度异常
# ============================================================================
class SchedulingException(DomainException):
"""Base exception for scheduling-related errors."""
"""调度相关错误的基础异常。"""
def __init__(self, message: str, code: str = "SCHEDULING_ERROR"):
super().__init__(message, code)
class TaskAlreadyScheduledException(SchedulingException):
"""Raised when trying to schedule an already scheduled task."""
"""当尝试调度已调度的任务时抛出。"""
def __init__(self, task_id: str):
self.task_id = task_id
super().__init__(f"Task already scheduled: {task_id}", "TASK_ALREADY_SCHEDULED")
super().__init__(f"任务已调度: {task_id}", "TASK_ALREADY_SCHEDULED")
class TaskNotFoundException(SchedulingException):
"""Raised when a scheduled task is not found."""
"""当找不到已调度的任务时抛出。"""
def __init__(self, task_id: str):
self.task_id = task_id
super().__init__(f"Scheduled task not found: {task_id}", "TASK_NOT_FOUND")
super().__init__(f"未找到已调度的任务: {task_id}", "TASK_NOT_FOUND")
# ============================================================================
# Validation Exceptions
# 验证异常
# ============================================================================
class ValidationException(DomainException):
"""Base exception for validation errors."""
"""验证错误的基础异常。"""
def __init__(self, message: str, field: str = "", code: str = "VALIDATION_ERROR"):
self.field = field
@@ -214,21 +213,21 @@ class ValidationException(DomainException):
class InvalidGroupIdException(ValidationException):
"""Raised when group ID is invalid."""
"""当群组 ID 无效时抛出。"""
def __init__(self, group_id: str):
super().__init__(f"Invalid group ID: {group_id}", "group_id", "INVALID_GROUP_ID")
super().__init__(f"无效的群组 ID: {group_id}", "group_id", "INVALID_GROUP_ID")
class InvalidUserIdException(ValidationException):
"""Raised when user ID is invalid."""
"""当用户 ID 无效时抛出。"""
def __init__(self, user_id: str):
super().__init__(f"Invalid user ID: {user_id}", "user_id", "INVALID_USER_ID")
super().__init__(f"无效的用户 ID: {user_id}", "user_id", "INVALID_USER_ID")
class InvalidMessageException(ValidationException):
"""Raised when message format is invalid."""
"""当消息格式无效时抛出。"""
def __init__(self, message: str = "Invalid message format"):
def __init__(self, message: str = "无效的消息格式"):
super().__init__(message, "message", "INVALID_MESSAGE")
src/domain/repositories/__init__.py
+1 -1
View File
@@ -1,4 +1,4 @@
# Repository Interfaces
# 仓储接口
from .message_repository import IMessageRepository, IMessageSender, IGroupInfoRepository
from .avatar_repository import IAvatarRepository
src/domain/repositories/avatar_repository.py
+22 -22
View File
@@ -1,5 +1,5 @@
"""
Avatar Repository Interface - Cross-platform avatar abstraction
头像仓储接口 - 跨平台头像抽象
"""
from abc import ABC, abstractmethod
@@ -8,13 +8,13 @@ from typing import List, Optional, Dict
class IAvatarRepository(ABC):
"""
Avatar repository interface
头像仓储接口
Different platforms have different ways to get avatars:
- QQ/OneBot: URL template (q1.qlogo.cn)
- Telegram: API call (getUserProfilePhotos + getFile)
- Discord: CDN URL template (cdn.discordapp.com)
- Slack: users.info API profile.image_* fields
不同平台获取头像的方式不同：
- QQ/OneBot: URL 模板 (q1.qlogo.cn)
- Telegram: API 调用 (getUserProfilePhotos + getFile)
- Discord: CDN URL 模板 (cdn.discordapp.com)
- Slack: users.info API profile.image_* 字段
"""
@abstractmethod
@@ -24,14 +24,14 @@ class IAvatarRepository(ABC):
size: int = 100,
) -> Optional[str]:
"""
Get user avatar URL
获取用户头像 URL
Args:
user_id: User ID
size: Desired avatar size (will pick closest available)
参数:
user_id: 用户 ID
size: 期望的头像尺寸（将选择最接近的可用尺寸）
Returns:
Avatar URL, or None if unavailable
返回:
头像 URL，如果不可用则返回 None
"""
pass
@@ -42,13 +42,13 @@ class IAvatarRepository(ABC):
size: int = 100,
) -> Optional[str]:
"""
Get user avatar as Base64 data
获取用户头像的 Base64 数据
For scenarios needing embedded images (e.g., HTML template rendering)
用于需要嵌入图片的场景（如 HTML 模板渲染）
Returns:
Base64 encoded image data (data:image/png;base64,...),
or None if unavailable
返回:
Base64 编码的图片数据 (data:image/png;base64,...)，
如果不可用则返回 None
"""
pass
@@ -58,7 +58,7 @@ class IAvatarRepository(ABC):
group_id: str,
size: int = 100,
) -> Optional[str]:
"""Get group avatar URL"""
"""获取群组头像 URL"""
pass
@abstractmethod
@@ -68,12 +68,12 @@ class IAvatarRepository(ABC):
size: int = 100,
) -> Dict[str, Optional[str]]:
"""
Batch get user avatar URLs
批量获取用户头像 URL
For report generation needing multiple avatars at once
用于报告生成时需要一次获取多个头像
"""
pass
def get_default_avatar_url(self) -> str:
"""Get default avatar URL (when user avatar unavailable)"""
"""获取默认头像 URL（当用户头像不可用时）"""
return "data:image/svg+xml;base64,PHN2ZyB4bWxucz0iaHR0cDovL3d3dy53My5vcmcvMjAwMC9zdmciIHZpZXdCb3g9IjAgMCAyNCAyNCI+PHBhdGggZD0iTTEyIDEyYzIuMjEgMCA0LTEuNzkgNC00cy0xLjc5LTQtNC00LTQgMS43OS00IDQgMS43OSA0IDQgNHptMCAyYy0yLjY3IDAtOCAxLjM0LTggNHYyaDE2di0yYzAtMi42Ni01LjMzLTQtOC00eiIvPjwvc3ZnPg=="
src/domain/repositories/message_repository.py
+23 -23
View File
@@ -1,5 +1,5 @@
"""
Message Repository Interfaces - Platform-agnostic abstractions
消息仓储接口 - 平台无关的抽象
"""
from abc import ABC, abstractmethod
@@ -12,10 +12,10 @@ from ..value_objects.unified_group import UnifiedGroup, UnifiedMember
class IMessageRepository(ABC):
"""
Message repository interface
消息仓储接口
Each platform adapter must implement this interface.
All methods return unified format, hiding platform differences.
每个平台适配器必须实现此接口。
所有方法返回统一格式，隐藏平台差异。
"""
@abstractmethod
@@ -27,32 +27,32 @@ class IMessageRepository(ABC):
before_id: Optional[str] = None,
) -> List[UnifiedMessage]:
"""
Fetch group message history
获取群组消息历史
Args:
group_id: Group ID
days: Fetch messages from last N days
max_count: Maximum message count
before_id: Fetch messages before this ID (for pagination)
参数:
group_id: 群组 ID
days: 获取最近 N 天的消息
max_count: 最大消息数量
before_id: 获取此 ID 之前的消息（用于分页）
Returns:
List of unified messages, sorted by time ascending
返回:
统一消息列表，按时间升序排列
"""
pass
@abstractmethod
def get_capabilities(self) -> PlatformCapabilities:
"""Get platform capabilities"""
"""获取平台能力"""
pass
@abstractmethod
def get_platform_name(self) -> str:
"""Get platform name"""
"""获取平台名称"""
pass
class IMessageSender(ABC):
"""Message sender interface"""
"""消息发送接口"""
@abstractmethod
async def send_text(
@@ -61,7 +61,7 @@ class IMessageSender(ABC):
text: str,
reply_to: Optional[str] = None,
) -> bool:
"""Send text message"""
"""发送文本消息"""
pass
@abstractmethod
@@ -71,7 +71,7 @@ class IMessageSender(ABC):
image_path: str,
caption: str = "",
) -> bool:
"""Send image message"""
"""发送图片消息"""
pass
@abstractmethod
@@ -81,26 +81,26 @@ class IMessageSender(ABC):
file_path: str,
filename: Optional[str] = None,
) -> bool:
"""Send file"""
"""发送文件"""
pass
class IGroupInfoRepository(ABC):
"""Group info repository interface"""
"""群组信息仓储接口"""
@abstractmethod
async def get_group_info(self, group_id: str) -> Optional[UnifiedGroup]:
"""Get group information"""
"""获取群组信息"""
pass
@abstractmethod
async def get_group_list(self) -> List[str]:
"""Get all group IDs the bot is in"""
"""获取机器人所在的所有群组 ID"""
pass
@abstractmethod
async def get_member_list(self, group_id: str) -> List[UnifiedMember]:
"""Get group member list"""
"""获取群组成员列表"""
pass
@abstractmethod
@@ -109,5 +109,5 @@ class IGroupInfoRepository(ABC):
group_id: str,
user_id: str,
) -> Optional[UnifiedMember]:
"""Get specific member info"""
"""获取指定成员信息"""
pass
src/domain/services/__init__.py
+3 -3
View File
@@ -1,8 +1,8 @@
"""
Domain Services - Business logic services for analysis
领域服务 - 分析业务逻辑服务
This module exports all domain services that encapsulate core business logic
for analyzing group chat data. These services are platform-agnostic.
该模块导出所有封装核心业务逻辑的领域服务，
用于分析群聊数据。这些服务是平台无关的。
"""
from .statistics_calculator import StatisticsCalculator
src/domain/services/report_generator.py
+117 -31
View File
@@ -1,8 +1,8 @@
"""
Report Generator - Domain service for generating analysis reports
报告生成器 - 生成分析报告的领域服务
This service generates formatted reports from analysis results.
It is platform-agnostic and produces text/markdown reports.
该服务从分析结果生成格式化报告。
它是平台无关的，生成文本/Markdown 报告。
"""
from datetime import datetime
@@ -16,19 +16,19 @@ from ..value_objects.statistics import GroupStatistics, TokenUsage
class ReportGenerator:
"""
Domain service for generating analysis reports.
生成分析报告的领域服务。
This service takes analysis results and produces formatted
text reports that can be sent to any platform.
该服务接收分析结果并生成格式化的
文本报告，可发送到任何平台。
"""
def __init__(self, group_name: str = "", date_str: str = ""):
"""
Initialize the report generator.
初始化报告生成器。
Args:
group_name: Name of the group for report header
date_str: Date string for the report
参数:
group_name: 报告标题中的群组名称
date_str: 报告的日期字符串
"""
self.group_name = group_name
self.date_str = date_str or datetime.now().strftime("%Y-%m-%d")
@@ -43,18 +43,18 @@ class ReportGenerator:
include_footer: bool = True,
) -> str:
"""
Generate a complete analysis report.
生成完整的分析报告。
Args:
statistics: Group chat statistics
topics: List of discussion topics
user_titles: List of user titles/badges
golden_quotes: List of golden quotes
include_header: Whether to include report header
include_footer: Whether to include report footer
参数:
statistics: 群聊统计
topics: 讨论话题列表
user_titles: 用户称号/徽章列表
golden_quotes: 金句列表
include_header: 是否包含报告头部
include_footer: 是否包含报告尾部
Returns:
Formatted report string
返回:
格式化的报告字符串
"""
sections = []
@@ -78,26 +78,112 @@ class ReportGenerator:
return "\n\n".join(sections)
def _generate_header(self) -> str:
"""Generate report header."""
title = f"📊 Group Analysis Report"
"""生成报告头部。"""
title = f"📊 群聊分析报告"
if self.group_name:
title += f" - {self.group_name}"
return f"{title}\n📅 Date: {self.date_str}\n{'=' * 40}"
return f"{title}\n📅 日期: {self.date_str}\n{'=' * 40}"
def _generate_statistics_section(self, stats: GroupStatistics) -> str:
"""Generate statistics section."""
"""生成统计部分。"""
lines = [
"📈 **Statistics Overview**",
f"• Total Messages: {stats.message_count}",
f"• Total Characters: {stats.total_characters}",
f"• Participants: {stats.participant_count}",
f"• Average Message Length: {stats.average_message_length:.1f} chars",
f"• Most Active Period: {stats.most_active_period}",
"📈 **统计概览**",
f"• 消息总数: {stats.message_count}",
f"• 字符总数: {stats.total_characters}",
f"• 参与人数: {stats.participant_count}",
f"• 平均消息长度: {stats.average_message_length:.1f} 字符",
f"• 最活跃时段: {stats.most_active_period}",
]
if stats.emoji_count > 0:
lines.append(f"• Emoji Used: {stats.emoji_count}")
lines.append(f"• 表情使用: {stats.emoji_count}")
return "\n".join(lines)
def _generate_topics_section(self, topics: List[Topic]) -> str:
"""生成话题部分。"""
lines = ["💬 **讨论话题**"]
for i, topic in enumerate(topics, 1):
contributors_str = ", ".join(topic.contributors[:3])
if len(topic.contributors) > 3:
contributors_str += f" 等{len(topic.contributors) - 3}人"
lines.append(f"\n{i}. **{topic.name}**")
lines.append(f" 参与者: {contributors_str}")
if topic.detail:
# 截断过长的详情
detail = topic.detail[:200] + "..." if len(topic.detail) > 200 else topic.detail
lines.append(f" {detail}")
return "\n".join(lines)
def _generate_user_titles_section(self, titles: List[UserTitle]) -> str:
"""生成用户称号部分。"""
lines = ["🏆 **用户称号与徽章**"]
for title in titles:
lines.append(f"\n👤 **{title.name}**")
lines.append(f" 🎖️ 称号: {title.title}")
if title.mbti:
lines.append(f" 🧠 MBTI: {title.mbti}")
if title.reason:
reason = title.reason[:150] + "..." if len(title.reason) > 150 else title.reason
lines.append(f" 💡 原因: {reason}")
return "\n".join(lines)
def _generate_golden_quotes_section(self, quotes: List[GoldenQuote]) -> str:
"""生成金句部分。"""
lines = ["✨ **金句集锦**"]
for i, quote in enumerate(quotes, 1):
lines.append(f'\n{i}. "{quote.content}"')
lines.append(f" — {quote.sender}")
if quote.reason:
reason = quote.reason[:100] + "..." if len(quote.reason) > 100 else quote.reason
lines.append(f" ({reason})")
return "\n".join(lines)
def _generate_footer(self, token_usage: Optional[TokenUsage] = None) -> str:
"""生成报告尾部。"""
lines = ["─" * 40]
lines.append(f"生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
if token_usage and token_usage.total_tokens > 0:
lines.append(f"令牌使用: {token_usage.total_tokens} tokens")
return "\n".join(lines)
def generate_summary_report(
self,
statistics: GroupStatistics,
top_topic: Optional[Topic] = None,
top_quote: Optional[GoldenQuote] = None,
) -> str:
"""
生成简要摘要报告。
参数:
statistics: 群聊统计
top_topic: 最重要的话题（可选）
top_quote: 最佳金句（可选）
返回:
简要摘要字符串
"""
lines = [
f"📊 每日摘要 ({self.date_str})",
f"消息: {statistics.message_count} | 参与: {statistics.participant_count}人",
]
if top_topic:
lines.append(f"🔥 热门话题: {top_topic.name}")
if top_quote:
lines.append(f'✨ 金句: "{top_quote.content}" — {top_quote.sender}')
return "\n".join(lines)
src/domain/services/statistics_calculator.py
+52 -52
View File
@@ -1,8 +1,8 @@
"""
Statistics Calculator - Domain service for computing chat statistics
统计计算器 - 计算聊天统计的领域服务
This service calculates various statistics from unified messages.
It is platform-agnostic and works with the domain value objects.
该服务从统一消息计算各种统计数据。
它是平台无关的，与领域值对象配合使用。
"""
from datetime import datetime
@@ -20,18 +20,18 @@ from ..value_objects.statistics import (
class StatisticsCalculator:
"""
Domain service for calculating group chat statistics.
计算群聊统计的领域服务。
This service processes UnifiedMessage objects and produces
platform-agnostic statistics.
该服务处理 UnifiedMessage 对象并生成
平台无关的统计数据。
"""
def __init__(self, bot_user_ids: Optional[List[str]] = None):
"""
Initialize the statistics calculator.
初始化统计计算器。
Args:
bot_user_ids: List of bot user IDs to filter out from statistics
参数:
bot_user_ids: 要从统计中过滤的机器人用户 ID 列表
"""
self.bot_user_ids = set(bot_user_ids or [])
@@ -41,19 +41,19 @@ class StatisticsCalculator:
token_usage: Optional[TokenUsage] = None,
) -> GroupStatistics:
"""
Calculate comprehensive group statistics from messages.
从消息计算综合群组统计。
Args:
messages: List of unified messages to analyze
token_usage: Optional token usage from LLM analysis
参数:
messages: 要分析的统一消息列表
token_usage: LLM 分析的可选令牌使用量
Returns:
GroupStatistics object with computed statistics
返回:
包含计算统计的 GroupStatistics 对象
"""
if not messages:
return GroupStatistics()
# Filter out bot messages
# 过滤机器人消息
filtered_messages = [
msg for msg in messages if msg.sender_id not in self.bot_user_ids
]
@@ -61,19 +61,19 @@ class StatisticsCalculator:
if not filtered_messages:
return GroupStatistics()
# Calculate basic statistics
# 计算基本统计
message_count = len(filtered_messages)
total_characters = sum(len(msg.text_content) for msg in filtered_messages)
unique_senders = set(msg.sender_id for msg in filtered_messages)
participant_count = len(unique_senders)
# Calculate emoji statistics
# 计算表情统计
emoji_stats = self._calculate_emoji_statistics(filtered_messages)
# Calculate activity visualization
# 计算活动可视化
activity_viz = self._calculate_activity_visualization(filtered_messages)
# Determine most active period
# 确定最活跃时段
most_active_period = self._determine_most_active_period(activity_viz)
return GroupStatistics(
@@ -90,18 +90,18 @@ class StatisticsCalculator:
self, messages: List[UnifiedMessage]
) -> Dict[str, UserStatistics]:
"""
Calculate per-user statistics from messages.
从消息计算单用户统计。
Args:
messages: List of unified messages to analyze
参数:
messages: 要分析的统一消息列表
Returns:
Dictionary mapping user_id to UserStatistics
返回:
user_id 到 UserStatistics 的映射字典
"""
user_stats: Dict[str, UserStatistics] = {}
for msg in messages:
# Skip bot messages
# 跳过机器人消息
if msg.sender_id in self.bot_user_ids:
continue
@@ -118,11 +118,11 @@ class StatisticsCalculator:
stats.char_count += len(msg.text_content)
stats.emoji_count += msg.emoji_count
# Count replies
# 计算回复数
if msg.reply_to_id:
stats.reply_count += 1
# Track hourly activity
# 跟踪每小时活动
hour = msg.timestamp.hour
stats.hours[hour] = stats.hours.get(hour, 0) + 1
@@ -135,15 +135,15 @@ class StatisticsCalculator:
min_messages: int = 5,
) -> List[Dict]:
"""
Get top users by message count.
按消息数获取活跃用户排行。
Args:
user_stats: Dictionary of user statistics
limit: Maximum number of users to return
min_messages: Minimum messages required to be included
参数:
user_stats: 用户统计字典
limit: 返回的最大用户数
min_messages: 被包含所需的最少消息数
Returns:
List of top user dictionaries sorted by message count
返回:
按消息数排序的活跃用户字典列表
"""
eligible_users = [
stats for stats in user_stats.values() if stats.message_count >= min_messages
@@ -157,7 +157,7 @@ class StatisticsCalculator:
{
"user_id": u.user_id,
"nickname": u.nickname,
"name": u.nickname, # Backward compatibility
"name": u.nickname, # 向后兼容
"message_count": u.message_count,
"avg_chars": round(u.average_chars, 1),
"emoji_ratio": round(u.emoji_ratio, 2),
@@ -170,7 +170,7 @@ class StatisticsCalculator:
def _calculate_emoji_statistics(
self, messages: List[UnifiedMessage]
) -> EmojiStatistics:
"""Calculate emoji usage statistics from messages."""
"""从消息计算表情使用统计。"""
standard_count = 0
custom_count = 0
animated_count = 0
@@ -208,28 +208,28 @@ class StatisticsCalculator:
def _calculate_activity_visualization(
self, messages: List[UnifiedMessage]
) -> ActivityVisualization:
"""Calculate activity visualization data from messages."""
"""从消息计算活动可视化数据。"""
hourly: Dict[int, int] = {h: 0 for h in range(24)}
daily: Dict[str, int] = {}
user_counts: Dict[str, int] = {}
for msg in messages:
# Hourly activity
# 每小时活动
hour = msg.timestamp.hour
hourly[hour] += 1
# Daily activity
# 每日活动
date_str = msg.timestamp.strftime("%Y-%m-%d")
daily[date_str] = daily.get(date_str, 0) + 1
# User activity
# 用户活动
user_counts[msg.sender_id] = user_counts.get(msg.sender_id, 0) + 1
# Calculate peak hours (top 3)
# 计算高峰时段（前 3 名）
sorted_hours = sorted(hourly.items(), key=lambda x: x[1], reverse=True)
peak_hours = [h for h, _ in sorted_hours[:3]]
# User activity ranking
# 用户活跃度排名
sorted_users = sorted(user_counts.items(), key=lambda x: x[1], reverse=True)
user_ranking = [
{"user_id": uid, "count": count} for uid, count in sorted_users[:20]
@@ -240,27 +240,27 @@ class StatisticsCalculator:
daily_activity=tuple(daily.items()),
user_activity_ranking=tuple(user_ranking),
peak_hours=tuple(peak_hours),
heatmap_data=tuple(), # Can be extended for heatmap visualization
heatmap_data=tuple(), # 可扩展用于热力图可视化
)
def _determine_most_active_period(
self, activity: ActivityVisualization
) -> str:
"""Determine the most active time period description."""
"""确定最活跃时间段描述。"""
hourly = dict(activity.hourly_activity)
if not hourly:
return "Unknown"
return "未知"
# Find peak hour
# 找到高峰时段
peak_hour = max(hourly, key=hourly.get)
# Categorize time periods
# 分类时间段
if 6 <= peak_hour < 12:
return "Morning (6:00-12:00)"
return "上午 (6:00-12:00)"
elif 12 <= peak_hour < 18:
return "Afternoon (12:00-18:00)"
return "下午 (12:00-18:00)"
elif 18 <= peak_hour < 24:
return "Evening (18:00-24:00)"
return "晚间 (18:00-24:00)"
else:
return "Late Night (0:00-6:00)"
return "深夜 (0:00-6:00)"
src/domain/value_objects/__init__.py
+4 -4
View File
@@ -1,4 +1,4 @@
# Value Objects
# 值对象
from .unified_message import UnifiedMessage, MessageContent, MessageContentType
from .platform_capabilities import PlatformCapabilities, PLATFORM_CAPABILITIES
from .unified_group import UnifiedGroup, UnifiedMember
@@ -14,7 +14,7 @@ from .statistics import (
)
__all__ = [
# Core platform abstractions
# 核心平台抽象
"UnifiedMessage",
"MessageContent",
"MessageContentType",
@@ -22,14 +22,14 @@ __all__ = [
"PLATFORM_CAPABILITIES",
"UnifiedGroup",
"UnifiedMember",
# Analysis value objects
# 分析值对象
"Topic",
"TopicCollection",
"UserTitle",
"UserTitleCollection",
"GoldenQuote",
"GoldenQuoteCollection",
# Statistics
# 统计
"TokenUsage",
"EmojiStatistics",
"ActivityVisualization",
src/domain/value_objects/golden_quote.py
+36 -36
View File
@@ -1,8 +1,8 @@
"""
GoldenQuote Value Object - Platform-agnostic golden quote representation
金句值对象 - 平台无关的金句表示
This value object represents a memorable quote extracted from group chat messages.
It is immutable and contains no platform-specific logic.
该值对象表示从群聊消息中提取的精彩语录。
它是不可变的，不包含任何平台特定的逻辑。
"""
from dataclasses import dataclass, field
@@ -12,16 +12,16 @@ from typing import List
@dataclass(frozen=True)
class GoldenQuote:
"""
GoldenQuote value object for group chat analysis.
群聊分析的金句值对象。
Represents a memorable/interesting quote from the chat.
Immutable by design (frozen=True).
表示聊天中令人难忘/有趣的语录。
设计上不可变 (frozen=True)。
Attributes:
content: The actual quote content
sender: Display name of the person who said it
reason: Why this quote was selected as golden
user_id: Platform-agnostic user identifier (stored as string)
属性:
content: 实际的语录内容
sender: 发言者的显示名称
reason: 该语录被选为金句的原因
user_id: 平台无关的用户标识符（存储为字符串）
"""
content: str
@@ -30,23 +30,23 @@ class GoldenQuote:
user_id: str = ""
def __post_init__(self):
"""Validate and normalize golden quote data after initialization."""
# Ensure user_id is always a string
"""初始化后验证和规范化金句数据。"""
# 确保 user_id 始终是字符串
if not isinstance(self.user_id, str):
object.__setattr__(self, "user_id", str(self.user_id))
@classmethod
def from_dict(cls, data: dict) -> "GoldenQuote":
"""
Create GoldenQuote from dictionary data.
从字典数据创建 GoldenQuote。
Args:
data: Dictionary with golden quote data
参数:
data: 包含金句数据的字典
Returns:
GoldenQuote instance
返回:
GoldenQuote 实例
"""
# Handle both 'qq' and 'user_id' keys for backward compatibility
# 同时处理 'qq' 和 'user_id' 键以保持向后兼容
user_id = data.get("user_id", data.get("qq", ""))
return cls(
@@ -58,29 +58,29 @@ class GoldenQuote:
def to_dict(self) -> dict:
"""
Convert GoldenQuote to dictionary.
将 GoldenQuote 转换为字典。
Returns:
Dictionary representation
返回:
字典表示
"""
return {
"content": self.content,
"sender": self.sender,
"reason": self.reason,
"user_id": self.user_id,
"qq": int(self.user_id) if self.user_id.isdigit() else 0, # Backward compat
"qq": int(self.user_id) if self.user_id.isdigit() else 0, # 向后兼容
}
@property
def is_valid(self) -> bool:
"""Check if golden quote has valid data."""
"""检查金句是否有有效数据。"""
return bool(
self.content and self.content.strip() and self.sender and self.sender.strip()
)
@property
def qq(self) -> int:
"""Get QQ number for backward compatibility."""
"""获取 QQ 号码以保持向后兼容。"""
try:
return int(self.user_id)
except (ValueError, TypeError):
@@ -88,15 +88,15 @@ class GoldenQuote:
def with_user_id(self, user_id: str) -> "GoldenQuote":
"""
Create a new GoldenQuote with updated user_id.
创建一个更新了 user_id 的新 GoldenQuote。
Since GoldenQuote is frozen, we need to create a new instance.
由于 GoldenQuote 是冻结的，需要创建新实例。
Args:
user_id: The user ID to set
参数:
user_id: 要设置的用户 ID
Returns:
New GoldenQuote instance with updated user_id
返回:
更新了 user_id 的新 GoldenQuote 实例
"""
return GoldenQuote(
content=self.content,
@@ -109,25 +109,25 @@ class GoldenQuote:
@dataclass
class GoldenQuoteCollection:
"""
Collection of golden quotes with utility methods.
带有实用方法的金句集合。
This is mutable to allow building up a collection of quotes.
这是可变的，以便逐步构建语录集合。
"""
quotes: List[GoldenQuote] = field(default_factory=list)
def add(self, quote: GoldenQuote) -> None:
"""Add a golden quote to the collection."""
"""添加金句到集合。"""
if quote.is_valid:
self.quotes.append(quote)
def add_from_dict(self, data: dict) -> None:
"""Add a golden quote from dictionary data."""
"""从字典数据添加金句。"""
quote = GoldenQuote.from_dict(data)
self.add(quote)
def to_list(self) -> List[dict]:
"""Convert all quotes to list of dictionaries."""
"""将所有语录转换为字典列表。"""
return [q.to_dict() for q in self.quotes]
def __len__(self) -> int:
src/domain/value_objects/platform_capabilities.py
+21 -21
View File
@@ -1,8 +1,8 @@
"""
Platform Capabilities Value Object - Runtime decision support
平台能力值对象 - 运行时决策支持
Each platform adapter declares its capabilities,
application layer decides operations based on capabilities.
每个平台适配器声明其能力，
应用层根据能力决定操作。
"""
from dataclasses import dataclass
@@ -12,30 +12,30 @@ from typing import Optional
@dataclass(frozen=True)
class PlatformCapabilities:
"""
Platform capability description
平台能力描述
Design principles:
1. All fields have default values (most conservative assumption)
2. Immutable
3. Provide convenient check methods
设计原则：
1. 所有字段都有默认值（最保守假设）
2. 不可变
3. 提供便捷的检查方法
"""
# Platform identification
# 平台标识
platform_name: str
platform_version: str = "unknown"
# Message retrieval capabilities
# 消息获取能力
supports_message_history: bool = False
max_message_history_days: int = 0
max_message_count: int = 0
supports_message_search: bool = False
# Group info capabilities
# 群组信息能力
supports_group_list: bool = False
supports_group_info: bool = False
supports_member_list: bool = False
supports_member_info: bool = False
# Message sending capabilities
# 消息发送能力
supports_text_message: bool = True
supports_image_message: bool = False
supports_file_message: bool = False
@@ -44,20 +44,20 @@ class PlatformCapabilities:
max_text_length: int = 4096
max_image_size_mb: float = 10.0
# Special capabilities
# 特殊能力
supports_at_all: bool = False
supports_recall: bool = False
supports_edit: bool = False
# Avatar capabilities
# 头像能力
supports_user_avatar: bool = True
supports_group_avatar: bool = False
avatar_needs_api_call: bool = False
avatar_sizes: tuple = (100,)
# Check methods
# 检查方法
def can_analyze(self) -> bool:
"""Whether supports group chat analysis (core capability)"""
"""是否支持群聊分析（核心能力）"""
return (
self.supports_message_history
and self.max_message_history_days > 0
@@ -65,7 +65,7 @@ class PlatformCapabilities:
)
def can_send_report(self, format: str = "image") -> bool:
"""Whether can send report"""
"""是否能发送报告"""
if format == "text":
return self.supports_text_message
elif format == "image":
@@ -75,11 +75,11 @@ class PlatformCapabilities:
return False
def get_effective_days(self, requested_days: int) -> int:
"""Get actual available days"""
"""获取实际可用天数"""
return min(requested_days, self.max_message_history_days)
def get_effective_count(self, requested_count: int) -> int:
"""Get actual available message count"""
"""获取实际可用消息数"""
return min(requested_count, self.max_message_count)
@@ -173,7 +173,7 @@ SLACK_CAPABILITIES = PlatformCapabilities(
avatar_sizes=(24, 32, 48, 72, 192, 512, 1024),
)
# Capability lookup table
# 能力查找表
PLATFORM_CAPABILITIES = {
"aiocqhttp": ONEBOT_V11_CAPABILITIES,
"onebot": ONEBOT_V11_CAPABILITIES,
@@ -184,5 +184,5 @@ PLATFORM_CAPABILITIES = {
def get_capabilities(platform_name: str) -> Optional[PlatformCapabilities]:
"""Get capabilities by platform name"""
"""根据平台名称获取能力"""
return PLATFORM_CAPABILITIES.get(platform_name.lower())
src/domain/value_objects/statistics.py
+68 -68
View File
@@ -1,8 +1,8 @@
"""
Statistics Value Objects - Platform-agnostic statistics representations
统计值对象 - 平台无关的统计数据表示
This module contains value objects for various statistics collected during
group chat analysis. All objects are immutable and platform-agnostic.
该模块包含群聊分析期间收集的各种统计数据的值对象。
所有对象都是不可变的和平台无关的。
"""
from dataclasses import dataclass, field
@@ -12,14 +12,14 @@ from typing import Dict, List
@dataclass(frozen=True)
class TokenUsage:
"""
Token usage statistics for LLM API calls.
LLM API 调用的令牌使用统计。
Immutable by design (frozen=True).
设计上不可变 (frozen=True)。
Attributes:
prompt_tokens: Number of tokens in the prompt
completion_tokens: Number of tokens in the completion
total_tokens: Total tokens used
属性:
prompt_tokens: 提示词中的令牌数
completion_tokens: 补全中的令牌数
total_tokens: 使用的总令牌数
"""
prompt_tokens: int = 0
@@ -28,7 +28,7 @@ class TokenUsage:
@classmethod
def from_dict(cls, data: dict) -> "TokenUsage":
"""Create TokenUsage from dictionary."""
"""从字典创建 TokenUsage。"""
return cls(
prompt_tokens=data.get("prompt_tokens", 0),
completion_tokens=data.get("completion_tokens", 0),
@@ -36,7 +36,7 @@ class TokenUsage:
)
def to_dict(self) -> dict:
"""Convert to dictionary."""
"""转换为字典。"""
return {
"prompt_tokens": self.prompt_tokens,
"completion_tokens": self.completion_tokens,
@@ -44,7 +44,7 @@ class TokenUsage:
}
def __add__(self, other: "TokenUsage") -> "TokenUsage":
"""Add two TokenUsage objects together."""
"""将两个 TokenUsage 对象相加。"""
if not isinstance(other, TokenUsage):
return NotImplemented
return TokenUsage(
@@ -57,18 +57,18 @@ class TokenUsage:
@dataclass(frozen=True)
class EmojiStatistics:
"""
Emoji usage statistics.
表情使用统计。
Platform-agnostic representation of emoji usage in messages.
Immutable by design (frozen=True).
消息中表情使用的平台无关表示。
设计上不可变 (frozen=True)。
Attributes:
standard_emoji_count: Standard unicode emoji count
custom_emoji_count: Platform-specific custom emoji count
animated_emoji_count: Animated emoji count
sticker_count: Sticker count
other_emoji_count: Other emoji types count
emoji_details: Detailed breakdown by emoji ID/name
属性:
standard_emoji_count: 标准 Unicode 表情数量
custom_emoji_count: 平台特定自定义表情数量
animated_emoji_count: 动态表情数量
sticker_count: 贴纸数量
other_emoji_count: 其他表情类型数量
emoji_details: 按表情 ID/名称的详细分类
"""
standard_emoji_count: int = 0
@@ -80,7 +80,7 @@ class EmojiStatistics:
@property
def total_count(self) -> int:
"""Get total emoji count."""
"""获取表情总数。"""
return (
self.standard_emoji_count
+ self.custom_emoji_count
@@ -91,7 +91,7 @@ class EmojiStatistics:
@classmethod
def from_dict(cls, data: dict) -> "EmojiStatistics":
"""Create EmojiStatistics from dictionary."""
"""从字典创建 EmojiStatistics。"""
details = data.get("face_details", data.get("emoji_details", {}))
if isinstance(details, dict):
details = tuple(details.items())
@@ -106,7 +106,7 @@ class EmojiStatistics:
)
def to_dict(self) -> dict:
"""Convert to dictionary."""
"""转换为字典。"""
return {
"standard_emoji_count": self.standard_emoji_count,
"custom_emoji_count": self.custom_emoji_count,
@@ -115,7 +115,7 @@ class EmojiStatistics:
"other_emoji_count": self.other_emoji_count,
"total_emoji_count": self.total_count,
"emoji_details": dict(self.emoji_details),
# Backward compatibility
# 向后兼容
"face_count": self.standard_emoji_count,
"mface_count": self.custom_emoji_count,
"bface_count": self.animated_emoji_count,
@@ -126,17 +126,17 @@ class EmojiStatistics:
@dataclass(frozen=True)
class ActivityVisualization:
"""
Activity visualization data.
活动可视化数据。
Platform-agnostic representation of chat activity patterns.
Immutable by design (frozen=True).
聊天活动模式的平台无关表示。
设计上不可变 (frozen=True)。
Attributes:
hourly_activity: Message count by hour (0-23)
daily_activity: Message count by date
user_activity_ranking: Ranked list of user activity
peak_hours: List of peak activity hours
heatmap_data: Data for activity heatmap visualization
属性:
hourly_activity: 按小时统计的消息数 (0-23)
daily_activity: 按日期统计的消息数
user_activity_ranking: 用户活跃度排名列表
peak_hours: 活动高峰时段列表
heatmap_data: 活动热力图可视化数据
"""
hourly_activity: tuple = field(default_factory=tuple)
@@ -147,7 +147,7 @@ class ActivityVisualization:
@classmethod
def from_dict(cls, data: dict) -> "ActivityVisualization":
"""Create ActivityVisualization from dictionary."""
"""从字典创建 ActivityVisualization。"""
hourly = data.get("hourly_activity", {})
daily = data.get("daily_activity", {})
ranking = data.get("user_activity_ranking", [])
@@ -163,7 +163,7 @@ class ActivityVisualization:
)
def to_dict(self) -> dict:
"""Convert to dictionary."""
"""转换为字典。"""
return {
"hourly_activity": dict(self.hourly_activity),
"daily_activity": dict(self.daily_activity),
@@ -176,19 +176,19 @@ class ActivityVisualization:
@dataclass(frozen=True)
class GroupStatistics:
"""
Comprehensive group chat statistics.
综合群聊统计。
Platform-agnostic representation of group chat statistics.
Immutable by design (frozen=True).
群聊统计数据的平台无关表示。
设计上不可变 (frozen=True)。
Attributes:
message_count: Total number of messages
total_characters: Total character count across all messages
participant_count: Number of unique participants
most_active_period: Description of the most active time period
emoji_statistics: Emoji usage statistics
activity_visualization: Activity pattern data
token_usage: LLM token usage for analysis
属性:
message_count: 消息总数
total_characters: 所有消息的总字符数
participant_count: 唯一参与者数量
most_active_period: 最活跃时间段描述
emoji_statistics: 表情使用统计
activity_visualization: 活动模式数据
token_usage: 分析使用的 LLM 令牌
"""
message_count: int = 0
@@ -201,22 +201,22 @@ class GroupStatistics:
@property
def average_message_length(self) -> float:
"""Calculate average message length."""
"""计算平均消息长度。"""
if self.message_count == 0:
return 0.0
return self.total_characters / self.message_count
@property
def emoji_count(self) -> int:
"""Get total emoji count for backward compatibility."""
"""获取表情总数以保持向后兼容。"""
return self.emoji_statistics.total_count
@classmethod
def from_dict(cls, data: dict) -> "GroupStatistics":
"""Create GroupStatistics from dictionary."""
"""从字典创建 GroupStatistics。"""
emoji_data = data.get("emoji_statistics", {})
if not emoji_data:
# Backward compatibility: construct from flat fields
# 向后兼容：从扁平字段构建
emoji_data = {
"face_count": data.get("emoji_count", 0),
}
@@ -235,13 +235,13 @@ class GroupStatistics:
)
def to_dict(self) -> dict:
"""Convert to dictionary."""
"""转换为字典。"""
return {
"message_count": self.message_count,
"total_characters": self.total_characters,
"participant_count": self.participant_count,
"most_active_period": self.most_active_period,
"emoji_count": self.emoji_count, # Backward compatibility
"emoji_count": self.emoji_count, # 向后兼容
"emoji_statistics": self.emoji_statistics.to_dict(),
"activity_visualization": self.activity_visualization.to_dict(),
"token_usage": self.token_usage.to_dict(),
@@ -251,16 +251,16 @@ class GroupStatistics:
@dataclass
class UserStatistics:
"""
Per-user statistics (mutable for accumulation during analysis).
单用户统计（可变以便在分析期间累积）。
Attributes:
user_id: Platform-agnostic user identifier
nickname: User's display name
message_count: Number of messages sent
char_count: Total characters sent
emoji_count: Number of emojis used
reply_count: Number of replies made
hours: Message count by hour (0-23)
属性:
user_id: 平台无关的用户标识符
nickname: 用户显示名称
message_count: 发送的消息数
char_count: 发送的总字符数
emoji_count: 使用的表情数
reply_count: 回复次数
hours: 按小时统计的消息数 (0-23)
"""
user_id: str
@@ -273,21 +273,21 @@ class UserStatistics:
@property
def average_chars(self) -> float:
"""Calculate average characters per message."""
"""计算每条消息的平均字符数。"""
if self.message_count == 0:
return 0.0
return self.char_count / self.message_count
@property
def emoji_ratio(self) -> float:
"""Calculate emoji per message ratio."""
"""计算每条消息的表情比率。"""
if self.message_count == 0:
return 0.0
return self.emoji_count / self.message_count
@property
def night_ratio(self) -> float:
"""Calculate night activity ratio (0-6 hours)."""
"""计算夜间活动比率 (0-6 点)。"""
if self.message_count == 0:
return 0.0
night_messages = sum(self.hours.get(h, 0) for h in range(6))
@@ -295,13 +295,13 @@ class UserStatistics:
@property
def reply_ratio(self) -> float:
"""Calculate reply ratio."""
"""计算回复比率。"""
if self.message_count == 0:
return 0.0
return self.reply_count / self.message_count
def to_dict(self) -> dict:
"""Convert to dictionary."""
"""转换为字典。"""
return {
"user_id": self.user_id,
"nickname": self.nickname,
src/domain/value_objects/topic.py
+28 -28
View File
@@ -1,8 +1,8 @@
"""
Topic Value Object - Platform-agnostic topic representation
话题值对象 - 平台无关的话题表示
This value object represents a discussion topic extracted from group chat messages.
It is immutable and contains no platform-specific logic.
该值对象表示从群聊消息中提取的讨论话题。
它是不可变的，不包含任何平台特定的逻辑。
"""
from dataclasses import dataclass, field
@@ -12,15 +12,15 @@ from typing import List
@dataclass(frozen=True)
class Topic:
"""
Topic value object for group chat analysis.
群聊分析的话题值对象。
Represents a discussion topic with contributors and details.
Immutable by design (frozen=True).
表示一个包含参与者和详情的讨论话题。
设计上不可变 (frozen=True)。
Attributes:
name: Topic title/name
contributors: List of usernames who participated in this topic
detail: Detailed description or summary of the topic discussion
属性:
name: 话题标题/名称
contributors: 参与该话题讨论的用户名列表
detail: 话题讨论的详细描述或摘要
"""
name: str
@@ -28,24 +28,24 @@ class Topic:
detail: str = ""
def __post_init__(self):
"""Validate topic data after initialization."""
"""初始化后验证话题数据。"""
if not self.name or not self.name.strip():
object.__setattr__(self, "name", "Unknown Topic")
object.__setattr__(self, "name", "未知话题")
# Ensure contributors is a tuple for immutability
# 确保 contributors 是元组以保证不可变性
if isinstance(self.contributors, list):
object.__setattr__(self, "contributors", tuple(self.contributors))
@classmethod
def from_dict(cls, data: dict) -> "Topic":
"""
Create Topic from dictionary data.
从字典数据创建 Topic。
Args:
data: Dictionary with topic data
参数:
data: 包含话题数据的字典
Returns:
Topic instance
返回:
Topic 实例
"""
contributors = data.get("contributors", [])
if isinstance(contributors, list):
@@ -59,10 +59,10 @@ class Topic:
def to_dict(self) -> dict:
"""
Convert Topic to dictionary.
将 Topic 转换为字典。
Returns:
Dictionary representation
返回:
字典表示
"""
return {
"topic": self.name,
@@ -72,37 +72,37 @@ class Topic:
@property
def contributor_count(self) -> int:
"""Get the number of contributors."""
"""获取参与者数量。"""
return len(self.contributors)
@property
def is_valid(self) -> bool:
"""Check if topic has valid data."""
"""检查话题是否有有效数据。"""
return bool(self.name and self.name.strip() and self.detail and self.detail.strip())
@dataclass
class TopicCollection:
"""
Collection of topics with utility methods.
带有实用方法的话题集合。
This is mutable to allow building up a collection of topics.
这是可变的，以便逐步构建话题集合。
"""
topics: List[Topic] = field(default_factory=list)
def add(self, topic: Topic) -> None:
"""Add a topic to the collection."""
"""添加话题到集合。"""
if topic.is_valid:
self.topics.append(topic)
def add_from_dict(self, data: dict) -> None:
"""Add a topic from dictionary data."""
"""从字典数据添加话题。"""
topic = Topic.from_dict(data)
self.add(topic)
def to_list(self) -> List[dict]:
"""Convert all topics to list of dictionaries."""
"""将所有话题转换为字典列表。"""
return [t.to_dict() for t in self.topics]
def __len__(self) -> int:
src/domain/value_objects/unified_group.py
+5 -8
View File
@@ -1,5 +1,5 @@
"""
Unified Group Value Objects - Cross-platform group abstraction
统一群组值对象 - 跨平台群组抽象
"""
from dataclasses import dataclass
@@ -8,22 +8,19 @@ from typing import Optional
@dataclass(frozen=True)
class UnifiedMember:
"""Unified member information"""
"""统一成员信息"""
user_id: str
nickname: str
card: Optional[str] = None # Group card
card: Optional[str] = None # 群名片
role: str = "member" # owner, admin, member
join_time: Optional[int] = None
avatar_url: Optional[str] = None
avatar_data: Optional[str] = None # Base64 for template rendering
def get_display_name(self) -> str:
return self.card or self.nickname or self.user_id
avatar_data: Optional[str] = None # Base64 用于模板渲染
@dataclass(frozen=True)
class UnifiedGroup:
"""Unified group information"""
"""统一群组信息"""
group_id: str
group_name: str
member_count: int = 0
src/domain/value_objects/unified_message.py
+27 -27
View File
@@ -1,7 +1,7 @@
"""
Unified Message Value Object - Cross-platform core abstraction
统一消息值对象 - 跨平台核心抽象
All platform messages are converted to this format for analysis.
所有平台消息都转换为此格式进行分析。
"""
from dataclasses import dataclass, field
@@ -11,7 +11,7 @@ from datetime import datetime
class MessageContentType(Enum):
"""Message content type enumeration"""
"""消息内容类型枚举"""
TEXT = "text"
IMAGE = "image"
FILE = "file"
@@ -28,9 +28,9 @@ class MessageContentType(Enum):
@dataclass(frozen=True)
class MessageContent:
"""
Message content segment value object
消息内容段值对象
Immutable, used to compose message chains
不可变，用于组成消息链
"""
type: MessageContentType
text: str = ""
@@ -50,60 +50,60 @@ class MessageContent:
@dataclass(frozen=True)
class UnifiedMessage:
"""
Unified message format - Cross-platform core value object
统一消息格式 - 跨平台核心值对象
Design principles:
1. Only keep fields needed for analysis
2. Use platform-agnostic types
3. Immutable (frozen=True) - thread-safe
4. All IDs are strings - avoid platform differences
设计原则：
1. 只保留分析所需的字段
2. 使用平台无关的类型
3. 不可变 (frozen=True) - 线程安全
4. 所有 ID 使用字符串 - 避免平台差异
"""
# Basic identification
# 基础标识
message_id: str
sender_id: str
sender_name: str
group_id: str
# Message content
text_content: str # Extracted plain text for LLM analysis
# 消息内容
text_content: str # 提取的纯文本用于 LLM 分析
contents: Tuple[MessageContent, ...] = field(default_factory=tuple)
# Time information
timestamp: int = 0 # Unix timestamp
# 时间信息
timestamp: int = 0 # Unix 时间戳
# Platform information
# 平台信息
platform: str = "unknown"
# Optional information
# 可选信息
reply_to_id: Optional[str] = None
sender_card: Optional[str] = None # Group card/nickname
sender_card: Optional[str] = None # 群名片/昵称
# Analysis helper methods
# 分析辅助方法
def has_text(self) -> bool:
"""Whether has text content"""
"""是否有文本内容"""
return bool(self.text_content.strip())
def get_display_name(self) -> str:
"""Get display name, prefer group card"""
"""获取显示名称，优先使用群名片"""
return self.sender_card or self.sender_name or self.sender_id
def get_emoji_count(self) -> int:
"""Get emoji count"""
"""获取表情数量"""
return sum(1 for c in self.contents if c.is_emoji())
def get_text_length(self) -> int:
"""Get text length"""
"""获取文本长度"""
return len(self.text_content)
def get_datetime(self) -> datetime:
"""Get message datetime"""
"""获取消息时间"""
return datetime.fromtimestamp(self.timestamp)
def to_analysis_format(self) -> str:
"""Convert to analysis format (for LLM)"""
"""转换为分析格式（供 LLM 使用）"""
name = self.get_display_name()
return f"[{name}]: {self.text_content}"
# Type alias
# 类型别名
MessageList = list[UnifiedMessage]
src/domain/value_objects/user_title.py
+32 -32
View File
@@ -1,8 +1,8 @@
"""
UserTitle Value Object - Platform-agnostic user title representation
用户称号值对象 - 平台无关的用户称号表示
This value object represents a user's title/badge assigned based on their
chat behavior analysis. It is immutable and contains no platform-specific logic.
该值对象表示基于聊天行为分析分配给用户的称号/徽章。
它是不可变的，不包含任何平台特定的逻辑。
"""
from dataclasses import dataclass, field
@@ -12,17 +12,17 @@ from typing import List
@dataclass(frozen=True)
class UserTitle:
"""
UserTitle value object for group chat analysis.
群聊分析的用户称号值对象。
Represents a title/badge assigned to a user based on their behavior.
Immutable by design (frozen=True).
表示基于用户行为分配的称号/徽章。
设计上不可变 (frozen=True)。
Attributes:
name: User's display name
user_id: Platform-agnostic user identifier (stored as string)
title: The title/badge assigned to the user
mbti: MBTI personality type assessment
reason: Explanation for why this title was assigned
属性:
name: 用户显示名称
user_id: 平台无关的用户标识符（存储为字符串）
title: 分配给用户的称号/徽章
mbti: MBTI 人格类型评估
reason: 分配该称号的原因说明
"""
name: str
@@ -32,23 +32,23 @@ class UserTitle:
reason: str = ""
def __post_init__(self):
"""Validate and normalize user title data after initialization."""
# Ensure user_id is always a string
"""初始化后验证和规范化用户称号数据。"""
# 确保 user_id 始终是字符串
if not isinstance(self.user_id, str):
object.__setattr__(self, "user_id", str(self.user_id))
@classmethod
def from_dict(cls, data: dict) -> "UserTitle":
"""
Create UserTitle from dictionary data.
从字典数据创建 UserTitle。
Args:
data: Dictionary with user title data
参数:
data: 包含用户称号数据的字典
Returns:
UserTitle instance
返回:
UserTitle 实例
"""
# Handle both 'qq' and 'user_id' keys for backward compatibility
# 同时处理 'qq' 和 'user_id' 键以保持向后兼容
user_id = data.get("user_id", data.get("qq", ""))
return cls(
@@ -61,15 +61,15 @@ class UserTitle:
def to_dict(self) -> dict:
"""
Convert UserTitle to dictionary.
将 UserTitle 转换为字典。
Returns:
Dictionary representation
返回:
字典表示
"""
return {
"name": self.name,
"user_id": self.user_id,
"qq": int(self.user_id) if self.user_id.isdigit() else 0, # Backward compat
"qq": int(self.user_id) if self.user_id.isdigit() else 0, # 向后兼容
"title": self.title,
"mbti": self.mbti,
"reason": self.reason,
@@ -77,7 +77,7 @@ class UserTitle:
@property
def is_valid(self) -> bool:
"""Check if user title has valid data."""
"""检查用户称号是否有有效数据。"""
return bool(
self.name
and self.name.strip()
@@ -88,7 +88,7 @@ class UserTitle:
@property
def qq(self) -> int:
"""Get QQ number for backward compatibility."""
"""获取 QQ 号码以保持向后兼容。"""
try:
return int(self.user_id)
except (ValueError, TypeError):
@@ -98,25 +98,25 @@ class UserTitle:
@dataclass
class UserTitleCollection:
"""
Collection of user titles with utility methods.
带有实用方法的用户称号集合。
This is mutable to allow building up a collection of titles.
这是可变的，以便逐步构建称号集合。
"""
titles: List[UserTitle] = field(default_factory=list)
def add(self, title: UserTitle) -> None:
"""Add a user title to the collection."""
"""添加用户称号到集合。"""
if title.is_valid:
self.titles.append(title)
def add_from_dict(self, data: dict) -> None:
"""Add a user title from dictionary data."""
"""从字典数据添加用户称号。"""
title = UserTitle.from_dict(data)
self.add(title)
def get_by_user_id(self, user_id: str) -> UserTitle | None:
"""Get title by user ID."""
"""根据用户 ID 获取称号。"""
user_id_str = str(user_id)
for title in self.titles:
if title.user_id == user_id_str:
@@ -124,7 +124,7 @@ class UserTitleCollection:
return None
def to_list(self) -> List[dict]:
"""Convert all titles to list of dictionaries."""
"""将所有称号转换为字典列表。"""
return [t.to_dict() for t in self.titles]
def __len__(self) -> int:
src/infrastructure/platform/__init__.py
+1 -1
View File
@@ -1,4 +1,4 @@
# Platform Adapters
# 平台适配器
from .factory import PlatformAdapterFactory
from .base import PlatformAdapter
from .adapters.onebot_adapter import OneBotAdapter
src/infrastructure/platform/base.py
+6 -6
View File
@@ -1,5 +1,5 @@
"""
Platform Adapter Base Class
平台适配器基类
"""
from abc import ABC, abstractmethod
@@ -24,10 +24,10 @@ class PlatformAdapter(
ABC
):
"""
Platform adapter base class
平台适配器基类
Combines message repository, message sender, group info, and avatar interfaces.
Each platform adapter inherits this class and implements all methods.
组合消息仓储、消息发送、群组信息和头像接口。
每个平台适配器继承此类并实现所有方法。
"""
def __init__(self, bot_instance: Any, config: dict = None):
@@ -37,14 +37,14 @@ class PlatformAdapter(
@property
def capabilities(self) -> PlatformCapabilities:
"""Platform capabilities (lazy initialization)"""
"""平台能力（延迟初始化）"""
if self._capabilities is None:
self._capabilities = self._init_capabilities()
return self._capabilities
@abstractmethod
def _init_capabilities(self) -> PlatformCapabilities:
"""Initialize platform capabilities, subclass must implement"""
"""初始化平台能力，子类必须实现"""
raise NotImplementedError
def get_capabilities(self) -> PlatformCapabilities:
src/infrastructure/platform/factory.py
+15 -15
View File
@@ -1,5 +1,5 @@
"""
Platform Adapter Factory
平台适配器工厂
"""
from typing import Optional, Any, Dict, Type
@@ -9,17 +9,17 @@ from .base import PlatformAdapter
class PlatformAdapterFactory:
"""
Platform adapter factory
平台适配器工厂
Creates adapter instances based on platform name.
Uses registry pattern for easy extension.
根据平台名称创建适配器实例。
使用注册表模式便于扩展。
"""
_adapters: Dict[str, Type[PlatformAdapter]] = {}
@classmethod
def register(cls, platform_name: str, adapter_class: Type[PlatformAdapter]):
"""Register a new adapter"""
"""注册新适配器"""
cls._adapters[platform_name.lower()] = adapter_class
@classmethod
@@ -30,15 +30,15 @@ class PlatformAdapterFactory:
config: dict = None,
) -> Optional[PlatformAdapter]:
"""
Create platform adapter
创建平台适配器
Args:
platform_name: Platform name (e.g., "aiocqhttp", "telegram")
bot_instance: AstrBot bot instance
config: Configuration dict
参数:
platform_name: 平台名称（如 "aiocqhttp"、"telegram"）
bot_instance: AstrBot 机器人实例
config: 配置字典
Returns:
Platform adapter instance, or None if unsupported
返回:
平台适配器实例，如果不支持则返回 None
"""
adapter_class = cls._adapters.get(platform_name.lower())
@@ -52,16 +52,16 @@ class PlatformAdapterFactory:
@classmethod
def get_supported_platforms(cls) -> list[str]:
"""Get all supported platform names"""
"""获取所有支持的平台名称"""
return list(cls._adapters.keys())
@classmethod
def is_supported(cls, platform_name: str) -> bool:
"""Check if platform is supported"""
"""检查平台是否被支持"""
return platform_name.lower() in cls._adapters
# Import adapters to register them
# 导入适配器以注册它们
def _register_adapters():
try:
from .adapters.onebot_adapter import OneBotAdapter
src/infrastructure/resilience/__init__.py
+1 -1
View File
@@ -1,5 +1,5 @@
"""
Resilience Module - Circuit breaker, rate limiter, and retry utilities
弹性模块 - 断路器、速率限制器和重试工具
"""
from .circuit_breaker import CircuitBreaker, CircuitState
src/infrastructure/resilience/circuit_breaker.py
+30 -31
View File
@@ -1,8 +1,7 @@
"""
Circuit Breaker - Prevents cascading failures
断路器 - 防止级联故障
Implements the circuit breaker pattern to prevent repeated calls
to failing services.
实现断路器模式，防止对失败服务的重复调用。
"""
import time
@@ -14,20 +13,20 @@ from astrbot.api import logger
class CircuitState(Enum):
"""Circuit breaker states."""
"""断路器状态。"""
CLOSED = "closed" # Normal operation
OPEN = "open" # Failing, reject calls
HALF_OPEN = "half_open" # Testing if service recovered
CLOSED = "closed" # 正常运行
OPEN = "open" # 故障中，拒绝调用
HALF_OPEN = "half_open" # 测试服务是否恢复
@dataclass
class CircuitBreaker:
"""
Circuit breaker implementation.
断路器实现。
Prevents cascading failures by tracking failure rates and
temporarily blocking calls to failing services.
通过跟踪故障率并临时阻止对故障服务的调用
来防止级联故障。
"""
name: str
@@ -35,7 +34,7 @@ class CircuitBreaker:
recovery_timeout: float = 30.0
half_open_max_calls: int = 3
# Internal state
# 内部状态
_state: CircuitState = field(default=CircuitState.CLOSED, init=False)
_failure_count: int = field(default=0, init=False)
_success_count: int = field(default=0, init=False)
@@ -44,14 +43,14 @@ class CircuitBreaker:
@property
def state(self) -> CircuitState:
"""Get current circuit state, checking for recovery."""
"""获取当前断路器状态，检查是否恢复。"""
if self._state == CircuitState.OPEN:
if time.time() - self._last_failure_time >= self.recovery_timeout:
self._transition_to(CircuitState.HALF_OPEN)
return self._state
def _transition_to(self, new_state: CircuitState) -> None:
"""Transition to a new state."""
"""转换到新状态。"""
old_state = self._state
self._state = new_state
@@ -61,20 +60,20 @@ class CircuitBreaker:
elif new_state == CircuitState.HALF_OPEN:
self._half_open_calls = 0
logger.debug(f"Circuit {self.name}: {old_state.value} -> {new_state.value}")
logger.debug(f"断路器 {self.name}: {old_state.value} -> {new_state.value}")
def record_success(self) -> None:
"""Record a successful call."""
"""记录成功调用。"""
if self._state == CircuitState.HALF_OPEN:
self._success_count += 1
if self._success_count >= self.half_open_max_calls:
self._transition_to(CircuitState.CLOSED)
elif self._state == CircuitState.CLOSED:
# Reset failure count on success
# 成功时重置故障计数
self._failure_count = 0
def record_failure(self) -> None:
"""Record a failed call."""
"""记录失败调用。"""
self._failure_count += 1
self._last_failure_time = time.time()
@@ -85,8 +84,8 @@ class CircuitBreaker:
self._transition_to(CircuitState.OPEN)
def can_execute(self) -> bool:
"""Check if a call can be executed."""
state = self.state # This may trigger state transition
"""检查是否可以执行调用。"""
state = self.state # 这可能触发状态转换
if state == CircuitState.CLOSED:
return True
@@ -99,7 +98,7 @@ class CircuitBreaker:
return False
def reset(self) -> None:
"""Reset the circuit breaker to closed state."""
"""重置断路器到关闭状态。"""
self._transition_to(CircuitState.CLOSED)
async def execute(
@@ -110,24 +109,24 @@ class CircuitBreaker:
**kwargs,
):
"""
Execute a function with circuit breaker protection.
使用断路器保护执行函数。
Args:
func: Async function to execute
*args: Function arguments
fallback: Optional fallback function if circuit is open
**kwargs: Function keyword arguments
参数:
func: 要执行的异步函数
*args: 函数参数
fallback: 断路器打开时的可选降级函数
**kwargs: 函数关键字参数
Returns:
Function result or fallback result
返回:
函数结果或降级结果
Raises:
Exception: If circuit is open and no fallback provided
异常:
Exception: 如果断路器打开且没有提供降级函数
"""
if not self.can_execute():
if fallback:
return await fallback(*args, **kwargs)
raise Exception(f"Circuit {self.name} is open")
raise Exception(f"断路器 {self.name} 已打开")
try:
result = await func(*args, **kwargs)
src/infrastructure/resilience/rate_limiter.py
+32 -32
View File
@@ -1,7 +1,7 @@
"""
Rate Limiter - Controls request rates
速率限制器 - 控制请求速率
Implements token bucket rate limiting to prevent overwhelming services.
实现令牌桶速率限制，防止服务过载。
"""
import asyncio
@@ -15,27 +15,27 @@ from astrbot.api import logger
@dataclass
class RateLimiter:
"""
Token bucket rate limiter.
令牌桶速率限制器。
Controls the rate of operations by using a token bucket algorithm.
使用令牌桶算法控制操作速率。
"""
name: str
rate: float # Tokens per second
burst: int # Maximum burst size (bucket capacity)
rate: float # 每秒令牌数
burst: int # 最大突发大小（桶容量）
# Internal state
# 内部状态
_tokens: float = field(default=0, init=False)
_last_update: float = field(default=0, init=False)
_lock: asyncio.Lock = field(default_factory=asyncio.Lock, init=False)
def __post_init__(self):
"""Initialize the token bucket."""
"""初始化令牌桶。"""
self._tokens = float(self.burst)
self._last_update = time.time()
def _refill(self) -> None:
"""Refill tokens based on elapsed time."""
"""根据经过的时间补充令牌。"""
now = time.time()
elapsed = now - self._last_update
self._tokens = min(self.burst, self._tokens + elapsed * self.rate)
@@ -43,14 +43,14 @@ class RateLimiter:
async def acquire(self, tokens: int = 1, timeout: Optional[float] = None) -> bool:
"""
Acquire tokens from the bucket.
从桶中获取令牌。
Args:
tokens: Number of tokens to acquire
timeout: Maximum time to wait (None = wait forever)
参数:
tokens: 要获取的令牌数
timeout: 最大等待时间（None = 无限等待）
Returns:
True if tokens acquired, False if timeout
返回:
如果获取到令牌返回 True，超时返回 False
"""
start_time = time.time()
@@ -67,7 +67,7 @@ class RateLimiter:
if elapsed >= timeout:
return False
# Calculate wait time for enough tokens
# 计算获取足够令牌的等待时间
tokens_needed = tokens - self._tokens
wait_time = tokens_needed / self.rate
@@ -80,13 +80,13 @@ class RateLimiter:
def try_acquire(self, tokens: int = 1) -> bool:
"""
Try to acquire tokens without waiting.
尝试获取令牌而不等待。
Args:
tokens: Number of tokens to acquire
参数:
tokens: 要获取的令牌数
Returns:
True if tokens acquired, False otherwise
返回:
如果获取到令牌返回 True，否则返回 False
"""
self._refill()
@@ -97,19 +97,19 @@ class RateLimiter:
@property
def available_tokens(self) -> float:
"""Get current available tokens."""
"""获取当前可用令牌数。"""
self._refill()
return self._tokens
def reset(self) -> None:
"""Reset the rate limiter to full capacity."""
"""重置速率限制器到满容量。"""
self._tokens = float(self.burst)
self._last_update = time.time()
class RateLimiterGroup:
"""
Group of rate limiters for different operations.
不同操作的速率限制器组。
"""
def __init__(self):
@@ -122,21 +122,21 @@ class RateLimiterGroup:
burst: int = 5,
) -> RateLimiter:
"""
Get or create a rate limiter.
获取或创建速率限制器。
Args:
name: Limiter name
rate: Tokens per second
burst: Maximum burst size
参数:
name: 限制器名称
rate: 每秒令牌数
burst: 最大突发大小
Returns:
RateLimiter instance
返回:
RateLimiter 实例
"""
if name not in self._limiters:
self._limiters[name] = RateLimiter(name=name, rate=rate, burst=burst)
return self._limiters[name]
def reset_all(self) -> None:
"""Reset all rate limiters."""
"""重置所有速率限制器。"""
for limiter in self._limiters.values():
limiter.reset()
src/infrastructure/resilience/retry.py
+41 -41
View File
@@ -1,7 +1,7 @@
"""
Retry - Retry utilities with exponential backoff
重试 - 带指数退避的重试工具
Provides retry decorators and utilities for handling transient failures.
提供用于处理瞬态故障的重试装饰器和工具。
"""
import asyncio
@@ -15,7 +15,7 @@ from astrbot.api import logger
@dataclass
class RetryConfig:
"""Configuration for retry behavior."""
"""重试行为配置。"""
max_attempts: int = 3
base_delay: float = 1.0
@@ -33,17 +33,17 @@ def calculate_delay(
jitter: bool,
) -> float:
"""
Calculate delay for a retry attempt.
计算重试尝试的延迟。
Args:
attempt: Current attempt number (0-based)
base_delay: Base delay in seconds
max_delay: Maximum delay in seconds
exponential_base: Base for exponential backoff
jitter: Whether to add random jitter
参数:
attempt: 当前尝试次数（从 0 开始）
base_delay: 基础延迟（秒）
max_delay: 最大延迟（秒）
exponential_base: 指数退避的基数
jitter: 是否添加随机抖动
Returns:
Delay in seconds
返回:
延迟时间（秒）
"""
delay = base_delay * (exponential_base**attempt)
delay = min(delay, max_delay)
@@ -64,19 +64,19 @@ def retry_async(
on_retry: Optional[Callable[[Exception, int], None]] = None,
):
"""
Decorator for retrying async functions with exponential backoff.
带指数退避的异步函数重试装饰器。
Args:
max_attempts: Maximum number of attempts
base_delay: Base delay between retries
max_delay: Maximum delay between retries
exponential_base: Base for exponential backoff
jitter: Whether to add random jitter
retry_exceptions: Tuple of exceptions to retry on
on_retry: Optional callback on retry (exception, attempt)
参数:
max_attempts: 最大尝试次数
base_delay: 重试之间的基础延迟
max_delay: 重试之间的最大延迟
exponential_base: 指数退避的基数
jitter: 是否添加随机抖动
retry_exceptions: 要重试的异常元组
on_retry: 重试时的可选回调（异常，尝试次数）
Returns:
Decorated function
返回:
装饰后的函数
"""
def decorator(func: Callable):
@@ -99,13 +99,13 @@ def retry_async(
on_retry(e, attempt + 1)
logger.debug(
f"Retry {attempt + 1}/{max_attempts} for {func.__name__} "
f"after {delay:.2f}s: {e}"
f"重试 {attempt + 1}/{max_attempts} {func.__name__} "
f"延迟 {delay:.2f}s: {e}"
)
await asyncio.sleep(delay)
else:
logger.warning(
f"All {max_attempts} attempts failed for {func.__name__}: {e}"
f"{func.__name__} 的所有 {max_attempts} 次尝试均失败: {e}"
)
raise last_exception
@@ -117,15 +117,15 @@ def retry_async(
class RetryExecutor:
"""
Executor for running functions with retry logic.
带重试逻辑的函数执行器。
"""
def __init__(self, config: Optional[RetryConfig] = None):
"""
Initialize the retry executor.
初始化重试执行器。
Args:
config: Retry configuration
参数:
config: 重试配置
"""
self.config = config or RetryConfig()
@@ -137,19 +137,19 @@ class RetryExecutor:
**kwargs,
):
"""
Execute a function with retry logic.
使用重试逻辑执行函数。
Args:
func: Async function to execute
*args: Function arguments
config: Optional override config
**kwargs: Function keyword arguments
参数:
func: 要执行的异步函数
*args: 函数参数
config: 可选的覆盖配置
**kwargs: 函数关键字参数
Returns:
Function result
返回:
函数结果
Raises:
Exception: If all retries fail
异常:
Exception: 如果所有重试都失败
"""
cfg = config or self.config
last_exception = None
@@ -169,7 +169,7 @@ class RetryExecutor:
cfg.jitter,
)
logger.debug(
f"Retry {attempt + 1}/{cfg.max_attempts} after {delay:.2f}s: {e}"
f"重试 {attempt + 1}/{cfg.max_attempts} 延迟 {delay:.2f}s: {e}"
)
await asyncio.sleep(delay)
src/shared/__init__.py
+2 -2
View File
@@ -1,5 +1,5 @@
"""
Shared Module - Common utilities and constants
共享模块 - 通用工具和常量
"""
from .constants import *
@@ -7,5 +7,5 @@ from .trace_context import TraceContext
__all__ = [
"TraceContext",
# Constants are exported via *
# 常量通过 * 导出
]
src/shared/constants.py
+21 -21
View File
@@ -1,12 +1,12 @@
"""
Constants - Shared constants used across the plugin
常量 - 插件中使用的共享常量
"""
# Plugin metadata
# 插件元数据
PLUGIN_NAME = "astrbot_plugin_qq_group_daily_analysis"
PLUGIN_VERSION = "2.0.0"
# Platform identifiers
# 平台标识符
PLATFORM_ONEBOT = "onebot"
PLATFORM_TELEGRAM = "telegram"
PLATFORM_DISCORD = "discord"
@@ -15,21 +15,21 @@ PLATFORM_LARK = "lark"
SUPPORTED_PLATFORMS = [
PLATFORM_ONEBOT,
# Future platforms
# 未来平台
# PLATFORM_TELEGRAM,
# PLATFORM_DISCORD,
# PLATFORM_SLACK,
# PLATFORM_LARK,
]
# Analysis defaults
# 分析默认值
DEFAULT_MAX_TOPICS = 5
DEFAULT_MAX_USER_TITLES = 10
DEFAULT_MAX_GOLDEN_QUOTES = 5
DEFAULT_MIN_MESSAGES = 50
DEFAULT_MAX_TOKENS = 2000
# Time periods
# 时间段
HOUR_RANGES = {
"morning": (6, 12),
"afternoon": (12, 18),
@@ -37,13 +37,13 @@ HOUR_RANGES = {
"night": (0, 6),
}
# Report formats
# 报告格式
REPORT_FORMAT_TEXT = "text"
REPORT_FORMAT_MARKDOWN = "markdown"
REPORT_FORMAT_IMAGE = "image"
REPORT_FORMAT_HTML = "html"
# Message content types
# 消息内容类型
CONTENT_TYPE_TEXT = "text"
CONTENT_TYPE_IMAGE = "image"
CONTENT_TYPE_EMOJI = "emoji"
@@ -55,37 +55,37 @@ CONTENT_TYPE_REPLY = "reply"
CONTENT_TYPE_AT = "at"
CONTENT_TYPE_UNKNOWN = "unknown"
# Analysis task states
# 分析任务状态
TASK_STATE_PENDING = "pending"
TASK_STATE_RUNNING = "running"
TASK_STATE_COMPLETED = "completed"
TASK_STATE_FAILED = "failed"
TASK_STATE_CANCELLED = "cancelled"
# Error codes
# 错误代码
ERROR_INSUFFICIENT_DATA = "INSUFFICIENT_DATA"
ERROR_LLM_FAILED = "LLM_FAILED"
ERROR_PLATFORM_ERROR = "PLATFORM_ERROR"
ERROR_CONFIG_ERROR = "CONFIG_ERROR"
ERROR_TIMEOUT = "TIMEOUT"
# Cache TTL (in seconds)
CACHE_TTL_SHORT = 60 # 1 minute
CACHE_TTL_MEDIUM = 300 # 5 minutes
CACHE_TTL_LONG = 3600 # 1 hour
CACHE_TTL_DAY = 86400 # 24 hours
# 缓存 TTL（秒）
CACHE_TTL_SHORT = 60 # 1 分钟
CACHE_TTL_MEDIUM = 300 # 5 分钟
CACHE_TTL_LONG = 3600 # 1 小时
CACHE_TTL_DAY = 86400 # 24 小时
# Rate limiting defaults
RATE_LIMIT_LLM_CALLS = 10 # calls per minute
RATE_LIMIT_API_CALLS = 60 # calls per minute
RATE_LIMIT_BURST = 5 # burst size
# 速率限制默认值
RATE_LIMIT_LLM_CALLS = 10 # 每分钟调用次数
RATE_LIMIT_API_CALLS = 60 # 每分钟调用次数
RATE_LIMIT_BURST = 5 # 突发大小
# Retry defaults
# 重试默认值
RETRY_MAX_ATTEMPTS = 3
RETRY_BASE_DELAY = 1.0
RETRY_MAX_DELAY = 30.0
# File paths
# 文件路径
HISTORY_DIR = "history"
CACHE_DIR = "cache"
TEMP_DIR = "temp"
src/shared/trace_context.py
+35 -36
View File
@@ -1,7 +1,7 @@
"""
Trace Context - Request tracing and correlation
追踪上下文 - 请求追踪和关联
Provides context for tracking requests across the plugin.
提供用于在插件中跟踪请求的上下文。
"""
import uuid
@@ -10,7 +10,7 @@ from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, Dict, Optional
# Context variable for current trace
# 当前追踪的上下文变量
_current_trace: ContextVar[Optional["TraceContext"]] = ContextVar(
"current_trace", default=None
)
@@ -19,10 +19,9 @@ _current_trace: ContextVar[Optional["TraceContext"]] = ContextVar(
@dataclass
class TraceContext:
"""
Context for tracing requests through the plugin.
用于在插件中追踪请求的上下文。
Provides correlation IDs and timing information for debugging
and monitoring.
提供用于调试和监控的关联 ID 和计时信息。
"""
trace_id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
@@ -32,27 +31,27 @@ class TraceContext:
start_time: datetime = field(default_factory=datetime.now)
metadata: Dict[str, Any] = field(default_factory=dict)
# Timing data
# 计时数据
_checkpoints: Dict[str, datetime] = field(default_factory=dict, init=False)
def checkpoint(self, name: str) -> None:
"""
Record a timing checkpoint.
记录计时检查点。
Args:
name: Checkpoint name
参数:
name: 检查点名称
"""
self._checkpoints[name] = datetime.now()
def elapsed_ms(self, from_checkpoint: Optional[str] = None) -> float:
"""
Get elapsed time in milliseconds.
获取经过的时间（毫秒）。
Args:
from_checkpoint: Optional checkpoint to measure from
参数:
from_checkpoint: 可选的起始检查点
Returns:
Elapsed time in milliseconds
返回:
经过的时间（毫秒）
"""
start = self.start_time
if from_checkpoint and from_checkpoint in self._checkpoints:
@@ -62,7 +61,7 @@ class TraceContext:
return delta.total_seconds() * 1000
def to_dict(self) -> Dict[str, Any]:
"""Convert trace context to dictionary."""
"""将追踪上下文转换为字典。"""
return {
"trace_id": self.trace_id,
"group_id": self.group_id,
@@ -75,17 +74,17 @@ class TraceContext:
}
def __enter__(self) -> "TraceContext":
"""Enter context manager."""
"""进入上下文管理器。"""
_current_trace.set(self)
return self
def __exit__(self, exc_type, exc_val, exc_tb) -> None:
"""Exit context manager."""
"""退出上下文管理器。"""
_current_trace.set(None)
@classmethod
def current(cls) -> Optional["TraceContext"]:
"""Get the current trace context."""
"""获取当前追踪上下文。"""
return _current_trace.get()
@classmethod
@@ -96,15 +95,15 @@ class TraceContext:
operation: str = "",
) -> "TraceContext":
"""
Get current trace or create a new one.
获取当前追踪或创建新追踪。
Args:
group_id: Group identifier
platform: Platform name
operation: Operation name
参数:
group_id: 群组标识符
platform: 平台名称
operation: 操作名称
Returns:
TraceContext instance
返回:
TraceContext 实例
"""
current = cls.current()
if current:
@@ -119,10 +118,10 @@ class TraceContext:
def get_trace_id() -> str:
"""
Get current trace ID or generate a new one.
获取当前追踪 ID 或生成新的。
Returns:
Trace ID string
返回:
追踪 ID 字符串
"""
trace = TraceContext.current()
if trace:
@@ -136,15 +135,15 @@ def with_trace(
operation: str = "",
):
"""
Decorator to add trace context to a function.
为函数添加追踪上下文的装饰器。
Args:
group_id: Group identifier
platform: Platform name
operation: Operation name
参数:
group_id: 群组标识符
platform: 平台名称
operation: 操作名称
Returns:
Decorated function
返回:
装饰后的函数
"""
def decorator(func):