139 Zeilen
5.3 KiB
Python
139 Zeilen
5.3 KiB
Python
"""
|
|
Platform Service - Business logic for platform management
|
|
"""
|
|
import os
|
|
import yaml
|
|
import logging
|
|
from typing import List, Dict, Any, Optional
|
|
from pathlib import Path
|
|
|
|
from domain.entities.platform import Platform, PlatformStatus
|
|
|
|
logger = logging.getLogger("platform_service")
|
|
|
|
|
|
class PlatformService:
|
|
"""
|
|
Service for managing platform configuration and availability.
|
|
Clean Architecture: Application Service
|
|
"""
|
|
|
|
def __init__(self, db_manager=None):
|
|
self.db_manager = db_manager
|
|
self._platforms_cache = None
|
|
self._config_path = Path(__file__).parent.parent.parent / "config" / "platforms.yaml"
|
|
|
|
def _load_platforms(self) -> Dict[str, Platform]:
|
|
"""Load platforms from configuration file"""
|
|
if self._platforms_cache is not None:
|
|
return self._platforms_cache
|
|
|
|
try:
|
|
with open(self._config_path, 'r', encoding='utf-8') as f:
|
|
config = yaml.safe_load(f)
|
|
|
|
platforms = {}
|
|
for platform_id, platform_data in config.get('platforms', {}).items():
|
|
platform = Platform(
|
|
id=platform_data.get('id', platform_id),
|
|
display_name=platform_data.get('display_name', platform_id.title()),
|
|
enabled=platform_data.get('enabled', False),
|
|
icon=platform_data.get('icon', f"{platform_id}.svg"),
|
|
color=platform_data.get('color', '#000000')
|
|
)
|
|
platforms[platform.id] = platform
|
|
|
|
self._platforms_cache = platforms
|
|
return platforms
|
|
|
|
except FileNotFoundError:
|
|
logger.warning(f"Platform config not found at {self._config_path}, using defaults")
|
|
return self._get_default_platforms()
|
|
except Exception as e:
|
|
logger.error(f"Error loading platform config: {e}")
|
|
return self._get_default_platforms()
|
|
|
|
def _get_default_platforms(self) -> Dict[str, Platform]:
|
|
"""Get default platform configuration"""
|
|
defaults = {
|
|
'instagram': Platform('instagram', 'Instagram', True, 'instagram.svg', '#E4405F'),
|
|
'facebook': Platform('facebook', 'Facebook', True, 'facebook.svg', '#1877F2'),
|
|
'tiktok': Platform('tiktok', 'TikTok', True, 'tiktok.svg', '#000000'),
|
|
'x': Platform('x', 'X', True, 'twitter.svg', '#000000'),
|
|
'gmail': Platform('gmail', 'Gmail', False, 'gmail.svg', '#EA4335'),
|
|
'vk': Platform('vk', 'VK', False, 'vk.svg', '#0077FF'),
|
|
'ok': Platform('ok', 'OK.ru', False, 'ok.svg', '#FF6600')
|
|
}
|
|
return defaults
|
|
|
|
def get_all_platforms(self) -> List[Platform]:
|
|
"""Get all platforms (enabled and disabled)"""
|
|
platforms = self._load_platforms()
|
|
return list(platforms.values())
|
|
|
|
def get_active_platforms(self) -> List[Platform]:
|
|
"""Get only enabled platforms for UI display"""
|
|
platforms = self._load_platforms()
|
|
return [p for p in platforms.values() if p.is_enabled]
|
|
|
|
def get_platform_by_id(self, platform_id: str) -> Optional[Platform]:
|
|
"""Get a specific platform by ID"""
|
|
platforms = self._load_platforms()
|
|
return platforms.get(platform_id.lower())
|
|
|
|
def is_platform_enabled(self, platform_id: str) -> bool:
|
|
"""Check if a platform is enabled"""
|
|
platform = self.get_platform_by_id(platform_id)
|
|
return platform.is_enabled if platform else False
|
|
|
|
def get_filter_platforms(self) -> List[Platform]:
|
|
"""
|
|
Get platforms for sidebar filters.
|
|
Shows:
|
|
- All enabled platforms
|
|
- Disabled platforms that have existing accounts
|
|
"""
|
|
platforms = self._load_platforms()
|
|
filter_platforms = []
|
|
|
|
for platform in platforms.values():
|
|
if platform.is_enabled:
|
|
# Always show enabled platforms
|
|
filter_platforms.append(platform)
|
|
elif self.db_manager:
|
|
# Show disabled platforms only if they have accounts
|
|
try:
|
|
count = self.db_manager.get_account_count(platform.id)
|
|
if count > 0:
|
|
filter_platforms.append(platform)
|
|
except Exception as e:
|
|
logger.error(f"Error checking accounts for {platform.id}: {e}")
|
|
|
|
return filter_platforms
|
|
|
|
def get_platforms_with_accounts(self) -> Dict[str, int]:
|
|
"""
|
|
Get platform IDs with their account counts.
|
|
Returns dict: {platform_id: count}
|
|
"""
|
|
if not self.db_manager:
|
|
return {}
|
|
|
|
platforms = self._load_platforms()
|
|
counts = {}
|
|
|
|
for platform_id in platforms.keys():
|
|
try:
|
|
count = self.db_manager.get_account_count(platform_id)
|
|
if count > 0:
|
|
counts[platform_id] = count
|
|
except Exception as e:
|
|
logger.error(f"Error getting count for {platform_id}: {e}")
|
|
counts[platform_id] = 0
|
|
|
|
return counts
|
|
|
|
def reload_configuration(self):
|
|
"""Force reload of platform configuration"""
|
|
self._platforms_cache = None
|
|
logger.info("Platform configuration reloaded") |