import os
import shutil
import hmac
import hashlib
import base64
import subprocess
import time
import async_timeout
import asyncio
import requests
import json
# Conditional imports to avoid circular dependencies
try:
from mysite.logger import logger, log_error
except ImportError:
# Fallback logger for standalone usage
import logging
logger = logging.getLogger(__name__)
log_error = logger.error
try:
import mysite.interpreter.interpreter_config
except ImportError:
pass
try:
from models.ride import test_set_lide
except ImportError:
def test_set_lide(prompt, url):
logger.warning("test_set_lide not available")
pass
try:
from mysite.libs.github import github
except ImportError:
def github(token, foldername):
logger.warning("github function not available")
return "https://github.com/placeholder"
GENERATION_TIMEOUT_SEC=60
def get_base_path():
"""
環境に応じて動的にベースパスを取得
"""
try:
# 環境変数から取得を試行
env_base_path = os.getenv("INTERPRETER_BASE_PATH")
if env_base_path:
# パスの正規化と存在確認
normalized_path = os.path.normpath(env_base_path)
if not normalized_path.endswith('/'):
normalized_path += '/'
# 親ディレクトリの存在確認
parent_dir = os.path.dirname(normalized_path.rstrip('/'))
if os.path.exists(parent_dir):
return normalized_path
logger.warning(f"Environment path parent not found: {parent_dir}")
# 現在の作業ディレクトリから推測
current_dir = os.getcwd()
logger.info(f"Current directory: {current_dir}")
# Codespaces環境の検出
if "/workspaces/" in current_dir:
path = os.path.join(current_dir, "app", "Http", "controller")
return os.path.normpath(path) + "/"
# Docker環境の検出
if "/home/user/app/" in current_dir or os.path.exists("/home/user/app/"):
return "/home/user/app/app/Http/controller/"
# ローカル開発環境
if "fastapi_django_main_live" in current_dir:
path = os.path.join(current_dir, "app", "Http", "controller")
return os.path.normpath(path) + "/"
# フォールバック: カレントディレクトリ下にcontrollerディレクトリを作成
fallback_path = os.path.join(current_dir, "temp_controller")
return os.path.normpath(fallback_path) + "/"
except Exception as e:
logger.error(f"Error in get_base_path: {str(e)}")
# 絶対フォールバック
return os.path.join(os.getcwd(), "temp_controller") + "/"
# 動的にベースパスを設定 - 遅延初期化
BASE_PATH = None
def get_base_path_safe():
"""
安全なベースパス取得(遅延初期化)
"""
global BASE_PATH
if BASE_PATH is None:
BASE_PATH = get_base_path()
return BASE_PATH
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):
# 改行文字を
タグに置き換える
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": "{}".format(link_text)
}
},
{
"buttons": [
{
"textButton": {
"text": "Open Link",
"onClick": {
"openLink": {
"url": link_url
}
}
}
}
]
},
{
"textParagraph": {
"text": "{}".format("ステップ2 テスト google apps script(その場で誰でもさくっと動くレベルがいい)")
}
},
{
"buttons": [
{
"textButton": {
"text": "Open Link",
"onClick": {
"openLink": {
"url": "https://script.google.com/d/1YJCQDrzGnrJiRcJcRjX9RuKR3gGrIj84p5kp4ZMiY-ls9UafjhDEUXIk/edit?usp=sharing"
}
}
}
}
]
},
{
"textParagraph": {
"text": "{}".format("ステップ3 テスト google colab(その場で誰でもさくっと動くレベルがいい)")
}
},
{
"buttons": [
{
"textButton": {
"text": "Open Link",
"onClick": {
"openLink": {
"url": "https://colab.research.google.com/drive/1LNR_Y9sk9DGYnWTxHP09s9x4JLe3GmqS?hl=ja"
}
}
}
}
]
},
{
"textParagraph": {
"text": "{}".format("ステップ4 workflow に組み込み(うまくいった 小さなものを組み合わせる)")
}
},
{
"buttons": [
{
"textButton": {
"text": "Open Link",
"onClick": {
"openLink": {
"url": "https://bpmboxesscom-46463613.hubspotpagebuilder.com/ja/"
}
}
}
}
]
},
{
"textParagraph": {
"text": "{}".format("ステップ5 カスタマーサービス内部で相談")
}
},
{
"buttons": [
{
"textButton": {
"text": "Open Link",
"onClick": {
"openLink": {
"url": "https://bpmboxesscom-46463613.hubspotpagebuilder.com/ja/"
}
}
}
}
]
},
{
"textParagraph": {
"text": "{}".format("ステップ6 過去のテンプレートを元に思ったものを作ってみる コピーペースト 繰り返し")
}
},
{
"buttons": [
{
"textButton": {
"text": "Open Link",
"onClick": {
"openLink": {
"url": "https://kenken999-php.hf.space/main_list.php"
}
}
}
}
]
},
{
"textParagraph": {
"text": "{}".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)
# Conditional import for google_chat functions
try:
from mysite.interpreter.google_chat import send_google_chat_card, send_google_chat_card_thread
except ImportError:
def send_google_chat_card(*args, **kwargs):
logger.warning("send_google_chat_card not available")
pass
def send_google_chat_card_thread(*args, **kwargs):
logger.warning("send_google_chat_card_thread not available")
pass
#プロセスの実行
def ensure_base_path_exists():
"""
ベースパスが存在することを確認し、必要に応じて作成
"""
global BASE_PATH
# 遅延初期化
if BASE_PATH is None:
BASE_PATH = get_base_path()
try:
os.makedirs(BASE_PATH, exist_ok=True)
return True
except Exception as e:
logger.error(f"Failed to create base path {BASE_PATH}: {str(e)}")
# フォールバックパスを試行
fallback_path = os.path.join(os.getcwd(), "temp_controller/")
try:
os.makedirs(fallback_path, exist_ok=True)
BASE_PATH = fallback_path
logger.info(f"Using fallback path: {BASE_PATH}")
return True
except Exception as fallback_error:
logger.error(f"Failed to create fallback path: {str(fallback_error)}")
return False
def no_process_file(prompt, foldername,thread_name=None):
set_environment_variables()
# ベースパスの存在確認
if not ensure_base_path_exists():
return "Error: Could not create or access base directory"
target_dir = f"{BASE_PATH}{foldername}"
try:
os.makedirs(target_dir, exist_ok=True)
except Exception as e:
return f"Error creating directory {target_dir}: {str(e)}"
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 = "ステップ1 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_nofile(prompt, foldername, token=None):
set_environment_variables()
# ベースパスの存在確認
if not ensure_base_path_exists():
return "Error: Could not create or access base directory"
target_dir = f"{BASE_PATH}{foldername}"
try:
os.makedirs(target_dir, exist_ok=True)
except Exception as e:
return f"Error creating directory {target_dir}: {str(e)}"
time.sleep(1)
# promptファイルの作成
prompt_file_path = f"{BASE_PATH}{foldername}/prompt"
try:
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")
token = os.getenv("github_token")
try:
url = github(token, foldername)
except Exception as e:
log_error(e)
url = "Error creating GitHub repo"
test_set_lide(prompt, url)
return f"Processed open github 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}\nMake Command Error:\n{e.stderr}"
def process_file(fileobj, prompt, foldername,token=None):
set_environment_variables()
# ベースパスの存在確認
if not ensure_base_path_exists():
return "Error: Could not create or access base directory"
target_dir = f"{BASE_PATH}{foldername}"
try:
os.makedirs(target_dir, exist_ok=True)
except Exception as e:
return f"Error creating directory {target_dir}: {str(e)}"
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")
token = os.getenv("github_token")
#githubでのソース作成
#log_error("github でエラーが起きました")
try:
url = github(token,foldername)
except Exception as e:
log_error(e)
#url = "test"
test_set_lide(prompt,url)
return f"Processed open github 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}"