潘多拉，一个让你呼吸顺畅的ChatGPT。Pandora, a ChatGPT that helps you breathe smoothly.
# -*- coding: utf-8 -*-

from setuptools import setup, find_packages

from src.pandora import __version__

with open('README.md', 'r', encoding='utf-8') as f:
    long_description = f.read()

with open('requirements.txt', 'r', encoding='utf-8') as f:
    requirements = f.read().split('\n')

with open('requirements_api.txt', 'r', encoding='utf-8') as f:
    requirements_api = f.read().split('\n')

setup(
    name='Pandora-ChatGPT',
    version=__version__,
    python_requires='>=3.7',
    author='Neo Peng',
    author_email='pengzhile@gmail.com',
    keywords='OpenAI ChatGPT ChatGPT-Plus gpt-3.5-turbo gpt-3.5-turbo-0301',
    description='A command-line interface to ChatGPT',
    long_description=long_description,
    long_description_content_type='text/markdown',
    url='https://github.com/pengzhile/pandora',
    packages=find_packages('src'),
    package_dir={'pandora': 'src/pandora'},
    include_package_data=True,
    install_requires=requirements,
    extras_require={
        'api': requirements_api,
        'cloud': ['pandora-cloud~=0.6.0'],
    },
    entry_points={
        'console_scripts': [
            'pandora = pandora.launcher:run',
            'pandora-cloud = pandora.cloud_launcher:run',
        ]
    },
    project_urls={
        'Source': 'https://github.com/pengzhile/pandora',
        'Tracker': 'https://github.com/pengzhile/pandora/issues',
    },
    classifiers=[
        'Development Status :: 5 - Production/Stable',

        'Environment :: Console',
        'Environment :: Web Environment',

        'Framework :: Flask',

        'Intended Audience :: Developers',
        'Intended Audience :: End Users/Desktop',

        'License :: OSI Approved :: GNU General Public License v2 (GPLv2)',

        'Natural Language :: English',
        'Natural Language :: Chinese (Simplified)',

        'Operating System :: MacOS',
        'Operating System :: Microsoft :: Windows',
        'Operating System :: POSIX :: Linux',

        'Programming Language :: SQL',
        'Programming Language :: JavaScript',
        'Programming Language :: Python :: 3.7',
        'Programming Language :: Python :: 3.8',
        'Programming Language :: Python :: 3.9',
        'Programming Language :: Python :: 3.10',
        'Programming Language :: Python :: 3.11',

        'Topic :: Communications :: Chat',
        'Topic :: Internet :: WWW/HTTP',
    ],
)

setup.py
cloud804d/mirrors-pengzhile-pandora
# -*- coding: utf-8 -*-

from pandora import launcher

if __name__ == '__main__':
    launcher.run()

src/pandora/__main__.py
cloud804d/mirrors-pengzhile-pandora
# -*- coding: utf-8 -*-

import re
import uuid

import pyperclip
from rich.prompt import Prompt, Confirm

from .. import __version__
from ..openai.utils import Console


class ChatPrompt:
    def __init__(self, prompt: str = None, parent_id=None, message_id=None):
        self.prompt = prompt
        self.parent_id = parent_id or self.gen_message_id()
        self.message_id = message_id or self.gen_message_id()

    @staticmethod
    def gen_message_id():
        return str(uuid.uuid4())


class State:
    def __init__(self, title=None, conversation_id=None, model_slug=None, user_prompt=ChatPrompt(),
                 chatgpt_prompt=ChatPrompt()):
        self.title = title
        self.conversation_id = conversation_id
        self.model_slug = model_slug
        self.user_prompt = user_prompt
        self.chatgpt_prompt = chatgpt_prompt
        self.user_prompts = []
        self.edit_index = None


class ChatBot:
    def __init__(self, chatgpt):
        self.chatgpt = chatgpt
        self.token_key = None
        self.state = None

    def run(self):
        self.token_key = self.__choice_token_key()

        conversation_base = self.__choice_conversation()
        if conversation_base:
            self.__load_conversation(conversation_base['id'])
        else:
            self.__new_conversation()

        self.__talk_loop()

    def __talk_loop(self):
        while True:
            Console.info_b('You{}:'.format(' (edit)' if self.state and self.state.edit_index else ''))

            prompt = self.__get_input()
            if not prompt:
                continue

            if '/' == prompt[0]:
                self.__process_command(prompt)
                continue

            self.__talk(prompt)

    @staticmethod
    def __get_input():
        lines = []
        while True:
            line = input()

            if not line:
                break

            if '/' == line[0]:
                return line

            lines.append(line)

        return '\n'.join(lines)

    def __process_command(self, command):
        command = command.strip().lower()

        if '/quit' == command or '/exit' == command or '/bye' == command:
            raise KeyboardInterrupt
        elif '/del' == command or '/delete' == command or '/remove' == command:
            self.__del_conversation(self.state)
        elif '/title' == command or '/set_title' == command or '/set-title' == command:
            self.__set_conversation_title(self.state)
        elif '/select' == command:
            self.run()
        elif '/refresh' == command or '/reload' == command:
            self.__load_conversation(self.state.conversation_id)
        elif '/new' == command:
            self.__new_conversation()
            self.__talk_loop()
        elif '/regen' == command or '/regenerate' == command:
            self.__regenerate_reply(self.state)
        elif '/goon' == command or '/continue' == command:
            self.__continue(self.state)
        elif '/edit' == command or '/modify' == command:
            self.__edit_choice()
        elif '/token' == command:
            self.__print_access_token()
        elif '/cls' == command or '/clear' == command:
            self.__clear_screen()
        elif '/copy' == command or '/cp' == command:
            self.__copy_text()
        elif '/copy_code' == command or "/cp_code" == command:
            self.__copy_code()
        elif '/ver' == command or '/version' == command:
            self.__print_version()
        else:
            self.__print_usage()

    @staticmethod
    def __print_usage():
        Console.info_b('\n#### Command list:')
        print('/?\t\tShow this help message.')
        print('/title\t\tSet the current conversation\'s title.')
        print('/select\t\tChoice a different conversation.')
        print('/reload\t\tReload the current conversation.')
        print('/regen\t\tRegenerate response.')
        print('/continue\t\tContinue generating.')
        print('/edit\t\tEdit one of your previous prompt.')
        print('/new\t\tStart a new conversation.')
        print('/del\t\tDelete the current conversation.')
        print('/token\t\tPrint your access token.')
        print('/copy\t\tCopy the last response to clipboard.')
        print('/copy_code\t\tCopy code from last response.')
        print('/clear\t\tClear your screen.')
        print('/version\tPrint the version of Pandora.')
        print('/exit\t\tExit Pandora.')
        print()

    def __edit_choice(self):
        if not self.state.user_prompts:
            return

        choices = []
        pattern = re.compile(r'\s+')
        Console.info_b('Choice your prompt to edit:')
        for idx, item in enumerate(self.state.user_prompts):
            number = str(idx + 1)
            choices.append(number)

            preview_prompt = re.sub(pattern, ' ', item.prompt)
            if len(preview_prompt) > 40:
                preview_prompt = '{}...'.format(preview_prompt[0:40])

            Console.info('  {}.\t{}'.format(number, preview_prompt))

        choices.append('c')
        Console.warn('  c.\t** Cancel')

        default_choice = None if len(choices) > 2 else '1'
        while True:
            choice = Prompt.ask('Your choice', choices=choices, show_choices=False, default=default_choice)
            if 'c' == choice:
                return

            self.state.edit_index = int(choice)
            return

    def __print_access_token(self):
        Console.warn_b('\n#### Your access token (keep it private)')
        Console.warn(self.chatgpt.get_access_token(token_key=self.token_key))
        print()

    def __clear_screen(self):
        Console.clear()

        if self.state:
            self.__print_conversation_title(self.state.title)

    @staticmethod
    def __print_version():
        Console.debug_bh('#### Version: {}'.format(__version__))
        print()

    def __new_conversation(self):
        self.state = State(model_slug=self.__choice_model()['slug'])

        self.state.title = 'New Chat'
        self.__print_conversation_title(self.state.title)

    @staticmethod
    def __print_conversation_title(title: str):
        Console.info_bh('==================== {} ===================='.format(title))
        Console.debug_h('Double enter to send. Type /? for help.')

    def __set_conversation_title(self, state: State):
        if not state.conversation_id:
            Console.error('#### Conversation has not been created.')
            return

        new_title = Prompt.ask('New title')
        if len(new_title) > 64:
            Console.error('#### Title too long.')
            return

        if self.chatgpt.set_conversation_title(state.conversation_id, new_title, token=self.token_key):
            state.title = new_title
            Console.debug('#### Set title success.')
        else:
            Console.error('#### Set title failed.')

    def __clear_conversations(self):
        if not Confirm.ask('Are you sure?', default=False):
            return

        if self.chatgpt.clear_conversations(token=self.token_key):
            self.run()
        else:
            Console.error('#### Clear conversations failed.')

    def __del_conversation(self, state: State):
        if not state.conversation_id:
            Console.error('#### Conversation has not been created.')
            return

        if not Confirm.ask('Are you sure?', default=False):
            return

        if self.chatgpt.del_conversation(state.conversation_id, token=self.token_key):
            self.run()
        else:
            Console.error('#### Delete conversation failed.')

    def __load_conversation(self, conversation_id):
        if not conversation_id:
            return

        self.state = State(conversation_id=conversation_id)

        nodes = []
        result = self.chatgpt.get_conversation(conversation_id, token=self.token_key)
        current_node_id = result['current_node']

        while True:
            node = result['mapping'][current_node_id]
            if not node.get('parent'):
                break

            nodes.insert(0, node)
            current_node_id = node['parent']

        self.state.title = result['title']
        self.__print_conversation_title(self.state.title)

        merge = False
        for node in nodes:
            message = node['message']
            if 'model_slug' in message['metadata']:
                self.state.model_slug = message['metadata']['model_slug']

            role = message['author']['role'] if 'author' in message else message['role']

            if 'user' == role:
                prompt = self.state.user_prompt
                self.state.user_prompts.append(ChatPrompt(message['content']['parts'][0], parent_id=node['parent']))

                Console.info_b('You:')
                Console.info(message['content']['parts'][0])
            elif 'assistant' == role:
                prompt = self.state.chatgpt_prompt

                if not merge:
                    Console.success_b('ChatGPT:')
                Console.success(message['content']['parts'][0])

                merge = 'end_turn' in message and message['end_turn'] is None
            else:
                continue

            prompt.prompt = message['content']['parts'][0]
            prompt.parent_id = node['parent']
            prompt.message_id = node['id']

            if not merge:
                print()

    def __talk(self, prompt):
        Console.success_b('ChatGPT:')

        first_prompt = not self.state.conversation_id

        if self.state.edit_index:
            idx = self.state.edit_index - 1
            user_prompt = self.state.user_prompts[idx]
            self.state.user_prompt = ChatPrompt(prompt, parent_id=user_prompt.parent_id)
            self.state.user_prompts = self.state.user_prompts[0:idx]

            self.state.edit_index = None
        else:
            self.state.user_prompt = ChatPrompt(prompt, parent_id=self.state.chatgpt_prompt.message_id)

        status, _, generator = self.chatgpt.talk(prompt, self.state.model_slug, self.state.user_prompt.message_id,
                                                 self.state.user_prompt.parent_id, self.state.conversation_id,
                                                 token=self.token_key)
        self.__print_reply(status, generator)

        self.state.user_prompts.append(self.state.user_prompt)

        if first_prompt:
            new_title = self.chatgpt.gen_conversation_title(self.state.conversation_id, self.state.model_slug,
                                                            self.state.chatgpt_prompt.message_id, token=self.token_key)
            self.state.title = new_title
            Console.debug_bh('#### Title generated: ' + new_title)

    def __regenerate_reply(self, state):
        if not state.conversation_id:
            Console.error('#### Conversation has not been created.')
            return

        status, _, generator = self.chatgpt.regenerate_reply(state.user_prompt.prompt, state.model_slug,
                                                             state.conversation_id, state.user_prompt.message_id,
                                                             state.user_prompt.parent_id, token=self.token_key)
        print()
        Console.success_b('ChatGPT:')
        self.__print_reply(status, generator)

    def __continue(self, state):
        if not state.conversation_id:
            Console.error('#### Conversation has not been created.')
            return

        status, _, generator = self.chatgpt.goon(state.model_slug, state.chatgpt_prompt.message_id,
                                                 state.conversation_id, token=self.token_key)
        print()
        Console.success_b('ChatGPT:')
        self.__print_reply(status, generator)

    def __print_reply(self, status, generator):
        if 200 != status:
            raise Exception(status, next(generator))

        p = 0
        for result in generator:
            if result['error']:
                raise Exception(result['error'])

            if not result['message']:
                raise Exception('miss message property.')

            text = None
            message = result['message']
            if 'assistant' == message['author']['role']:
                text = message['content']['parts'][0][p:]
                p += len(text)

            self.state.conversation_id = result['conversation_id']
            self.state.chatgpt_prompt.prompt = message['content']['parts'][0]
            self.state.chatgpt_prompt.parent_id = self.state.user_prompt.message_id
            self.state.chatgpt_prompt.message_id = message['id']

            if 'system' == message['author']['role']:
                self.state.user_prompt.parent_id = message['id']

            if text:
                Console.success(text, end='')

        print('\n')

    def __choice_conversation(self, page=1, page_size=20):
        conversations = self.chatgpt.list_conversations((page - 1) * page_size, page_size, token=self.token_key)
        if not conversations['total']:
            return None

        choices = ['c', 'r', 'dd']
        items = conversations['items']
        first_page = 0 == conversations['offset']
        last_page = (conversations['offset'] + conversations['limit']) >= conversations['total']

        Console.info_b('Choice conversation (Page {}):'.format(page))
        for idx, item in enumerate(items):
            number = str(idx + 1)
            choices.append(number)
            choices.append('t' + number)
            choices.append('d' + number)
            Console.info('  {}.\t{}'.format(number, item['title'].replace('\n', ' ')))

        if not last_page:
            choices.append('n')
            Console.warn('  n.\t>> Next page')

        if not first_page:
            choices.append('p')
            Console.warn('  p.\t<< Previous page')

        Console.warn('  t?.\tSet title for the conversation, eg: t1')
        Console.warn('  d?.\tDelete the conversation, eg: d1')
        Console.warn('  dd.\t!! Clear all conversations')
        Console.warn('  r.\tRefresh conversation list')

        if len(self.chatgpt.list_token_keys()) > 1:
            choices.append('k')
            Console.warn('  k.\tChoice access token')

        Console.warn('  c.\t** Start new chat')

        while True:
            choice = Prompt.ask('Your choice', choices=choices, show_choices=False)
            if 'c' == choice:
                return None

            if 'k' == choice:
                self.run()
                return

            if 'r' == choice:
                return self.__choice_conversation(page, page_size)

            if 'n' == choice:
                return self.__choice_conversation(page + 1, page_size)

            if 'p' == choice:
                return self.__choice_conversation(page - 1, page_size)

            if 'dd' == choice:
                self.__clear_conversations()
                continue

            if 't' == choice[0]:
                self.__set_conversation_title(State(conversation_id=items[int(choice[1:]) - 1]['id']))
                return self.__choice_conversation(page, page_size)

            if 'd' == choice[0]:
                self.__del_conversation(State(conversation_id=items[int(choice[1:]) - 1]['id']))
                continue

            return items[int(choice) - 1]

    def __choice_token_key(self):
        tokens = self.chatgpt.list_token_keys()

        size = len(tokens)
        if 1 == size:
            return None

        choices = ['r']
        Console.info_b('Choice access token:')
        for idx, item in enumerate(tokens):
            number = str(idx + 1)
            choices.append(number)
            Console.info('  {}.\t{}'.format(number, item))

        while True:
            choice = Prompt.ask('Your choice', choices=choices, show_choices=False)

            return tokens[int(choice) - 1]

    def __choice_model(self):
        models = self.chatgpt.list_models(token=self.token_key)

        size = len(models)
        if 1 == size:
            return models[0]

        choices = ['r']
        Console.info_b('Choice model:')
        for idx, item in enumerate(models):
            number = str(idx + 1)
            choices.append(number)
            Console.info('  {}.\t{} - {} - {}'.format(number, item['title'], item['description'],
                                                      '|'.join(item['tags'])))

        Console.warn('  r.\tRefresh model list')

        while True:
            choice = Prompt.ask('Your choice', choices=choices, show_choices=False)
            if 'r' == choice:
                return self.__choice_model()

            return models[int(choice) - 1]

    def __copy_text(self):
        pyperclip.copy(self.state.chatgpt_prompt.prompt)
        Console.info("已将上一次返回结果复制到剪切板。")
        pass

    def __copy_code(self):
        text = self.state.chatgpt_prompt.prompt
        pattern = re.compile(r'```.*\s([\s\S]*?)\s```')
        result = re.findall(pattern, text)
        if len(result) == 0:
            Console.info("未找到代码。")
            return
        else:
            code = '\n=======================================================\n'.join(result)
            pyperclip.copy(code)
            Console.info("已将上一次生成的代码复制到剪切板。")
        pass

src/pandora/bots/legacy.py
cloud804d/mirrors-pengzhile-pandora
# -*- coding: utf-8 -*-

import logging
from datetime import timedelta
from os.path import join, abspath, dirname

from flask import Flask, jsonify, make_response, request, Response, render_template
from flask_cors import CORS
from waitress import serve
from werkzeug.exceptions import default_exceptions
from werkzeug.middleware.proxy_fix import ProxyFix
from werkzeug.serving import WSGIRequestHandler

from .. import __version__
from ..exts.hooks import hook_logging
from ..openai.api import API


class ChatBot:
    __default_ip = '127.0.0.1'
    __default_port = 8008

    def __init__(self, chatgpt, debug=False, sentry=False):
        self.chatgpt = chatgpt
        self.debug = debug
        self.sentry = sentry
        self.log_level = logging.DEBUG if debug else logging.WARN

        hook_logging(level=self.log_level, format='[%(asctime)s] %(levelname)s in %(module)s: %(message)s')
        self.logger = logging.getLogger('waitress')

    def run(self, bind_str, threads=8):
        host, port = self.__parse_bind(bind_str)

        resource_path = abspath(join(dirname(__file__), '..', 'flask'))
        app = Flask(__name__, static_url_path='',
                    static_folder=join(resource_path, 'static'),
                    template_folder=join(resource_path, 'templates'))
        app.wsgi_app = ProxyFix(app.wsgi_app, x_port=1)
        app.after_request(self.__after_request)

        CORS(app, resources={r'/api/*': {'supports_credentials': True, 'expose_headers': [
            'Content-Type',
            'Authorization',
            'X-Requested-With',
            'Accept',
            'Origin',
            'Access-Control-Request-Method',
            'Access-Control-Request-Headers',
            'Content-Disposition',
        ], 'max_age': 600}})

        for ex in default_exceptions:
            app.register_error_handler(ex, self.__handle_error)

        app.route('/api/models')(self.list_models)
        app.route('/api/conversations')(self.list_conversations)
        app.route('/api/conversations', methods=['DELETE'])(self.clear_conversations)
        app.route('/api/conversation/<conversation_id>')(self.get_conversation)
        app.route('/api/conversation/<conversation_id>', methods=['DELETE'])(self.del_conversation)
        app.route('/api/conversation/<conversation_id>', methods=['PATCH'])(self.set_conversation_title)
        app.route('/api/conversation/gen_title/<conversation_id>', methods=['POST'])(self.gen_conversation_title)
        app.route('/api/conversation/talk', methods=['POST'])(self.talk)
        app.route('/api/conversation/regenerate', methods=['POST'])(self.regenerate)
        app.route('/api/conversation/goon', methods=['POST'])(self.goon)

        app.route('/api/auth/session')(self.session)
        app.route('/api/accounts/check')(self.check)
        app.route('/_next/data/olf4sv64FWIcQ_zCGl90t/chat.json')(self.chat_info)

        app.route('/')(self.chat)
        app.route('/chat')(self.chat)
        app.route('/chat/<conversation_id>')(self.chat)

        if not self.debug:
            self.logger.warning('Serving on http://{}:{}'.format(host, port))

        WSGIRequestHandler.protocol_version = 'HTTP/1.1'
        serve(app, host=host, port=port, ident=None, threads=threads)

    @staticmethod
    def __after_request(resp):
        resp.headers['X-Server'] = 'pandora/{}'.format(__version__)

        return resp

    def __parse_bind(self, bind_str):
        sections = bind_str.split(':', 2)
        if len(sections) < 2:
            try:
                port = int(sections[0])
                return self.__default_ip, port
            except ValueError:
                return sections[0], self.__default_port

        return sections[0], int(sections[1])

    def __handle_error(self, e):
        self.logger.error(e)

        return make_response(jsonify({
            'code': e.code,
            'message': str(e.original_exception if self.debug and hasattr(e, 'original_exception') else e.name)
        }), 500)

    @staticmethod
    def __set_cookie(resp, token_key, max_age):
        resp.set_cookie('token-key', token_key, max_age=max_age, path='/', domain=None, httponly=True, samesite='Lax')

    @staticmethod
    def __get_token_key():
        return request.headers.get('X-Use-Token', request.cookies.get('token-key'))

    def chat(self, conversation_id=None):
        query = {'chatId': [conversation_id]} if conversation_id else {}

        token_key = request.args.get('token')
        rendered = render_template('chat.html', pandora_base=request.url_root.strip('/'), query=query)
        resp = make_response(rendered)

        if token_key:
            self.__set_cookie(resp, token_key, timedelta(days=30))

        return resp

    @staticmethod
    def session():
        ret = {
            'user': {
                'id': 'user-000000000000000000000000',
                'name': 'admin@openai.com',
                'email': 'admin@openai.com',
                'image': None,
                'picture': None,
                'groups': []
            },
            'expires': '2089-08-08T23:59:59.999Z',
            'accessToken': 'secret',
        }

        return jsonify(ret)

    @staticmethod
    def chat_info():
        ret = {
            'pageProps': {
                'user': {
                    'id': 'user-000000000000000000000000',
                    'name': 'admin@openai.com',
                    'email': 'admin@openai.com',
                    'image': None,
                    'picture': None,
                    'groups': []
                },
                'serviceStatus': {},
                'userCountry': 'US',
                'geoOk': True,
                'serviceAnnouncement': {
                    'paid': {},
                    'public': {}
                },
                'isUserInCanPayGroup': True
            },
            '__N_SSP': True
        }

        return jsonify(ret)

    @staticmethod
    def check():
        ret = {
            'account_plan': {
                'is_paid_subscription_active': True,
                'subscription_plan': 'chatgptplusplan',
                'account_user_role': 'account-owner',
                'was_paid_customer': True,
                'has_customer_object': True,
                'subscription_expires_at_timestamp': 3774355199
            },
            'user_country': 'US',
            'features': [
                'model_switcher',
                'dfw_message_feedback',
                'dfw_inline_message_regen_comparison',
                'model_preview',
                'system_message',
                'can_continue',
            ],
        }

        return jsonify(ret)

    def list_models(self):
        return self.__proxy_result(self.chatgpt.list_models(True, self.__get_token_key()))

    def list_conversations(self):
        offset = request.args.get('offset', '0')
        limit = request.args.get('limit', '20')

        return self.__proxy_result(self.chatgpt.list_conversations(offset, limit, True, self.__get_token_key()))

    def get_conversation(self, conversation_id):
        return self.__proxy_result(self.chatgpt.get_conversation(conversation_id, True, self.__get_token_key()))

    def del_conversation(self, conversation_id):
        return self.__proxy_result(self.chatgpt.del_conversation(conversation_id, True, self.__get_token_key()))

    def clear_conversations(self):
        return self.__proxy_result(self.chatgpt.clear_conversations(True, self.__get_token_key()))

    def set_conversation_title(self, conversation_id):
        title = request.json['title']

        return self.__proxy_result(
            self.chatgpt.set_conversation_title(conversation_id, title, True, self.__get_token_key()))

    def gen_conversation_title(self, conversation_id):
        payload = request.json
        model = payload['model']
        message_id = payload['message_id']

        return self.__proxy_result(
            self.chatgpt.gen_conversation_title(conversation_id, model, message_id, True, self.__get_token_key()))

    def talk(self):
        payload = request.json
        prompt = payload['prompt']
        model = payload['model']
        message_id = payload['message_id']
        parent_message_id = payload['parent_message_id']
        conversation_id = payload.get('conversation_id')
        stream = payload.get('stream', True)

        return self.__process_stream(
            *self.chatgpt.talk(prompt, model, message_id, parent_message_id, conversation_id, stream,
                               self.__get_token_key()), stream)

    def goon(self):
        payload = request.json
        model = payload['model']
        parent_message_id = payload['parent_message_id']
        conversation_id = payload.get('conversation_id')
        stream = payload.get('stream', True)

        return self.__process_stream(
            *self.chatgpt.goon(model, parent_message_id, conversation_id, stream, self.__get_token_key()), stream)

    def regenerate(self):
        payload = request.json

        conversation_id = payload.get('conversation_id')
        if not conversation_id:
            return self.talk()

        prompt = payload['prompt']
        model = payload['model']
        message_id = payload['message_id']
        parent_message_id = payload['parent_message_id']
        stream = payload.get('stream', True)

        return self.__process_stream(
            *self.chatgpt.regenerate_reply(prompt, model, conversation_id, message_id, parent_message_id, stream,
                                           self.__get_token_key()), stream)

    @staticmethod
    def __process_stream(status, headers, generator, stream):
        if stream:
            return Response(API.wrap_stream_out(generator, status), mimetype=headers['Content-Type'], status=status)

        last_json = None
        for json in generator:
            last_json = json

        return make_response(last_json, status)

    @staticmethod
    def __proxy_result(remote_resp):
        resp = make_response(remote_resp.text)
        resp.content_type = remote_resp.headers['Content-Type']
        resp.status_code = remote_resp.status_code

        return resp

src/pandora/bots/server.py
cloud804d/mirrors-pengzhile-pandora
# -*- coding: utf-8 -*-

import argparse

from loguru import logger

from . import __version__
from .exts.hooks import hook_except_handle
from .openai.utils import Console

__show_verbose = False


def main():
    global __show_verbose

    Console.debug_b(
        '''
            Pandora-Cloud - A web interface to ChatGPT
            Github: https://github.com/pengzhile/pandora
            Version: {}, Mode: cloud, Engine: free
          '''.format(__version__)
    )

    parser = argparse.ArgumentParser()
    parser.add_argument(
        '-p',
        '--proxy',
        help='Use a proxy. Format: protocol://user:pass@ip:port',
        required=False,
        type=str,
        default=None,
    )
    parser.add_argument(
        '-s',
        '--server',
        help='Specific server bind. Format: ip:port, default: 127.0.0.1:8018',
        required=False,
        type=str,
        default='127.0.0.1:8018',
    )
    parser.add_argument(
        '--threads',
        help='Define the number of server workers, default: 4',
        required=False,
        type=int,
        default=4,
    )
    parser.add_argument(
        '-l',
        '--local',
        help='Login locally. Pay attention to the risk control of the login ip!',
        action='store_true',
    )
    parser.add_argument(
        '-v',
        '--verbose',
        help='Show exception traceback.',
        action='store_true',
    )
    args, _ = parser.parse_known_args()
    __show_verbose = args.verbose

    try:
        from pandora_cloud.server import ChatBot as CloudServer

        return CloudServer(args.proxy, args.verbose, login_local=args.local).run(args.server, args.threads)
    except (ImportError, ModuleNotFoundError):
        Console.error_bh('### You need `pip install Pandora-ChatGPT[cloud]` to support cloud mode.')


def run():
    hook_except_handle()

    try:
        main()
    except Exception as e:
        Console.error_bh('### Error occurred: ' + str(e))

        if __show_verbose:
            logger.exception('Exception occurred.')

src/pandora/cloud_launcher.py
cloud804d/mirrors-pengzhile-pandora
# -*- coding: utf-8 -*-

from datetime import datetime, timedelta
from os import getenv
from os.path import join

from appdirs import user_config_dir

USER_CONFIG_DIR = getenv('USER_CONFIG_DIR', user_config_dir('Pandora-ChatGPT'))
DATABASE_URI = getenv('DATABASE_URI',
                      'sqlite:///{}?check_same_thread=False'.format(join(USER_CONFIG_DIR, 'pandora-chatgpt.db')))


def default_api_prefix():
    return 'https://ai-{}.fakeopen.com'.format((datetime.now() - timedelta(days=1)).strftime('%Y%m%d'))

src/pandora/exts/config.py
cloud804d/mirrors-pengzhile-pandora
# -*- coding: utf-8 -*-

import logging
import sys

from loguru import logger


def __exception_handle(e_type, e_value, e_traceback):
    if issubclass(e_type, KeyboardInterrupt):
        print('\nBye...')
        sys.exit(0)

    sys.__excepthook__(e_type, e_value, e_traceback)


class __InterceptHandler(logging.Handler):
    def emit(self, record):
        try:
            level = logger.level(record.levelname).name
        except ValueError:
            level = record.levelno

        frame, depth = logging.currentframe(), 2
        while frame.f_code.co_filename == logging.__file__:
            frame = frame.f_back
            depth += 1

        logger.opt(depth=depth, exception=record.exc_info).log(
            level, record.getMessage()
        )


def hook_except_handle():
    sys.excepthook = __exception_handle


def hook_logging(**kwargs):
    logging.basicConfig(handlers=[__InterceptHandler()], **kwargs)

src/pandora/exts/hooks.py
cloud804d/mirrors-pengzhile-pandora
# -*- coding: utf-8 -*-

from jwt import decode

from ..openai.utils import Console

__public_key = b'-----BEGIN PUBLIC KEY-----\n' \
               b'MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA27rOErDOPvPc3mOADYtQ\n' \
               b'BeenQm5NS5VHVaoO/Zmgsf1M0Wa/2WgLm9jX65Ru/K8Az2f4MOdpBxxLL686ZS+K\n' \
               b'7eJC/oOnrxCRzFYBqQbYo+JMeqNkrCn34yed4XkX4ttoHi7MwCEpVfb05Qf/ZAmN\n' \
               b'I1XjecFYTyZQFrd9LjkX6lr05zY6aM/+MCBNeBWp35pLLKhiq9AieB1wbDPcGnqx\n' \
               b'lXuU/bLgIyqUltqLkr9JHsf/2T4VrXXNyNeQyBq5wjYlRkpBQDDDNOcdGpx1buRr\n' \
               b'Z2hFyYuXDRrMcR6BQGC0ur9hI5obRYlchDFhlb0ElsJ2bshDDGRk5k3doHqbhj2I\n' \
               b'gQIDAQAB\n' \
               b'-----END PUBLIC KEY-----'


def check_access_token(access_token, api=False):
    if access_token.startswith('fk-'):
        return True

    if api and (access_token.startswith('sk-') or access_token.startswith('pk-')):
        return True

    payload = (decode(access_token, key=__public_key, algorithms='RS256', audience=[
        "https://api.openai.com/v1",
        "https://openai.openai.auth0app.com/userinfo"
    ], issuer='https://auth0.openai.com/'))

    if 'scope' not in payload:
        raise Exception('miss scope')

    scope = payload['scope']
    if 'model.read' not in scope or 'model.request' not in scope:
        raise Exception('invalid scope')

    if 'https://api.openai.com/auth' not in payload or 'https://api.openai.com/profile' not in payload:
        raise Exception('belonging to an unregistered user.')

    return payload


def check_access_token_out(access_token, api=False):
    try:
        return check_access_token(access_token, api)
    except Exception as e:
        Console.error('### Invalid access token: {}'.format(str(e)))
        return False

src/pandora/exts/token.py
cloud804d/mirrors-pengzhile-pandora
# -*- coding: utf-8 -*-

import argparse
import os
from os import getenv

from loguru import logger
from rich.prompt import Prompt, Confirm

from . import __version__
from .bots.legacy import ChatBot as ChatBotLegacy
from .bots.server import ChatBot as ChatBotServer
from .exts.config import USER_CONFIG_DIR, default_api_prefix
from .exts.hooks import hook_except_handle
from .exts.token import check_access_token_out
from .openai.api import ChatGPT
from .openai.auth import Auth0
from .openai.utils import Console

if 'nt' == os.name:
    import pyreadline3 as readline
else:
    import readline

    readline.set_completer_delims('')
    readline.set_auto_history(False)

__show_verbose = False


def read_access_token(token_file):
    with open(token_file, 'r') as f:
        return f.read().strip()


def save_access_token(access_token):
    token_file = os.path.join(USER_CONFIG_DIR, 'access_token.dat')

    if not os.path.exists(USER_CONFIG_DIR):
        os.makedirs(USER_CONFIG_DIR)

    with open(token_file, 'w') as f:
        f.write(access_token)

    if __show_verbose:
        Console.debug_b('\nThe access token has been saved to the file:')
        Console.debug(token_file)
        print()


def confirm_access_token(token_file=None, silence=False, api=False):
    app_token_file = os.path.join(USER_CONFIG_DIR, 'access_token.dat')

    app_token_file_exists = os.path.isfile(app_token_file)
    if app_token_file_exists and __show_verbose:
        Console.debug_b('Found access token file: ', end='')
        Console.debug(app_token_file)

    if token_file:
        if not os.path.isfile(token_file):
            raise Exception('Error: {} is not a file.'.format(token_file))

        access_token = read_access_token(token_file)
        if os.path.isfile(app_token_file) and access_token == read_access_token(app_token_file):
            return access_token, False

        return access_token, True

    if app_token_file_exists:
        confirm = 'y' if silence else Prompt.ask('A saved access token has been detected. Do you want to use it?',
                                                 choices=['y', 'n', 'del'], default='y')
        if 'y' == confirm:
            access_token = read_access_token(app_token_file)
            if not check_access_token_out(access_token, api):
                os.remove(app_token_file)
                return None, True

            return access_token, False
        elif 'del' == confirm:
            os.remove(app_token_file)

    return None, True


def parse_access_tokens(tokens_file, api=False):
    if not os.path.isfile(tokens_file):
        raise Exception('Error: {} is not a file.'.format(tokens_file))

    import json
    with open(tokens_file, 'r') as f:
        tokens = json.load(f)

    valid_tokens = {}
    for key, value in tokens.items():
        if not check_access_token_out(value, api=api):
            Console.error('### Access token id: {}'.format(key))
            continue
        valid_tokens[key] = value

    if not valid_tokens:
        Console.error('### No valid access tokens.')
        return None

    return valid_tokens


def main():
    global __show_verbose

    api_prefix = getenv('CHATGPT_API_PREFIX', default_api_prefix())

    Console.debug_b(
        '''
            Pandora - A command-line interface to ChatGPT
            Github: https://github.com/pengzhile/pandora
            Get access token: {}/auth
            Version: {}'''.format(api_prefix, __version__), end=''
    )

    parser = argparse.ArgumentParser()
    parser.add_argument(
        '-p',
        '--proxy',
        help='Use a proxy. Format: protocol://user:pass@ip:port',
        required=False,
        type=str,
        default=None,
    )
    parser.add_argument(
        '-t',
        '--token_file',
        help='Specify an access token file and login with your access token.',
        required=False,
        type=str,
        default=None,
    )
    parser.add_argument(
        '--tokens_file',
        help='Specify an access tokens json file.',
        required=False,
        type=str,
        default=None,
    )
    parser.add_argument(
        '-s',
        '--server',
        help='Start as a proxy server. Format: ip:port, default: 127.0.0.1:8008',
        required=False,
        type=str,
        default=None,
        action='store',
        nargs='?',
        const='127.0.0.1:8008',
    )
    parser.add_argument(
        '--threads',
        help='Define the number of server workers, default: 8',
        required=False,
        type=int,
        default=8,
    )
    parser.add_argument(
        '-a',
        '--api',
        help='Use gpt-3.5-turbo chat api. Note: OpenAI will bill you.',
        action='store_true',
    )
    parser.add_argument(
        '-l',
        '--local',
        help='Login locally. Pay attention to the risk control of the login ip!',
        action='store_true',
    )
    parser.add_argument(
        '-v',
        '--verbose',
        help='Show exception traceback.',
        action='store_true',
    )
    args, _ = parser.parse_known_args()
    __show_verbose = args.verbose

    Console.debug_b(''', Mode: {}, Engine: {}
        '''.format('server' if args.server else 'cli', 'turbo' if args.api else 'free'))

    if args.api:
        try:
            from .openai.token import gpt_num_tokens
            from .migrations.migrate import do_migrate

            do_migrate()
        except (ImportError, ModuleNotFoundError):
            Console.error_bh('### You need `pip install Pandora-ChatGPT[api]` to support API mode.')
            return

    access_tokens = parse_access_tokens(args.tokens_file, args.api) if args.tokens_file else None

    if not access_tokens:
        access_token, need_save = confirm_access_token(args.token_file, args.server, args.api)
        if not access_token:
            Console.info_b('Please enter your email and password to log in ChatGPT!')
            if not args.local:
                Console.warn('We login via {}'.format(api_prefix))

            email = getenv('OPENAI_EMAIL') or Prompt.ask('  Email')
            password = getenv('OPENAI_PASSWORD') or Prompt.ask('  Password', password=True)
            mfa = getenv('OPENAI_MFA_CODE') or Prompt.ask('  MFA Code(Optional if not set)')
            Console.warn('### Do login, please wait...')
            access_token = Auth0(email, password, args.proxy, mfa=mfa).auth(args.local)

        if not check_access_token_out(access_token, args.api):
            return

        if need_save:
            if args.server or Confirm.ask('Do you want to save your access token for the next login?', default=True):
                save_access_token(access_token)

        access_tokens = {'default': access_token}

    if args.api:
        from .turbo.chat import TurboGPT

        chatgpt = TurboGPT(access_tokens, args.proxy)
    else:
        chatgpt = ChatGPT(access_tokens, args.proxy)

    if args.server:
        return ChatBotServer(chatgpt, args.verbose).run(args.server, args.threads)

    ChatBotLegacy(chatgpt).run()


def run():
    hook_except_handle()

    try:
        main()
    except Exception as e:
        Console.error_bh('### Error occurred: ' + str(e))

        if __show_verbose:
            logger.exception('Exception occurred.')

src/pandora/launcher.py
cloud804d/mirrors-pengzhile-pandora
# -*- coding: utf-8 -*-

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from ..exts.config import DATABASE_URI

engine = create_engine(DATABASE_URI, echo=False)

Session = sessionmaker(bind=engine)

session = Session()

src/pandora/migrations/database.py
cloud804d/mirrors-pengzhile-pandora
# -*- coding: utf-8 -*-

from os import makedirs, path
from os.path import abspath, join, dirname

from yoyo import get_backend
from yoyo import read_migrations

from ..exts.config import DATABASE_URI, USER_CONFIG_DIR


def do_migrate():
    if not path.exists(USER_CONFIG_DIR):
        makedirs(USER_CONFIG_DIR)

    url = 'mysql:{}'.format(DATABASE_URI[14:]) if 'mysql+pymysql:' == DATABASE_URI[0:14] else DATABASE_URI
    backend = get_backend(url)
    migrations = read_migrations(abspath(join(dirname(__file__), 'scripts')))

    with backend.lock():
        backend.apply_migrations(backend.to_apply(migrations))

src/pandora/migrations/migrate.py
cloud804d/mirrors-pengzhile-pandora
# -*- coding: utf-8 -*-

from datetime import datetime as dt

from sqlalchemy import func, Column, Text, Integer
from sqlalchemy.orm import DeclarativeBase

from ..migrations.database import session


class Base(DeclarativeBase):
    pass


class ConversationOfficial(Base):
    __tablename__ = 'conversation_official'

    conversation_id = Column(Text, primary_key=True, autoincrement=False)
    title = Column(Text, nullable=False)
    create_time = Column(Integer, nullable=False)

    @staticmethod
    def get_list(offset, limit):
        total = session.query(func.count(ConversationOfficial.conversation_id)).scalar()
        return total, session.query(ConversationOfficial).order_by(ConversationOfficial.create_time.desc()).limit(
            limit).offset(offset).all()

    @staticmethod
    def get(conversation_id):
        return session.query(ConversationOfficial).get(conversation_id)

    def save(self):
        session.commit()
        return self

    def new(self):
        session.add(self)
        session.commit()

        return self

    @staticmethod
    def delete(conversation_id):
        session.query(ConversationOfficial).filter(ConversationOfficial.conversation_id == conversation_id).delete()
        session.commit()

    @staticmethod
    def clear():
        session.query(ConversationOfficial).delete()
        session.commit()

    @staticmethod
    def new_conversation(conversation_id, title=None):
        conv = ConversationOfficial.get(conversation_id)

        if not conv:
            conv = ConversationOfficial()
            conv.conversation_id = conversation_id
            conv.title = title or 'New chat'
            conv.create_time = dt.now().timestamp()
            conv.new()
        else:
            conv.title = title or 'New chat'
            conv.save()

    @staticmethod
    def wrap_conversation_list(offset, limit):
        total, items = ConversationOfficial.get_list(offset, limit)

        stripped = []
        for item in items:
            stripped.append({
                'id': item.conversation_id,
                'title': item.title,
                'create_time': dt.utcfromtimestamp(item.create_time).isoformat(),
            })

        return {'items': stripped, 'total': total, 'limit': limit, 'offset': offset}


class ConversationInfo(Base):
    __tablename__ = 'conversation_info'

    conversation_id = Column(Text, primary_key=True, autoincrement=False)
    title = Column(Text, nullable=False)
    create_time = Column(Integer, nullable=False)
    current_node = Column(Text, nullable=True)

    @staticmethod
    def get_list(offset, limit):
        total = session.query(func.count(ConversationInfo.conversation_id)).scalar()
        return total, session.query(ConversationInfo).order_by(ConversationInfo.create_time.desc()).limit(
            limit).offset(offset).all()

    @staticmethod
    def get(conversation_id):
        return session.query(ConversationInfo).get(conversation_id)

    def new(self):
        session.add(self)
        session.commit()

        return self

    @staticmethod
    def delete(conversation_id):
        session.query(ConversationInfo).filter(ConversationInfo.conversation_id == conversation_id).delete()
        session.commit()

    @staticmethod
    def clear():
        session.query(ConversationInfo).delete()
        session.commit()


class PromptInfo(Base):
    __tablename__ = 'prompt_info'

    prompt_id = Column(Text, primary_key=True, autoincrement=False)
    conversation_id = Column(Text, primary_key=True, autoincrement=False)
    model = Column(Text, nullable=True)
    parent_id = Column(Text, nullable=True)
    role = Column(Text, nullable=True)
    content = Column(Text, nullable=True)
    create_time = Column(Integer, nullable=False)

    @staticmethod
    def list_by_conversation_id(conversation_id):
        return session.query(PromptInfo).filter(PromptInfo.conversation_id == conversation_id).all()

    def new(self):
        session.add(self)
        session.commit()

        return self

    @staticmethod
    def clear():
        session.query(PromptInfo).delete()
        session.commit()

src/pandora/migrations/models.py
cloud804d/mirrors-pengzhile-pandora
# -*- coding: utf-8 -*-

import asyncio
import json
import queue as block_queue
import threading
from os import getenv

import httpx
import requests
from certifi import where

from .. import __version__
from ..exts.config import default_api_prefix


class API:
    def __init__(self, proxy, ca_bundle):
        self.proxy = proxy
        self.ca_bundle = ca_bundle

    @staticmethod
    def wrap_stream_out(generator, status):
        if status != 200:
            for line in generator:
                yield json.dumps(line)

            return

        for line in generator:
            yield b'data: ' + json.dumps(line).encode('utf-8') + b'\n\n'

        yield b'data: [DONE]\n\n'

    async def __process_sse(self, resp):
        yield resp.status_code
        yield resp.headers

        if resp.status_code != 200:
            yield await self.__process_sse_except(resp)
            return

        async for utf8_line in resp.aiter_lines():
            if 'data: [DONE]' == utf8_line[0:12]:
                break

            if 'data: {"message":' == utf8_line[0:17] or 'data: {"id":' == utf8_line[0:12]:
                yield json.loads(utf8_line[6:])

    @staticmethod
    async def __process_sse_except(resp):
        result = b''
        async for line in resp.aiter_bytes():
            result += line

        return json.loads(result.decode('utf-8'))

    @staticmethod
    def __generate_wrap(queue, thread, event):
        while True:
            try:
                item = queue.get()
                if item is None:
                    break

                yield item
            except BaseException as e:
                event.set()
                thread.join()

                if isinstance(e, GeneratorExit):
                    raise e

    async def _do_request_sse(self, url, headers, data, queue, event):
        async with httpx.AsyncClient(verify=self.ca_bundle, proxies=self.proxy) as client:
            async with client.stream('POST', url, json=data, headers=headers, timeout=600) as resp:
                async for line in self.__process_sse(resp):
                    queue.put(line)

                    if event.is_set():
                        await client.aclose()
                        break

                queue.put(None)

    def _request_sse(self, url, headers, data):
        queue, e = block_queue.Queue(), threading.Event()
        t = threading.Thread(target=asyncio.run, args=(self._do_request_sse(url, headers, data, queue, e),))
        t.start()

        return queue.get(), queue.get(), self.__generate_wrap(queue, t, e)


class ChatGPT(API):
    def __init__(self, access_tokens: dict, proxy=None):
        self.access_tokens = access_tokens
        self.access_token_key_list = list(access_tokens)
        self.default_token_key = self.access_token_key_list[0]
        self.session = requests.Session()
        self.req_kwargs = {
            'proxies': {
                'http': proxy,
                'https': proxy,
            } if proxy else None,
            'verify': where(),
            'timeout': 100,
            'allow_redirects': False,
        }

        self.user_agent = 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) ' \
                          'Pandora/{} Safari/537.36'.format(__version__)

        super().__init__(proxy, self.req_kwargs['verify'])

    def __get_headers(self, token_key=None):
        return {
            'Authorization': 'Bearer ' + self.get_access_token(token_key),
            'User-Agent': self.user_agent,
            'Content-Type': 'application/json',
        }

    @staticmethod
    def __get_api_prefix():
        return getenv('CHATGPT_API_PREFIX', default_api_prefix())

    def get_access_token(self, token_key=None):
        return self.access_tokens[token_key or self.default_token_key]

    def list_token_keys(self):
        return self.access_token_key_list

    def list_models(self, raw=False, token=None):
        url = '{}/api/models'.format(self.__get_api_prefix())
        resp = self.session.get(url=url, headers=self.__get_headers(token), **self.req_kwargs)

        if raw:
            return resp

        if resp.status_code != 200:
            raise Exception('list models failed: ' + self.__get_error(resp))

        result = resp.json()
        if 'models' not in result:
            raise Exception('list models failed: ' + resp.text)

        return result['models']

    def list_conversations(self, offset, limit, raw=False, token=None):
        url = '{}/api/conversations?offset={}&limit={}'.format(self.__get_api_prefix(), offset, limit)
        resp = self.session.get(url=url, headers=self.__get_headers(token), **self.req_kwargs)

        if raw:
            return resp

        if resp.status_code != 200:
            raise Exception('list conversations failed: ' + self.__get_error(resp))

        return resp.json()

    def get_conversation(self, conversation_id, raw=False, token=None):
        url = '{}/api/conversation/{}'.format(self.__get_api_prefix(), conversation_id)
        resp = self.session.get(url=url, headers=self.__get_headers(token), **self.req_kwargs)

        if raw:
            return resp

        if resp.status_code != 200:
            raise Exception('get conversation failed: ' + self.__get_error(resp))

        return resp.json()

    def clear_conversations(self, raw=False, token=None):
        data = {
            'is_visible': False,
        }

        url = '{}/api/conversations'.format(self.__get_api_prefix())
        resp = self.session.patch(url=url, headers=self.__get_headers(token), json=data, **self.req_kwargs)

        if raw:
            return resp

        if resp.status_code != 200:
            raise Exception('clear conversations failed: ' + self.__get_error(resp))

        result = resp.json()
        if 'success' not in result:
            raise Exception('clear conversations failed: ' + resp.text)

        return result['success']

    def del_conversation(self, conversation_id, raw=False, token=None):
        data = {
            'is_visible': False,
        }

        return self.__update_conversation(conversation_id, data, raw, token)

    def gen_conversation_title(self, conversation_id, model, message_id, raw=False, token=None):
        url = '{}/api/conversation/gen_title/{}'.format(self.__get_api_prefix(), conversation_id)
        data = {
            'model': model,
            'message_id': message_id,
        }
        resp = self.session.post(url=url, headers=self.__get_headers(token), json=data, **self.req_kwargs)

        if raw:
            return resp

        if resp.status_code != 200:
            raise Exception('gen title failed: ' + self.__get_error(resp))

        result = resp.json()
        if 'title' not in result:
            raise Exception('gen title failed: ' + resp.text)

        return result['title']

    def set_conversation_title(self, conversation_id, title, raw=False, token=None):
        data = {
            'title': title,
        }

        return self.__update_conversation(conversation_id, data, raw, token)

    def talk(self, prompt, model, message_id, parent_message_id, conversation_id=None, stream=True, token=None):
        data = {
            'action': 'next',
            'messages': [
                {
                    'id': message_id,
                    'role': 'user',
                    'author': {
                        'role': 'user',
                    },
                    'content': {
                        'content_type': 'text',
                        'parts': [prompt],
                    },
                }
            ],
            'model': model,
            'parent_message_id': parent_message_id,
        }

        if conversation_id:
            data['conversation_id'] = conversation_id

        return self.__request_conversation(data, token)

    def goon(self, model, parent_message_id, conversation_id, stream=True, token=None):
        data = {
            'action': 'continue',
            'conversation_id': conversation_id,
            'model': model,
            'parent_message_id': parent_message_id,
        }

        return self.__request_conversation(data, token)

    def regenerate_reply(self, prompt, model, conversation_id, message_id, parent_message_id, stream=True, token=None):
        data = {
            'action': 'variant',
            'messages': [
                {
                    'id': message_id,
                    'role': 'user',
                    'author': {
                        'role': 'user',
                    },
                    'content': {
                        'content_type': 'text',
                        'parts': [prompt],
                    },
                }
            ],
            'model': model,
            'conversation_id': conversation_id,
            'parent_message_id': parent_message_id,
        }

        return self.__request_conversation(data, token)

    def __request_conversation(self, data, token=None):
        url = '{}/api/conversation'.format(self.__get_api_prefix())
        headers = {**self.session.headers, **self.__get_headers(token), 'Accept': 'text/event-stream'}

        return self._request_sse(url, headers, data)

    def __update_conversation(self, conversation_id, data, raw=False, token=None):
        url = '{}/api/conversation/{}'.format(self.__get_api_prefix(), conversation_id)
        resp = self.session.patch(url=url, headers=self.__get_headers(token), json=data, **self.req_kwargs)

        if raw:
            return resp

        if resp.status_code != 200:
            raise Exception('update conversation failed: ' + self.__get_error(resp))

        result = resp.json()
        if 'success' not in result:
            raise Exception('update conversation failed: ' + resp.text)

        return result['success']

    @staticmethod
    def __get_error(resp):
        try:
            return str(resp.json()['detail'])
        except:
            return resp.text


class ChatCompletion(API):
    def __init__(self, proxy=None):
        self.session = requests.Session()
        self.req_kwargs = {
            'proxies': {
                'http': proxy,
                'https': proxy,
            } if proxy else None,
            'verify': where(),
            'timeout': 600,
            'allow_redirects': False,
        }

        self.user_agent = 'pandora/{}'.format(__version__)

        super().__init__(proxy, self.req_kwargs['verify'])

    def __get_headers(self, api_key):
        return {
            'Authorization': 'Bearer ' + api_key,
            'User-Agent': self.user_agent,
            'Content-Type': 'application/json',
        }

    def request(self, api_key, model, messages, stream=True, **kwargs):
        data = {
            'model': model,
            'messages': messages,
            **kwargs,
            'stream': stream,
        }

        return self.__request_conversation(api_key, data, stream)

    def __request_conversation(self, api_key, data, stream):
        default = default_api_prefix()

        if api_key.startswith('fk-') or api_key.startswith('pk-'):
            prefix = default
        else:
            prefix = getenv('OPENAI_API_PREFIX', default)
        url = '{}/v1/chat/completions'.format(prefix)

        if stream:
            headers = {**self.__get_headers(api_key), 'Accept': 'text/event-stream'}
            return self._request_sse(url, headers, data)

        resp = self.session.post(url=url, headers=self.__get_headers(api_key), json=data, **self.req_kwargs)

        def __generate_wrap():
            yield resp.json()

        return resp.status_code, resp.headers, __generate_wrap()

src/pandora/openai/api.py
cloud804d/mirrors-pengzhile-pandora
# -*- coding: utf-8 -*-

import datetime
import re
from datetime import datetime as dt
from urllib.parse import urlparse, parse_qs

import requests
from certifi import where

from ..exts.config import default_api_prefix


class Auth0:
    def __init__(self, email: str, password: str, proxy: str = None, use_cache: bool = True, mfa: str = None):
        self.session_token = None
        self.email = email
        self.password = password
        self.use_cache = use_cache
        self.mfa = mfa
        self.session = requests.Session()
        self.req_kwargs = {
            'proxies': {
                'http': proxy,
                'https': proxy,
            } if proxy else None,
            'verify': where(),
            'timeout': 100,
        }
        self.access_token = None
        self.refresh_token = None
        self.expires = None
        self.user_agent = 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) ' \
                          'Chrome/109.0.0.0 Safari/537.36'

    @staticmethod
    def __check_email(email: str):
        regex = r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,7}\b'
        return re.fullmatch(regex, email)

    def auth(self, login_local=False) -> str:
        if self.use_cache and self.access_token and self.expires and self.expires > dt.now():
            return self.access_token

        if not self.__check_email(self.email) or not self.password:
            raise Exception('invalid email or password.')

        return self.__part_one() if login_local else self.get_access_token_proxy()

    def get_refresh_token(self):
        return self.refresh_token

    def __part_one(self) -> str:
        url = '{}/auth/preauth'.format(default_api_prefix())
        resp = self.session.get(url, allow_redirects=False, **self.req_kwargs)

        if resp.status_code == 200:
            json = resp.json()
            if 'preauth_cookie' not in json or not json['preauth_cookie']:
                raise Exception('Get preauth cookie failed.')

            return self.__part_two(json['preauth_cookie'])
        else:
            raise Exception('Error request preauth.')

    def __part_two(self, preauth: str) -> str:
        code_challenge = 'w6n3Ix420Xhhu-Q5-mOOEyuPZmAsJHUbBpO8Ub7xBCY'
        code_verifier = 'yGrXROHx_VazA0uovsxKfE263LMFcrSrdm4SlC-rob8'

        url = 'https://auth0.openai.com/authorize?client_id=pdlLIX2Y72MIl2rhLhTE9VV9bN905kBh&audience=https%3A%2F' \
              '%2Fapi.openai.com%2Fv1&redirect_uri=com.openai.chat%3A%2F%2Fauth0.openai.com%2Fios%2Fcom.openai.chat' \
              '%2Fcallback&scope=openid%20email%20profile%20offline_access%20model.request%20model.read' \
              '%20organization.read%20offline&response_type=code&code_challenge={}' \
              '&code_challenge_method=S256&prompt=login&preauth_cookie={}'.format(code_challenge, preauth)
        return self.__part_three(code_verifier, url)

    def __part_three(self, code_verifier, url: str) -> str:
        headers = {
            'User-Agent': self.user_agent,
            'Referer': 'https://ios.chat.openai.com/',
        }
        resp = self.session.get(url, headers=headers, allow_redirects=True, **self.req_kwargs)

        if resp.status_code == 200:
            try:
                url_params = parse_qs(urlparse(resp.url).query)
                state = url_params['state'][0]
                return self.__part_four(code_verifier, state)
            except IndexError as exc:
                raise Exception('Rate limit hit.') from exc
        else:
            raise Exception('Error request login url.')

    def __part_four(self, code_verifier: str, state: str) -> str:
        url = 'https://auth0.openai.com/u/login/identifier?state=' + state
        headers = {
            'User-Agent': self.user_agent,
            'Referer': url,
            'Origin': 'https://auth0.openai.com',
        }
        data = {
            'state': state,
            'username': self.email,
            'js-available': 'true',
            'webauthn-available': 'true',
            'is-brave': 'false',
            'webauthn-platform-available': 'false',
            'action': 'default',
        }
        resp = self.session.post(url, headers=headers, data=data, allow_redirects=False, **self.req_kwargs)

        if resp.status_code == 302:
            return self.__part_five(code_verifier, state)
        else:
            raise Exception('Error check email.')

    def __part_five(self, code_verifier: str, state: str) -> str:
        url = 'https://auth0.openai.com/u/login/password?state=' + state
        headers = {
            'User-Agent': self.user_agent,
            'Referer': url,
            'Origin': 'https://auth0.openai.com',
        }
        data = {
            'state': state,
            'username': self.email,
            'password': self.password,
            'action': 'default',
        }

        resp = self.session.post(url, headers=headers, data=data, allow_redirects=False, **self.req_kwargs)
        if resp.status_code == 302:
            location = resp.headers['Location']
            if not location.startswith('/authorize/resume?'):
                raise Exception('Login failed.')

            return self.__part_six(code_verifier, location, url)

        if resp.status_code == 400:
            raise Exception('Wrong email or password.')
        else:
            raise Exception('Error login.')

    def __part_six(self, code_verifier: str, location: str, ref: str) -> str:
        url = 'https://auth0.openai.com' + location
        headers = {
            'User-Agent': self.user_agent,
            'Referer': ref,
        }

        resp = self.session.get(url, headers=headers, allow_redirects=False, **self.req_kwargs)
        if resp.status_code == 302:
            location = resp.headers['Location']
            if location.startswith('/u/mfa-otp-challenge?'):
                if not self.mfa:
                    raise Exception('MFA required.')
                return self.__part_seven(code_verifier, location)

            if not location.startswith('com.openai.chat://auth0.openai.com/ios/com.openai.chat/callback?'):
                raise Exception('Login callback failed.')

            return self.get_access_token(code_verifier, resp.headers['Location'])

        raise Exception('Error login.')

    def __part_seven(self, code_verifier: str, location: str) -> str:
        url = 'https://auth0.openai.com' + location
        data = {
            'state': parse_qs(urlparse(url).query)['state'][0],
            'code': self.mfa,
            'action': 'default',
        }
        headers = {
            'User-Agent': self.user_agent,
            'Referer': url,
            'Origin': 'https://auth0.openai.com',
        }

        resp = self.session.post(url, headers=headers, data=data, allow_redirects=False, **self.req_kwargs)
        if resp.status_code == 302:
            location = resp.headers['Location']
            if not location.startswith('/authorize/resume?'):
                raise Exception('MFA failed.')

            return self.__part_six(code_verifier, location, url)

        if resp.status_code == 400:
            raise Exception('Wrong MFA code.')
        else:
            raise Exception('Error login.')

    def __parse_access_token(self, resp):
        if resp.status_code == 200:
            json = resp.json()
            if 'access_token' not in json:
                raise Exception('Get access token failed, maybe you need a proxy.')

            if 'refresh_token' in json:
                self.refresh_token = json['refresh_token']

            self.access_token = json['access_token']
            self.expires = dt.utcnow() + datetime.timedelta(seconds=json['expires_in']) - datetime.timedelta(minutes=5)
            return self.access_token
        else:
            raise Exception(resp.text)

    def get_access_token(self, code_verifier: str, callback_url: str) -> str:
        url_params = parse_qs(urlparse(callback_url).query)

        if 'error' in url_params:
            error = url_params['error'][0]
            error_description = url_params['error_description'][0] if 'error_description' in url_params else ''
            raise Exception('{}: {}'.format(error, error_description))

        if 'code' not in url_params:
            raise Exception('Error get code from callback url.')

        url = 'https://auth0.openai.com/oauth/token'
        headers = {
            'User-Agent': self.user_agent,
        }
        data = {
            'redirect_uri': 'com.openai.chat://auth0.openai.com/ios/com.openai.chat/callback',
            'grant_type': 'authorization_code',
            'client_id': 'pdlLIX2Y72MIl2rhLhTE9VV9bN905kBh',
            'code': url_params['code'][0],
            'code_verifier': code_verifier,
        }
        resp = self.session.post(url, headers=headers, json=data, allow_redirects=False, **self.req_kwargs)

        return self.__parse_access_token(resp)

    def get_access_token_proxy(self) -> str:
        url = '{}/auth/login'.format(default_api_prefix())
        headers = {
            'User-Agent': self.user_agent,
        }
        data = {
            'username': self.email,
            'password': self.password,
            'mfa_code': self.mfa,
        }
        resp = self.session.post(url=url, headers=headers, data=data, allow_redirects=False, **self.req_kwargs)

        return self.__parse_access_token(resp)

src/pandora/openai/auth.py
cloud804d/mirrors-pengzhile-pandora
# -*- coding: utf-8 -*-

import tiktoken


def gpt_num_tokens(messages, model='gpt-3.5-turbo'):
    encoding = tiktoken.encoding_for_model(model)

    num_tokens = 0
    for message in messages:
        num_tokens += 4
        for key, value in message.items():
            num_tokens += len(encoding.encode(value))
            if 'name' == key:
                num_tokens -= 1
    num_tokens += 2

    return num_tokens

src/pandora/openai/token.py
cloud804d/mirrors-pengzhile-pandora
# -*- coding: utf-8 -*-

import os

from rich.console import Console as RichConsole
from rich.theme import Theme


class Console:
    __theme = Theme({
        'info': 'white',
        'info_b': 'white bold',
        'debug': 'cyan',
        'debug_b': 'cyan bold',
        'warn': 'yellow',
        'warn_b': 'yellow bold',
        'error': 'red',
        'error_b': 'red bold',
        'success': 'green',
        'success_b': 'green bold',
    })

    __console = RichConsole(theme=__theme)

    @staticmethod
    def clear():
        os.system('cls' if 'nt' == os.name else 'clear')

    @staticmethod
    def print(msg):
        Console.__console.print(msg)

    @staticmethod
    def info(text: str, highlight=False, bold=False, end='\n'):
        Console.__console.print(text, style='info_b' if bold else 'info', highlight=highlight, end=end, markup=False)

    @staticmethod
    def info_b(text: str, highlight=False, end='\n'):
        Console.info(text, highlight, True, end)

    @staticmethod
    def info_h(text: str, bold=False, end='\n'):
        Console.info(text, True, bold, end)

    @staticmethod
    def info_bh(text: str, end='\n'):
        Console.info(text, True, True, end)

    @staticmethod
    def debug(text: str, highlight=False, bold=False, end='\n'):
        Console.__console.print(text, style='debug_b' if bold else 'debug', highlight=highlight, end=end, markup=False)

    @staticmethod
    def debug_b(text: str, highlight=False, end='\n'):
        Console.debug(text, highlight, True, end)

    @staticmethod
    def debug_h(text: str, bold=False, end='\n'):
        Console.debug(text, True, bold, end)

    @staticmethod
    def debug_bh(text: str, end='\n'):
        Console.debug(text, True, True, end)

    @staticmethod
    def error(text: str, highlight=False, bold=False, end='\n'):
        Console.__console.print(text, style='error_b' if bold else 'error', highlight=highlight, end=end, markup=False)

    @staticmethod
    def error_b(text: str, highlight=False, end='\n'):
        Console.error(text, highlight, True, end)

    @staticmethod
    def error_h(text: str, bold=False, end='\n'):
        Console.error(text, True, bold, end)

    @staticmethod
    def error_bh(text: str, end='\n'):
        Console.error(text, True, True, end)

    @staticmethod
    def success(text: str, highlight=False, bold=False, end='\n'):
        Console.__console.print(text, style='success_b' if bold else 'success', highlight=highlight, end=end,
                                markup=False)

    @staticmethod
    def success_b(text: str, highlight=False, end='\n'):
        Console.success(text, highlight, True, end)

    @staticmethod
    def success_h(text: str, bold=False, end='\n'):
        Console.success(text, True, bold, end)

    @staticmethod
    def success_bh(text: str, end='\n'):
        Console.success(text, True, True, end)

    @staticmethod
    def warn(text: str, highlight=False, bold=False, end='\n'):
        Console.__console.print(text, style='warn_b' if bold else 'warn', highlight=highlight, end=end, markup=False)

    @staticmethod
    def warn_b(text: str, highlight=False, end='\n'):
        Console.warn(text, highlight, True, end)

    @staticmethod
    def warn_h(text: str, bold=False, end='\n'):
        Console.warn(text, True, bold, end)

    @staticmethod
    def warn_bh(text: str, end='\n'):
        Console.warn(text, True, True, end)

src/pandora/openai/utils.py
cloud804d/mirrors-pengzhile-pandora
# -*- coding: utf-8 -*-

import uuid
from datetime import datetime as dt


class Prompt:
    def __init__(self, prompt_id=None, role=None, content=None, parent=None):
        self.prompt_id = prompt_id or str(uuid.uuid4())
        self.parent_id = None
        self.role = role
        self.content = content
        self.children = []
        self.create_time = dt.now().timestamp()

        if parent:
            self.parent_id = parent.prompt_id
            parent.add_child(self.prompt_id)

    def add_child(self, prompt_id):
        self.children.append(prompt_id)

    def get_message(self, end=True):
        return None

    def get_info(self):
        return {
            'id': self.prompt_id,
            'message': self.get_message(),
            'parent': self.parent_id,
            'children': self.children
        }


class SystemPrompt(Prompt):
    def __init__(self, content, parent):
        super().__init__(role='system', content=content, parent=parent)

    def get_message(self, end=True):
        return {
            'id': self.prompt_id,
            'author': {
                'role': self.role,
                'name': None,
                'metadata': {}
            },
            'create_time': self.create_time,
            'update_time': None,
            'content': {
                'content_type': 'text',
                'parts': ['']
            },
            'end_turn': True,
            'weight': 1.0,
            'metadata': {},
            'recipient': 'all'
        }


class UserPrompt(Prompt):
    def __init__(self, prompt_id, content, parent):
        super().__init__(prompt_id=prompt_id, role='user', content=content, parent=parent)

    def get_message(self, end=True):
        return {
            'id': self.prompt_id,
            'author': {
                'role': self.role,
                'name': None,
                'metadata': {}
            },
            'create_time': self.create_time,
            'update_time': None,
            'content': {
                'content_type': 'text',
                'parts': [self.content]
            },
            'end_turn': None,
            'weight': 1.0,
            'metadata': {
                'timestamp_': 'absolute',
                'message_type': None
            },
            'recipient': 'all'
        }


class GptPrompt(Prompt):
    def __init__(self, parent, model):
        super().__init__(role='assistant', content='', parent=parent)
        self.model = model

    def append_content(self, content):
        self.content += content

        return self

    def get_message(self, end=True):
        return {
            'id': self.prompt_id,
            'author': {
                'role': self.role,
                'name': None,
                'metadata': {}
            },
            'create_time': self.create_time,
            'update_time': None,
            'content': {
                'content_type': 'text',
                'parts': [self.content]
            },
            'end_turn': False if end else None,
            'weight': 1.0,
            'metadata': {
                'message_type': None,
                'model_slug': self.model,
                'finish_details': {
                    'type': 'stop'
                } if end else None,
                'timestamp_': 'absolute'
            },
            'recipient': 'all'
        }


class Conversation:
    def __init__(self):
        self.conversation_id = str(uuid.uuid4())
        self.title = 'New chat'
        self.create_time = dt.now().timestamp()
        self.current_node = None
        self.prompts = {}

    def add_prompt(self, prompt):
        self.prompts[prompt.prompt_id] = prompt
        self.current_node = prompt.prompt_id

        return prompt

    def get_prompt(self, prompt_id):
        return self.prompts.get(prompt_id)

    def get_prompts(self):
        return self.prompts

    def set_title(self, title):
        self.title = title

    def get_title(self):
        return self.title

    def get_messages_directly(self, message_id):
        messages = []
        while True:
            prompt = self.get_prompt(message_id)
            if not prompt.parent_id:
                break

            messages.insert(0, {
                'role': prompt.role,
                'content': prompt.content
            })
            message_id = prompt.parent_id

        return messages

    def get_messages(self, message_id, model):
        messages = []
        user_prompt = None
        while True:
            prompt = self.get_prompt(message_id)
            if not prompt.parent_id:
                break

            if not user_prompt and isinstance(prompt, UserPrompt):
                user_prompt = prompt

            messages.insert(0, {
                'role': prompt.role,
                'content': prompt.content
            })
            message_id = prompt.parent_id

        return user_prompt, self.add_prompt(GptPrompt(user_prompt, model)), messages

    def get_info(self):
        mapping = {}
        for prompt_id in self.prompts:
            mapping[prompt_id] = self.prompts[prompt_id].get_info()

        return {
            'title': self.title,
            'create_time': self.create_time,
            'mapping': mapping,
            'moderation_results': [],
            'current_node': self.current_node,
        }


class Conversations:
    def __init__(self):
        self.__data = []

    def list(self, offset, limit):
        return len(self.__data), self.__data[offset: limit]

    def clear(self):
        self.__data = []

    def delete(self, conversation):
        self.__data = [x for x in self.__data if conversation.conversation_id != x.conversation_id]

    def new(self):
        conversation = Conversation()
        self.__data.insert(0, conversation)

        return conversation

    def get(self, conversation_id):
        for x in self.__data:
            if x.conversation_id == conversation_id:
                return x

        return None

    def guard_get(self, conversation_id):
        conversation = self.get(conversation_id)
        if not conversation:
            raise Exception('Can\'t load conversation {}'.format(conversation_id))

        return conversation

src/pandora/turbo/base.py
cloud804d/mirrors-pengzhile-pandora
# -*- coding: utf-8 -*-

import json
from datetime import datetime as dt
from os import getenv

from requests import Response

from .base import Conversations, UserPrompt, Prompt, SystemPrompt
from ..openai.api import ChatCompletion
from ..openai.token import gpt_num_tokens


class TurboGPT:
    DEFAULT_SYSTEM_PROMPT = 'You are ChatGPT, a large language model trained by OpenAI. ' \
                            'Answer as concisely as possible.\nKnowledge cutoff: 2021-09-01\n' \
                            'Current date: {}'.format(dt.now().strftime('%Y-%m-%d'))
    TITLE_PROMPT = 'Generate a brief title for our conversation.'
    MAX_TOKENS = {
        'gpt-3.5-turbo': 4096,
        'gpt-4': 8192,
        'gpt-4-32k': 32768,
    }
    FAKE_TOKENS = {
        'gpt-3.5-turbo': 8191,
        'gpt-4': 4095,
        'gpt-4-32k': 8195,
    }

    def __init__(self, api_keys: dict, proxy=None):
        self.api_keys = api_keys
        self.api_keys_key_list = list(api_keys)
        self.default_api_keys_key = self.api_keys_key_list[0]

        self.api = ChatCompletion(proxy)
        self.conversations_map = {}
        self.system_prompt = getenv('API_SYSTEM_PROMPT', self.DEFAULT_SYSTEM_PROMPT)

    def __get_conversations(self, api_keys_key=None):
        if api_keys_key is None:
            api_keys_key = self.default_api_keys_key

        if api_keys_key not in self.conversations_map:
            self.conversations_map[api_keys_key] = Conversations()

        return self.conversations_map[api_keys_key]

    def __is_fake_api(self, token=None):
        api_key = self.get_access_token(token)
        return api_key.startswith('fk-') or api_key.startswith('pk-')


    def get_access_token(self, token_key=None):
        return self.api_keys[token_key or self.default_api_keys_key]

    def list_token_keys(self):
        return self.api_keys_key_list

    def list_models(self, raw=False, token=None):
        fake_api = self.__is_fake_api(token)

        models = {
            'models': [
                {
                    'slug': 'gpt-3.5-turbo',
                    'max_tokens': self.FAKE_TOKENS['gpt-3.5-turbo'] if fake_api else self.MAX_TOKENS['gpt-3.5-turbo'],
                    'title': 'GPT-3.5',
                    'description': 'Turbo is the api model that powers ChatGPT',
                    'tags': []
                },
                {
                    'slug': 'gpt-4',
                    'max_tokens': self.FAKE_TOKENS['gpt-4'] if fake_api else self.MAX_TOKENS['gpt-4'],
                    'title': 'GPT-4',
                    'description': 'More capable than any GPT-3.5, able to do complex tasks, and optimized for chat',
                    'tags': []
                },
                {
                    'slug': 'gpt-4-32k',
                    'max_tokens': self.FAKE_TOKENS['gpt-4-32k'] if fake_api else self.MAX_TOKENS['gpt-4-32k'],
                    'title': 'GPT-4 32K',
                    'description': 'Same capabilities as the base gpt-4 mode but with 4x the context length',
                    'tags': []
                }
            ]
        }

        if raw:
            return self.__wrap_response(models)

        return models['models']

    def list_conversations(self, offset, limit, raw=False, token=None):
        offset = int(offset)
        limit = int(limit)
        total, items = self.__get_conversations(token).list(offset, limit)

        stripped = []
        for item in items:
            stripped.append({
                'id': item.conversation_id,
                'title': item.title,
                'create_time': dt.utcfromtimestamp(item.create_time).isoformat(),
            })

        result = {'items': stripped, 'total': total, 'limit': limit, 'offset': offset}

        if raw:
            return self.__wrap_response(result)

        return result

    def get_conversation(self, conversation_id, raw=False, token=None):
        def __shadow():
            try:
                conversation = self.__get_conversations(token).guard_get(conversation_id)
            except Exception as e:
                return self.__out_error(str(e), 404)

            return self.__wrap_response(conversation.get_info())

        resp = __shadow()

        if raw:
            return resp

        if resp.status_code != 200:
            raise Exception('get conversation failed: ' + resp.json()['detail'])

        return resp.json()

    def clear_conversations(self, raw=False, token=None):
        def __shadow():
            self.__get_conversations(token).clear()

            result = {
                'success': True
            }

            return self.__wrap_response(result)

        resp = __shadow()

        if raw:
            return resp

        return resp.json()['success']

    def del_conversation(self, conversation_id, raw=False, token=None):
        def __shadow():
            conversations = self.__get_conversations(token)

            try:
                conversation = conversations.guard_get(conversation_id)
            except Exception as e:
                return self.__out_error(str(e), 404)

            conversations.delete(conversation)

            result = {
                'success': True
            }

            return self.__wrap_response(result)

        resp = __shadow()

        if raw:
            return resp

        if resp.status_code != 200:
            raise Exception('delete conversation failed: ' + resp.json()['detail'])

        return resp.json()['success']

    def gen_conversation_title(self, conversation_id, model, message_id, raw=False, token=None):
        def __shadow():
            conversation = self.__get_conversations(token).get(conversation_id)
            if not conversation:
                return self.__out_error('Conversation not found', 404)

            if 'New chat' != conversation.title:
                message = {
                    'message': 'Conversation {} already has title \'{}\''.format(conversation_id, conversation.title)
                }
                return self.__wrap_response(message)

            messages = conversation.get_messages_directly(message_id)
            messages.append({'role': 'user', 'content': self.TITLE_PROMPT})

            status, header, generator = self.api.request(self.get_access_token(token), model, messages, False)
            last_ok, last = self.__get_completion(status, next(generator))

            if not last_ok:
                return self.__out_error(last['detail'], status)

            conversation.set_title(last.strip('"'))

            result = {
                'title': conversation.title
            }

            return self.__wrap_response(result)

        resp = __shadow()

        if raw:
            return resp

        if resp.status_code != 200:
            raise Exception('generate title failed: ' + resp.text)

        return resp.json()['title']

    def set_conversation_title(self, conversation_id, title, raw=False, token=None):
        def __shadow():
            try:
                conversation = self.__get_conversations(token).guard_get(conversation_id)
            except Exception as e:
                return self.__out_error(str(e), 404)

            conversation.set_title(title)

            result = {
                'success': True
            }

            return self.__wrap_response(result)

        resp = __shadow()

        if raw:
            return resp

        if resp.status_code != 200:
            raise Exception('update conversation failed: ' + resp.json()['detail'])

        return resp.json()['success']

    def talk(self, content, model, message_id, parent_message_id, conversation_id=None, stream=True, token=None):
        system_prompt = None
        if conversation_id:
            conversation = self.__get_conversations(token).get(conversation_id)
            if not conversation:
                return self.__out_error_stream('Conversation not found', 404)

            parent = conversation.get_prompt(parent_message_id)
        else:
            conversation = self.__get_conversations(token).new()
            parent = conversation.add_prompt(Prompt(parent_message_id))
            parent = system_prompt = conversation.add_prompt(SystemPrompt(self.system_prompt, parent))

        conversation.add_prompt(UserPrompt(message_id, content, parent))

        user_prompt, gpt_prompt, messages = conversation.get_messages(message_id, model)
        try:
            status, headers, generator = self.api.request(self.get_access_token(token), model,
                                                          self.__reduce_messages(messages, model, token), stream)
        except Exception as e:
            return self.__out_error_stream(str(e))

        def __out_generator():
            if 200 == status and system_prompt and stream:
                yield self.__out_stream(conversation, system_prompt)
                yield self.__out_stream(conversation, user_prompt)

            for line in generator:
                yield self.__map_conversation(status, conversation, gpt_prompt, line)

        return status, headers, __out_generator()

    def goon(self, model, parent_message_id, conversation_id, stream=True, token=None):
        return self.regenerate_reply(None, model, conversation_id, parent_message_id, None, stream, token)

    def regenerate_reply(self, prompt, model, conversation_id, message_id, parent_message_id, stream=True, token=None):
        if not conversation_id:
            return self.__out_error_stream('Miss conversation_id', 400)

        conversation = self.__get_conversations(token).get(conversation_id)
        if not conversation:
            return self.__out_error_stream('Conversation not found', 404)

        user_prompt, gpt_prompt, messages = conversation.get_messages(message_id, model)
        try:
            status, headers, generator = self.api.request(self.get_access_token(token), model,
                                                          self.__reduce_messages(messages, model, token), stream)
        except Exception as e:
            return self.__out_error_stream(str(e))

        def __out_generator():
            for line in generator:
                yield self.__map_conversation(status, conversation, gpt_prompt, line)

        return status, headers, __out_generator()

    def __reduce_messages(self, messages, model, token=None):
        max_tokens = self.FAKE_TOKENS[model] if self.__is_fake_api(token) else self.MAX_TOKENS[model]

        while gpt_num_tokens(messages) > max_tokens - 200:
            if len(messages) < 2:
                raise Exception('prompt too long')

            messages.pop(1)

        return messages

    def __out_error(self, error, status=500):
        result = {
            'detail': error
        }

        return self.__wrap_response(result, status)

    def __out_error_stream(self, error, status=500):
        resp = self.__out_error(error, status)

        def __generator():
            yield resp.json()

        return resp.status_code, resp.headers, __generator()

    @staticmethod
    def __out_stream(conversation, prompt, end=True):
        return {
            'message': prompt.get_message(end),
            'conversation_id': conversation.conversation_id,
            'error': None,
        }

    @staticmethod
    def __wrap_response(data, status=200):
        resp = Response()
        resp.status_code = status
        resp._content = json.dumps(data).encode('utf-8')
        resp.headers['Content-Type'] = 'application/json'

        return resp

    @staticmethod
    def __get_completion(status, data):
        if status != 200:
            error = data['error']['message'] if 'error' in data else 'Unknown error'
            result = {
                'detail': error
            }
            return False, result

        choice = data['choices'][0]
        if 'message' in choice:
            text = choice['message'].get('content', '')
        else:
            text = choice['delta'].get('content', '')

        return True, text

    def __map_conversation(self, status, conversation, gpt_prompt, data):
        success, result = self.__get_completion(status, data)
        if not success:
            return result

        choice = data['choices'][0]
        is_stop = 'stop' == choice['finish_reason']

        return self.__out_stream(conversation, gpt_prompt.append_content(result), is_stop)

src/pandora/turbo/chat.py
cloud804d/mirrors-pengzhile-pandora