| |
| |
| |
| |
|
|
| 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__) |
|
|
| |
| |
| |
| HF_TOKEN = os.environ.get("HF_TOKEN", "") |
| HF_MODEL = os.environ.get("HF_MODEL", "mistralai/Mistral-7B-Instruct-v0.3") |
| 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) |
|
|
|
|
| |
| |
| |
| 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 |
|
|
|
|
| |
| |
| |
| @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) |
