85 lines
2.6 KiB
Python
85 lines
2.6 KiB
Python
"""数据管理器 - MySQL数据库存储"""
|
|
|
|
import warnings
|
|
warnings.filterwarnings('ignore', category=DeprecationWarning)
|
|
|
|
from datetime import datetime
|
|
from typing import List, Dict, Optional
|
|
from loguru import logger
|
|
from config import Config
|
|
|
|
# 导入MySQL数据库管理器
|
|
try:
|
|
from db_manager import SiteManager, ClickManager, InteractionManager, StatisticsManager
|
|
logger.info("使用MySQL数据库存储")
|
|
except Exception as e:
|
|
logger.error(f"MySQL数据库初始化失败: {str(e)}")
|
|
raise
|
|
|
|
|
|
class DataManager:
|
|
"""数据管理器 - MySQL数据库存储"""
|
|
|
|
def __init__(self, data_file: str = None):
|
|
"""初始化MySQL数据库管理器"""
|
|
self.site_mgr = SiteManager()
|
|
self.click_mgr = ClickManager()
|
|
self.interaction_mgr = InteractionManager()
|
|
self.stats_mgr = StatisticsManager()
|
|
logger.info("初始化MySQL数据库管理器")
|
|
|
|
|
|
def add_url(self, url: str) -> bool:
|
|
"""添加新URL到MySQL数据库"""
|
|
site_id = self.site_mgr.add_site(
|
|
site_url=url,
|
|
site_name=url,
|
|
site_dimension='MIP广告'
|
|
)
|
|
return site_id is not None
|
|
|
|
def record_click(self, url: str, has_reply: bool = False):
|
|
"""记录一次点击到MySQL数据库"""
|
|
try:
|
|
site = self.site_mgr.get_site_by_url(url)
|
|
if not site:
|
|
logger.error(f"URL不存在: {url}")
|
|
return
|
|
|
|
click_id = self.click_mgr.record_click(
|
|
site_id=site['id'],
|
|
site_url=url
|
|
)
|
|
|
|
if click_id and has_reply:
|
|
self.interaction_mgr.record_interaction(
|
|
site_id=site['id'],
|
|
click_id=click_id,
|
|
interaction_type='reply',
|
|
is_successful=True,
|
|
response_received=True
|
|
)
|
|
|
|
logger.info(f"记录点击成功: {url}, has_reply={has_reply}")
|
|
|
|
except Exception as e:
|
|
logger.error(f"记录点击失败: {str(e)}")
|
|
|
|
|
|
def get_active_urls(self) -> List[Dict]:
|
|
"""获取所有活跃的URL"""
|
|
return self.site_mgr.get_active_sites()
|
|
|
|
def get_url_info(self, url: str) -> Optional[Dict]:
|
|
"""获取URL详细信息"""
|
|
return self.site_mgr.get_site_by_url(url)
|
|
|
|
def get_all_urls(self) -> List[Dict]:
|
|
"""获取所有URL"""
|
|
return self.site_mgr.get_all_sites()
|
|
|
|
def get_statistics(self) -> Dict:
|
|
"""获取统计数据"""
|
|
return self.stats_mgr.get_statistics()
|
|
|