Spaces:
Running
on
Zero
Running
on
Zero
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 | |
from mysite.interpreter.interpreter import chat_with_interpreter | |
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 datetime import datetime | |
#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)) | |
""" | |
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を作成してください | |
必要なテーブルのCreate文も作成してして下さい | |
## リファスタの住所 | |
〒170-0013 東京都豊島区東池袋1丁目25−14 アルファビルディング 4F | |
### 買取テンプレート | |
・ブランド名: | |
・モデル名: | |
・型番や品番: | |
・購入店: | |
・購入時期: | |
・購入金額: | |
・付属品: | |
・コンディション: | |
(10段階評価厳しめ) | |
・貴金属品位: | |
・貴金属重量: | |
(キッチンスケールでも(sparkle)) | |
・ダイヤや宝石の鑑定書はお写真で! | |
・イニシャル:あり なし | |
## リファスタのサイト | |
(monitor)24h対応事前査定 | |
https://kinkaimasu.jp/estimate/?openExternalBrowser=1&utm_source=LINE | |
(open book)買取システムナビ | |
https://kinkaimasu.jp/system/?openExternalBrowser=1&utm_source=LINE | |
(car)店舗アクセス | |
https://goo.gl/veQZ03 | |
(?)よくある質問 | |
https://kinkaimasu.jp/faq/?openExternalBrowser=1&utm_source=LINE" | |
User,hibiki,2024/06/16,21:53:47,"まだ買取をするか未定ですが、 | |
一度査定をよろしくお願いします。" | |
Account,応答メッセージ,2024/06/16,21:53:47,"(clock)ただ今対応時間外(clock) | |
営業時間:11:00~20:00 | |
※年中無休 | |
翌営業日に順次対応致しますので、お写真や情報はいつでもお送りください(moon wink) | |
(monitor)24h対応事前査定 | |
https://kinkaimasu.jp/estimate/?openExternalBrowser=1&utm_source=LINE | |
(open book)買取システムナビ | |
https://kinkaimasu.jp/system/?openExternalBrowser=1&utm_source=LINE | |
(car)店舗アクセス | |
https://goo.gl/veQZ03 | |
(?)よくある質問 | |
https://kinkaimasu.jp/faq/?openExternalBrowser=1&utm_source=LINE" | |
""" | |
promps,prompt_res = prompt_genalate(text,prompt2) | |
#test_set_lide(text,"a1") | |
#no_process_file(text, "ai") | |
title = f""" プロンプト作成 {promps}""" | |
subtitle = prompt_res | |
link_text = "データを確認する" | |
link_url = "https://kenken999-php.hf.space/diamondprice_list.php" | |
#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,text) | |
send_google_chat_card(webhook_url, "プロンプトテスト "+first_line, str(res), link_text, link_url) | |
now = datetime.now() | |
yyyymmddhis = now.strftime('%Y%m%d%H%M%S') | |
print(yyyymmddhis) | |
no_process_file("下記のコードをdjangoで作成"+res, "gpt_enginner"+ yyyymmddhis) | |
chat_with_interpreter(res) | |
# | |
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)) | |