MavareeSwimmingPool's picture
Update app/main.py
7d0607a verified
raw
history blame
9.79 kB
import os
import sys
import csv
from datetime import datetime
from pathlib import Path
from typing import Optional, Tuple, List, Dict
import gradio as gr
import requests
# Опционально: подтягиваем .env при локальном запуске
try:
from dotenv import load_dotenv
except Exception:
load_dotenv = None
# --- Fix imports when запуск: `python app\main.py` из корня проекта ---
PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
if load_dotenv:
load_dotenv(PROJECT_ROOT / ".env")
# --- Paths ---
DATA_DIR = PROJECT_ROOT / "data"
FEEDBACK_DIR = DATA_DIR / "feedback"
FEEDBACK_CSV = FEEDBACK_DIR / "feedback.csv"
# --- Telegram integration ---
def send_telegram_message(text: str) -> Tuple[bool, str]:
"""
Отправляет сообщение в Telegram (если TELEGRAM_BOT_TOKEN и TELEGRAM_CHAT_ID заданы).
Возвращает (ok, details).
"""
token = os.getenv("TELEGRAM_BOT_TOKEN", "").strip()
chat_id = os.getenv("TELEGRAM_CHAT_ID", "").strip()
if not token or not chat_id:
return False, "Telegram: пропущено (нет TELEGRAM_BOT_TOKEN или TELEGRAM_CHAT_ID)"
url = f"https://api.telegram.org/bot{token}/sendMessage"
try:
r = requests.post(url, data={"chat_id": chat_id, "text": text}, timeout=10)
if r.status_code == 200:
return True, "Telegram: отправлено ✅"
return False, f"Telegram: ошибка {r.status_code}: {r.text[:200]}"
except Exception as e:
return False, f"Telegram: исключение: {e}"
# --- Feedback storage ---
def ensure_feedback_csv_exists() -> None:
FEEDBACK_DIR.mkdir(parents=True, exist_ok=True)
if not FEEDBACK_CSV.exists():
with FEEDBACK_CSV.open("w", newline="", encoding="utf-8") as f:
w = csv.writer(f)
w.writerow(["timestamp", "category", "rating", "contact", "feedback_text"])
def append_feedback_row(category: str, rating: int, contact: str, feedback_text: str) -> None:
ensure_feedback_csv_exists()
ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
with FEEDBACK_CSV.open("a", newline="", encoding="utf-8") as f:
w = csv.writer(f)
w.writerow([ts, category, rating, (contact or "").strip(), (feedback_text or "").strip()])
# --- UI logic: feedback ---
def save_feedback(category: str, rating: int, contact: str, feedback_text: str):
if not feedback_text or not feedback_text.strip():
return "⚠️ Введите текст обратной связи.", gr.update(value=feedback_text), None
if rating is None:
return "⚠️ Выберите оценку (1–5).", gr.update(value=feedback_text), None
try:
append_feedback_row(category, int(rating), contact or "", feedback_text)
except Exception as e:
return f"❌ Не удалось сохранить в CSV: {e}", gr.update(value=feedback_text), None
# Telegram уведомление (не блокирует сохранение)
tg_text = (
"📝 2MOOD Feedback\n"
f"Категория: {category}\n"
f"Оценка: {rating}\n"
f"Контакт: {contact or '-'}\n"
f"Текст: {feedback_text.strip()}"
)
tg_ok, tg_details = send_telegram_message(tg_text)
status = "✅ Отзыв сохранён в CSV.\n\n"
status += f"Файл: `{FEEDBACK_CSV.as_posix()}`\n\n"
status += ("✅ " if tg_ok else "⚠️ ") + tg_details
# очистим поле текста + выдадим файл (чтобы сразу можно было скачать)
return status, gr.update(value=""), str(FEEDBACK_CSV)
def get_feedback_csv() -> Tuple[Optional[str], str]:
if FEEDBACK_CSV.exists() and FEEDBACK_CSV.stat().st_size > 0:
return str(FEEDBACK_CSV), "✅ Готово: файл доступен для скачивания."
return None, "⚠️ Файл ещё не создан или пуст."
# --- Messenger logic ---
def messenger_reply(user_message: str, history):
"""
Универсально: работаем как с history=None, так и с любым форматом.
Для простоты добавим пары (user, assistant).
"""
msg = (user_message or "").strip()
if not msg:
return "", history
if history is None:
history = []
# Gradio Chatbot обычно хранит list[tuple[str,str]] для старого формата
try:
history.append((msg, f"Echo: {msg}"))
except Exception:
# fallback на список словарей (если вдруг другой формат)
if isinstance(history, list):
history.append({"role": "user", "content": msg})
history.append({"role": "assistant", "content": f"Echo: {msg}"})
return "", history
# --- Knowledge Base (stub) ---
def index_pdf(file) -> str:
if file is None:
return "❌ Файл не выбран"
# Пока тестовый режим: делаем вид, что добавили
# Позже здесь будет реальная индексация (FAISS/Qdrant + embeddings)
name = getattr(file, "name", "uploaded.pdf")
return f"✅ Файл `{name}` успешно добавлен в базу (тестовый режим)."
def build_interface():
ensure_feedback_csv_exists()
with gr.Blocks(title="2MOOD AI Workspace (Pilot)") as demo:
gr.Markdown("# 2MOOD AI Workspace")
gr.Markdown("База знаний + Messenger + Обратная связь")
# -------- TAB: Knowledge Base --------
with gr.Tab("📚 База знаний"):
gr.Markdown("### Загрузка PDF и индексирование (пока заглушка)")
pdf_file = gr.File(label="Загрузить PDF")
add_btn = gr.Button("Добавить в базу", variant="primary")
status_kb = gr.Markdown()
add_btn.click(
fn=index_pdf,
inputs=pdf_file,
outputs=status_kb
)
# -------- TAB: Messenger --------
with gr.Tab("💬 2MOOD Messenger"):
chat = gr.Chatbot(height=420) # без type="messages" для совместимости
msg = gr.Textbox(label="Сообщение", placeholder="Напишите сообщение…")
send = gr.Button("Отправить", variant="primary")
send.click(messenger_reply, inputs=[msg, chat], outputs=[msg, chat])
# -------- TAB: Feedback --------
with gr.Tab("📝 Обратная связь"):
category = gr.Dropdown(
label="Категория",
choices=["Другое", "Messenger", "База знаний", "UI/UX", "Ошибка/Баг", "Идея/Feature"],
value="Другое",
)
rating = gr.Radio(
label="Оценка",
choices=[1, 2, 3, 4, 5],
value=5,
)
contact = gr.Textbox(label="Контакт (опционально): Telegram/почта")
feedback_text = gr.Textbox(
label="Текст обратной связи",
lines=6,
placeholder="Например: не хватает кнопки…, ошибка…, хочу экспорт в Word…",
)
submit = gr.Button("Отправить", variant="primary")
status_fb = gr.Markdown()
download_btn = gr.Button("📥 Скачать все отзывы")
download_file = gr.File(label="Файл с отзывами", interactive=False)
download_status = gr.Markdown()
gr.Markdown(f"Файл логов: `{FEEDBACK_CSV.as_posix()}`")
submit.click(
save_feedback,
inputs=[category, rating, contact, feedback_text],
outputs=[status_fb, feedback_text, download_file],
)
download_btn.click(
get_feedback_csv,
inputs=None,
outputs=[download_file, download_status],
)
return demo
# Локальный запуск (на HF не используется, но пусть будет)
def launch_with_port_fallback(demo: gr.Blocks):
os.environ.setdefault("NO_PROXY", "localhost,127.0.0.1")
os.environ.setdefault("no_proxy", "localhost,127.0.0.1")
env_port = os.getenv("GRADIO_SERVER_PORT") or os.getenv("PORT") or ""
ports: List[int] = []
if str(env_port).isdigit():
ports.append(int(env_port))
ports.extend([7860] + list(range(7861, 7871)))
last_error: Optional[Exception] = None
for port in ports:
try:
demo.launch(server_name="127.0.0.1", server_port=port, share=False, inbrowser=True)
return
except Exception as e:
last_error = e
msg = str(e)
if "localhost is not accessible" in msg or "shareable link must be created" in msg:
try:
demo.launch(server_name="0.0.0.0", server_port=port, share=True, inbrowser=False)
return
except Exception as e2:
last_error = e2
continue
raise RuntimeError(f"Не удалось запустить Gradio ни на одном порту. Последняя ошибка: {last_error}")
if __name__ == "__main__":
app = build_interface()
launch_with_port_fallback(app)