diff --git a/src/application/__init__.py b/src/application/__init__.py index 6426f74..960a7fc 100644 --- a/src/application/__init__.py +++ b/src/application/__init__.py @@ -1,5 +1,12 @@ # Application Layer - Orchestration and Use Cases from .analysis_orchestrator import AnalysisOrchestrator from .message_converter import MessageConverter +from .scheduling_service import SchedulingService +from .reporting_service import ReportingService -__all__ = ["AnalysisOrchestrator", "MessageConverter"] +__all__ = [ + "AnalysisOrchestrator", + "MessageConverter", + "SchedulingService", + "ReportingService", +] diff --git a/src/application/reporting_service.py b/src/application/reporting_service.py new file mode 100644 index 0000000..6cdd5ea --- /dev/null +++ b/src/application/reporting_service.py @@ -0,0 +1,263 @@ +""" +Reporting Service - Application service for generating and sending reports + +This service coordinates report generation and delivery to groups. +""" + +from datetime import datetime +from typing import Any, Dict, List, Optional + +from astrbot.api import logger + +from ..domain.services import ReportGenerator +from ..domain.value_objects.topic import Topic +from ..domain.value_objects.user_title import UserTitle +from ..domain.value_objects.golden_quote import GoldenQuote +from ..domain.value_objects.statistics import GroupStatistics +from ..infrastructure.config import ConfigManager +from ..infrastructure.persistence import HistoryRepository + + +class ReportingService: + """ + Application service for generating and managing reports. + + This service coordinates between domain services and infrastructure + to produce and deliver analysis reports. + """ + + def __init__( + self, + config: ConfigManager, + history_repository: HistoryRepository, + ): + """ + Initialize the reporting service. + + Args: + config: Configuration manager + history_repository: Repository for storing reports + """ + self.config = config + self.history = history_repository + + def generate_report( + self, + group_id: str, + group_name: str, + statistics: GroupStatistics, + topics: List[Topic], + user_titles: List[UserTitle], + golden_quotes: List[GoldenQuote], + date_str: Optional[str] = None, + ) -> str: + """ + Generate a complete analysis report. + + Args: + group_id: Group identifier + group_name: Group display name + statistics: Group statistics + topics: List of discussion topics + user_titles: List of user titles + golden_quotes: List of golden quotes + date_str: Report date (defaults to today) + + Returns: + Formatted report string + """ + date_str = date_str or datetime.now().strftime("%Y-%m-%d") + + generator = ReportGenerator( + group_name=group_name, + date_str=date_str, + ) + + # Generate report based on configuration + report = generator.generate_full_report( + statistics=statistics, + topics=topics if self.config.get_include_topics() else [], + user_titles=user_titles if self.config.get_include_user_titles() else [], + golden_quotes=golden_quotes if self.config.get_include_golden_quotes() else [], + include_header=True, + include_footer=True, + ) + + return report + + def generate_summary( + self, + group_id: str, + statistics: GroupStatistics, + top_topic: Optional[Topic] = None, + top_quote: Optional[GoldenQuote] = None, + date_str: Optional[str] = None, + ) -> str: + """ + Generate a brief summary report. + + Args: + group_id: Group identifier + statistics: Group statistics + top_topic: Most significant topic + top_quote: Best golden quote + date_str: Report date + + Returns: + Brief summary string + """ + date_str = date_str or datetime.now().strftime("%Y-%m-%d") + + generator = ReportGenerator(date_str=date_str) + return generator.generate_summary_report( + statistics=statistics, + top_topic=top_topic, + top_quote=top_quote, + ) + + def save_report( + self, + group_id: str, + report_data: Dict[str, Any], + date_str: Optional[str] = None, + ) -> bool: + """ + Save a report to history. + + Args: + group_id: Group identifier + report_data: Report data dictionary + date_str: Report date + + Returns: + True if saved successfully + """ + date_str = date_str or datetime.now().strftime("%Y-%m-%d") + + return self.history.save_analysis_result( + group_id=group_id, + result=report_data, + date_str=date_str, + ) + + def get_report( + self, + group_id: str, + date_str: str, + ) -> Optional[Dict[str, Any]]: + """ + Get a saved report. + + Args: + group_id: Group identifier + date_str: Report date + + Returns: + Report data or None + """ + return self.history.get_analysis_result(group_id, date_str) + + def get_recent_reports( + self, + group_id: str, + limit: int = 7, + ) -> List[Dict[str, Any]]: + """ + Get recent reports for a group. + + Args: + group_id: Group identifier + limit: Maximum number of reports + + Returns: + List of report data dictionaries + """ + return self.history.get_recent_results(group_id, limit) + + def has_report_for_today(self, group_id: str) -> bool: + """ + Check if a report exists for today. + + Args: + group_id: Group identifier + + Returns: + True if report exists + """ + today = datetime.now().strftime("%Y-%m-%d") + return self.history.has_analysis_for_date(group_id, today) + + def format_for_platform( + self, + report: str, + platform: str, + format_type: Optional[str] = None, + ) -> str: + """ + Format a report for a specific platform. + + Args: + report: Raw report text + platform: Target platform + format_type: Override format type + + Returns: + Platform-formatted report + """ + format_type = format_type or self.config.get_report_format() + + # For now, return as-is. Can be extended for platform-specific formatting + if format_type == "markdown": + return report + elif format_type == "text": + # Strip markdown formatting + return self._strip_markdown(report) + else: + return report + + def _strip_markdown(self, text: str) -> str: + """Strip markdown formatting from text.""" + # Simple markdown stripping + import re + + # Remove bold + text = re.sub(r"\*\*(.*?)\*\*", r"\1", text) + # Remove italic + text = re.sub(r"\*(.*?)\*", r"\1", text) + # Remove headers + text = re.sub(r"^#+\s*", "", text, flags=re.MULTILINE) + + return text + + def create_report_data( + self, + group_id: str, + group_name: str, + statistics: GroupStatistics, + topics: List[Topic], + user_titles: List[UserTitle], + golden_quotes: List[GoldenQuote], + ) -> Dict[str, Any]: + """ + Create a report data dictionary for storage. + + Args: + group_id: Group identifier + group_name: Group display name + statistics: Group statistics + topics: List of topics + user_titles: List of user titles + golden_quotes: List of golden quotes + + Returns: + Report data dictionary + """ + return { + "group_id": group_id, + "group_name": group_name, + "timestamp": datetime.now().isoformat(), + "statistics": statistics.to_dict(), + "topics": [t.to_dict() for t in topics], + "user_titles": [u.to_dict() for u in user_titles], + "golden_quotes": [q.to_dict() for q in golden_quotes], + } diff --git a/src/application/scheduling_service.py b/src/application/scheduling_service.py new file mode 100644 index 0000000..77ec456 --- /dev/null +++ b/src/application/scheduling_service.py @@ -0,0 +1,264 @@ +""" +Scheduling Service - Application service for scheduled analysis + +This service manages scheduled analysis tasks and coordinates +with the analysis orchestrator. +""" + +import asyncio +from datetime import datetime, timedelta +from typing import Any, Callable, Dict, List, Optional, Set + +from astrbot.api import logger + +from ..infrastructure.config import ConfigManager +from ..shared.constants import TASK_STATE_PENDING, TASK_STATE_RUNNING, TASK_STATE_COMPLETED + + +class ScheduledTask: + """Represents a scheduled analysis task.""" + + def __init__( + self, + task_id: str, + group_id: str, + scheduled_time: str, # HH:MM format + callback: Callable, + enabled: bool = True, + ): + self.task_id = task_id + self.group_id = group_id + self.scheduled_time = scheduled_time + self.callback = callback + self.enabled = enabled + self.last_run: Optional[datetime] = None + self.next_run: Optional[datetime] = None + self._calculate_next_run() + + def _calculate_next_run(self) -> None: + """Calculate the next run time.""" + if not self.enabled: + self.next_run = None + return + + try: + hours, minutes = map(int, self.scheduled_time.split(":")) + now = datetime.now() + next_run = now.replace(hour=hours, minute=minutes, second=0, microsecond=0) + + # If the time has passed today, schedule for tomorrow + if next_run <= now: + next_run += timedelta(days=1) + + self.next_run = next_run + except ValueError: + logger.error(f"Invalid scheduled time format: {self.scheduled_time}") + self.next_run = None + + def should_run(self) -> bool: + """Check if the task should run now.""" + if not self.enabled or not self.next_run: + return False + + now = datetime.now() + + # Check if we're within the execution window (5 minute tolerance) + if self.next_run <= now <= self.next_run + timedelta(minutes=5): + # Check if we haven't run today + if self.last_run is None or self.last_run.date() != now.date(): + return True + + return False + + def mark_completed(self) -> None: + """Mark the task as completed and schedule next run.""" + self.last_run = datetime.now() + self._calculate_next_run() + + +class SchedulingService: + """ + Application service for managing scheduled analysis tasks. + + This service runs a background loop that checks for and + executes scheduled tasks. + """ + + def __init__(self, config: ConfigManager): + """ + Initialize the scheduling service. + + Args: + config: Configuration manager + """ + self.config = config + self._tasks: Dict[str, ScheduledTask] = {} + self._running = False + self._task: Optional[asyncio.Task] = None + self._callbacks: Dict[str, Callable] = {} + + def register_callback(self, name: str, callback: Callable) -> None: + """ + Register a callback for scheduled tasks. + + Args: + name: Callback name + callback: Async callback function + """ + self._callbacks[name] = callback + + def add_task( + self, + group_id: str, + scheduled_time: Optional[str] = None, + callback_name: str = "analyze", + ) -> str: + """ + Add a scheduled task for a group. + + Args: + group_id: Group identifier + scheduled_time: Time in HH:MM format (uses config default if not provided) + callback_name: Name of registered callback to use + + Returns: + Task ID + """ + scheduled_time = scheduled_time or self.config.get_analysis_time() + task_id = f"task_{group_id}" + + callback = self._callbacks.get(callback_name) + if not callback: + logger.warning(f"Callback '{callback_name}' not registered") + return task_id + + task = ScheduledTask( + task_id=task_id, + group_id=group_id, + scheduled_time=scheduled_time, + callback=callback, + enabled=True, + ) + + self._tasks[task_id] = task + logger.info(f"Added scheduled task {task_id} for {scheduled_time}") + + return task_id + + def remove_task(self, task_id: str) -> bool: + """ + Remove a scheduled task. + + Args: + task_id: Task identifier + + Returns: + True if task was removed + """ + if task_id in self._tasks: + del self._tasks[task_id] + logger.info(f"Removed scheduled task {task_id}") + return True + return False + + def enable_task(self, task_id: str) -> bool: + """Enable a scheduled task.""" + if task_id in self._tasks: + self._tasks[task_id].enabled = True + self._tasks[task_id]._calculate_next_run() + return True + return False + + def disable_task(self, task_id: str) -> bool: + """Disable a scheduled task.""" + if task_id in self._tasks: + self._tasks[task_id].enabled = False + self._tasks[task_id].next_run = None + return True + return False + + def get_task_status(self, task_id: str) -> Optional[Dict[str, Any]]: + """ + Get status of a scheduled task. + + Args: + task_id: Task identifier + + Returns: + Task status dictionary or None + """ + task = self._tasks.get(task_id) + if not task: + return None + + return { + "task_id": task.task_id, + "group_id": task.group_id, + "scheduled_time": task.scheduled_time, + "enabled": task.enabled, + "last_run": task.last_run.isoformat() if task.last_run else None, + "next_run": task.next_run.isoformat() if task.next_run else None, + } + + def list_tasks(self) -> List[Dict[str, Any]]: + """List all scheduled tasks.""" + return [self.get_task_status(tid) for tid in self._tasks.keys()] + + async def start(self) -> None: + """Start the scheduling service.""" + if self._running: + return + + self._running = True + self._task = asyncio.create_task(self._run_loop()) + logger.info("Scheduling service started") + + async def stop(self) -> None: + """Stop the scheduling service.""" + self._running = False + if self._task: + self._task.cancel() + try: + await self._task + except asyncio.CancelledError: + pass + logger.info("Scheduling service stopped") + + async def _run_loop(self) -> None: + """Main scheduling loop.""" + while self._running: + try: + await self._check_and_run_tasks() + # Check every minute + await asyncio.sleep(60) + except asyncio.CancelledError: + break + except Exception as e: + logger.error(f"Error in scheduling loop: {e}") + await asyncio.sleep(60) + + async def _check_and_run_tasks(self) -> None: + """Check for and execute due tasks.""" + for task in list(self._tasks.values()): + if task.should_run(): + try: + logger.info(f"Executing scheduled task {task.task_id}") + await task.callback(task.group_id) + task.mark_completed() + logger.info(f"Completed scheduled task {task.task_id}") + except Exception as e: + logger.error(f"Failed to execute task {task.task_id}: {e}") + + def setup_from_config(self) -> None: + """Set up scheduled tasks from configuration.""" + if not self.config.get_auto_analysis_enabled(): + logger.info("Auto analysis is disabled") + return + + enabled_groups = self.config.get_enabled_groups() + analysis_time = self.config.get_analysis_time() + + for group_id in enabled_groups: + self.add_task(group_id, analysis_time) + + logger.info(f"Set up {len(enabled_groups)} scheduled tasks") diff --git a/src/domain/exceptions.py b/src/domain/exceptions.py new file mode 100644 index 0000000..ce23d57 --- /dev/null +++ b/src/domain/exceptions.py @@ -0,0 +1,234 @@ +""" +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 + self.code = code + super().__init__(self.message) + + +# ============================================================================ +# 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"): + super().__init__(message, "INSUFFICIENT_DATA") + + +class AnalysisTimeoutException(AnalysisException): + """Raised when analysis takes too long.""" + + def __init__(self, message: str = "Analysis timed out"): + super().__init__(message, "ANALYSIS_TIMEOUT") + + +class LLMException(AnalysisException): + """Raised when LLM API call fails.""" + + def __init__(self, message: str = "LLM API call failed", provider: str = ""): + self.provider = provider + super().__init__(f"{message} (provider: {provider})" if provider else message, "LLM_ERROR") + + +class LLMRateLimitException(LLMException): + """Raised when LLM API rate limit is exceeded.""" + + def __init__(self, message: str = "LLM rate limit exceeded", provider: str = ""): + super().__init__(message, provider) + self.code = "LLM_RATE_LIMIT" + + +class LLMQuotaExceededException(LLMException): + """Raised when LLM API quota is exceeded.""" + + def __init__(self, message: str = "LLM quota exceeded", 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 + super().__init__(f"[{platform}] {message}" if platform else message, code) + + +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") + + +class PlatformConnectionException(PlatformException): + """Raised when connection to platform fails.""" + + def __init__(self, message: str = "Failed to connect to platform", platform: str = ""): + super().__init__(message, platform, "PLATFORM_CONNECTION_ERROR") + + +class PlatformAPIException(PlatformException): + """Raised when platform API call fails.""" + + def __init__(self, message: str = "Platform API call failed", 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 = ""): + self.group_id = group_id + super().__init__(f"{message} (group: {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 = ""): + self.group_id = group_id + super().__init__(f"{message} (group: {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 = ""): + 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") + + +# ============================================================================ +# 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 = ""): + 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") + + +class DataPersistenceException(RepositoryException): + """Raised when data persistence fails.""" + + def __init__(self, message: str = "Failed to persist data"): + 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") + + +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") + + +# ============================================================================ +# Validation Exceptions +# ============================================================================ + + +class ValidationException(DomainException): + """Base exception for validation errors.""" + + def __init__(self, message: str, field: str = "", code: str = "VALIDATION_ERROR"): + self.field = field + super().__init__(f"{field}: {message}" if field else message, code) + + +class InvalidGroupIdException(ValidationException): + """Raised when group ID is invalid.""" + + def __init__(self, group_id: str): + super().__init__(f"Invalid group ID: {group_id}", "group_id", "INVALID_GROUP_ID") + + +class InvalidUserIdException(ValidationException): + """Raised when user ID is invalid.""" + + def __init__(self, user_id: str): + super().__init__(f"Invalid user 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"): + super().__init__(message, "message", "INVALID_MESSAGE") diff --git a/src/domain/services/__init__.py b/src/domain/services/__init__.py new file mode 100644 index 0000000..678384e --- /dev/null +++ b/src/domain/services/__init__.py @@ -0,0 +1,14 @@ +""" +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 +from .report_generator import ReportGenerator + +__all__ = [ + "StatisticsCalculator", + "ReportGenerator", +] diff --git a/src/domain/services/report_generator.py b/src/domain/services/report_generator.py new file mode 100644 index 0000000..76467c5 --- /dev/null +++ b/src/domain/services/report_generator.py @@ -0,0 +1,188 @@ +""" +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. +""" + +from datetime import datetime +from typing import List, Optional + +from ..value_objects.topic import Topic +from ..value_objects.user_title import UserTitle +from ..value_objects.golden_quote import GoldenQuote +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 + """ + self.group_name = group_name + self.date_str = date_str or datetime.now().strftime("%Y-%m-%d") + + def generate_full_report( + self, + statistics: GroupStatistics, + topics: List[Topic], + user_titles: List[UserTitle], + golden_quotes: List[GoldenQuote], + include_header: bool = True, + 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 + + Returns: + Formatted report string + """ + sections = [] + + if include_header: + sections.append(self._generate_header()) + + sections.append(self._generate_statistics_section(statistics)) + + if topics: + sections.append(self._generate_topics_section(topics)) + + if user_titles: + sections.append(self._generate_user_titles_section(user_titles)) + + if golden_quotes: + sections.append(self._generate_golden_quotes_section(golden_quotes)) + + if include_footer: + sections.append(self._generate_footer(statistics.token_usage)) + + return "\n\n".join(sections) + + def _generate_header(self) -> str: + """Generate report header.""" + title = f"šŸ“Š Group Analysis Report" + if self.group_name: + title += f" - {self.group_name}" + + return f"{title}\nšŸ“… Date: {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}", + ] + + if stats.emoji_count > 0: + lines.append(f"ā€¢ Emoji Used: {stats.emoji_count}") + + return "\n".join(lines) + + def _generate_topics_section(self, topics: List[Topic]) -> str: + """Generate topics section.""" + lines = ["šŸ’¬ **Discussion Topics**"] + + 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} more" + + lines.append(f"\n{i}. **{topic.name}**") + lines.append(f" Contributors: {contributors_str}") + if topic.detail: + # Truncate long details + 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: + """Generate user titles section.""" + lines = ["šŸ† **User Titles & Badges**"] + + for title in titles: + lines.append(f"\nšŸ‘¤ **{title.name}**") + lines.append(f" šŸŽ–ļø Title: {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: {reason}") + + return "\n".join(lines) + + def _generate_golden_quotes_section(self, quotes: List[GoldenQuote]) -> str: + """Generate golden quotes section.""" + lines = ["āœØ **Golden Quotes**"] + + 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: + """Generate report footer.""" + lines = ["ā”€" * 40] + lines.append(f"Generated at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") + + if token_usage and token_usage.total_tokens > 0: + lines.append(f"Token Usage: {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: + """ + Generate a brief summary report. + + Args: + statistics: Group chat statistics + top_topic: Most significant topic (optional) + top_quote: Best golden quote (optional) + + Returns: + Brief summary string + """ + lines = [ + f"šŸ“Š Daily Summary ({self.date_str})", + f"Messages: {statistics.message_count} | Participants: {statistics.participant_count}", + ] + + if top_topic: + lines.append(f"šŸ”„ Hot Topic: {top_topic.name}") + + if top_quote: + lines.append(f"āœØ Quote: \"{top_quote.content}\" ā€” {top_quote.sender}") + + return "\n".join(lines) diff --git a/src/domain/services/statistics_calculator.py b/src/domain/services/statistics_calculator.py new file mode 100644 index 0000000..5c4f25e --- /dev/null +++ b/src/domain/services/statistics_calculator.py @@ -0,0 +1,266 @@ +""" +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 +from typing import Dict, List, Optional + +from ..value_objects import UnifiedMessage +from ..value_objects.statistics import ( + GroupStatistics, + UserStatistics, + EmojiStatistics, + ActivityVisualization, + TokenUsage, +) + + +class StatisticsCalculator: + """ + Domain service for calculating group chat statistics. + + This service processes UnifiedMessage objects and produces + platform-agnostic statistics. + """ + + 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 + """ + self.bot_user_ids = set(bot_user_ids or []) + + def calculate_group_statistics( + self, + messages: List[UnifiedMessage], + 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 + + Returns: + GroupStatistics object with computed statistics + """ + 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 + ] + + 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( + message_count=message_count, + total_characters=total_characters, + participant_count=participant_count, + most_active_period=most_active_period, + emoji_statistics=emoji_stats, + activity_visualization=activity_viz, + token_usage=token_usage or TokenUsage(), + ) + + def calculate_user_statistics( + self, messages: List[UnifiedMessage] + ) -> Dict[str, UserStatistics]: + """ + Calculate per-user statistics from messages. + + Args: + messages: List of unified messages to analyze + + Returns: + Dictionary mapping user_id to UserStatistics + """ + user_stats: Dict[str, UserStatistics] = {} + + for msg in messages: + # Skip bot messages + if msg.sender_id in self.bot_user_ids: + continue + + user_id = msg.sender_id + + if user_id not in user_stats: + user_stats[user_id] = UserStatistics( + user_id=user_id, + nickname=msg.sender_name, + ) + + stats = user_stats[user_id] + stats.message_count += 1 + 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 + + return user_stats + + def get_top_users( + self, + user_stats: Dict[str, UserStatistics], + limit: int = 10, + 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 + + 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 + ] + + sorted_users = sorted( + eligible_users, key=lambda x: x.message_count, reverse=True + ) + + return [ + { + "user_id": u.user_id, + "nickname": u.nickname, + "name": u.nickname, # Backward compatibility + "message_count": u.message_count, + "avg_chars": round(u.average_chars, 1), + "emoji_ratio": round(u.emoji_ratio, 2), + "night_ratio": round(u.night_ratio, 2), + "reply_ratio": round(u.reply_ratio, 2), + } + for u in sorted_users[:limit] + ] + + def _calculate_emoji_statistics( + self, messages: List[UnifiedMessage] + ) -> EmojiStatistics: + """Calculate emoji usage statistics from messages.""" + standard_count = 0 + custom_count = 0 + animated_count = 0 + sticker_count = 0 + other_count = 0 + emoji_details: Dict[str, int] = {} + + for msg in messages: + for content in msg.contents: + if content.type.value == "emoji": + emoji_id = content.metadata.get("emoji_id", "unknown") + emoji_details[emoji_id] = emoji_details.get(emoji_id, 0) + 1 + + emoji_type = content.metadata.get("emoji_type", "standard") + if emoji_type == "standard": + standard_count += 1 + elif emoji_type == "custom": + custom_count += 1 + elif emoji_type == "animated": + animated_count += 1 + elif emoji_type == "sticker": + sticker_count += 1 + else: + other_count += 1 + + return EmojiStatistics( + standard_emoji_count=standard_count, + custom_emoji_count=custom_count, + animated_emoji_count=animated_count, + sticker_count=sticker_count, + other_emoji_count=other_count, + emoji_details=tuple(emoji_details.items()), + ) + + 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) + 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] + ] + + return ActivityVisualization( + hourly_activity=tuple(hourly.items()), + 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 + ) + + 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" + + # Find peak hour + peak_hour = max(hourly, key=hourly.get) + + # Categorize time periods + if 6 <= peak_hour < 12: + return "Morning (6:00-12:00)" + elif 12 <= peak_hour < 18: + return "Afternoon (12:00-18:00)" + elif 18 <= peak_hour < 24: + return "Evening (18:00-24:00)" + else: + return "Late Night (0:00-6:00)" diff --git a/src/domain/value_objects/__init__.py b/src/domain/value_objects/__init__.py index b7c0182..3085ae0 100644 --- a/src/domain/value_objects/__init__.py +++ b/src/domain/value_objects/__init__.py @@ -2,13 +2,37 @@ from .unified_message import UnifiedMessage, MessageContent, MessageContentType from .platform_capabilities import PlatformCapabilities, PLATFORM_CAPABILITIES from .unified_group import UnifiedGroup, UnifiedMember +from .topic import Topic, TopicCollection +from .user_title import UserTitle, UserTitleCollection +from .golden_quote import GoldenQuote, GoldenQuoteCollection +from .statistics import ( + TokenUsage, + EmojiStatistics, + ActivityVisualization, + GroupStatistics, + UserStatistics, +) __all__ = [ + # Core platform abstractions "UnifiedMessage", - "MessageContent", + "MessageContent", "MessageContentType", "PlatformCapabilities", "PLATFORM_CAPABILITIES", "UnifiedGroup", "UnifiedMember", + # Analysis value objects + "Topic", + "TopicCollection", + "UserTitle", + "UserTitleCollection", + "GoldenQuote", + "GoldenQuoteCollection", + # Statistics + "TokenUsage", + "EmojiStatistics", + "ActivityVisualization", + "GroupStatistics", + "UserStatistics", ] diff --git a/src/domain/value_objects/golden_quote.py b/src/domain/value_objects/golden_quote.py new file mode 100644 index 0000000..29286af --- /dev/null +++ b/src/domain/value_objects/golden_quote.py @@ -0,0 +1,137 @@ +""" +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 +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). + + 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: str + sender: str + reason: str = "" + user_id: str = "" + + def __post_init__(self): + """Validate and normalize golden quote data after initialization.""" + # Ensure user_id is always a string + 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. + + Args: + data: Dictionary with golden quote data + + Returns: + GoldenQuote instance + """ + # Handle both 'qq' and 'user_id' keys for backward compatibility + user_id = data.get("user_id", data.get("qq", "")) + + return cls( + content=data.get("content", "").strip(), + sender=data.get("sender", "").strip(), + reason=data.get("reason", "").strip(), + user_id=str(user_id) if user_id else "", + ) + + def to_dict(self) -> dict: + """ + Convert GoldenQuote to dictionary. + + 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 + } + + @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.""" + try: + return int(self.user_id) + except (ValueError, TypeError): + return 0 + + def with_user_id(self, user_id: str) -> "GoldenQuote": + """ + Create a new GoldenQuote with updated user_id. + + Since GoldenQuote is frozen, we need to create a new instance. + + Args: + user_id: The user ID to set + + Returns: + New GoldenQuote instance with updated user_id + """ + return GoldenQuote( + content=self.content, + sender=self.sender, + reason=self.reason, + user_id=str(user_id), + ) + + +@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: + return len(self.quotes) + + def __iter__(self): + return iter(self.quotes) diff --git a/src/domain/value_objects/statistics.py b/src/domain/value_objects/statistics.py new file mode 100644 index 0000000..6c5c36a --- /dev/null +++ b/src/domain/value_objects/statistics.py @@ -0,0 +1,317 @@ +""" +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 +from typing import Dict, List + + +@dataclass(frozen=True) +class TokenUsage: + """ + Token usage statistics for LLM API calls. + + Immutable by design (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: int = 0 + completion_tokens: int = 0 + total_tokens: int = 0 + + @classmethod + def from_dict(cls, data: dict) -> "TokenUsage": + """Create TokenUsage from dictionary.""" + return cls( + prompt_tokens=data.get("prompt_tokens", 0), + completion_tokens=data.get("completion_tokens", 0), + total_tokens=data.get("total_tokens", 0), + ) + + def to_dict(self) -> dict: + """Convert to dictionary.""" + return { + "prompt_tokens": self.prompt_tokens, + "completion_tokens": self.completion_tokens, + "total_tokens": self.total_tokens, + } + + def __add__(self, other: "TokenUsage") -> "TokenUsage": + """Add two TokenUsage objects together.""" + if not isinstance(other, TokenUsage): + return NotImplemented + return TokenUsage( + prompt_tokens=self.prompt_tokens + other.prompt_tokens, + completion_tokens=self.completion_tokens + other.completion_tokens, + total_tokens=self.total_tokens + other.total_tokens, + ) + + +@dataclass(frozen=True) +class EmojiStatistics: + """ + Emoji usage statistics. + + Platform-agnostic representation of emoji usage in messages. + Immutable by design (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: int = 0 + custom_emoji_count: int = 0 + animated_emoji_count: int = 0 + sticker_count: int = 0 + other_emoji_count: int = 0 + emoji_details: tuple = field(default_factory=tuple) + + @property + def total_count(self) -> int: + """Get total emoji count.""" + return ( + self.standard_emoji_count + + self.custom_emoji_count + + self.animated_emoji_count + + self.sticker_count + + self.other_emoji_count + ) + + @classmethod + def from_dict(cls, data: dict) -> "EmojiStatistics": + """Create EmojiStatistics from dictionary.""" + details = data.get("face_details", data.get("emoji_details", {})) + if isinstance(details, dict): + details = tuple(details.items()) + + return cls( + standard_emoji_count=data.get("face_count", data.get("standard_emoji_count", 0)), + custom_emoji_count=data.get("mface_count", data.get("custom_emoji_count", 0)), + animated_emoji_count=data.get("bface_count", data.get("animated_emoji_count", 0)), + sticker_count=data.get("sface_count", data.get("sticker_count", 0)), + other_emoji_count=data.get("other_emoji_count", 0), + emoji_details=details, + ) + + def to_dict(self) -> dict: + """Convert to dictionary.""" + return { + "standard_emoji_count": self.standard_emoji_count, + "custom_emoji_count": self.custom_emoji_count, + "animated_emoji_count": self.animated_emoji_count, + "sticker_count": self.sticker_count, + "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, + "sface_count": self.sticker_count, + } + + +@dataclass(frozen=True) +class ActivityVisualization: + """ + Activity visualization data. + + Platform-agnostic representation of chat activity patterns. + Immutable by design (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: tuple = field(default_factory=tuple) + daily_activity: tuple = field(default_factory=tuple) + user_activity_ranking: tuple = field(default_factory=tuple) + peak_hours: tuple = field(default_factory=tuple) + heatmap_data: tuple = field(default_factory=tuple) + + @classmethod + def from_dict(cls, data: dict) -> "ActivityVisualization": + """Create ActivityVisualization from dictionary.""" + hourly = data.get("hourly_activity", {}) + daily = data.get("daily_activity", {}) + ranking = data.get("user_activity_ranking", []) + peaks = data.get("peak_hours", []) + heatmap = data.get("activity_heatmap_data", data.get("heatmap_data", {})) + + return cls( + hourly_activity=tuple(hourly.items()) if isinstance(hourly, dict) else tuple(hourly), + daily_activity=tuple(daily.items()) if isinstance(daily, dict) else tuple(daily), + user_activity_ranking=tuple(ranking), + peak_hours=tuple(peaks), + heatmap_data=tuple(heatmap.items()) if isinstance(heatmap, dict) else tuple(heatmap), + ) + + def to_dict(self) -> dict: + """Convert to dictionary.""" + return { + "hourly_activity": dict(self.hourly_activity), + "daily_activity": dict(self.daily_activity), + "user_activity_ranking": list(self.user_activity_ranking), + "peak_hours": list(self.peak_hours), + "activity_heatmap_data": dict(self.heatmap_data), + } + + +@dataclass(frozen=True) +class GroupStatistics: + """ + Comprehensive group chat statistics. + + Platform-agnostic representation of group chat statistics. + Immutable by design (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: int = 0 + total_characters: int = 0 + participant_count: int = 0 + most_active_period: str = "" + emoji_statistics: EmojiStatistics = field(default_factory=EmojiStatistics) + activity_visualization: ActivityVisualization = field(default_factory=ActivityVisualization) + token_usage: TokenUsage = field(default_factory=TokenUsage) + + @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.""" + 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), + } + + activity_data = data.get("activity_visualization", {}) + token_data = data.get("token_usage", {}) + + return cls( + message_count=data.get("message_count", 0), + total_characters=data.get("total_characters", 0), + participant_count=data.get("participant_count", 0), + most_active_period=data.get("most_active_period", ""), + emoji_statistics=EmojiStatistics.from_dict(emoji_data), + activity_visualization=ActivityVisualization.from_dict(activity_data), + token_usage=TokenUsage.from_dict(token_data), + ) + + 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_statistics": self.emoji_statistics.to_dict(), + "activity_visualization": self.activity_visualization.to_dict(), + "token_usage": self.token_usage.to_dict(), + } + + +@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: str + nickname: str = "" + message_count: int = 0 + char_count: int = 0 + emoji_count: int = 0 + reply_count: int = 0 + hours: Dict[int, int] = field(default_factory=lambda: {h: 0 for h in range(24)}) + + @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).""" + if self.message_count == 0: + return 0.0 + night_messages = sum(self.hours.get(h, 0) for h in range(6)) + return night_messages / self.message_count + + @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, + "message_count": self.message_count, + "char_count": self.char_count, + "emoji_count": self.emoji_count, + "reply_count": self.reply_count, + "avg_chars": round(self.average_chars, 1), + "emoji_ratio": round(self.emoji_ratio, 2), + "night_ratio": round(self.night_ratio, 2), + "reply_ratio": round(self.reply_ratio, 2), + "hours": self.hours, + } diff --git a/src/domain/value_objects/topic.py b/src/domain/value_objects/topic.py new file mode 100644 index 0000000..cf37c52 --- /dev/null +++ b/src/domain/value_objects/topic.py @@ -0,0 +1,112 @@ +""" +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 +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). + + Attributes: + name: Topic title/name + contributors: List of usernames who participated in this topic + detail: Detailed description or summary of the topic discussion + """ + + name: str + contributors: tuple[str, ...] = field(default_factory=tuple) + 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") + + # Ensure contributors is a tuple for immutability + 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. + + Args: + data: Dictionary with topic data + + Returns: + Topic instance + """ + contributors = data.get("contributors", []) + if isinstance(contributors, list): + contributors = tuple(contributors) + + return cls( + name=data.get("topic", data.get("name", "")).strip(), + contributors=contributors, + detail=data.get("detail", "").strip(), + ) + + def to_dict(self) -> dict: + """ + Convert Topic to dictionary. + + Returns: + Dictionary representation + """ + return { + "topic": self.name, + "contributors": list(self.contributors), + "detail": self.detail, + } + + @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: + return len(self.topics) + + def __iter__(self): + return iter(self.topics) diff --git a/src/domain/value_objects/user_title.py b/src/domain/value_objects/user_title.py new file mode 100644 index 0000000..a088505 --- /dev/null +++ b/src/domain/value_objects/user_title.py @@ -0,0 +1,134 @@ +""" +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 +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). + + 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: str + user_id: str + title: str + mbti: str = "" + reason: str = "" + + def __post_init__(self): + """Validate and normalize user title data after initialization.""" + # Ensure user_id is always a string + 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. + + Args: + data: Dictionary with user title data + + Returns: + UserTitle instance + """ + # Handle both 'qq' and 'user_id' keys for backward compatibility + user_id = data.get("user_id", data.get("qq", "")) + + return cls( + name=data.get("name", "").strip(), + user_id=str(user_id), + title=data.get("title", "").strip(), + mbti=data.get("mbti", "").strip().upper(), + reason=data.get("reason", "").strip(), + ) + + def to_dict(self) -> dict: + """ + Convert UserTitle to dictionary. + + 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 + "title": self.title, + "mbti": self.mbti, + "reason": self.reason, + } + + @property + def is_valid(self) -> bool: + """Check if user title has valid data.""" + return bool( + self.name + and self.name.strip() + and self.title + and self.title.strip() + and self.user_id + ) + + @property + def qq(self) -> int: + """Get QQ number for backward compatibility.""" + try: + return int(self.user_id) + except (ValueError, TypeError): + return 0 + + +@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.""" + user_id_str = str(user_id) + for title in self.titles: + if title.user_id == user_id_str: + return title + 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: + return len(self.titles) + + def __iter__(self): + return iter(self.titles) diff --git a/src/infrastructure/__init__.py b/src/infrastructure/__init__.py index cd05840..17737ec 100644 --- a/src/infrastructure/__init__.py +++ b/src/infrastructure/__init__.py @@ -1 +1,24 @@ # Infrastructure Layer +from .platform import PlatformAdapter, PlatformAdapterFactory, OneBotAdapter +from .persistence import HistoryRepository +from .llm import LLMClient +from .config import ConfigManager +from .resilience import CircuitBreaker, RateLimiter, retry_async, RetryConfig + +__all__ = [ + # Platform + "PlatformAdapter", + "PlatformAdapterFactory", + "OneBotAdapter", + # Persistence + "HistoryRepository", + # LLM + "LLMClient", + # Config + "ConfigManager", + # Resilience + "CircuitBreaker", + "RateLimiter", + "retry_async", + "RetryConfig", +] diff --git a/src/infrastructure/config/__init__.py b/src/infrastructure/config/__init__.py new file mode 100644 index 0000000..e814246 --- /dev/null +++ b/src/infrastructure/config/__init__.py @@ -0,0 +1,7 @@ +""" +Config Module - Configuration management +""" + +from .config_manager import ConfigManager + +__all__ = ["ConfigManager"] diff --git a/src/infrastructure/config/config_manager.py b/src/infrastructure/config/config_manager.py new file mode 100644 index 0000000..12b60be --- /dev/null +++ b/src/infrastructure/config/config_manager.py @@ -0,0 +1,239 @@ +""" +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 + +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 + """ + 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 + + Returns: + Configuration value or default + """ + try: + keys = key.split(".") + value = self._config + for k in keys: + if isinstance(value, dict): + value = value.get(k) + else: + return default + if value is None: + return default + return value + except Exception: + return default + + def set(self, key: str, value: Any) -> None: + """ + Set a configuration value. + + Args: + key: Configuration key + value: Value to set + """ + keys = key.split(".") + config = self._config + for k in keys[:-1]: + if k not in config: + config[k] = {} + config = config[k] + config[keys[-1]] = value + + # ======================================================================== + # Group Configuration + # ======================================================================== + + def get_enabled_groups(self) -> List[str]: + """Get list of enabled group IDs.""" + 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 + + def get_bot_qq_ids(self) -> List[str]: + """Get list of bot QQ IDs to filter out.""" + 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 + # ======================================================================== + + def get_topic_provider_id(self) -> Optional[str]: + """Get provider ID for topic analysis.""" + return self.get("topic_provider_id") + + def get_user_title_provider_id(self) -> Optional[str]: + """Get provider ID for user title analysis.""" + return self.get("user_title_provider_id") + + def get_golden_quote_provider_id(self) -> Optional[str]: + """Get provider ID for golden quote analysis.""" + return self.get("golden_quote_provider_id") + + def get_topic_max_tokens(self) -> int: + """Get max tokens for topic analysis.""" + return int(self.get("topic_max_tokens", 2000)) + + def get_user_title_max_tokens(self) -> int: + """Get max tokens for user title analysis.""" + return int(self.get("user_title_max_tokens", 2000)) + + def get_golden_quote_max_tokens(self) -> int: + """Get max tokens for golden quote analysis.""" + 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).""" + 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).""" + 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 + """ + 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") + + 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") + + 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") + + # 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)") + except ValueError: + errors.append("analysis_time must be in HH:MM format") + + return errors diff --git a/src/infrastructure/llm/__init__.py b/src/infrastructure/llm/__init__.py new file mode 100644 index 0000000..845fc7a --- /dev/null +++ b/src/infrastructure/llm/__init__.py @@ -0,0 +1,7 @@ +""" +LLM Module - LLM client implementations +""" + +from .llm_client import LLMClient + +__all__ = ["LLMClient"] diff --git a/src/infrastructure/llm/llm_client.py b/src/infrastructure/llm/llm_client.py new file mode 100644 index 0000000..7c1f433 --- /dev/null +++ b/src/infrastructure/llm/llm_client.py @@ -0,0 +1,186 @@ +""" +LLM Client - Wrapper for AstrBot's LLM provider system + +This module provides a clean interface to AstrBot's LLM capabilities, +abstracting away the provider management details. +""" + +from typing import Any, Dict, List, Optional, Tuple + +from astrbot.api import logger + +from ...domain.value_objects.statistics import TokenUsage +from ...domain.exceptions import LLMException, LLMRateLimitException + + +class LLMClient: + """ + Client for interacting with LLM providers. + + This class wraps AstrBot's provider system and provides + a clean interface for making LLM calls. + """ + + def __init__(self, context: Any): + """ + Initialize the LLM client. + + Args: + context: AstrBot plugin context with provider access + """ + 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. + + Args: + provider_id: Specific provider ID, or None for default + + Returns: + Provider instance + + Raises: + LLMException: If provider not found + """ + try: + if provider_id and provider_id in self._provider_cache: + return self._provider_cache[provider_id] + + 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") + provider = providers[0] + + if provider: + self._provider_cache[provider_id or "default"] = provider + + return provider + + except Exception as e: + raise LLMException(f"Failed to get provider: {e}") + + async def chat_completion( + self, + prompt: str, + provider_id: Optional[str] = None, + max_tokens: int = 2000, + temperature: float = 0.7, + 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 + + Returns: + Tuple of (response_text, token_usage) + + Raises: + LLMException: If the request fails + """ + try: + provider = self.get_provider(provider_id) + if not provider: + raise LLMException("No provider available", 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 + ) + + # Extract response text + if hasattr(response, "completion_text"): + response_text = response.completion_text + elif isinstance(response, dict): + response_text = response.get("completion_text", response.get("text", "")) + else: + response_text = str(response) + + # Extract token usage + token_usage = TokenUsage() + if hasattr(response, "usage"): + usage = response.usage + if hasattr(usage, "prompt_tokens"): + token_usage = TokenUsage( + prompt_tokens=usage.prompt_tokens or 0, + completion_tokens=usage.completion_tokens or 0, + total_tokens=usage.total_tokens or 0, + ) + + return response_text, token_usage + + except Exception as e: + 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") + + async def analyze_with_json_output( + self, + prompt: str, + provider_id: Optional[str] = None, + max_tokens: int = 2000, + temperature: float = 0.7, + ) -> Tuple[str, TokenUsage]: + """ + Make a completion request expecting JSON output. + + Args: + prompt: The analysis prompt + provider_id: Specific provider ID (optional) + max_tokens: Maximum tokens in response + temperature: Sampling temperature + + Returns: + Tuple of (response_text, token_usage) + """ + # Add JSON instruction to prompt if not present + json_instruction = "\nRespond with valid JSON only." + if "json" not in prompt.lower(): + prompt = prompt + json_instruction + + return await self.chat_completion( + prompt=prompt, + provider_id=provider_id, + max_tokens=max_tokens, + temperature=temperature, + ) + + def list_available_providers(self) -> List[Dict[str, str]]: + """ + List all available LLM providers. + + Returns: + List of provider info dictionaries + """ + try: + providers = self.context.get_all_providers() + return [ + { + "id": getattr(p, "id", str(i)), + "name": getattr(p, "name", f"Provider {i}"), + "type": getattr(p, "type", "unknown"), + } + for i, p in enumerate(providers) + ] + except Exception as e: + logger.error(f"Failed to list providers: {e}") + return [] diff --git a/src/infrastructure/persistence/__init__.py b/src/infrastructure/persistence/__init__.py new file mode 100644 index 0000000..1e7e9d3 --- /dev/null +++ b/src/infrastructure/persistence/__init__.py @@ -0,0 +1,7 @@ +""" +Persistence Module - Data storage implementations +""" + +from .history_repository import HistoryRepository + +__all__ = ["HistoryRepository"] diff --git a/src/infrastructure/persistence/history_repository.py b/src/infrastructure/persistence/history_repository.py new file mode 100644 index 0000000..f7155b2 --- /dev/null +++ b/src/infrastructure/persistence/history_repository.py @@ -0,0 +1,212 @@ +""" +History Repository - Implementation for storing analysis history + +This module provides persistent storage for analysis results and history. +It wraps the existing history_manager functionality. +""" + +import json +import os +from datetime import datetime +from pathlib import Path +from typing import Any, Dict, List, Optional + +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. + """ + + def __init__(self, data_dir: str): + """ + Initialize the history repository. + + Args: + data_dir: Base directory for storing history data + """ + 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( + self, + group_id: str, + result: Dict[str, Any], + 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) + + Returns: + True if saved successfully + """ + 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}") + return True + + except Exception as e: + logger.error(f"Failed to save analysis result: {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 + + Returns: + History dictionary + """ + try: + history_path = self._get_group_history_path(group_id) + if history_path.exists(): + with open(history_path, "r", encoding="utf-8") as f: + return json.load(f) + return {"daily": {}, "group_id": group_id} + except Exception as e: + logger.error(f"Failed to load group history: {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) + + Returns: + Analysis result or None if not found + """ + history = self.load_group_history(group_id) + return history.get("daily", {}).get(date_str) + + def get_recent_results( + 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 + + 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) + + Returns: + True if analysis exists + """ + 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 + + 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) + 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: + del daily[date] + + if dates_to_delete: + history["daily"] = daily + 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) + + return len(dates_to_delete) + + except Exception as e: + logger.error(f"Failed to delete old history: {e}") + return 0 + + def list_groups_with_history(self) -> List[str]: + """ + List all groups that have history. + + Returns: + List of group IDs + """ + try: + groups = [] + for file_path in self.history_dir.glob("group_*.json"): + group_id = file_path.stem.replace("group_", "") + groups.append(group_id) + return groups + except Exception as e: + logger.error(f"Failed to list groups: {e}") + return [] diff --git a/src/infrastructure/platform/__init__.py b/src/infrastructure/platform/__init__.py index 67ac024..cc5a9f6 100644 --- a/src/infrastructure/platform/__init__.py +++ b/src/infrastructure/platform/__init__.py @@ -1,5 +1,6 @@ # Platform Adapters from .factory import PlatformAdapterFactory from .base import PlatformAdapter +from .adapters.onebot_adapter import OneBotAdapter -__all__ = ["PlatformAdapterFactory", "PlatformAdapter"] +__all__ = ["PlatformAdapterFactory", "PlatformAdapter", "OneBotAdapter"] diff --git a/src/infrastructure/resilience/__init__.py b/src/infrastructure/resilience/__init__.py new file mode 100644 index 0000000..bd22ef4 --- /dev/null +++ b/src/infrastructure/resilience/__init__.py @@ -0,0 +1,15 @@ +""" +Resilience Module - Circuit breaker, rate limiter, and retry utilities +""" + +from .circuit_breaker import CircuitBreaker, CircuitState +from .rate_limiter import RateLimiter +from .retry import retry_async, RetryConfig + +__all__ = [ + "CircuitBreaker", + "CircuitState", + "RateLimiter", + "retry_async", + "RetryConfig", +] diff --git a/src/infrastructure/resilience/circuit_breaker.py b/src/infrastructure/resilience/circuit_breaker.py new file mode 100644 index 0000000..9bd84c0 --- /dev/null +++ b/src/infrastructure/resilience/circuit_breaker.py @@ -0,0 +1,138 @@ +""" +Circuit Breaker - Prevents cascading failures + +Implements the circuit breaker pattern to prevent repeated calls +to failing services. +""" + +import time +from dataclasses import dataclass, field +from enum import Enum +from typing import Callable, Optional + +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 + + +@dataclass +class CircuitBreaker: + """ + Circuit breaker implementation. + + Prevents cascading failures by tracking failure rates and + temporarily blocking calls to failing services. + """ + + name: str + failure_threshold: int = 5 + 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) + _last_failure_time: float = field(default=0, init=False) + _half_open_calls: int = field(default=0, init=False) + + @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 + + if new_state == CircuitState.CLOSED: + self._failure_count = 0 + self._success_count = 0 + elif new_state == CircuitState.HALF_OPEN: + self._half_open_calls = 0 + + logger.debug(f"Circuit {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() + + if self._state == CircuitState.HALF_OPEN: + self._transition_to(CircuitState.OPEN) + elif self._state == CircuitState.CLOSED: + if self._failure_count >= self.failure_threshold: + 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 + + if state == CircuitState.CLOSED: + return True + elif state == CircuitState.OPEN: + return False + elif state == CircuitState.HALF_OPEN: + self._half_open_calls += 1 + return self._half_open_calls <= self.half_open_max_calls + + return False + + def reset(self) -> None: + """Reset the circuit breaker to closed state.""" + self._transition_to(CircuitState.CLOSED) + + async def execute( + self, + func: Callable, + *args, + fallback: Optional[Callable] = None, + **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 + + Returns: + Function result or fallback result + + Raises: + Exception: If circuit is open and no fallback provided + """ + if not self.can_execute(): + if fallback: + return await fallback(*args, **kwargs) + raise Exception(f"Circuit {self.name} is open") + + try: + result = await func(*args, **kwargs) + self.record_success() + return result + except Exception as e: + self.record_failure() + raise diff --git a/src/infrastructure/resilience/rate_limiter.py b/src/infrastructure/resilience/rate_limiter.py new file mode 100644 index 0000000..8a1af9b --- /dev/null +++ b/src/infrastructure/resilience/rate_limiter.py @@ -0,0 +1,142 @@ +""" +Rate Limiter - Controls request rates + +Implements token bucket rate limiting to prevent overwhelming services. +""" + +import asyncio +import time +from dataclasses import dataclass, field +from typing import Optional + +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) + + # 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) + self._last_update = now + + 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) + + Returns: + True if tokens acquired, False if timeout + """ + start_time = time.time() + + async with self._lock: + while True: + self._refill() + + if self._tokens >= tokens: + self._tokens -= tokens + return True + + if timeout is not None: + elapsed = time.time() - start_time + if elapsed >= timeout: + return False + + # Calculate wait time for enough tokens + tokens_needed = tokens - self._tokens + wait_time = tokens_needed / self.rate + + if timeout is not None: + remaining = timeout - (time.time() - start_time) + wait_time = min(wait_time, remaining) + + if wait_time > 0: + await asyncio.sleep(wait_time) + + def try_acquire(self, tokens: int = 1) -> bool: + """ + Try to acquire tokens without waiting. + + Args: + tokens: Number of tokens to acquire + + Returns: + True if tokens acquired, False otherwise + """ + self._refill() + + if self._tokens >= tokens: + self._tokens -= tokens + return True + return False + + @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): + self._limiters: dict[str, RateLimiter] = {} + + def get_or_create( + self, + name: str, + rate: float = 1.0, + burst: int = 5, + ) -> RateLimiter: + """ + Get or create a rate limiter. + + Args: + name: Limiter name + rate: Tokens per second + burst: Maximum burst size + + Returns: + RateLimiter instance + """ + 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() diff --git a/src/infrastructure/resilience/retry.py b/src/infrastructure/resilience/retry.py new file mode 100644 index 0000000..09341ec --- /dev/null +++ b/src/infrastructure/resilience/retry.py @@ -0,0 +1,176 @@ +""" +Retry - Retry utilities with exponential backoff + +Provides retry decorators and utilities for handling transient failures. +""" + +import asyncio +import random +from dataclasses import dataclass +from functools import wraps +from typing import Callable, Optional, Tuple, Type, Union + +from astrbot.api import logger + + +@dataclass +class RetryConfig: + """Configuration for retry behavior.""" + + max_attempts: int = 3 + base_delay: float = 1.0 + max_delay: float = 60.0 + exponential_base: float = 2.0 + jitter: bool = True + retry_exceptions: Tuple[Type[Exception], ...] = (Exception,) + + +def calculate_delay( + attempt: int, + base_delay: float, + max_delay: float, + exponential_base: float, + 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 + + Returns: + Delay in seconds + """ + delay = base_delay * (exponential_base**attempt) + delay = min(delay, max_delay) + + if jitter: + delay = delay * (0.5 + random.random()) + + return delay + + +def retry_async( + max_attempts: int = 3, + base_delay: float = 1.0, + max_delay: float = 60.0, + exponential_base: float = 2.0, + jitter: bool = True, + retry_exceptions: Tuple[Type[Exception], ...] = (Exception,), + 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) + + Returns: + Decorated function + """ + + def decorator(func: Callable): + @wraps(func) + async def wrapper(*args, **kwargs): + last_exception = None + + for attempt in range(max_attempts): + try: + return await func(*args, **kwargs) + except retry_exceptions as e: + last_exception = e + + if attempt < max_attempts - 1: + delay = calculate_delay( + attempt, base_delay, max_delay, exponential_base, jitter + ) + + if on_retry: + on_retry(e, attempt + 1) + + logger.debug( + f"Retry {attempt + 1}/{max_attempts} for {func.__name__} " + f"after {delay:.2f}s: {e}" + ) + await asyncio.sleep(delay) + else: + logger.warning( + f"All {max_attempts} attempts failed for {func.__name__}: {e}" + ) + + raise last_exception + + return wrapper + + return decorator + + +class RetryExecutor: + """ + Executor for running functions with retry logic. + """ + + def __init__(self, config: Optional[RetryConfig] = None): + """ + Initialize the retry executor. + + Args: + config: Retry configuration + """ + self.config = config or RetryConfig() + + async def execute( + self, + func: Callable, + *args, + config: Optional[RetryConfig] = None, + **kwargs, + ): + """ + Execute a function with retry logic. + + Args: + func: Async function to execute + *args: Function arguments + config: Optional override config + **kwargs: Function keyword arguments + + Returns: + Function result + + Raises: + Exception: If all retries fail + """ + cfg = config or self.config + last_exception = None + + for attempt in range(cfg.max_attempts): + try: + return await func(*args, **kwargs) + except cfg.retry_exceptions as e: + last_exception = e + + if attempt < cfg.max_attempts - 1: + delay = calculate_delay( + attempt, + cfg.base_delay, + cfg.max_delay, + cfg.exponential_base, + cfg.jitter, + ) + logger.debug( + f"Retry {attempt + 1}/{cfg.max_attempts} after {delay:.2f}s: {e}" + ) + await asyncio.sleep(delay) + + raise last_exception diff --git a/src/shared/__init__.py b/src/shared/__init__.py new file mode 100644 index 0000000..ec718e0 --- /dev/null +++ b/src/shared/__init__.py @@ -0,0 +1,11 @@ +""" +Shared Module - Common utilities and constants +""" + +from .constants import * +from .trace_context import TraceContext + +__all__ = [ + "TraceContext", + # Constants are exported via * +] diff --git a/src/shared/constants.py b/src/shared/constants.py new file mode 100644 index 0000000..901b190 --- /dev/null +++ b/src/shared/constants.py @@ -0,0 +1,91 @@ +""" +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" +PLATFORM_SLACK = "slack" +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), + "evening": (18, 24), + "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" +CONTENT_TYPE_STICKER = "sticker" +CONTENT_TYPE_FILE = "file" +CONTENT_TYPE_AUDIO = "audio" +CONTENT_TYPE_VIDEO = "video" +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 + +# 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 + +# 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" diff --git a/src/shared/trace_context.py b/src/shared/trace_context.py new file mode 100644 index 0000000..2199cdc --- /dev/null +++ b/src/shared/trace_context.py @@ -0,0 +1,161 @@ +""" +Trace Context - Request tracing and correlation + +Provides context for tracking requests across the plugin. +""" + +import uuid +from contextvars import ContextVar +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 +) + + +@dataclass +class TraceContext: + """ + Context for tracing requests through the plugin. + + Provides correlation IDs and timing information for debugging + and monitoring. + """ + + trace_id: str = field(default_factory=lambda: str(uuid.uuid4())[:8]) + group_id: str = "" + platform: str = "" + operation: str = "" + 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 + """ + 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 + + Returns: + Elapsed time in milliseconds + """ + start = self.start_time + if from_checkpoint and from_checkpoint in self._checkpoints: + start = self._checkpoints[from_checkpoint] + + delta = datetime.now() - start + 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, + "platform": self.platform, + "operation": self.operation, + "start_time": self.start_time.isoformat(), + "elapsed_ms": self.elapsed_ms(), + "metadata": self.metadata, + "checkpoints": {k: v.isoformat() for k, v in self._checkpoints.items()}, + } + + 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 + def get_or_create( + cls, + group_id: str = "", + platform: str = "", + operation: str = "", + ) -> "TraceContext": + """ + Get current trace or create a new one. + + Args: + group_id: Group identifier + platform: Platform name + operation: Operation name + + Returns: + TraceContext instance + """ + current = cls.current() + if current: + return current + + return cls( + group_id=group_id, + platform=platform, + operation=operation, + ) + + +def get_trace_id() -> str: + """ + Get current trace ID or generate a new one. + + Returns: + Trace ID string + """ + trace = TraceContext.current() + if trace: + return trace.trace_id + return str(uuid.uuid4())[:8] + + +def with_trace( + group_id: str = "", + platform: str = "", + operation: str = "", +): + """ + Decorator to add trace context to a function. + + Args: + group_id: Group identifier + platform: Platform name + operation: Operation name + + Returns: + Decorated function + """ + + def decorator(func): + async def wrapper(*args, **kwargs): + with TraceContext( + group_id=group_id, + platform=platform, + operation=operation or func.__name__, + ): + return await func(*args, **kwargs) + + return wrapper + + return decorator