# =============================================
# 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("""
RobotAI v8.5
🤖 RobotAI v8.5 — All-in-One Server
Server is running successfully! Use /chat API or chat below.
{% if reply %}
Robot: {{ reply }}
{% if tts_url %}{% endif %}
{% endif %}
""", 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("""
RobotAI v8.5
🤖 RobotAI v8.5 — Chat
Robot: {{ reply }}
{% if tts_url %}{% endif %}
""", 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/")
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)