mirror of
https://github.com/Nezumi-2711/astrbot_plugin_qq_group_daily_analysis.git
synced 2026-09-22 13:38:43 +00:00
Refactor: Translate comments and logs to Chinese in infrastructure layer
This commit is contained in:
SXP-Simon
@@ -1,8 +1,8 @@
|
||||
"""
|
||||
Config Manager - Centralized configuration management
|
||||
配置管理器 - 集中化配置管理
|
||||
|
||||
This module provides a unified interface for accessing plugin configuration,
|
||||
wrapping the existing config module with additional validation and defaults.
|
||||
该模块提供了一个访问插件配置的统一接口，
|
||||
封装了现有的配置模块，并增加了验证和默认值功能。
|
||||
"""
|
||||
|
||||
from typing import Any, Dict, List, Optional
|
||||
@@ -12,31 +12,30 @@ from astrbot.api import logger
|
||||
|
||||
class ConfigManager:
|
||||
"""
|
||||
Centralized configuration manager for the plugin.
|
||||
插件的集中配置管理器。
|
||||
|
||||
Provides typed access to configuration values with defaults
|
||||
and validation.
|
||||
提供带有默认值和验证的配置值类型化访问。
|
||||
"""
|
||||
|
||||
def __init__(self, config: Dict[str, Any]):
|
||||
"""
|
||||
Initialize the configuration manager.
|
||||
初始化配置管理器。
|
||||
|
||||
Args:
|
||||
config: Raw configuration dictionary
|
||||
config: 原始配置字典
|
||||
"""
|
||||
self._config = config or {}
|
||||
|
||||
def get(self, key: str, default: Any = None) -> Any:
|
||||
"""
|
||||
Get a configuration value.
|
||||
获取配置值。
|
||||
|
||||
Args:
|
||||
key: Configuration key (supports dot notation)
|
||||
default: Default value if key not found
|
||||
key: 配置键（支持点号表示法）
|
||||
default: 如果键未找到则返回默认值
|
||||
|
||||
Returns:
|
||||
Configuration value or default
|
||||
配置值或默认值
|
||||
"""
|
||||
try:
|
||||
keys = key.split(".")
|
||||
@@ -54,11 +53,11 @@ class ConfigManager:
|
||||
|
||||
def set(self, key: str, value: Any) -> None:
|
||||
"""
|
||||
Set a configuration value.
|
||||
设置配置值。
|
||||
|
||||
Args:
|
||||
key: Configuration key
|
||||
value: Value to set
|
||||
key: 配置键
|
||||
value: 要设置的值
|
||||
"""
|
||||
keys = key.split(".")
|
||||
config = self._config
|
||||
@@ -69,171 +68,169 @@ class ConfigManager:
|
||||
config[keys[-1]] = value
|
||||
|
||||
# ========================================================================
|
||||
# Group Configuration
|
||||
# 群组配置
|
||||
# ========================================================================
|
||||
|
||||
def get_enabled_groups(self) -> List[str]:
|
||||
"""Get list of enabled group IDs."""
|
||||
"""获取启用的群组 ID 列表。"""
|
||||
groups = self.get("enabled_groups", [])
|
||||
return [str(g) for g in groups] if groups else []
|
||||
|
||||
def is_group_enabled(self, group_id: str) -> bool:
|
||||
"""Check if a group is enabled for analysis."""
|
||||
"""检查群组是否启用了分析。"""
|
||||
enabled = self.get_enabled_groups()
|
||||
return str(group_id) in enabled or not enabled # Empty means all enabled
|
||||
return str(group_id) in enabled or not enabled # 空列表意味着全部启用
|
||||
|
||||
def get_bot_qq_ids(self) -> List[str]:
|
||||
"""Get list of bot QQ IDs to filter out."""
|
||||
"""获取要过滤掉的机器人 QQ ID 列表。"""
|
||||
ids = self.get("bot_qq_ids", [])
|
||||
return [str(i) for i in ids] if ids else []
|
||||
|
||||
# ========================================================================
|
||||
# Analysis Configuration
|
||||
# 分析配置
|
||||
# ========================================================================
|
||||
|
||||
def get_max_topics(self) -> int:
|
||||
"""Get maximum number of topics to extract."""
|
||||
"""获取要提取的最大话题数。"""
|
||||
return int(self.get("max_topics", 5))
|
||||
|
||||
def get_max_user_titles(self) -> int:
|
||||
"""Get maximum number of user titles to generate."""
|
||||
"""获取要生成的最大用户称号数。"""
|
||||
return int(self.get("max_user_titles", 10))
|
||||
|
||||
def get_max_golden_quotes(self) -> int:
|
||||
"""Get maximum number of golden quotes to extract."""
|
||||
"""获取要提取的最大金句数。"""
|
||||
return int(self.get("max_golden_quotes", 5))
|
||||
|
||||
def get_min_messages_for_analysis(self) -> int:
|
||||
"""Get minimum messages required for analysis."""
|
||||
"""获取分析所需的最小消息数。"""
|
||||
return int(self.get("min_messages", 50))
|
||||
|
||||
# ========================================================================
|
||||
# LLM Configuration
|
||||
# ========================================================================
|
||||
|
||||
# LLM 配置
|
||||
def get_topic_provider_id(self) -> Optional[str]:
|
||||
"""Get provider ID for topic analysis."""
|
||||
"""获取话题分析的提供商 ID"""
|
||||
return self.get("topic_provider_id")
|
||||
|
||||
def get_user_title_provider_id(self) -> Optional[str]:
|
||||
"""Get provider ID for user title analysis."""
|
||||
"""获取用户称号分析的提供商 ID"""
|
||||
return self.get("user_title_provider_id")
|
||||
|
||||
def get_golden_quote_provider_id(self) -> Optional[str]:
|
||||
"""Get provider ID for golden quote analysis."""
|
||||
"""获取金句分析的提供商 ID"""
|
||||
return self.get("golden_quote_provider_id")
|
||||
|
||||
def get_topic_max_tokens(self) -> int:
|
||||
"""Get max tokens for topic analysis."""
|
||||
"""获取话题分析的最大 token 数"""
|
||||
return int(self.get("topic_max_tokens", 2000))
|
||||
|
||||
def get_user_title_max_tokens(self) -> int:
|
||||
"""Get max tokens for user title analysis."""
|
||||
"""获取用户称号分析的最大 token 数"""
|
||||
return int(self.get("user_title_max_tokens", 2000))
|
||||
|
||||
def get_golden_quote_max_tokens(self) -> int:
|
||||
"""Get max tokens for golden quote analysis."""
|
||||
"""获取金句分析的最大 token 数"""
|
||||
return int(self.get("golden_quote_max_tokens", 1500))
|
||||
|
||||
# ========================================================================
|
||||
# Prompt Configuration
|
||||
# 提示词配置
|
||||
# ========================================================================
|
||||
|
||||
def get_topic_analysis_prompt(self) -> Optional[str]:
|
||||
"""Get custom prompt template for topic analysis."""
|
||||
"""获取话题分析的自定义提示词模板"""
|
||||
return self.get("prompts.topic_analysis")
|
||||
|
||||
def get_user_title_analysis_prompt(self) -> Optional[str]:
|
||||
"""Get custom prompt template for user title analysis."""
|
||||
"""获取用户称号分析的自定义提示词模板"""
|
||||
return self.get("prompts.user_title_analysis")
|
||||
|
||||
def get_golden_quote_analysis_prompt(self) -> Optional[str]:
|
||||
"""Get custom prompt template for golden quote analysis."""
|
||||
"""获取金句分析的自定义提示词模板"""
|
||||
return self.get("prompts.golden_quote_analysis")
|
||||
|
||||
# ========================================================================
|
||||
# Scheduling Configuration
|
||||
# 调度配置
|
||||
# ========================================================================
|
||||
|
||||
def get_auto_analysis_enabled(self) -> bool:
|
||||
"""Check if auto analysis is enabled."""
|
||||
"""检查是否启用了自动分析"""
|
||||
return bool(self.get("auto_analysis_enabled", False))
|
||||
|
||||
def get_analysis_time(self) -> str:
|
||||
"""Get scheduled analysis time (HH:MM format)."""
|
||||
"""获取计划分析时间 (HH:MM 格式)"""
|
||||
return str(self.get("analysis_time", "23:00"))
|
||||
|
||||
def get_analysis_timezone(self) -> str:
|
||||
"""Get timezone for scheduled analysis."""
|
||||
"""获取计划分析的时区"""
|
||||
return str(self.get("timezone", "Asia/Shanghai"))
|
||||
|
||||
# ========================================================================
|
||||
# Report Configuration
|
||||
# 报告配置
|
||||
# ========================================================================
|
||||
|
||||
def get_report_format(self) -> str:
|
||||
"""Get report format (text, markdown, image)."""
|
||||
"""获取报告格式 (text, markdown, image)"""
|
||||
return str(self.get("report_format", "text"))
|
||||
|
||||
def get_include_statistics(self) -> bool:
|
||||
"""Check if statistics should be included in reports."""
|
||||
"""检查是否在报告中包含统计信息"""
|
||||
return bool(self.get("include_statistics", True))
|
||||
|
||||
def get_include_topics(self) -> bool:
|
||||
"""Check if topics should be included in reports."""
|
||||
"""检查是否在报告中包含话题"""
|
||||
return bool(self.get("include_topics", True))
|
||||
|
||||
def get_include_user_titles(self) -> bool:
|
||||
"""Check if user titles should be included in reports."""
|
||||
"""检查是否在报告中包含用户称号"""
|
||||
return bool(self.get("include_user_titles", True))
|
||||
|
||||
def get_include_golden_quotes(self) -> bool:
|
||||
"""Check if golden quotes should be included in reports."""
|
||||
"""检查是否在报告中包含金句"""
|
||||
return bool(self.get("include_golden_quotes", True))
|
||||
|
||||
# ========================================================================
|
||||
# Utility Methods
|
||||
# 工具方法
|
||||
# ========================================================================
|
||||
|
||||
def to_dict(self) -> Dict[str, Any]:
|
||||
"""Get the raw configuration dictionary."""
|
||||
"""获取原始配置字典"""
|
||||
return self._config.copy()
|
||||
|
||||
def update(self, updates: Dict[str, Any]) -> None:
|
||||
"""
|
||||
Update configuration with new values.
|
||||
使用新值更新配置
|
||||
|
||||
Args:
|
||||
updates: Dictionary of updates to apply
|
||||
updates: 要应用的更新字典
|
||||
"""
|
||||
self._config.update(updates)
|
||||
|
||||
def validate(self) -> List[str]:
|
||||
"""
|
||||
Validate the configuration.
|
||||
验证配置
|
||||
|
||||
Returns:
|
||||
List of validation error messages (empty if valid)
|
||||
验证错误消息列表（如果有效则为空）
|
||||
"""
|
||||
errors = []
|
||||
|
||||
# Validate numeric ranges
|
||||
# 验证数值范围
|
||||
if self.get_max_topics() < 1 or self.get_max_topics() > 20:
|
||||
errors.append("max_topics must be between 1 and 20")
|
||||
errors.append("max_topics 必须在 1 到 20 之间")
|
||||
|
||||
if self.get_max_user_titles() < 1 or self.get_max_user_titles() > 50:
|
||||
errors.append("max_user_titles must be between 1 and 50")
|
||||
errors.append("max_user_titles 必须在 1 到 50 之间")
|
||||
|
||||
if self.get_max_golden_quotes() < 1 or self.get_max_golden_quotes() > 20:
|
||||
errors.append("max_golden_quotes must be between 1 and 20")
|
||||
errors.append("max_golden_quotes 必须在 1 到 20 之间")
|
||||
|
||||
# Validate time format
|
||||
# 验证时间格式
|
||||
time_str = self.get_analysis_time()
|
||||
try:
|
||||
hours, minutes = time_str.split(":")
|
||||
if not (0 <= int(hours) <= 23 and 0 <= int(minutes) <= 59):
|
||||
errors.append("analysis_time must be in HH:MM format (00:00-23:59)")
|
||||
errors.append("analysis_time 必须是 HH:MM 格式 (00:00-23:59)")
|
||||
except ValueError:
|
||||
errors.append("analysis_time must be in HH:MM format")
|
||||
errors.append("analysis_time 必须是 HH:MM 格式")
|
||||
|
||||
return errors
|
||||
|
||||
@@ -1,8 +1,8 @@
|
||||
"""
|
||||
LLM Client - Wrapper for AstrBot's LLM provider system
|
||||
LLM 客户端 - 包装 AstrBot 的 LLM 提供商系统
|
||||
|
||||
This module provides a clean interface to AstrBot's LLM capabilities,
|
||||
abstracting away the provider management details.
|
||||
该模块提供了一个访问 AstrBot LLM 功能的清晰接口，
|
||||
抽象了提供商管理的细节。
|
||||
"""
|
||||
|
||||
from typing import Any, Dict, List, Optional, Tuple
|
||||
@@ -15,34 +15,34 @@ from ...domain.exceptions import LLMException, LLMRateLimitException
|
||||
|
||||
class LLMClient:
|
||||
"""
|
||||
Client for interacting with LLM providers.
|
||||
用于与 LLM 提供商交互的客户端。
|
||||
|
||||
This class wraps AstrBot's provider system and provides
|
||||
a clean interface for making LLM calls.
|
||||
该类包装了 AstrBot 的提供商系统，并提供了一个
|
||||
清晰的接口来进行 LLM 调用。
|
||||
"""
|
||||
|
||||
def __init__(self, context: Any):
|
||||
"""
|
||||
Initialize the LLM client.
|
||||
初始化 LLM 客户端。
|
||||
|
||||
Args:
|
||||
context: AstrBot plugin context with provider access
|
||||
context: 具有提供商访问权限的 AstrBot 插件上下文
|
||||
"""
|
||||
self.context = context
|
||||
self._provider_cache: Dict[str, Any] = {}
|
||||
|
||||
def get_provider(self, provider_id: Optional[str] = None) -> Any:
|
||||
"""
|
||||
Get an LLM provider by ID.
|
||||
通过 ID 获取 LLM 提供商。
|
||||
|
||||
Args:
|
||||
provider_id: Specific provider ID, or None for default
|
||||
provider_id: 特定的提供商 ID，None 表示默认
|
||||
|
||||
Returns:
|
||||
Provider instance
|
||||
提供商实例
|
||||
|
||||
Raises:
|
||||
LLMException: If provider not found
|
||||
LLMException: 如果未找到提供商
|
||||
"""
|
||||
try:
|
||||
if provider_id and provider_id in self._provider_cache:
|
||||
@@ -51,10 +51,10 @@ class LLMClient:
|
||||
if provider_id:
|
||||
provider = self.context.get_provider_by_id(provider_id)
|
||||
else:
|
||||
# Get default provider
|
||||
# 获取默认提供商
|
||||
providers = self.context.get_all_providers()
|
||||
if not providers:
|
||||
raise LLMException("No LLM providers available")
|
||||
raise LLMException("无可用 LLM 提供商")
|
||||
provider = providers[0]
|
||||
|
||||
if provider:
|
||||
@@ -63,7 +63,7 @@ class LLMClient:
|
||||
return provider
|
||||
|
||||
except Exception as e:
|
||||
raise LLMException(f"Failed to get provider: {e}")
|
||||
raise LLMException(f"获取提供商失败: {e}")
|
||||
|
||||
async def chat_completion(
|
||||
self,
|
||||
@@ -74,39 +74,39 @@ class LLMClient:
|
||||
system_prompt: Optional[str] = None,
|
||||
) -> Tuple[str, TokenUsage]:
|
||||
"""
|
||||
Make a chat completion request.
|
||||
发起聊天完成请求。
|
||||
|
||||
Args:
|
||||
prompt: The user prompt
|
||||
provider_id: Specific provider ID (optional)
|
||||
max_tokens: Maximum tokens in response
|
||||
temperature: Sampling temperature
|
||||
system_prompt: Optional system prompt
|
||||
prompt: 用户提示词
|
||||
provider_id: 特定的提供商 ID (可选)
|
||||
max_tokens: 响应中的最大 token 数
|
||||
temperature: 采样温度
|
||||
system_prompt: 可选的系统提示词
|
||||
|
||||
Returns:
|
||||
Tuple of (response_text, token_usage)
|
||||
(response_text, token_usage) 元组
|
||||
|
||||
Raises:
|
||||
LLMException: If the request fails
|
||||
LLMException: 如果请求失败
|
||||
"""
|
||||
try:
|
||||
provider = self.get_provider(provider_id)
|
||||
if not provider:
|
||||
raise LLMException("No provider available", provider_id or "default")
|
||||
raise LLMException("无可用提供商", provider_id or "default")
|
||||
|
||||
# Build messages
|
||||
# 构建消息
|
||||
messages = []
|
||||
if system_prompt:
|
||||
messages.append({"role": "system", "content": system_prompt})
|
||||
messages.append({"role": "user", "content": prompt})
|
||||
|
||||
# Make the request
|
||||
# 发起请求
|
||||
response = await provider.text_chat(
|
||||
prompt=prompt,
|
||||
session_id=None, # Stateless
|
||||
messages=messages,
|
||||
session_id=None, # 无状态
|
||||
)
|
||||
|
||||
# Extract response text
|
||||
# 提取响应文本
|
||||
if hasattr(response, "completion_text"):
|
||||
response_text = response.completion_text
|
||||
elif isinstance(response, dict):
|
||||
@@ -114,7 +114,7 @@ class LLMClient:
|
||||
else:
|
||||
response_text = str(response)
|
||||
|
||||
# Extract token usage
|
||||
# 提取 token 使用情况
|
||||
token_usage = TokenUsage()
|
||||
if hasattr(response, "usage"):
|
||||
usage = response.usage
|
||||
@@ -131,7 +131,7 @@ class LLMClient:
|
||||
error_msg = str(e).lower()
|
||||
if "rate limit" in error_msg or "429" in error_msg:
|
||||
raise LLMRateLimitException(str(e), provider_id or "default")
|
||||
raise LLMException(f"Chat completion failed: {e}", provider_id or "default")
|
||||
raise LLMException(f"聊天完成请求失败: {e}", provider_id or "default")
|
||||
|
||||
async def analyze_with_json_output(
|
||||
self,
|
||||
@@ -141,18 +141,18 @@ class LLMClient:
|
||||
temperature: float = 0.7,
|
||||
) -> Tuple[str, TokenUsage]:
|
||||
"""
|
||||
Make a completion request expecting JSON output.
|
||||
发起期望 JSON 输出的完成请求。
|
||||
|
||||
Args:
|
||||
prompt: The analysis prompt
|
||||
provider_id: Specific provider ID (optional)
|
||||
max_tokens: Maximum tokens in response
|
||||
temperature: Sampling temperature
|
||||
prompt: 分析提示词
|
||||
provider_id: 特定的提供商 ID (可选)
|
||||
max_tokens: 响应中的最大 token 数
|
||||
temperature: 采样温度
|
||||
|
||||
Returns:
|
||||
Tuple of (response_text, token_usage)
|
||||
(response_text, token_usage) 元组
|
||||
"""
|
||||
# Add JSON instruction to prompt if not present
|
||||
# 如果提示词中没有 JSON 指令，则添加
|
||||
json_instruction = "\nRespond with valid JSON only."
|
||||
if "json" not in prompt.lower():
|
||||
prompt = prompt + json_instruction
|
||||
@@ -166,10 +166,10 @@ class LLMClient:
|
||||
|
||||
def list_available_providers(self) -> List[Dict[str, str]]:
|
||||
"""
|
||||
List all available LLM providers.
|
||||
列出所有可用的 LLM 提供商。
|
||||
|
||||
Returns:
|
||||
List of provider info dictionaries
|
||||
提供商信息字典列表
|
||||
"""
|
||||
try:
|
||||
providers = self.context.get_all_providers()
|
||||
@@ -182,5 +182,5 @@ class LLMClient:
|
||||
for i, p in enumerate(providers)
|
||||
]
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to list providers: {e}")
|
||||
logger.error(f"列出提供商失败: {e}")
|
||||
return []
|
||||
|
||||
@@ -1,8 +1,8 @@
|
||||
"""
|
||||
History Repository - Implementation for storing analysis history
|
||||
历史仓库 - 存储分析历史的实现
|
||||
|
||||
This module provides persistent storage for analysis results and history.
|
||||
It wraps the existing history_manager functionality.
|
||||
该模块提供分析结果和历史记录的持久化存储。
|
||||
它封装了现有的 history_manager 功能。
|
||||
"""
|
||||
|
||||
import json
|
||||
@@ -16,29 +16,29 @@ from astrbot.api import logger
|
||||
|
||||
class HistoryRepository:
|
||||
"""
|
||||
Repository for storing and retrieving analysis history.
|
||||
用于存储和检索分析历史的仓库。
|
||||
|
||||
This implementation stores history as JSON files, maintaining
|
||||
backward compatibility with the existing history_manager.
|
||||
此实现将历史记录存储为 JSON 文件，保持
|
||||
与现有 history_manager 的向后兼容性。
|
||||
"""
|
||||
|
||||
def __init__(self, data_dir: str):
|
||||
"""
|
||||
Initialize the history repository.
|
||||
初始化历史仓库。
|
||||
|
||||
Args:
|
||||
data_dir: Base directory for storing history data
|
||||
data_dir: 存储历史数据的基础目录
|
||||
"""
|
||||
self.data_dir = Path(data_dir)
|
||||
self.history_dir = self.data_dir / "history"
|
||||
self._ensure_directories()
|
||||
|
||||
def _ensure_directories(self) -> None:
|
||||
"""Ensure required directories exist."""
|
||||
"""确保所需目录存在。"""
|
||||
self.history_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
def _get_group_history_path(self, group_id: str) -> Path:
|
||||
"""Get the history file path for a group."""
|
||||
"""获取群组的历史文件路径。"""
|
||||
return self.history_dir / f"group_{group_id}.json"
|
||||
|
||||
def save_analysis_result(
|
||||
@@ -48,52 +48,52 @@ class HistoryRepository:
|
||||
date_str: Optional[str] = None,
|
||||
) -> bool:
|
||||
"""
|
||||
Save an analysis result to history.
|
||||
保存分析结果到历史记录。
|
||||
|
||||
Args:
|
||||
group_id: The group identifier
|
||||
result: Analysis result dictionary
|
||||
date_str: Date string (defaults to today)
|
||||
group_id: 群组标识符
|
||||
result: 分析结果字典
|
||||
date_str: 日期字符串（默认为今天）
|
||||
|
||||
Returns:
|
||||
True if saved successfully
|
||||
如果保存成功则返回 True
|
||||
"""
|
||||
try:
|
||||
date_str = date_str or datetime.now().strftime("%Y-%m-%d")
|
||||
history = self.load_group_history(group_id)
|
||||
|
||||
# Add timestamp if not present
|
||||
# 如果不存在则添加时间戳
|
||||
if "timestamp" not in result:
|
||||
result["timestamp"] = datetime.now().isoformat()
|
||||
|
||||
# Store by date
|
||||
# 按日期存储
|
||||
if "daily" not in history:
|
||||
history["daily"] = {}
|
||||
|
||||
history["daily"][date_str] = result
|
||||
history["last_updated"] = datetime.now().isoformat()
|
||||
|
||||
# Write to file
|
||||
# 写入文件
|
||||
history_path = self._get_group_history_path(group_id)
|
||||
with open(history_path, "w", encoding="utf-8") as f:
|
||||
json.dump(history, f, ensure_ascii=False, indent=2)
|
||||
|
||||
logger.debug(f"Saved analysis result for group {group_id} on {date_str}")
|
||||
logger.debug(f"已保存群组 {group_id} 在 {date_str} 的分析结果")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to save analysis result: {e}")
|
||||
logger.error(f"保存分析结果失败: {e}")
|
||||
return False
|
||||
|
||||
def load_group_history(self, group_id: str) -> Dict[str, Any]:
|
||||
"""
|
||||
Load history for a group.
|
||||
加载群组历史记录。
|
||||
|
||||
Args:
|
||||
group_id: The group identifier
|
||||
group_id: 群组标识符
|
||||
|
||||
Returns:
|
||||
History dictionary
|
||||
历史记录字典
|
||||
"""
|
||||
try:
|
||||
history_path = self._get_group_history_path(group_id)
|
||||
@@ -102,21 +102,21 @@ class HistoryRepository:
|
||||
return json.load(f)
|
||||
return {"daily": {}, "group_id": group_id}
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to load group history: {e}")
|
||||
logger.error(f"加载群组历史记录失败: {e}")
|
||||
return {"daily": {}, "group_id": group_id}
|
||||
|
||||
def get_analysis_result(
|
||||
self, group_id: str, date_str: str
|
||||
) -> Optional[Dict[str, Any]]:
|
||||
"""
|
||||
Get analysis result for a specific date.
|
||||
获取特定日期的分析结果。
|
||||
|
||||
Args:
|
||||
group_id: The group identifier
|
||||
date_str: Date string (YYYY-MM-DD format)
|
||||
group_id: 群组标识符
|
||||
date_str: 日期字符串 (YYYY-MM-DD 格式)
|
||||
|
||||
Returns:
|
||||
Analysis result or None if not found
|
||||
分析结果，如果未找到则返回 None
|
||||
"""
|
||||
history = self.load_group_history(group_id)
|
||||
return history.get("daily", {}).get(date_str)
|
||||
@@ -125,58 +125,58 @@ class HistoryRepository:
|
||||
self, group_id: str, limit: int = 7
|
||||
) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Get recent analysis results.
|
||||
获取最近的分析结果。
|
||||
|
||||
Args:
|
||||
group_id: The group identifier
|
||||
limit: Maximum number of results to return
|
||||
group_id: 群组标识符
|
||||
limit: 返回的最大结果数
|
||||
|
||||
Returns:
|
||||
List of recent analysis results
|
||||
最近分析结果列表
|
||||
"""
|
||||
history = self.load_group_history(group_id)
|
||||
daily = history.get("daily", {})
|
||||
|
||||
# Sort by date descending
|
||||
# 按日期降序排序
|
||||
sorted_dates = sorted(daily.keys(), reverse=True)[:limit]
|
||||
return [daily[date] for date in sorted_dates]
|
||||
|
||||
def has_analysis_for_date(self, group_id: str, date_str: str) -> bool:
|
||||
"""
|
||||
Check if analysis exists for a specific date.
|
||||
检查特定日期是否存在分析结果。
|
||||
|
||||
Args:
|
||||
group_id: The group identifier
|
||||
date_str: Date string (YYYY-MM-DD format)
|
||||
group_id: 群组标识符
|
||||
date_str: 日期字符串 (YYYY-MM-DD 格式)
|
||||
|
||||
Returns:
|
||||
True if analysis exists
|
||||
如果分析结果存在则返回 True
|
||||
"""
|
||||
result = self.get_analysis_result(group_id, date_str)
|
||||
return result is not None
|
||||
|
||||
def delete_old_history(self, group_id: str, keep_days: int = 30) -> int:
|
||||
"""
|
||||
Delete history older than specified days.
|
||||
删除超过指定天数的历史记录。
|
||||
|
||||
Args:
|
||||
group_id: The group identifier
|
||||
keep_days: Number of days of history to keep
|
||||
group_id: 群组标识符
|
||||
keep_days: 保留历史记录的天数
|
||||
|
||||
Returns:
|
||||
Number of entries deleted
|
||||
删除的条目数
|
||||
"""
|
||||
try:
|
||||
history = self.load_group_history(group_id)
|
||||
daily = history.get("daily", {})
|
||||
|
||||
cutoff_date = datetime.now().strftime("%Y-%m-%d")
|
||||
# Calculate cutoff (simple string comparison works for YYYY-MM-DD format)
|
||||
# 计算截止日期（简单的字符串比较适用于 YYYY-MM-DD 格式）
|
||||
from datetime import timedelta
|
||||
|
||||
cutoff = (datetime.now() - timedelta(days=keep_days)).strftime("%Y-%m-%d")
|
||||
|
||||
# Find dates to delete
|
||||
# 查找要删除的日期
|
||||
dates_to_delete = [date for date in daily.keys() if date < cutoff]
|
||||
|
||||
for date in dates_to_delete:
|
||||
@@ -191,15 +191,15 @@ class HistoryRepository:
|
||||
return len(dates_to_delete)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to delete old history: {e}")
|
||||
logger.error(f"删除旧历史记录失败: {e}")
|
||||
return 0
|
||||
|
||||
def list_groups_with_history(self) -> List[str]:
|
||||
"""
|
||||
List all groups that have history.
|
||||
列出所有有历史记录的群组。
|
||||
|
||||
Returns:
|
||||
List of group IDs
|
||||
群组 ID 列表
|
||||
"""
|
||||
try:
|
||||
groups = []
|
||||
@@ -208,5 +208,5 @@ class HistoryRepository:
|
||||
groups.append(group_id)
|
||||
return groups
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to list groups: {e}")
|
||||
logger.error(f"列出群组失败: {e}")
|
||||
return []
|
||||
|
||||
Reference in New Issue
Block a user