大更新！！！

增加 4 个答题平台支持。分别是 UC-疯狂夺金、蘑菇街-大富翁、掌阅-百万文豪、映客-芝士超人。

开放www.answerot.com试用。目前服务器配置较低，可能会影响速度，仅给有兴趣的朋友试用。如果后续有需求，再考虑升级服务器。后续功能同步更新 answerot、AnswerotX、服务器。

欢迎大家试用，提供宝贵的意见和建议，谢谢。

[Python]如何更优雅地实现直播答题功能？新增4个平台支持并开放服务器

zlyuanteng 1楼作者

8 小时 0 回复惨案

eggper 2楼

要优雅地实现直播答题功能并支持多平台，关键在于设计一个清晰、可扩展的架构。核心是分离业务逻辑与平台适配，用抽象层处理通用流程，用具体实现对接不同平台API。

下面是一个基础框架示例，展示了如何通过抽象基类定义接口，然后为每个平台创建适配器：

from abc import ABC, abstractmethod
from typing import Dict, Any, List
import asyncio

# 1. 定义抽象基类 - 这是优雅扩展的关键
class LiveQuizPlatform(ABC):
    """直播答题平台抽象接口"""
    
    @abstractmethod
    async def connect(self) -> bool:
        """连接到直播平台"""
        pass
    
    @abstractmethod
    async def listen_for_questions(self) -> List[Dict[str, Any]]:
        """监听题目推送"""
        pass
    
    @abstractmethod
    async def submit_answer(self, question_id: str, answer: str) -> Dict[str, Any]:
        """提交答案"""
        pass
    
    @abstractmethod
    async def get_result(self, question_id: str) -> Dict[str, Any]:
        """获取答题结果"""
        pass

# 2. 具体平台实现 - 新增平台只需继承并实现接口
class DouyinLiveQuiz(LiveQuizPlatform):
    """抖音直播答题实现"""
    
    def __init__(self, room_id: str, token: str):
        self.room_id = room_id
        self.token = token
        self.connected = False
    
    async def connect(self) -> bool:
        # 实现抖音API连接逻辑
        print(f"连接到抖音直播间 {self.room_id}")
        self.connected = True
        return True
    
    async def listen_for_questions(self) -> List[Dict[str, Any]]:
        # 监听抖音题目推送
        # 这里应该是实际的API调用
        return [
            {
                "question_id": "q1",
                "question": "Python诞生于哪一年？",
                "options": ["1989", "1991", "1995", "2000"],
                "timeout": 10
            }
        ]
    
    async def submit_answer(self, question_id: str, answer: str) -> Dict[str, Any]:
        # 提交答案到抖音
        return {"success": True, "question_id": question_id}
    
    async def get_result(self, question_id: str) -> Dict[str, Any]:
        # 从抖音获取结果
        return {"correct": True, "correct_answer": "1991"}

# 3. 其他平台类似实现
class KuaishouLiveQuiz(LiveQuizPlatform):
    """快手直播答题实现"""
    # 实现快手特定逻辑...

class BilibiliLiveQuiz(LiveQuizPlatform):
    """B站直播答题实现"""
    # 实现B站特定逻辑...

# 4. 统一的答题管理器 - 处理核心业务逻辑
class LiveQuizManager:
    """统一管理多平台答题"""
    
    def __init__(self):
        self.platforms: Dict[str, LiveQuizPlatform] = {}
    
    def add_platform(self, name: str, platform: LiveQuizPlatform):
        """添加平台支持"""
        self.platforms[name] = platform
    
    async def run_quiz_session(self, platform_name: str):
        """运行答题会话"""
        platform = self.platforms.get(platform_name)
        if not platform:
            raise ValueError(f"平台 {platform_name} 未配置")
        
        # 连接平台
        if await platform.connect():
            print(f"已连接到 {platform_name}")
            
            # 监听题目
            questions = await platform.listen_for_questions()
            for question in questions:
                print(f"收到题目: {question['question']}")
                
                # 这里可以加入自动答题逻辑
                answer = "1991"  # 示例答案
                result = await platform.submit_answer(
                    question["question_id"], 
                    answer
                )
                
                if result["success"]:
                    final_result = await platform.get_result(question["question_id"])
                    print(f"答题结果: {final_result}")

# 5. 使用示例
async def main():
    manager = LiveQuizManager()
    
    # 添加支持的平台
    manager.add_platform("douyin", DouyinLiveQuiz("123456", "token_abc"))
    manager.add_platform("kuaishou", KuaishouLiveQuiz("789012", "token_def"))
    # 继续添加其他平台...
    
    # 运行抖音答题
    await manager.run_quiz_session("douyin")

if __name__ == "__main__":
    asyncio.run(main())

关键设计要点：