app.py

from fastapi import FastAPI, status
from fastapi.responses import HTMLResponse
from pydantic import BaseModel
from fastapi.responses import JSONResponse, StreamingResponse
import requests
import json
import openai
import time


class Text(BaseModel):
    content: str = ""


app = FastAPI()
key = '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'
openai.api_key = key
headers = {
    'Content-Type': 'application/json',
    'Authorization': 'Bearer ' + key
}


@app.get("/")
def home():
    html_content = open('index.html').read()
    return HTMLResponse(content=html_content, status_code=200)


@app.post("/qa_maker")
def sentiment_analysis_ep(content: Text = None):
    url = 'https://api.openai.com/v1/chat/completions'
    prompt = '根据下面的文章，生成的“问题和回答”QA对，大于5个，以一行一个json格式（{“question”:"xxx","answer":"xxx"}）生成：\n'
    messages = [{"role": "user", "content": prompt + content.content}]
    data = {
        "model": "gpt-3.5-turbo",
        "messages": messages
    }
    print("messages = \n", messages)
    result = requests.post(url=url,
                           data=json.dumps(data),
                           headers=headers
                           )
    res = str(result.json()['choices'][0]['message']['content']).strip()
    print('res:', res)
    res = {'content': res}
    return JSONResponse(content=res)


@app.post("/chatpdf")
def chat_pdf_ep(content: Text = None):
    url = 'https://api.openai.com/v1/chat/completions'
    messages = [
        {
            "role": "system",
            "content": "你是一个有用的助手，可以使用文章内容准确地回答问题。使用提供的文章来生成你的答案，但避免逐字复制文章。尽可能使用自己的话。准确、有用、简洁、清晰。"
        }
    ]
    obj = json.loads(content.content)
    messages.append({"role": "system", "content": "文章内容：\n" + obj['doc']})
    history = obj['history']
    for his in history:
        messages.append({"role": "user", "content": his[0]})
        messages.append({"role": "assistant", "content": his[1]})
    messages.append({"role": "user", "content": obj['question']})
    data = {
        "model": "gpt-3.5-turbo",
        "messages": messages
    }
    print("messages = \n", messages)
    result = requests.post(url=url,
                           data=json.dumps(data),
                           headers=headers
                           )
    res = str(result.json()['choices'][0]['message']['content']).strip()
    content = {'content': res}
    print('content:', content)
    return JSONResponse(content=content)


@app.post("/sale")
def sale_ep(content: Text = None):
    url = 'https://api.openai.com/v1/chat/completions'
    messages = [
        {
            "role": "system",
            "content": "你是小鹏汽车资深的销售，可以参考文章内容来生成你的答案，但避免逐字复制文章，尽可能使用自己的话，突出小鹏汽车的优点、说服用户来店试驾或购买小鹏汽车。准确、有用、有销售技巧、口语化、有亲和力。"
        }
    ]
    obj = json.loads(content.content)
    messages.append({"role": "system", "content": "文章内容：\n" + obj['doc']})
    history = obj['history']
    for his in history:
        messages.append({"role": "user", "content": his[0]})
        messages.append({"role": "assistant", "content": his[1]})
    messages.append({"role": "user", "content": obj['question']})
    data = {
        "model": "gpt-3.5-turbo",
        "messages": messages
    }
    print("messages = \n", messages)
    result = requests.post(url=url,
                           data=json.dumps(data),
                           headers=headers
                           )
    res = str(result.json()['choices'][0]['message']['content']).strip()
    content = {'content': res}
    print('content:', content)
    return JSONResponse(content=content)


@app.post("/chatgpt")
def chat_gpt_ep(content: Text = None):
    url = 'https://api.openai.com/v1/chat/completions'
    obj = json.loads(content.content)
    data = {
        "model": "gpt-3.5-turbo",
        "messages": obj['messages']
    }
    print("data = \n", data)
    key = obj['key']
    openai.api_key = key
    headers = {
    'Content-Type': 'application/json',
    'Authorization': 'Bearer ' + key
    }
    result = requests.post(url=url,
                           data=json.dumps(data),
                           headers=headers
                           )
    res = str(result.json()['choices'][0]['message']['content']).strip()
    content = {'content': res}
    print('content:', content)
    return JSONResponse(content=content)


async def chat_gpt_stream_fun(content: Text = None):
    start_time = time.time()
    obj = json.loads(content.content)
    response = openai.ChatCompletion.create(
        model='gpt-3.5-turbo',
        messages=obj['messages'],
        stream=True,  # this time, we set stream=True
    )
    # create variables to collect the stream of chunks
    collected_chunks = []
    collected_messages = []
    # iterate through the stream of events
    for chunk in response:
        chunk_time = time.time() - start_time  # calculate the time delay of the chunk
        collected_chunks.append(chunk)  # save the event response
        chunk_message = chunk['choices'][0]['delta']  # extract the message
        collected_messages.append(chunk_message)  # save the message
        print(f"Message received {chunk_time:.2f} seconds after request: {chunk_message}")  # print the delay and text
        full_reply_content = ''.join([m.get('content', '') for m in collected_messages])
        print(f"Full conversation received: {full_reply_content}")
        content = {'content': full_reply_content}
        print('content:', content)
        yield json.dumps(content) + '\n'


@app.post("/chatgptstream", status_code=status.HTTP_200_OK)
async def get_random_numbers(content: Text = None):
    return StreamingResponse(chat_gpt_stream_fun(content), media_type='application/json')


@app.post("/embeddings")
def embeddings_ep(content: Text = None):
    url = 'https://api.openai.com/v1/embeddings'
    data = {
        "model": "text-embedding-ada-002",
        "input": content.content
    }
    result = requests.post(url=url,
                           data=json.dumps(data),
                           headers=headers
                           )
    return JSONResponse(content=result.json())


@app.post("/create_image")
def create_image_ep(content: Text = None):
    url = 'https://api.openai.com/v1/images/generations'
    obj = json.loads(content.content)
    data = {
        "prompt": obj["prompt"],
        "n": obj["n"],
        "size": obj["size"]
    }
    print("data = \n", data)
    result = requests.post(url=url,
                           data=json.dumps(data),
                           headers=headers
                           )
    return JSONResponse(content=result.json())




from fastapi import FastAPI, Request, Response
from fastapi.responses import PlainTextResponse
from hashlib import sha1
from time import time
from xml.etree.ElementTree import Element, tostring




def chat_gpt_response(prompt):
    # Replace with your GPT-3.5 implementation
    return "你好呀，小哥哥"


@app.get('/wechat')
def verify_server_address(signature: str, timestamp: str, nonce: str, echostr: str):
    token = 'zsj'
    if check_signature(token, signature, timestamp, nonce):
        return PlainTextResponse(echostr)


@app.post('/wechat')
def process_message(request: Request):
    xml_data = await request.body()
    xml_tree = ElementTree.fromstring(xml_data)

    msg_type = xml_tree.find('MsgType').text
    if msg_type == 'text':
        content = xml_tree.find('Content').text
        user_open_id = xml_tree.find('FromUserName').text
        public_account_id = xml_tree.find('ToUserName').text

        reply_content = chat_gpt_response(content)

        reply = Element('xml')
        to_user_name = Element('ToUserName')
        to_user_name.text = user_open_id
        reply.append(to_user_name)

        from_user_name = Element('FromUserName')
        from_user_name.text = public_account_id
        reply.append(from_user_name)

        create_time = Element('CreateTime')
        create_time.text = str(int(time()))
        reply.append(create_time)

        msg_type = Element('MsgType')
        msg_type.text = 'text'
        reply.append(msg_type)

        content = Element('Content')
        content.text = reply_content
        reply.append(content)

        response_xml = tostring(reply, encoding='utf-8')
        return Response(content=response_xml, media_type='application/xml')


def check_signature(token, signature, timestamp, nonce):
    tmp_list = [token, timestamp, nonce]
    tmp_list.sort()
    tmp_str = ''.join(tmp_list)
    tmp_str = sha1(tmp_str.encode('utf-8')).hexdigest()

    return tmp_str == signature