multi-platform-media-manage/server/python/platforms/base.py

# -*- coding: utf-8 -*-
"""
平台发布基类
提供通用的发布接口和工具方法
"""

import asyncio
import json
import os
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime
from typing import List, Optional, Callable, Dict, Any
from playwright.async_api import async_playwright, Browser, BrowserContext, Page


@dataclass
class PublishParams:
    """发布参数"""
    title: str
    video_path: str
    description: str = ""
    cover_path: Optional[str] = None
    tags: List[str] = field(default_factory=list)
    publish_date: Optional[datetime] = None
    location: str = "重庆市"
    
    def __post_init__(self):
        if not self.description:
            self.description = self.title


@dataclass
class PublishResult:
    """发布结果"""
    success: bool
    platform: str
    video_id: str = ""
    video_url: str = ""
    message: str = ""
    error: str = ""
    need_captcha: bool = False  # 是否需要验证码
    captcha_type: str = ""  # 验证码类型: phone, slider, image
    screenshot_base64: str = ""  # 页面截图（Base64）
    page_url: str = ""  # 当前页面 URL
    status: str = ""  # 状态: uploading, processing, success, failed, need_captcha, need_action


@dataclass
class WorkItem:
    """作品数据"""
    work_id: str
    title: str
    cover_url: str = ""
    video_url: str = ""
    duration: int = 0  # 秒
    status: str = "published"  # published, reviewing, rejected, draft
    publish_time: str = ""
    play_count: int = 0
    like_count: int = 0
    comment_count: int = 0
    share_count: int = 0
    collect_count: int = 0
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "work_id": self.work_id,
            "title": self.title,
            "cover_url": self.cover_url,
            "video_url": self.video_url,
            "duration": self.duration,
            "status": self.status,
            "publish_time": self.publish_time,
            "play_count": self.play_count,
            "like_count": self.like_count,
            "comment_count": self.comment_count,
            "share_count": self.share_count,
            "collect_count": self.collect_count,
        }


@dataclass
class CommentItem:
    """评论数据"""
    comment_id: str
    work_id: str
    content: str
    author_id: str = ""
    author_name: str = ""
    author_avatar: str = ""
    like_count: int = 0
    reply_count: int = 0
    create_time: str = ""
    is_author: bool = False  # 是否是作者的评论
    replies: List['CommentItem'] = field(default_factory=list)
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "comment_id": self.comment_id,
            "work_id": self.work_id,
            "content": self.content,
            "author_id": self.author_id,
            "author_name": self.author_name,
            "author_avatar": self.author_avatar,
            "like_count": self.like_count,
            "reply_count": self.reply_count,
            "create_time": self.create_time,
            "is_author": self.is_author,
            "replies": [r.to_dict() for r in self.replies],
        }


@dataclass
class WorksResult:
    """作品列表结果"""
    success: bool
    platform: str
    works: List[WorkItem] = field(default_factory=list)
    total: int = 0
    has_more: bool = False
    error: str = ""
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "success": self.success,
            "platform": self.platform,
            "works": [w.to_dict() for w in self.works],
            "total": self.total,
            "has_more": self.has_more,
            "error": self.error,
        }


@dataclass
class CommentsResult:
    """评论列表结果"""
    success: bool
    platform: str
    work_id: str
    comments: List[CommentItem] = field(default_factory=list)
    total: int = 0
    has_more: bool = False
    error: str = ""
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "success": self.success,
            "platform": self.platform,
            "work_id": self.work_id,
            "comments": [c.to_dict() for c in self.comments],
            "total": self.total,
            "has_more": self.has_more,
            "error": self.error,
        }


class BasePublisher(ABC):
    """
    平台发布基类
    所有平台发布器都需要继承此类
    """
    
    platform_name: str = "base"
    login_url: str = ""
    publish_url: str = ""
    cookie_domain: str = ""
    
    def __init__(self, headless: bool = True):
        self.headless = headless
        self.browser: Optional[Browser] = None
        self.context: Optional[BrowserContext] = None
        self.page: Optional[Page] = None
        self.on_progress: Optional[Callable[[int, str], None]] = None
    
    def set_progress_callback(self, callback: Callable[[int, str], None]):
        """设置进度回调"""
        self.on_progress = callback
    
    def report_progress(self, progress: int, message: str):
        """报告进度"""
        print(f"[{self.platform_name}] [{progress}%] {message}")
        if self.on_progress:
            self.on_progress(progress, message)
    
    @staticmethod
    def parse_cookies(cookies_str: str) -> list:
        """解析 cookie 字符串为列表"""
        try:
            cookies = json.loads(cookies_str)
            if isinstance(cookies, list):
                return cookies
        except json.JSONDecodeError:
            pass
        
        # 字符串格式: name=value; name2=value2
        cookies = []
        for item in cookies_str.split(';'):
            item = item.strip()
            if '=' in item:
                name, value = item.split('=', 1)
                cookies.append({
                    'name': name.strip(),
                    'value': value.strip(),
                    'domain': '',
                    'path': '/'
                })
        return cookies
    
    @staticmethod
    def cookies_to_string(cookies: list) -> str:
        """将 cookie 列表转换为字符串"""
        return '; '.join([f"{c['name']}={c['value']}" for c in cookies])
    
    async def init_browser(self, storage_state: str = None):
        """初始化浏览器"""
        print(f"[{self.platform_name}] init_browser: headless={self.headless}", flush=True)
        playwright = await async_playwright().start()
        self.browser = await playwright.chromium.launch(headless=self.headless)
        
        if storage_state and os.path.exists(storage_state):
            self.context = await self.browser.new_context(storage_state=storage_state)
        else:
            self.context = await self.browser.new_context()
        
        self.page = await self.context.new_page()
        return self.page
    
    async def set_cookies(self, cookies: list):
        """设置 cookies"""
        if not self.context:
            raise Exception("Browser context not initialized")
        
        # 设置默认域名
        for cookie in cookies:
            if 'domain' not in cookie or not cookie['domain']:
                cookie['domain'] = self.cookie_domain
        
        await self.context.add_cookies(cookies)
    
    async def close_browser(self):
        """关闭浏览器"""
        if self.context:
            await self.context.close()
        if self.browser:
            await self.browser.close()
    
    async def save_cookies(self, file_path: str):
        """保存 cookies 到文件"""
        if self.context:
            await self.context.storage_state(path=file_path)

    async def capture_screenshot(self) -> str:
        """截取当前页面截图，返回 Base64 编码"""
        import base64
        if not self.page:
            return ""
        try:
            screenshot_bytes = await self.page.screenshot(type="jpeg", quality=80)
            return base64.b64encode(screenshot_bytes).decode('utf-8')
        except Exception as e:
            print(f"[{self.platform_name}] 截图失败: {e}")
            return ""

    async def get_page_url(self) -> str:
        """获取当前页面 URL"""
        if not self.page:
            return ""
        try:
            return self.page.url
        except:
            return ""

    async def check_publish_status(self) -> dict:
        """
        检查发布状态
        返回: {status, screenshot_base64, page_url, message}
        """
        if not self.page:
            return {"status": "error", "message": "页面未初始化"}
        
        try:
            screenshot = await self.capture_screenshot()
            page_url = await self.get_page_url()
            
            # 检查常见的成功/失败标志
            page_content = await self.page.content()
            
            # 检查成功标志
            success_keywords = ['发布成功', '上传成功', '发表成功', '提交成功']
            for keyword in success_keywords:
                if keyword in page_content:
                    return {
                        "status": "success",
                        "screenshot_base64": screenshot,
                        "page_url": page_url,
                        "message": "发布成功"
                    }
            
            # 检查验证码标志
            captcha_keywords = ['验证码', '身份验证', '请完成验证', '滑动验证', '图形验证']
            for keyword in captcha_keywords:
                if keyword in page_content:
                    return {
                        "status": "need_captcha",
                        "screenshot_base64": screenshot,
                        "page_url": page_url,
                        "message": f"检测到{keyword}"
                    }
            
            # 检查失败标志
            fail_keywords = ['发布失败', '上传失败', '提交失败', '操作失败']
            for keyword in fail_keywords:
                if keyword in page_content:
                    return {
                        "status": "failed",
                        "screenshot_base64": screenshot,
                        "page_url": page_url,
                        "message": keyword
                    }
            
            # 默认返回处理中
            return {
                "status": "processing",
                "screenshot_base64": screenshot,
                "page_url": page_url,
                "message": "处理中"
            }
        except Exception as e:
            return {
                "status": "error",
                "screenshot_base64": "",
                "page_url": "",
                "message": str(e)
            }

    async def wait_for_upload_complete(self, success_selector: str, timeout: int = 300):
        """等待上传完成"""
        if not self.page:
            raise Exception("Page not initialized")
        
        for _ in range(timeout // 3):
            try:
                count = await self.page.locator(success_selector).count()
                if count > 0:
                    return True
            except:
                pass
            await asyncio.sleep(3)
            self.report_progress(30, "正在上传视频...")
        
        return False
    
    @abstractmethod
    async def publish(self, cookies: str, params: PublishParams) -> PublishResult:
        """
        发布视频 - 子类必须实现
        
        Args:
            cookies: cookie 字符串或 JSON
            params: 发布参数
            
        Returns:
            PublishResult: 发布结果
        """
        pass
    
    async def get_works(self, cookies: str, page: int = 0, page_size: int = 20) -> WorksResult:
        """
        获取作品列表 - 子类可覆盖实现
        
        Args:
            cookies: cookie 字符串或 JSON
            page: 页码（从0开始）
            page_size: 每页数量
            
        Returns:
            WorksResult: 作品列表结果
        """
        return WorksResult(
            success=False,
            platform=self.platform_name,
            error="该平台暂不支持获取作品列表"
        )
    
    async def get_comments(self, cookies: str, work_id: str, cursor: str = "") -> CommentsResult:
        """
        获取作品评论 - 子类可覆盖实现
        
        Args:
            cookies: cookie 字符串或 JSON
            work_id: 作品ID
            cursor: 分页游标
            
        Returns:
            CommentsResult: 评论列表结果
        """
        return CommentsResult(
            success=False,
            platform=self.platform_name,
            work_id=work_id,
            error="该平台暂不支持获取评论"
        )
    
    async def run(self, cookies: str, params: PublishParams) -> PublishResult:
        """
        运行发布任务
        包装了 publish 方法，添加了异常处理和资源清理
        """
        try:
            return await self.publish(cookies, params)
        except Exception as e:
            import traceback
            traceback.print_exc()
            return PublishResult(
                success=False,
                platform=self.platform_name,
                error=str(e)
            )
        finally:
            await self.close_browser()
    
    async def run_get_works(self, cookies: str, page: int = 0, page_size: int = 20) -> WorksResult:
        """
        运行获取作品任务
        """
        try:
            return await self.get_works(cookies, page, page_size)
        except Exception as e:
            import traceback
            traceback.print_exc()
            return WorksResult(
                success=False,
                platform=self.platform_name,
                error=str(e)
            )
        finally:
            await self.close_browser()
    
    async def run_get_comments(self, cookies: str, work_id: str, cursor: str = "") -> CommentsResult:
        """
        运行获取评论任务
        """
        try:
            return await self.get_comments(cookies, work_id, cursor)
        except Exception as e:
            import traceback
            traceback.print_exc()
            return CommentsResult(
                success=False,
                platform=self.platform_name,
                work_id=work_id,
                error=str(e)
            )
        finally:
            await self.close_browser()
    
    async def check_login_status(self, cookies: str) -> dict:
        """
        检查 Cookie 登录状态（通过浏览器访问后台页面检测）
        
        Args:
            cookies: cookie 字符串或 JSON
            
        Returns:
            dict: {
                "success": True,
                "valid": True/False,
                "need_login": True/False,
                "message": "状态描述"
            }
        """
        try:
            await self.init_browser()
            cookie_list = self.parse_cookies(cookies)
            await self.set_cookies(cookie_list)
            
            if not self.page:
                raise Exception("Page not initialized")
            
            # 访问平台后台首页
            home_url = self.login_url
            print(f"[{self.platform_name}] 访问后台页面: {home_url}")
            await self.page.goto(home_url, wait_until='domcontentloaded', timeout=30000)
            await asyncio.sleep(3)
            
            # 检查当前 URL 是否被重定向到登录页
            current_url = self.page.url
            print(f"[{self.platform_name}] 当前 URL: {current_url}")
            
            # 登录页特征
            login_indicators = ['login', 'passport', 'signin', 'auth']
            is_login_page = any(indicator in current_url.lower() for indicator in login_indicators)
            
            # 检查页面是否有登录弹窗
            need_login = is_login_page
            
            if not need_login:
                # 检查页面内容是否有登录提示
                login_selectors = [
                    'text="请先登录"',
                    'text="登录后继续"',
                    'text="请登录"',
                    '[class*="login-modal"]',
                    '[class*="login-dialog"]',
                    '[class*="login-popup"]',
                ]
                for selector in login_selectors:
                    try:
                        if await self.page.locator(selector).count() > 0:
                            need_login = True
                            print(f"[{self.platform_name}] 检测到登录弹窗: {selector}")
                            break
                    except:
                        pass
            
            if need_login:
                return {
                    "success": True,
                    "valid": False,
                    "need_login": True,
                    "message": "Cookie 已过期，需要重新登录"
                }
            else:
                return {
                    "success": True,
                    "valid": True,
                    "need_login": False,
                    "message": "登录状态有效"
                }
                
        except Exception as e:
            import traceback
            traceback.print_exc()
            return {
                "success": False,
                "valid": False,
                "need_login": True,
                "error": str(e)
            }
        finally:
            await self.close_browser()