Python中如何更优雅地进行直播答题？使用answerot和answertX实现
直播答题小能手的自白
先放为敬：
- answerot: https://github.com/anhkgg/answerot
- AnswerotX: https://github.com/anhkgg/answerotx
增加本地支持。
将题目检索和题目拉取功能分离，服务端只负责题目的识别和搜索以及展示结果，而本地功能负责拉取手机中的题目信息，这样可以部署一次服务端，再任意电脑开启本地功能即可使用。本地功能需要 AnserotX 的支持，详情见AnserotX介绍。如果不想自己部署服务端，可以偷懒使用我提供的服务端 http://answerot.com (暂时还未部署)，然后只需下载本地功能模块AnserotX即可。
增加扩展搜索功能。
有些时候题目没有明显关键信息，如下列那个是正确说法？，这种题目是无法拿到结果的，此时其实可以通过答案来搜索确认答案，但是 10 秒时间肯定时不允许你来完成手工搜索的，那么搜索来帮你完成，点击答案旁边的搜或者加问搜(问题和当前答案合并搜索)即可完成快速搜索，然后确认答案。
欢迎使用！
Python中如何更优雅地进行直播答题？使用answerot和answertX实现
5 回复
我理解你想在Python中实现直播答题功能，并且提到了answerot和answertX这两个库。不过说实话，这两个库在Python生态中并不常见，可能是特定项目或内部工具。
如果你想要优雅地实现直播答题，我建议考虑更成熟的方案。这里给你一个基于WebSocket的完整实现示例：
import asyncio
import websockets
import json
from typing import Dict, Set
class LiveQuizServer:
def __init__(self):
self.connected_clients: Set = set()
self.questions = [
{"id": 1, "question": "Python是哪年创建的？", "options": ["1989", "1991", "1995", "2000"], "answer": 1},
{"id": 2, "question": "下列哪个不是Python框架？", "options": ["Django", "Flask", "Spring", "FastAPI"], "answer": 2}
]
self.current_question = 0
self.user_answers: Dict[str, int] = {}
async def handle_client(self, websocket, path):
self.connected_clients.add(websocket)
try:
async for message in websocket:
data = json.loads(message)
if data["type"] == "join":
await websocket.send(json.dumps({
"type": "welcome",
"message": "欢迎参加直播答题！",
"total_questions": len(self.questions)
}))
elif data["type"] == "answer":
question_id = data["question_id"]
answer = data["answer"]
self.user_answers[websocket.id] = answer
# 广播答题统计
stats = self.calculate_stats()
await self.broadcast({
"type": "stats",
"stats": stats
})
elif data["type"] == "next_question":
await self.send_question()
finally:
self.connected_clients.remove(websocket)
async def send_question(self):
if self.current_question < len(self.questions):
question = self.questions[self.current_question]
await self.broadcast({
"type": "question",
"question": question
})
self.current_question += 1
def calculate_stats(self):
# 计算答题统计
total = len(self.user_answers)
correct = sum(1 for ans in self.user_answers.values()
if ans == self.questions[self.current_question-1]["answer"])
return {"total": total, "correct": correct, "percentage": correct/total*100 if total > 0 else 0}
async def broadcast(self, message):
if self.connected_clients:
await asyncio.gather(*[
client.send(json.dumps(message))
for client in self.connected_clients
])
# 启动服务器
async def main():
server = LiveQuizServer()
async with websockets.serve(server.handle_client, "localhost", 8765):
print("直播答题服务器已启动，端口8765")
await asyncio.Future() # 永久运行
if __name__ == "__main__":
asyncio.run(main())
客户端可以这样写：
import asyncio
import websockets
import json
async def quiz_client():
async with websockets.connect("ws://localhost:8765") as websocket:
# 加入答题
await websocket.send(json.dumps({"type": "join"}))
while True:
message = await websocket.recv()
data = json.loads(message)
if data["type"] == "question":
print(f"\n问题：{data['question']['question']}")
for i, option in enumerate(data['question']['options']):
print(f"{i}. {option}")
answer = int(input("请输入答案编号："))
await websocket.send(json.dumps({
"type": "answer",
"question_id": data['question']['id'],
"answer": answer
}))
elif data["type"] == "stats":
print(f"答题统计：{data['stats']}")
asyncio.run(quiz_client())
这个方案使用WebSocket实现实时通信，结构清晰且易于扩展。如果你确实需要answerot或answertX的具体用法，可能需要提供更多关于这两个库的信息。
建议： 用成熟的WebSocket方案替代不常见的库。
