Feat support poe

README.md

@@ -40,6 +40,7 @@
 * [x] 支持 ChatGPT API
 * [x] 支持 Bing 聊天
 * [x] 支持 Google bard
+* [x] 支持 poe.com 网页版
 * [x] 支持 文心一言 网页版
 
 **平台兼容情况**  
@@ -355,6 +356,13 @@ auto_remove_old_conversations = true
 
 # === OpenAI 账号部分结束
 
+# === Poe 账号部分开始
+# 如果你没有 Poe 账号，可以直接删除这部分
+[poe]
+[[poe.accounts]]
+# 登陆 poe.com 网站后，通过开发者工具查看Cookie获取
+p_b = "V4j***"
+# === Poe 账号部分结束
 
 # === Bing 设置部分开始
 # 如果你没有 Bing 账号，可以直接删除这部分
@@ -816,6 +824,16 @@ title_pattern="qq-{session_id}"
 4. 找到 控制台（或 Console），输入 `document.cookie` 然后回车
 5. 复制接下来出现的一段文本，这就是你的 Cookie
 
+### Poe 账号 Cookie 获取方法
+
+你需要通过电脑浏览器来获得 Poe Cookie，如果你有别的手段能获得 cookie 的话也是可以的。  
+
+1. 确认能科学上网
+2. 打开 https://poe.com 并登陆
+3. 按下 F12，打开开发者工具（DevTools）
+4. 找到 `应用程序 - 存储` 或 `应用 - 存储` 或 `存储`， 查看Cookie下 https://poe.com 域名下的 "p-b" 
+5. 复制值即可
+
 ### 文心一言 账号 Cookie 获取方法
 
 请参考这里：[wiki](https://github.com/lss233/chatgpt-mirai-qq-bot/wiki/%E6%96%87%E5%BF%83%E4%B8%80%E8%A8%80-Cookie-%E8%8E%B7%E5%8F%96%E6%95%99%E7%A8%8B)

adapter/botservice.py

@@ -1,5 +1,7 @@
 from typing import Generator
 
+from loguru import logger
+
 
 class BotAdapter:
     """定义所有 Chatbot 的通用接口"""
@@ -22,8 +24,25 @@ async def rollback(self): ...
     async def on_reset(self): ...
     """当会话被重置时，此函数被调用"""
 
-    async def preset_ask(self, role: str, text: str): ...
-    """以预设方式进行提问"""
+    def use_default_preset_ask(self) -> bool:
+        """使用默认预设逻辑"""
+        return False
+
+    async def preset_ask(self, role: str, text: str):
+        """以预设方式进行提问"""
+        if self.use_default_preset_ask():
+            if role.endswith('bot') or role in ['assistant', 'chatgpt']:
+                logger.debug(f"[预设] 响应：{text}")
+                yield text
+            else:
+                logger.debug(f"[预设] 发送：{text}")
+                item = None
+                async for item in self.ask(text): ...
+                if item:
+                    logger.debug(f"[预设] Chatbot 回应：{item}")
+                pass  # 不发送 AI 的回应，免得串台
+        else:
+            yield None
 
     async def switch_model(self, model_name): ...
     """切换模型"""

adapter/chatgpt/web.py

@@ -52,7 +52,6 @@ async def switch_model(self, model_name):
         # self.current_model = model_name
         raise Exception("此 AI 暂不支持切换模型的操作！")
 
-
     async def rollback(self):
         if len(self.parent_id_prev_queue) > 0:
             self.parent_id = self.parent_id_prev_queue.pop()
@@ -113,17 +112,5 @@ async def ask(self, prompt: str) -> Generator[str, None, None]:
                 raise ConcurrentMessageException()
             raise e
 
-    async def preset_ask(self, role: str, text: str):
-        if role.endswith('bot') or role in ['assistant', 'chatgpt']:
-            logger.debug(f"[预设] 响应：{text}")
-            yield text
-        else:
-            logger.debug(f"[预设] 发送：{text}")
-            item = None
-            async for item in self.ask(text): ...
-            if item:
-                logger.debug(f"[预设] Chatbot 回应：{item}")
-            pass  # 不发送 AI 的回应，免得串台
-
     def get_queue_info(self):
         return self.bot.queue

adapter/google/bard.py

@@ -92,14 +92,6 @@ async def ask(self, prompt: str) -> Generator[str, None, None]:
             await self.on_reset()
             return
 
-    async def preset_ask(self, role: str, text: str):
-        if role.endswith('bot') or role in ['assistant', 'bard']:
-            logger.debug(f"[预设] 响应：{text}")
-            yield text
-        else:
-            logger.debug(f"[预设] 发送：{text}")
-            item = None
-            async for item in self.ask(text): ...
-            if item:
-                logger.debug(f"[预设] Chatbot 回应：{item}")
-            pass  # 不发送 AI 的回应，免得串台
+    def use_default_preset_ask(self) -> bool:
+        """使用默认预设逻辑"""
+        return True

adapter/ms/bing.py

@@ -74,7 +74,3 @@ async def ask(self, prompt: str) -> Generator[str, None, None]:
             yield "Bing 已结束本次会话。继续发送消息将重新开启一个新会话。"
             await self.on_reset()
             return
-
-    async def preset_ask(self, role: str, text: str):
-        # 不会给 Bing 提供预设
-        yield None

adapter/openai/api.py

@@ -38,9 +38,6 @@ async def on_reset(self):
     async def ask(self, prompt: str) -> Generator[str, None, None]:
         yield None
 
-    async def preset_ask(self, role: str, text: str):
-        yield None
-
     async def image_creation(self, prompt: str):
         logger.debug(f"[OpenAI Image] Prompt: {prompt}")
         response = await openai.Image.acreate(

adapter/quora/poe.py

@@ -0,0 +1,57 @@
+from enum import Enum
+from typing import Generator
+
+from adapter.botservice import BotAdapter
+from constants import botManager
+
+
+class PoeBot(Enum):
+    """Poe 支持的机器人：{'capybara': 'Sage', 'beaver': 'GPT-4', 'a2_2': 'Claude+','a2': 'Claude', 'chinchilla': 'ChatGPT',
+    'nutria': 'Dragonfly'} """
+    Sage = "capybara"
+    GPT4 = "beaver"
+    Claude2 = "a2_2"
+    Claude = "a2"
+    ChatGPT = "chinchilla"
+    Dragonfly = "nutria"
+
+    @staticmethod
+    def parse(bot_name: str):
+        tmp_name = bot_name.lower()
+        for bot in PoeBot:
+            if str(bot.name).lower() == tmp_name or str(bot.value).lower() == tmp_name \
+                    or f"poe-{str(bot.name).lower()}" == tmp_name:
+                return bot
+        return None
+
+
+class PoeAdapter(BotAdapter):
+
+    def __init__(self, session_id: str = "unknown", poe_bot: PoeBot = None):
+        """获取内部队列"""
+        super().__init__(session_id)
+        self.session_id = session_id
+        self.poe_bot = poe_bot if poe_bot else PoeBot.ChatGPT
+        self.poe_client = botManager.pick("poe-web")
+
+    async def ask(self, msg: str) -> Generator[str, None, None]:
+        """向 AI 发送消息"""
+        final_resp = None
+        for final_resp in self.poe_client.send_message(chatbot=self.poe_bot.value, message=msg):
+            pass
+        if final_resp is None:
+            raise Exception("OpenAI 在返回结果时出现了错误")
+        resp = final_resp["text"]
+        yield resp
+
+    async def rollback(self):
+        """回滚对话"""
+        self.poe_client.purge_conversation(self.poe_bot.value, 2)
+
+    async def on_reset(self):
+        """当会话被重置时，此函数被调用"""
+        self.poe_client.send_chat_break(self.poe_bot.value)
+
+    def use_default_preset_ask(self) -> bool:
+        """使用默认预设逻辑"""
+        return True

config.py

@@ -120,18 +120,30 @@ class OpenAIAPIKey(OpenAIAuthBase):
     """OpenAI 的 api_key"""
 
 
+class PoeCookieAuth(BaseModel):
+    p_b: str
+    """登陆 poe.com 后 Cookie 中 p_b 的值"""
+
+
 class BingCookiePath(BaseModel):
     cookie_content: str
     """Bing 的 Cookie 文件内容"""
     proxy: Optional[str] = None
     """可选的代理地址，留空则检测系统代理"""
 
+
 class BardCookiePath(BaseModel):
     cookie_content: str
     """Bard 的 Cookie 文件内容"""
     proxy: Optional[str] = None
     """可选的代理地址，留空则检测系统代理"""
 
+
+class PoeAuths(BaseModel):
+    accounts: List[PoeCookieAuth] = []
+    """Poe 的账号列表"""
+
+
 class TTSAccounts(BaseModel):
     speech_key: str
     """TTS KEY"""
@@ -148,24 +160,29 @@ class BingAuths(BaseModel):
     accounts: List[BingCookiePath] = []
     """Bing 的账号列表"""
 
+
 class BardAuths(BaseModel):
     accounts: List[BardCookiePath] = []
     """Bard 的账号列表"""
 
+
 class AzureAuths(BaseModel):
     tts_accounts: List[TTSAccounts] = []
     """Azure 的账号列表"""
 
+
 class YiyanCookiePath(BaseModel):
     cookie_content: str
     """"文心一言网站的 Cookie 内容"""
     proxy: Optional[str] = None
     """可选的代理地址，留空则检测系统代理"""
 
+
 class YiyanAuths(BaseModel):
     accounts: List[YiyanCookiePath] = []
     """文心一言的账号列表"""
 
+
 class TextToImage(BaseModel):
     always: bool = False
     """强制开启，设置后所有的会话强制以图片发送"""
@@ -340,6 +357,7 @@ class Config(BaseModel):
     bard: BardAuths = BardAuths()
     azure: AzureAuths = AzureAuths()
     yiyan: YiyanAuths = YiyanAuths()
+    poe: PoeAuths = PoeAuths()
     text_to_image: TextToImage = TextToImage()
     trigger: Trigger = Trigger()
     response: Response = Response()

conversation.py

@@ -16,6 +16,7 @@
 from adapter.ms.bing import BingAdapter
 from adapter.google.bard import BardAdapter
 from adapter.openai.api import OpenAIAPIAdapter
+from adapter.quora.poe import PoeBot, PoeAdapter
 from constants import config
 from exceptions import PresetNotFoundException, BotTypeNotFoundException, NoAvailableBotException, \
     CommandRefusedException
@@ -63,6 +64,8 @@ def __init__(self, _type: str, session_id: str):
             self.adapter = ChatGPTWebAdapter(self.session_id)
         elif _type == 'chatgpt-api':
             self.adapter = ChatGPTAPIAdapter(self.session_id)
+        elif PoeBot.parse(_type):
+            self.adapter = PoeAdapter(self.session_id, PoeBot.parse(_type))
         elif _type == 'bing':
             self.adapter = BingAdapter(self.session_id)
         elif _type == 'bing-c':

manager/bot.py

@@ -20,14 +20,15 @@
 
 from chatbot.Unofficial import AsyncChatbot as BrowserChatbot
 from loguru import logger
-from config import OpenAIAuthBase, OpenAIAPIKey, Config, BingCookiePath, BardCookiePath, YiyanCookiePath
+from config import OpenAIAuthBase, OpenAIAPIKey, Config, BingCookiePath, BardCookiePath, YiyanCookiePath, PoeCookieAuth
 import OpenAIAuth
 import urllib3.exceptions
 import utils.network as network
 from tinydb import TinyDB, Query
 import hashlib
 import datetime
 from dateutil.relativedelta import relativedelta
+from poe import Client as PoeClient
 
 
 class BotManager:
@@ -36,6 +37,7 @@ class BotManager:
     bots: Dict[str, List] = {
         "chatgpt-web": [],
         "openai-api": [],
+        "poe-web": [],
         "bing-cookie": [],
         "bard-cookie": [],
         "yiyan-cookie": [],
@@ -51,6 +53,9 @@ class BotManager:
     bard: List[BardCookiePath]
     """Bard Account Infos"""
 
+    poe: List[PoeCookieAuth]
+    """Poe Account infos"""
+
     yiyan: List[YiyanCookiePath]
     """Yiyan Account Infos"""
 
@@ -61,6 +66,7 @@ def __init__(self, config: Config) -> None:
         self.openai = config.openai.accounts if config.openai else []
         self.bing = config.bing.accounts if config.bing else []
         self.bard = config.bard.accounts if config.bard else []
+        self.poe = config.poe.accounts if config.poe else []
         self.yiyan = config.yiyan.accounts if config.yiyan else []
         try:
             os.mkdir('data')
@@ -74,13 +80,16 @@ async def login(self):
         self.bots = {
             "chatgpt-web": [],
             "openai-api": [],
+            "poe-web": [],
             "bing-cookie": [],
             "bard-cookie": [],
             "yiyan-cookie": [],
         }
         self.__setup_system_proxy()
         if len(self.bing) > 0:
             self.login_bing()
+        if len(self.poe) > 0:
+            self.login_poe()
         if len(self.bard) > 0:
             self.login_bard()
         if len(self.openai) > 0:
@@ -105,7 +114,9 @@ async def login(self):
                 logger.info(f"AI 类型：{k} - 可用账号： {len(v)} 个")
         # 自动推测默认 AI
         if not self.config.response.default_ai:
-            if len(self.bots['chatgpt-web']) > 0:
+            if len(self.bots['poe-web']) > 0:
+                self.config.response.default_ai = 'poe-chatgpt'
+            elif len(self.bots['chatgpt-web']) > 0:
                 self.config.response.default_ai = 'chatgpt-web'
             elif len(self.bots['openai-api']) > 0:
                 self.config.response.default_ai = 'chatgpt-api'
@@ -148,6 +159,28 @@ def login_bard(self):
             logger.error("所有 Bard 账号均解析失败！")
         logger.success(f"成功解析 {len(self.bots['bard-cookie'])}/{len(self.bing)} 个 Bard 账号！")
 
+    def login_poe(self):
+        def poe_check_auth(client: PoeClient) -> bool:
+            try:
+                response = client.get_bot_names()
+                logger.debug(f"poe bot is running. bot names -> {response}")
+                return True
+            except KeyError:
+                return False
+        try:
+            for i, account in enumerate(self.poe):
+                logger.info("正在解析第 {i} 个 poe web 账号", i=i + 1)
+                bot = PoeClient(token=account.p_b)
+                if poe_check_auth(bot):
+                    self.bots["poe-web"].append(bot)
+                    logger.success("解析成功！", i=i + 1)
+        except Exception as e:
+            logger.error("解析失败：")
+            logger.exception(e)
+        if len(self.bots["poe-web"]) < 1:
+            logger.error("所有 Poe 账号均解析失败！")
+        logger.success(f"成功解析 {len(self.bots['poe-web'])}/{len(self.bing)} 个 poe web 账号！")
+
     def login_yiyan(self):
         for i, account in enumerate(self.yiyan):
             logger.info("正在解析第 {i} 个 文心一言 账号", i=i + 1)

requirements.txt

@@ -31,3 +31,4 @@ undetected_chromedriver
 python-dateutil
 discord.py
 azure-cognitiveservices-speech
+poe-api~=0.2.0