""" 发布服务 (基于 social-auto-upload 架构) """ from datetime import datetime from pathlib import Path from typing import Optional, List from loguru import logger from app.core.config import settings # Import platform uploaders from .uploader.bilibili_uploader import BilibiliUploader from .uploader.douyin_uploader import DouyinUploader from .uploader.xiaohongshu_uploader import XiaohongshuUploader class PublishService: """Social media publishing service""" PLATFORMS = { "bilibili": {"name": "B站", "url": "https://member.bilibili.com/platform/upload/video/frame"}, "douyin": {"name": "抖音", "url": "https://creator.douyin.com/"}, "xiaohongshu": {"name": "小红书", "url": "https://creator.xiaohongshu.com/"}, "weixin": {"name": "微信视频号", "url": "https://channels.weixin.qq.com/"}, "kuaishou": {"name": "快手", "url": "https://cp.kuaishou.com/"}, } def __init__(self): self.cookies_dir = settings.BASE_DIR / "cookies" self.cookies_dir.mkdir(exist_ok=True) # 存储活跃的登录会话，用于跟踪登录状态 self.active_login_sessions = {} def get_accounts(self): """Get list of platform accounts with login status""" accounts = [] for pid, pinfo in self.PLATFORMS.items(): cookie_file = self.cookies_dir / f"{pid}_cookies.json" accounts.append({ "platform": pid, "name": pinfo["name"], "logged_in": cookie_file.exists(), "enabled": True }) return accounts async def publish( self, video_path: str, platform: str, title: str, tags: List[str], description: str = "", publish_time: Optional[datetime] = None, **kwargs ): """ Publish video to specified platform Args: video_path: Path to video file platform: Platform ID (bilibili, douyin, etc.) title: Video title tags: List of tags description: Video description publish_time: Scheduled publish time (None = immediate) **kwargs: Additional platform-specific parameters Returns: dict: Publish result """ # Validate platform if platform not in self.PLATFORMS: logger.error(f"[发布] 不支持的平台: {platform}") return { "success": False, "message": f"不支持的平台: {platform}", "platform": platform } # Get account file path account_file = self.cookies_dir / f"{platform}_cookies.json" logger.info(f"[发布] 平台: {self.PLATFORMS[platform]['name']}") logger.info(f"[发布] 视频: {video_path}") logger.info(f"[发布] 标题: {title}") try: # Select appropriate uploader if platform == "bilibili": uploader = BilibiliUploader( title=title, file_path=str(settings.BASE_DIR.parent / video_path), # Convert to absolute path tags=tags, publish_date=publish_time, account_file=str(account_file), description=description, tid=kwargs.get('tid', 122), # Category ID copyright=kwargs.get('copyright', 1) # 1=original ) elif platform == "douyin": uploader = DouyinUploader( title=title, file_path=str(settings.BASE_DIR.parent / video_path), tags=tags, publish_date=publish_time, account_file=str(account_file), description=description ) elif platform == "xiaohongshu": uploader = XiaohongshuUploader( title=title, file_path=str(settings.BASE_DIR.parent / video_path), tags=tags, publish_date=publish_time, account_file=str(account_file), description=description ) else: logger.warning(f"[发布] {platform} 上传功能尚未实现") return { "success": False, "message": f"{self.PLATFORMS[platform]['name']} 上传功能开发中", "platform": platform } # Execute upload result = await uploader.main() result['platform'] = platform return result except Exception as e: logger.exception(f"[发布] 上传异常: {e}") return { "success": False, "message": f"上传异常: {str(e)}", "platform": platform } async def login(self, platform: str): """ 启动QR码登录流程 Returns: dict: 包含二维码base64图片 """ if platform not in self.PLATFORMS: return {"success": False, "message": "不支持的平台"} try: from .qr_login_service import QRLoginService # 创建QR登录服务 qr_service = QRLoginService(platform, self.cookies_dir) # 存储活跃会话 self.active_login_sessions[platform] = qr_service # 启动登录并获取二维码 result = await qr_service.start_login() return result except Exception as e: logger.exception(f"[登录] QR码登录失败: {e}") return { "success": False, "message": f"登录失败: {str(e)}" } def get_login_session_status(self, platform: str): """获取活跃登录会话的状态""" # 1. 如果有活跃的扫码会话，优先检查它 if platform in self.active_login_sessions: qr_service = self.active_login_sessions[platform] status = qr_service.get_login_status() # 如果登录成功且Cookie已保存，清理会话 if status["success"] and status["cookies_saved"]: del self.active_login_sessions[platform] return {"success": True, "message": "登录成功"} return {"success": False, "message": "等待扫码..."} # 2. 如果没有活跃会话，检查本地Cookie文件是否存在 (用于页面初始加载) # 注意：这无法检测Cookie是否过期，只能检测文件在不在 # 在扫码流程中，前端应该依赖上面第1步的返回 cookie_file = self.cookies_dir / f"{platform}_cookies.json" if cookie_file.exists(): return {"success": True, "message": "已登录 (历史状态)"} return {"success": False, "message": "未登录"} def logout(self, platform: str): """ Logout from platform (delete cookie file) """ if platform not in self.PLATFORMS: return {"success": False, "message": "不支持的平台"} try: # 1. 移除活跃会话 if platform in self.active_login_sessions: del self.active_login_sessions[platform] # 2. 删除Cookie文件 cookie_file = self.cookies_dir / f"{platform}_cookies.json" if cookie_file.exists(): cookie_file.unlink() logger.info(f"[登出] {platform} Cookie已删除") return {"success": True, "message": "已注销"} except Exception as e: logger.exception(f"[登出] 失败: {e}") return {"success": False, "message": f"注销失败: {str(e)}"} async def save_cookie_string(self, platform: str, cookie_string: str): """ 保存从客户端浏览器提取的Cookie字符串 Args: platform: 平台ID cookie_string: document.cookie 格式的Cookie字符串 """ try: account_file = self.cookies_dir / f"{platform}_cookies.json" # 解析Cookie字符串 cookie_dict = {} for item in cookie_string.split('; '): if '=' in item: name, value = item.split('=', 1) cookie_dict[name] = value # 对B站进行特殊处理，提取biliup需要的字段 if platform == "bilibili": bilibili_cookies = {} required_fields = ['SESSDATA', 'bili_jct', 'DedeUserID', 'DedeUserID__ckMd5'] for field in required_fields: if field in cookie_dict: bilibili_cookies[field] = cookie_dict[field] if len(bilibili_cookies) < 3: # 至少需要3个关键字段 return { "success": False, "message": "Cookie不完整，请确保已登录" } cookie_dict = bilibili_cookies # 保存Cookie import json with open(account_file, 'w', encoding='utf-8') as f: json.dump(cookie_dict, f, indent=2) logger.success(f"[登录] {platform} Cookie已保存") return { "success": True, "message": f"{self.PLATFORMS[platform]['name']} 登录成功" } except Exception as e: logger.exception(f"[登录] Cookie保存失败: {e}") return { "success": False, "message": f"Cookie保存失败: {str(e)}" }