Spaces:
Sleeping
Sleeping
File size: 8,846 Bytes
164aab4 54216e7 e485f1f 062317a 3f66c9b 164aab4 33413a1 164aab4 3f66c9b aba7131 164aab4 aba7131 164aab4 aaeedb3 3f66c9b 4f694e2 3f66c9b 164aab4 4f694e2 3f2ea6b 4f694e2 23450bf 886362f c440cac 23450bf c440cac 3dc5c8f 5798728 bdd4bbb 30364af 8ec430f 30364af 062317a dc8c1fd 062317a dc8c1fd 7207909 062317a 3f5bb3b 062317a 164aab4 3217144 164aab4 3217144 33413a1 164aab4 07c1d41 164aab4 3f66c9b 164aab4 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 |
import os
import subprocess
import logging
from fastapi import FastAPI, Request, HTTPException
import requests
import json
from datetime import datetime
import importlib
import os
import pkgutil
from mysite.libs.utilities import validate_signature, no_process_file
#from mysite.database.database import ride,create_ride
from controllers.gra_04_database.rides import test_set_lide
from mysite.interpreter.prompt import prompt_genalate,test_prompt
from mysite.interpreter.google_chat import send_google_chat_card
logger = logging.getLogger(__name__)
"""
router
"""
def include_routers(app):
package_dir = "/home/user/app/routers"
if not os.path.exists(package_dir):
logger.error(f"Package directory {package_dir} does not exist.")
return
for module_info in pkgutil.iter_modules([package_dir]):
try:
if module_info.ispkg:
sub_package_dir = os.path.join(package_dir, module_info.name)
for sub_module_info in pkgutil.iter_modules([sub_package_dir]):
module_name = (
f"routers.{module_info.name}.{sub_module_info.name}"
if sub_module_info.ispkg
else f"routers.{module_info.name}.{sub_module_info.name}"
)
module = importlib.import_module(module_name)
if hasattr(module, "router"):
app.include_router(module.router)
else:
module_name = f"routers.{module_info.name}"
module = importlib.import_module(module_name)
if hasattr(module, "router"):
app.include_router(module.router)
except ModuleNotFoundError as e:
logger.error(f"Module not found: {e}")
except Exception as e:
logger.error(f"An error occurred: {e}")
#from routers.webhooks import router
def setup_webhook_routes(app: FastAPI):
from polls.routers import register_routers
register_routers(app)
"""
@app.post("/webhooks")
def get_choices(
messages
):
logger.info("[Start] ====== LINE webhook ======")
try:
now = datetime.now().strftime("%Y%m%d%H%M%S")
user_id_with_timestamp = messages[:10]
#user_id_with_timestamp = messages#f"{now}_{title}_{user_id}"
no_process_file(messages, user_id_with_timestamp)
#db登録
test_set_lide(messages, user_id_with_timestamp)
except Exception as e:
logger.error("Error: %s", str(e))
"""
@app.post("/webhook")
async def webhook(request: Request):
#logger.info("[Start] ====== LINE webhook ======")
body = await request.body()
received_headers = dict(request.headers)
body_str = body.decode("utf-8")
#logger.info("Received Body: %s", body_str)
body_json = json.loads(body_str)
events = body_json.get("events", [])
webhook_url = os.getenv("chat_url")
token = os.getenv("token")
#url = github(token,foldername)
try:
for event in events:
if event["type"] == "message" and event["message"]["type"] == "text":
user_id = event["source"]["userId"]
text = event["message"]["text"]
#logger.info("------------------------------------------")
first_line = text.split('\n')[0]
#logger.info(f"User ID: {user_id}, Text: {text}")
prompt = """
1, Q&Aのテーブルを作成してください
2, 質問が来た際には、まず質問に対しての答えを過去のデータから探します
3, Q&Aから役割を作成します
質問に対しての答えを出す、シナリオを考える
4, 実際にテストして正しい答えがでるか確認
5, 出ない場合は再度作成しなおします
1から6を繰り返し、答えが出たプロンプトを登録します
7, 成功した場合それを保存します
8, 同じ質問が来たら質問別にプロンプトを変更します
9, 上記をラインの質問に内部の方が納得いくまで、日々修正していきます
"""
prompt2 = f""" 下記について日本語で答えて下さい
金、ダイヤモンド、商品を売りにきた顧客います
売りに来た顧客の質問内容は {text}
この質問を買取店の査定人に対して、理解がしやすい わかりやすい質問に変更してください
会社にはデータベースがあり質問内容から、商品を検索するSQLを作成してください
"""
promps,prompt_res = prompt_genalate(text,prompt2)
#test_set_lide(text,"a1")
#no_process_file(text, "ai")
title = """ プロンプト作成 """
subtitle = promps
link_text = "test"
link_url = "url"
#test_set_lide(subtitle, text)
send_google_chat_card(webhook_url, title, subtitle, link_text, link_url)
#test case
first_line = text.split('\n')[0]
#test_prompt
res = test_prompt(prompt_res,first_line)
send_google_chat_card(webhook_url, "プロンプトテスト"+first_line, str(res), link_text, link_url)
#
return
for event in events:
if event["type"] == "message" and event["message"]["type"] == "text":
user_id = event["source"]["userId"]
text = event["message"]["text"]
logger.info(event)
logger.info(f"User ID: {user_id}, Text: {text}")
now = datetime.now().strftime("%Y%m%d%H%M%S")
title = text[:10]
user_id_with_timestamp = title#f"{now}_{title}_{user_id}"
no_process_file(text, user_id_with_timestamp)
#db登録
test_set_lide(text, user_id_with_timestamp)
logger.info("Received Headers: %s", received_headers)
logger.info("Received Body: %s", body.decode("utf-8"))
line_signature = received_headers.get("x-line-signature")
if not line_signature:
raise HTTPException(status_code=400, detail="X-Line-Signature header is missing.")
if not validate_signature(body.decode("utf-8"), line_signature, os.getenv("ChannelSecret")):
raise HTTPException(status_code=400, detail="Invalid signature.")
if not os.getenv("WEBHOOK_URL") or not os.getenv("WEBHOOK_URL").startswith("https://"):
raise HTTPException(status_code=400, detail="Invalid webhook URL")
headers = {
"Content-Type": "application/json",
"X-Line-Signature": line_signature,
"Authorization": f"Bearer {os.getenv('ChannelAccessToken')}",
}
logger.info("Forwarding to URL: %s", os.getenv("WEBHOOK_URL"))
logger.info("Forwarding Headers: %s", headers)
logger.info("Forwarding Body: %s", body.decode("utf-8"))
response = requests.post(os.getenv("WEBHOOK_URL"), headers=headers, data=body)
responses = requests.post(os.getenv("WEBHOOKGAS"), headers=headers, data=body)
logger.info("Response Code: %s", response.status_code)
logger.info("Response Content: %s", response.text)
logger.info("Response Headers: %s", response.headers)
return {"status": "success", "response_content": response.text}, response.status_code
except Exception as e:
promps = prompt_genalate(str(e))
#test_set_lide(text,"a1")
#no_process_file(text, "ai")
title = """本番テスト 入力内容のみ設定 プロンプトも付け足してはテスト """
subtitle = promps
link_text = "test"
link_url = "url"
#test_set_lide(subtitle, text)
send_google_chat_card(webhook_url, title, subtitle, link_text, link_url)
logger.error("Error: %s", str(e))
raise HTTPException(status_code=500, detail=str(e))
|