File size: 5,387 Bytes
63de067 3848b94 63de067 0b3e626 a35fb27 1728f2c 3848b94 63de067 1728f2c 63de067 1728f2c 63de067 3848b94 63de067 3848b94 63de067 3848b94 63de067 3848b94 63de067 3848b94 1728f2c 3848b94 63de067 3848b94 baaa9d1 3848b94 63de067 3848b94 1728f2c 3848b94 1728f2c 3848b94 1728f2c 3848b94 1728f2c 63de067 3848b94 63de067 3848b94 1728f2c 63de067 3848b94 63de067 3848b94 63de067 1728f2c 3848b94 | 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 | # =============================================
# RobotAI v8.5 — All-in-One (UI + ESP32 + Telegram)
# Author: GPT-5 for Cường Phan
# =============================================
import os
import json
import requests
from flask import Flask, request, jsonify, render_template_string
from gtts import gTTS
from datetime import datetime
app = Flask(__name__)
# ======================
# Config
# ======================
HF_TOKEN = os.environ.get("HF_TOKEN", "")
HF_MODEL = os.environ.get("HF_MODEL", "mistralai/Mistral-7B-Instruct-v0.3") # model public
HF_TTS_MODEL = os.environ.get("HF_TTS_MODEL", "espnet/kan-bayashi_ljspeech_vits")
TELEGRAM_TOKEN = os.environ.get("TELEGRAM_TOKEN", "")
TELEGRAM_CHAT_ID = os.environ.get("TELEGRAM_CHAT_ID", "")
AUDIO_DIR = "audio_cache"
os.makedirs(AUDIO_DIR, exist_ok=True)
# ======================
# Utils
# ======================
def log(msg):
print(f"[{datetime.now().strftime('%H:%M:%S')}] {msg}")
def send_telegram(text):
if TELEGRAM_TOKEN and TELEGRAM_CHAT_ID:
url = f"https://api.telegram.org/bot{TELEGRAM_TOKEN}/sendMessage"
requests.post(url, json={"chat_id": TELEGRAM_CHAT_ID, "text": text})
def detect_lang(text):
vi_chars = "ăâđêôơưáàảãạấầẩẫậắằẳẵặéèẻẽẹếềểễệóòỏõọốồổỗộớờởỡợúùủũụứừửữựýỳỷỹỵ"
return "vi" if any(c in text for c in vi_chars) else "en"
def query_huggingface(prompt, max_new_tokens=200):
url = f"https://api-inference.huggingface.co/models/{HF_MODEL}"
headers = {"Authorization": f"Bearer {HF_TOKEN}"} if HF_TOKEN else {}
payload = {"inputs": prompt, "parameters": {"max_new_tokens": max_new_tokens, "temperature": 0.7}}
try:
r = requests.post(url, headers=headers, json=payload, timeout=90)
if r.status_code == 200:
out = r.json()
if isinstance(out, list) and len(out) > 0:
return out[0].get("generated_text", out[0].get("text", ""))
elif isinstance(out, dict):
return out.get("generated_text", str(out))
elif r.status_code == 404:
return "⚠️ Model not found. Please check HF_MODEL name."
else:
return f"⚠️ Error: {r.status_code} - {r.text}"
except Exception as e:
return f"⚠️ Exception: {e}"
def gen_tts(text, lang="en"):
try:
path = os.path.join(AUDIO_DIR, f"tts_{int(datetime.now().timestamp())}.mp3")
tts = gTTS(text=text, lang="vi" if lang == "vi" else "en")
tts.save(path)
return f"/tts/{os.path.basename(path)}"
except Exception as e:
log(f"TTS error: {e}")
return None
# ======================
# ROUTES
# ======================
@app.route("/")
def home():
return render_template_string("""
<html><head><title>RobotAI v8.5</title></head>
<body style="font-family:sans-serif;text-align:center;">
<h2>🤖 RobotAI v8.5 — All-in-One Server</h2>
<p>Server is running successfully! Use /chat API or chat below.</p>
<form method="post" action="/chat_ui">
<textarea name="text" rows="3" cols="50" placeholder="Nhập tiếng Việt hoặc English..."></textarea><br>
<button type="submit">Gửi</button>
</form>
{% if reply %}
<p><b>Robot:</b> {{ reply }}</p>
{% if tts_url %}<audio controls autoplay><source src="{{ tts_url }}" type="audio/mpeg"></audio>{% endif %}
{% endif %}
</body></html>
""", reply=None)
@app.route("/chat_ui", methods=["POST"])
def chat_ui():
text = request.form.get("text", "")
lang = detect_lang(text)
prefix = "Trả lời bằng tiếng Việt:" if lang == "vi" else "Answer in English:"
reply = query_huggingface(f"{prefix}\n\n{text}")
tts_url = gen_tts(reply, lang)
return render_template_string("""
<html><head><title>RobotAI v8.5</title></head>
<body style="font-family:sans-serif;text-align:center;">
<h2>🤖 RobotAI v8.5 — Chat</h2>
<form method="post" action="/chat_ui">
<textarea name="text" rows="3" cols="50">{{text}}</textarea><br>
<button type="submit">Gửi</button>
</form>
<p><b>Robot:</b> {{ reply }}</p>
{% if tts_url %}<audio controls autoplay><source src="{{ tts_url }}" type="audio/mpeg"></audio>{% endif %}
</body></html>
""", text=text, reply=reply, tts_url=tts_url)
@app.route("/api/chat", methods=["POST"])
def api_chat():
data = request.get_json(force=True)
text = data.get("text", "")
lang = detect_lang(text)
reply = query_huggingface(text)
tts_url = gen_tts(reply, lang)
return jsonify({"reply": reply, "tts": tts_url, "lang": lang})
@app.route("/api/presence", methods=["POST"])
def api_presence():
data = request.get_json(force=True)
note = data.get("note", "🚶 Có người vừa đi qua!")
send_telegram(note)
reply = "Xin chào! Tôi là RobotAI. Có thể giúp gì cho bạn?"
tts_url = gen_tts(reply, "vi")
return jsonify({"reply": reply, "tts": tts_url})
@app.route("/tts/<fname>")
def serve_tts(fname):
return app.send_static_file(os.path.join(AUDIO_DIR, fname))
@app.route("/model_check")
def model_check():
return jsonify({"HF_MODEL": HF_MODEL, "Token": bool(HF_TOKEN)})
if __name__ == "__main__":
app.run(host="0.0.0.0", port=int(os.environ.get("PORT", 7860)), debug=False) |