CocoPi_upgrade/preset/drivers/pylib/GPTrequest.py

##################################
#            GPTrequest          #
##################################
#Version date   :       2024-07-08
#region         :       CH

import uuid
import requests
import json
import base64
import os

try:
    class LazyImport:
        def __init__(self, module_name):
            self.module_name = module_name
            self.module = None
        def __getattr__(self, name):
            if self.module is None:
                self.module = __import__(self.module_name)
            return getattr(self.module, name)
    uuid = LazyImport("uuid")
    requests = LazyImport("requests")
    json = LazyImport("json")
    base64 = LazyImport("base64")

except ModuleNotFoundError as e:
    pass

class GPTrequest():
    def __init__(self):
        self.uid = str(uuid.uuid1().hex)
        #获取hardware mac地址
        cmd="cat /sys/class/sunxi_info/sys_info"
        rec=os.popen(cmd).read()
        self.hwMac=rec[rec.find("sunxi_serial      : ")+20:rec.find("sunxi_serial      : ")+52]
        print(self.hwMac)
       
    # 文字生成图片
    def get_text_to_img(self,text,model="cogview-3"):
        uid = str(uuid.uuid1().hex)
        param = {
            "size": "1024*1024" if model.find("cogview-3")>-1 else "1024x1024",
            "quality": "standard",
            "n": 1,
            "prompt": text,
            "style":"natural",
            "uid": uid,
            "model":model
        }
        headers={
            "Content-Type":"application/json",
            "hwMac":self.hwMac
        }
        imgUrl = ""
        res = requests.post("https://gpt4.cocorobo.cn/getImageV831", headers=headers, json=param)
        try:
            url = (json.loads(res.text))["FunctionResponse"]["image_url_list"][0]
            r = requests.get("https://gpt4.cocorobo.cn"+url)
            with open("/root/user/img/" + uid + ".png","wb") as f:
                f.write(r.content)
            imgUrl = "/root/user/img/" + uid + ".png"
        except:
            imgUrl = ""
        return imgUrl
     
    # 图片转base64
    def encode_image(self,image_path):
        img_data = ""
        try:
            with open(image_path, "rb") as f:
                img_data = f.read()
        except:
            pass
        return str(base64.b64encode(img_data), "utf-8")

    # 通过图片获取文字
    def get_image_to_text(self,imgUrl,textDesc,model="gpt-4o"):
        uid = str(uuid.uuid1().hex)
        strBase = self.encode_image(imgUrl)
        params = {
            "max_tokens": 4096,
            "messages":[
                {
                    "role": "user",
                    "content":[
                        {"text": textDesc,"type": "text"},
                        {"image_url": {"url": f"data:image/jpeg;base64,{strBase}"},"type": "image_url"}
                    ]
                }
            ],
            "uid": uid,
            "stream": False,
            "model":model
        }
        headers={
            "Content-Type":"application/json",
            "hwMac":self.hwMac
        }
        textContent = ""
        res = requests.post("https://gpt4.cocorobo.cn/imageAnalyse", headers=headers, data=json.dumps(params))
        try:
            if res.status_code == 200:
                textContent = json.loads(res.text)["FunctionResponse"]["choices"][0]["message"]["content"]
                
        except:
            textContent = ""
        return textContent
    
    # 文字生成音頻文件
    def get_text_to_voice(self,text,model="tts-1"):
        uid = str(uuid.uuid1().hex)
        url = "https://gpt4.cocorobo.cn/getV831Audio"
        payload = json.dumps({"input": text,"response_format":"mp3","voice": "shimmer","uid":uid,"model":model})
        
        headers={
            "Content-Type":"application/json",
            "hwMac":self.hwMac
        }
        response=requests.request("POST", url, headers=headers, data=payload)
        voiceFilePath = ""
        try:
            if response.status_code == 200:
                # print(json.loads(response.text)["FunctionResponse"]["url"])
                a = "https://gpt4.cocorobo.cn" + json.loads(response.text)["FunctionResponse"]["url"]
                r = requests.get(a)
                with open("/root/preset/audio/" + uid+".wav","wb") as f:
                    f.write(r.content)
                voiceFilePath = "/root/preset/audio/" + uid+".wav"
        except:
            voiceFilePath = ""
        return voiceFilePath
    
    # 音频转文字
    def get_voice_to_text(self,voicePath,model=""):
        voiceData = open(voicePath,'rb')
        data = {"file": voiceData,"model":model}
        param = {}
        headers={
            "hwMac":self.hwMac
        }
        res = requests.post("https://gpt4.cocorobo.cn/transcribe_file_stream",headers=headers, files=data, data=param)
        try:
            voiceResult = (json.loads(res.text))["FunctionResponse"]
        except:
            voiceResult = ""
        return voiceResult

    # gpt 對話，並生成音頻文件
    def get_post_chatgpt(self,datas,model="deepseek-chat"):
        url = "https://gpt4.cocorobo.cn/v831AskForTopicNew"
        payload = json.dumps({"topic": datas,"model":model})
        headers={
            "Content-Type":"application/json",
            "hwMac":self.hwMac
        }
        response=requests.request("POST", url, headers=headers, data=payload)
        result = ""
        try:
            result = json.loads(response.text)["FunctionResponse"]["result"]
        except:
            result = ""
        # print("resule:"+str(result))
        return result
    
    
    def get_post_AI_Intelligence(self,id,text,model="deepseek-chat"):
        url = "https://gpt4.cocorobo.cn/ai_agent_chat"
        params = {
            "id":id,
            "userId":self.hwMac,
            "message":text,
            "file_ids": [],
            "model":model
        }
        headers={
            "Content-Type":"application/json",
            "hwMac":self.hwMac
        }
        textContent = ""
        res = requests.post(url, headers=headers, data=json.dumps(params))
        try:
            textContent = json.loads(res.text)["FunctionResponse"]["message"]
        except:
            textContent = ""
        return textContent