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)