mysite/interpreter/process.py

import os
import shutil
import hmac
import hashlib
import base64
import subprocess
import time
from mysite.logger import logger
import async_timeout
import asyncio
import mysite.interpreter.interpreter_config 
from models.ride import test_set_lide
from mysite.libs.github import github
import requests
import json
from mysite.logger import log_error

GENERATION_TIMEOUT_SEC=60
BASE_PATH = "/home/user/app/controllers/"

def set_environment_variables():
    os.environ["OPENAI_API_BASE"] = "https://api.groq.com/openai/v1"
    os.environ["OPENAI_API_KEY"] = os.getenv("api_key")
    os.environ["MODEL_NAME"] = "llama3-8b-8192"
    os.environ["LOCAL_MODEL"] = "true"

def convert_newlines_to_google_chat_format(text):
    # 改行文字を <br> タグに置き換える
    return text.replace('\\n', '\\\n')

def send_google_chat_card(webhook_url, title, subtitle, link_text, link_url):
    headers = {
        'Content-Type': 'application/json; charset=UTF-8'
    }

    subtitle = convert_newlines_to_google_chat_format(subtitle)

    card_message = {
        "cards": [
            {
                "header": {
                    "title": title,
                },
                "sections": [
                    {
                        "widgets": [
                            {
                                "textParagraph": {
                                    "text": subtitle
                                }
                            },
                            {
                                "textParagraph": {
                                    "text": "<b>{}</b>".format(link_text)
                                }
                            },
                            {
                                "buttons": [
                                    {
                                        "textButton": {
                                            "text": "Open Link",
                                            "onClick": {
                                                "openLink": {
                                                    "url": link_url
                                                }
                                            }
                                        }
                                    }
                                ]
                            },
                            {
                                "textParagraph": {
                                    "text": "<b>{}</b>".format("ステップ２　テスト　google apps script(その場で誰でもさくっと動くレベルがいい)")
                                }
                            },
                            {
                                "buttons": [
                                    {
                                        "textButton": {
                                            "text": "Open Link",
                                            "onClick": {
                                                "openLink": {
                                                    "url": "https://script.google.com/d/1YJCQDrzGnrJiRcJcRjX9RuKR3gGrIj84p5kp4ZMiY-ls9UafjhDEUXIk/edit?usp=sharing"
                                                }
                                            }
                                        }
                                    }
                                ]
                            },
                            {
                                "textParagraph": {
                                    "text": "<b>{}</b>".format("ステップ３　テスト　google colab(その場で誰でもさくっと動くレベルがいい)")
                                }
                            },
                            {
                                "buttons": [
                                    {
                                        "textButton": {
                                            "text": "Open Link",
                                            "onClick": {
                                                "openLink": {
                                                    "url": "https://colab.research.google.com/drive/1LNR_Y9sk9DGYnWTxHP09s9x4JLe3GmqS?hl=ja"
                                                }
                                            }
                                        }
                                    }
                                ]
                            },
                            {
                                "textParagraph": {
                                    "text": "<b>{}</b>".format("ステップ４　workflow　に組み込み(うまくいった　小さなものを組み合わせる)")
                                }
                            },
                            {
                                "buttons": [
                                    {
                                        "textButton": {
                                            "text": "Open Link",
                                            "onClick": {
                                                "openLink": {
                                                    "url": "https://bpmboxesscom-46463613.hubspotpagebuilder.com/ja/"
                                                }
                                            }
                                        }
                                    }
                                ]
                            },
                            {
                                "textParagraph": {
                                    "text": "<b>{}</b>".format("ステップ５　カスタマーサービス内部で相談")
                                }
                            },
                            {
                                "buttons": [
                                    {
                                        "textButton": {
                                            "text": "Open Link",
                                            "onClick": {
                                                "openLink": {
                                                    "url": "https://bpmboxesscom-46463613.hubspotpagebuilder.com/ja/"
                                                }
                                            }
                                        }
                                    }
                                ]
                            },
                             {
                                "textParagraph": {
                                    "text": "<b>{}</b>".format("ステップ６　過去のテンプレートを元に思ったものを作ってみる　コピーペースト 繰り返し")
                                }
                            },
                            {
                                "buttons": [
                                    {
                                        "textButton": {
                                            "text": "Open Link",
                                            "onClick": {
                                                "openLink": {
                                                    "url": "https://kenken999-php.hf.space/main_list.php"
                                                }
                                            }
                                        }
                                    }
                                ]
                            }, 
                             {
                                "textParagraph": {
                                    "text": "<b>{}</b>".format("リファスタチャットボット確認")
                                }
                            },
                            {
                                "buttons": [
                                    {
                                        "textButton": {
                                            "text": "Open Link",
                                            "onClick": {
                                                "openLink": {
                                                    "url": "https://www.coze.com/store/bot/7372534558954176520?panel=1&bid=6cqv73j0k7g0i"
                                                }
                                            }
                                        }
                                    }
                                ]
                            },                                                                                                                

                        ]
                    }
                ]
            }
        ]
    }


    response = requests.post(webhook_url, headers=headers, data=json.dumps(card_message))

    if response.status_code == 200:
        print("Message posted successfully.")
    else:
        print(f"Failed to post message: {response.status_code}, {response.text}")



def validate_signature(body: str, signature: str, secret: str) -> bool:
    if secret is None:
        logger.error("Secret is None")
        return False

    hash = hmac.new(
        secret.encode("utf-8"), body.encode("utf-8"), hashlib.sha256
    ).digest()
    expected_signature = base64.b64encode(hash).decode("utf-8")
    return hmac.compare_digest(expected_signature, signature)
from mysite.interpreter.google_chat import send_google_chat_card,send_google_chat_card_thread
#プロセスの実行
def no_process_file(prompt, foldername,thread_name=None):
    set_environment_variables()
    try:
        proc = subprocess.Popen(["mkdir", f"{BASE_PATH}{foldername}"])
    except subprocess.CalledProcessError as e:
        return f"Processed Content:\n{e.stdout}\n\nMake Command Error:\n{e.stderr}"

    no_extension_path = f"{BASE_PATH}{foldername}/prompt"
    time.sleep(1)
    with open(no_extension_path, "a") as f:
        f.write(prompt)
    time.sleep(1)
    try:
        prompt_file_path = no_extension_path
        with open(prompt_file_path, "a") as prompt_file:
            prompt_file.write(prompt)
    except Exception as e:
        return f"Error writing prompt to file: {str(e)}"
    time.sleep(1)

    try:
        proc = subprocess.Popen(
            ["make", "run", foldername],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
        )
        stdout, stderr = proc.communicate(input="n\ny\ny\n")
        webhook_url = os.getenv("chat_url")
        token = os.getenv("token")
        #githubでのソース作成
        #log_error("github でエラーが起きました")
        try:
            url = github(token,foldername)
        except Exception as e:
            log_error(e)

        title = """ラインで作るオープンシステム
        お客様の質問内容の
        プログラムを作成しました"""
        subtitle = prompt
        link_text = "ステップ１　githubで　プログラムを確認する githubだと相談者が沢山いるのでわからなければ聞いてみる "
        link_url = url

        send_google_chat_card,send_google_chat_card_thread(webhook_url, title, subtitle, link_text, link_url,thread_name)
        #insert data to db
        #test_set_lide(prompt,url)
        print(f"Processed Content:success ")
        return f"Processed Content:success "#\n{stdout}\n\nMake Command Output:\n{stdout}\n\nMake Command Error:\n{stderr}"
    except subprocess.CalledProcessError as e:
        print(f"Processed Content:false ")
        return f"Processed Content:error"# {str(e)}"#\n{e.stdout}\n\nMake Command Error:\n{e.stderr}"



def process_file(fileobj, prompt, foldername,token=None):
    set_environment_variables()
    try:
        proc = subprocess.Popen(["mkdir", f"{BASE_PATH}{foldername}"])
    except subprocess.CalledProcessError as e:
        return f"Processed Content:\n{e.stdout}\n\nMake Command Error:\n{e.stderr}"
    time.sleep(2)
    path = f"{BASE_PATH}{foldername}/" + os.path.basename(fileobj)
    shutil.copyfile(fileobj.name, path)
    base_name = os.path.splitext(os.path.basename(fileobj))[0]
    no_extension_path = f"{BASE_PATH}{foldername}/{base_name}"
    shutil.copyfile(fileobj, no_extension_path)
    with open(no_extension_path, "a") as f:
        f.write(prompt)
    try:
        prompt_file_path = no_extension_path
        with open(prompt_file_path, "w") as prompt_file:
            prompt_file.write(prompt)
    except Exception as e:
        return f"Error writing prompt to file: {str(e)}"
    time.sleep(1)
    #foldernameの登録
    test_set_lide(prompt,foldername)
    try:
        proc = subprocess.Popen(
            ["make", "run", foldername],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
        )
        stdout, stderr = proc.communicate(input="n\ny\ny\n")
        
        test_set_lide(prompt,url)
        return f"Processed url:{url}\nContent:\n{stdout}\n\nMake Command Output:\n{stdout}\n\nMake Command Error:\n{stderr}"
    except subprocess.CalledProcessError as e:
        return f"Processed url:{url}\nContent:\n{stdout}\n\nMake Command Error:\n{e.stderr}"