import gradio as gr import pandas as pd import io import re from typing import List, Optional, Tuple import logging # Настройка логирования logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class WebsiteCategorizerApp: def __init__(self): self.sheet_url = "" self.sheet_data = [] self.current_index = 0 self.categories = ["новости", "коммерция", "другое"] self.results_data = [] # Для сохранения результатов def convert_google_sheet_url(self, sheet_url: str) -> str: """Конвертирует URL Google таблицы в CSV экспорт URL""" try: # Различные форматы URL Google таблиц if "/edit#gid=" in sheet_url: csv_url = sheet_url.replace("/edit#gid=", "/export?format=csv&gid=") elif "/edit?usp=sharing" in sheet_url: csv_url = sheet_url.replace("/edit?usp=sharing", "/export?format=csv") elif "/edit" in sheet_url: csv_url = sheet_url.replace("/edit", "/export?format=csv") else: # Если URL уже в формате экспорта csv_url = sheet_url return csv_url except Exception as e: logger.error(f"Ошибка конвертации URL: {e}") return "" def connect_to_sheet(self, sheet_url: str) -> Tuple[str, str]: """Подключается к Google таблице через CSV экспорт""" try: if not sheet_url: return "❌ Ошибка: Введите URL Google таблицы", "" # Конвертируем URL в CSV формат csv_url = self.convert_google_sheet_url(sheet_url) if not csv_url: return "❌ Ошибка: Неверный формат URL", "" # Загружаем данные через pandas df = pd.read_csv(csv_url) if df.empty: return "❌ Ошибка: Таблица пуста", "" # Убедимся что есть минимум 2 столбца if len(df.columns) < 2: return "❌ Ошибка: Нужно минимум 2 столбца (URL и категория)", "" # Сохранение данных self.sheet_data = [] self.results_data = [] # Получаем названия столбцов url_column = df.columns[0] category_column = df.columns[1] for index, row in df.iterrows(): url = str(row[url_column]).strip() if pd.notna(row[url_column]) else "" category = str(row[category_column]).strip() if pd.notna(row[category_column]) else "" if url and url.lower() not in ['url', 'nan']: self.sheet_data.append({ "index": index, "url": url, "category": category if category.lower() != 'nan' else "" }) # Добавляем в результаты self.results_data.append({ "url": url, "category": category if category.lower() != 'nan' else "" }) if not self.sheet_data: return "❌ Ошибка: Не найдены валидные URL", "" self.current_index = 0 self.sheet_url = sheet_url success_msg = f"✅ Подключено успешно! Найдено {len(self.sheet_data)} записей" first_url = self.get_current_url_for_display() return success_msg, first_url except Exception as e: logger.error(f"Ошибка подключения к таблице: {e}") error_msg = f"❌ Ошибка: {str(e)}" error_msg += "\n\nУбедитесь что:\n- Таблица публичная (доступна всем по ссылке)\n- URL корректный" return error_msg, "" def get_current_url_for_display(self) -> str: """Возвращает текущий URL для отображения в iframe""" if not self.sheet_data or self.current_index >= len(self.sheet_data): return "" url = self.sheet_data[self.current_index]["url"] # Добавляем http:// если протокол не указан if url and not url.startswith(("http://", "https://")): url = "http://" + url return url def get_current_info(self) -> Tuple[str, str, str]: """Возвращает информацию о текущей записи""" if not self.sheet_data: return "", "", "Нет данных" if self.current_index >= len(self.sheet_data): self.current_index = 0 current = self.sheet_data[self.current_index] url = current["url"] category = current["category"] info = f"Запись {self.current_index + 1} из {len(self.sheet_data)}" return url, category, info def navigate_to_index(self, index: int) -> Tuple[str, str, str, str]: """Переходит к записи по индексу""" if not self.sheet_data: return "", "", "", "Нет данных" # Ограничиваем индекс допустимыми значениями index = max(0, min(index, len(self.sheet_data) - 1)) self.current_index = index url, category, info = self.get_current_info() iframe_url = self.get_current_url_for_display() return url, category, info, iframe_url def previous_record(self) -> Tuple[str, str, str, str]: """Переход к предыдущей записи""" if not self.sheet_data: return "", "", "", "Нет данных" if self.current_index > 0: self.current_index -= 1 else: self.current_index = len(self.sheet_data) - 1 return self.navigate_to_index(self.current_index) def next_record(self) -> Tuple[str, str, str, str]: """Переход к следующей записи""" if not self.sheet_data: return "", "", "", "Нет данных" if self.current_index < len(self.sheet_data) - 1: self.current_index += 1 else: self.current_index = 0 return self.navigate_to_index(self.current_index) def save_category(self, category: str) -> Tuple[str, str]: """Сохраняет категорию локально и возвращает CSV для скачивания""" if not self.sheet_data: return "❌ Нет данных для сохранения", "" try: # Обновляем локальные данные self.sheet_data[self.current_index]["category"] = category self.results_data[self.current_index]["category"] = category # Создаем CSV файл с результатами df_results = pd.DataFrame(self.results_data) csv_buffer = io.StringIO() df_results.to_csv(csv_buffer, index=False, encoding='utf-8') csv_content = csv_buffer.getvalue() status_msg = f"✅ Категория '{category}' сохранена локально" return status_msg, csv_content except Exception as e: logger.error(f"Ошибка сохранения категории: {e}") return f"❌ Ошибка: {str(e)}", "" def export_results(self) -> str: """Экспортирует все результаты в CSV""" if not self.results_data: return "" df_results = pd.DataFrame(self.results_data) csv_buffer = io.StringIO() df_results.to_csv(csv_buffer, index=False, encoding='utf-8') return csv_buffer.getvalue() # Создаем экземпляр приложения app = WebsiteCategorizerApp() # Создание Gradio интерфейса with gr.Blocks( title="Категоризатор веб-сайтов", theme=gr.themes.Soft() ) as demo: gr.HTML("

🌐 Категоризатор веб-сайтов

") gr.HTML("

Подключите публичную Google таблицу с URL и назначайте категории сайтам

") # Инструкция для пользователя with gr.Accordion("📖 Как использовать", open=False): gr.HTML("""

Подготовка Google таблицы:

  1. Откройте вашу Google таблицу
  2. Нажмите "Поделиться" → "Доступ по ссылке" → "Просматривать могут все"
  3. Скопируйте ссылку

Структура таблицы:

Использование:

  1. Вставьте ссылку на таблицу и подключитесь
  2. Просматривайте сайты в окне предпросмотра
  3. Выбирайте категорию из списка
  4. Скачайте результаты в конце работы
""") with gr.Row(): with gr.Column(scale=1): # Панель подключения gr.HTML("

📊 Подключение к таблице

") sheet_url_input = gr.Textbox( label="URL Google таблицы", placeholder="https://docs.google.com/spreadsheets/d/.../edit?usp=sharing", value="", info="Таблица должна быть публичной (доступна всем по ссылке)" ) connect_btn = gr.Button("🔗 Подключиться", variant="primary", size="lg") connection_status = gr.HTML("") # Панель навигации with gr.Group(): gr.HTML("

🧭 Навигация

") with gr.Row(): prev_btn = gr.Button("⬅️ Предыдущий", size="sm") next_btn = gr.Button("Следующий ➡️", size="sm") record_info = gr.HTML("") current_url_display = gr.Textbox( label="Текущий URL", interactive=False ) # Панель категоризации with gr.Group(): gr.HTML("

🏷️ Категоризация

") category_dropdown = gr.Dropdown( choices=app.categories, label="Выберите категорию", value="" ) save_status = gr.HTML("") # Панель экспорта with gr.Group(): gr.HTML("

💾 Экспорт результатов

") export_btn = gr.Button("📥 Скачать результаты (CSV)", variant="secondary") export_file = gr.File(label="Файл с результатами", visible=False) with gr.Column(scale=2): gr.HTML("

🌍 Предварительный просмотр сайта

") website_viewer = gr.HTML( value="""

🚀 Добро пожаловать!

Подключите Google таблицу для просмотра сайтов

Здесь будет отображаться предварительный просмотр сайтов

""" ) gr.HTML("""

⚠️ Некоторые сайты могут блокировать отображение в iframe по соображениям безопасности

""") # Скрытое состояние для CSV данных csv_data = gr.State("") # Обработчики событий def handle_connect(url): """Обработчик подключения к таблице""" status, iframe_url = app.connect_to_sheet(url) if "✅" in status: # Успешное подключение url_display, category, info = app.get_current_info() if iframe_url: iframe_html = f'' else: iframe_html = "

❌ URL не найден или некорректен

" info_html = f'
{info}
' if info else "" return ( gr.HTML(f'
{status}
'), iframe_html, url_display, category, info_html ) else: # Ошибка return ( gr.HTML(f'
{status}
'), website_viewer.value, "", "", gr.HTML("") ) def handle_navigation(direction): """Обработчик навигации""" if direction == "next": url_display, category, info, iframe_url = app.next_record() else: url_display, category, info, iframe_url = app.previous_record() if iframe_url: iframe_html = f'' else: iframe_html = "

❌ URL не найден

" info_html = f'
{info}
' if info else "" return ( iframe_html, url_display, category, info_html ) def handle_category_change(category): """Обработчик изменения категории""" if category: status, csv_content = app.save_category(category) if "✅" in status: status_html = f'
{status}
' else: status_html = f'
{status}
' return status_html, csv_content return gr.HTML(""), "" def handle_export(): """Обработчик экспорта результатов""" csv_content = app.export_results() if csv_content: # Сохраняем во временный файл with open("results.csv", "w", encoding="utf-8") as f: f.write(csv_content) return gr.File(value="results.csv", visible=True) return gr.File(visible=False) # Привязка событий connect_btn.click( handle_connect, inputs=[sheet_url_input], outputs=[connection_status, website_viewer, current_url_display, category_dropdown, record_info] ) next_btn.click( lambda: handle_navigation("next"), outputs=[website_viewer, current_url_display, category_dropdown, record_info] ) prev_btn.click( lambda: handle_navigation("previous"), outputs=[website_viewer, current_url_display, category_dropdown, record_info] ) category_dropdown.change( handle_category_change, inputs=[category_dropdown], outputs=[save_status, csv_data] ) export_btn.click( handle_export, outputs=[export_file] ) # Запуск приложения if __name__ == "__main__": demo.launch()