services/google_sheets.py - master - Shtirlitc/TickBot

services/google_sheets.py
308 строк13 KB
Shtirlyts
Initial commit: TickBot attendance bot with stale-status fixes
31 июл 2026, 23:40
31 июл 2026, 23:4030e955c
import logging
import time
from typing import Optional, Dict, List
from datetime import datetime, timezone, timedelta

import gspread
from config import get_google_credentials, GROUPS
from aiocache import Cache
from aiocache.serializers import PickleSerializer

logger = logging.getLogger(__name__)

# Единый кэш через aiocache
cache = Cache(Cache.MEMORY, serializer=PickleSerializer())

# ===========================
# GOOGLE SHEETS CLIENT
# ===========================
class GoogleSheetsClient:
    """Клиент для работы с Google Sheets"""

    def __init__(self):
        self.gc = None
        self.spreadsheets: Dict[str, gspread.Spreadsheet] = {}
        self._connect()

    def _connect(self):
        """Подключение к Google Sheets"""
        try:
            creds = get_google_credentials()

            if creds:
                self.gc = gspread.service_account_from_dict(creds)
                logger.info("✅ Подключение к Google Sheets (из env)")
            else:
                self.gc = gspread.service_account(filename='data/credentials.json')
                logger.info("✅ Подключение к Google Sheets (из файла)")

            for group_name, info in GROUPS.items():
                try:
                    spreadsheet = self.gc.open_by_url(info['spreadsheet_url'])
                    self.spreadsheets[group_name] = spreadsheet
                    logger.info(f"✅ Группа {group_name} подключена")
                except Exception as e:
                    logger.error(f"❌ Ошибка подключения группы {group_name}: {e}")

        except Exception as e:
            logger.error(f"❌ Ошибка подключения к Google Sheets: {e}")
            raise

    def get_students_data(self, group_name: str) -> list:
        """Загружает данные студентов из Google Sheets (синхронно)."""
        try:
            if group_name not in GROUPS:
                logger.error(f"❌ Группа {group_name} не найдена в конфиге")
                return []

            spreadsheet_url = GROUPS[group_name]["spreadsheet_url"]
            
            # Синхронный вызов — без await!
            sh = self.gc.open_by_url(spreadsheet_url)
            worksheet = sh.worksheet("Студенты")  # ← имя листа

            all_rows = worksheet.get_all_values()  # ← синхронно

            if len(all_rows) < 2:
                return []

            expected_headers = ["ФИО", "№", "Подгруппа", "Telegram ID"]
            header_row = all_rows[0]

            col_index = {}
            for h in expected_headers:
                try:
                    col_index[h] = header_row.index(h)
                except ValueError:
                    col_index[h] = -1  # не найдено

            students = []
            for row in all_rows[1:]:
                if not any(cell.strip() for cell in row):
                    continue

                student = {
                    "ФИО": row[col_index["ФИО"]].strip() if col_index["ФИО"] != -1 else "",
                    "№": row[col_index["№"]].strip() if col_index["№"] != -1 else "",
                    "Подгруппа": int(row[col_index["Подгруппа"]]) if col_index["Подгруппа"] != -1 and row[col_index["Подгруппа"]].isdigit() else 1,
                    "Telegram ID": row[col_index["Telegram ID"]].strip() if col_index["Telegram ID"] != -1 else ""
                }

                if student["ФИО"] and student["№"]:
                    students.append(student)

            logger.info(f"✅ Загружено {len(students)} студентов из GS для {group_name}")
            return students

        except Exception as e:
            logger.error(f"❌ Ошибка загрузки студентов из GS для {group_name}: {e}")
            return []
    
    async def find_student_by_telegram_id(self, telegram_id: int) -> Optional[dict]:
        """Найти студента по Telegram ID во всех группах"""
        cache_key = f"gs:student:tg:{telegram_id}"
        cached = await cache.get(cache_key)
        if cached:
            logger.debug(f"📦 Кэш: студент {telegram_id} найден")
            return cached

        for group_name, spreadsheet in self.spreadsheets.items():
            try:
                students_sheet = spreadsheet.worksheet("Студенты")
                all_records = students_sheet.get_all_records()

                for student in all_records:
                    existing_id = str(student.get('Telegram ID', '')).strip()
                    if existing_id and existing_id.isdigit() and int(existing_id) == telegram_id:
                        result = {
                            'fio': student['ФИО'],
                            'student_number': str(student['№']),
                            'subgroup': int(student['Подгруппа']),
                            'group_name': group_name,
                            'telegram_id': telegram_id
                        }
                        await cache.set(cache_key, result, ttl=3600)
                        logger.info(f"✅ Найден студент {telegram_id} в группе {group_name}")
                        return result
            except Exception as e:
                logger.error(f"❌ Ошибка поиска в группе {group_name}: {e}")
                continue

        logger.info(f"🔍 Студент {telegram_id} не найден")
        return None

    async def find_student_by_fio(self, fio: str, group_name: str) -> Optional[dict]:
        """Найти студента по ФИО в конкретной группе"""
        cache_key = f"gs:student:fio:{group_name}:{fio.lower()}"
        cached = await cache.get(cache_key)
        if cached:
            return cached

        if group_name not in self.spreadsheets:
            logger.error(f"❌ Группа {group_name} не подключена")
            return None

        try:
            spreadsheet = self.spreadsheets[group_name]
            students_sheet = spreadsheet.worksheet("Студенты")
            all_records = students_sheet.get_all_records()

            for student in all_records:
                if student['ФИО'].lower() == fio.lower():
                    result = {
                        'fio': student['ФИО'],
                        'student_number': str(student['№']),
                        'subgroup': int(student['Подгруппа']),
                        'telegram_id': int(student.get('Telegram ID', 0)) if str(student.get('Telegram ID', '')).strip().isdigit() else None
                    }
                    await cache.set(cache_key, result, ttl=3600)
                    logger.info(f"✅ Найден студент '{fio}' в группе {group_name}")
                    return result

            logger.info(f"🔍 Студент '{fio}' не найден")
            return None
        except Exception as e:
            logger.error(f"❌ Ошибка поиска ФИО: {e}")
            return None

    async def update_student_telegram_id(self, group_name: str, student_number: str, telegram_id: int):
        """Обновить Telegram ID студента в Google Sheets"""
        if group_name not in self.spreadsheets:
            logger.error(f"❌ Группа {group_name} не подключена")
            return

        try:
            spreadsheet = self.spreadsheets[group_name]
            students_sheet = spreadsheet.worksheet("Студенты")
            cell = students_sheet.find(student_number)
            if cell:
                students_sheet.update_cell(cell.row, 4, str(telegram_id))
                logger.info(f"✅ Обновлён Telegram ID для студента {student_number}")

                # Инвалидируем кэш
                await cache.delete(f"gs:students:{group_name}")
                await cache.delete(f"gs:student:tg:{telegram_id}")
            else:
                logger.warning(f"⚠️ Студент {student_number} не найден")
        except Exception as e:
            logger.error(f"❌ Ошибка обновления Telegram ID: {e}")
            raise

    async def get_schedule_data(self, group_name: str, subgroup: int) -> List[list]:
        """Получить расписание подгруппы"""
        cache_key = f"schedule:{group_name}:subgroup_{subgroup}"
        cached = await cache.get(cache_key)
        if cached:
            return cached

        if group_name not in self.spreadsheets:
            return []

        try:
            spreadsheet = self.spreadsheets[group_name]
            schedule_sheet = spreadsheet.worksheet(f"{subgroup} подгруппа")
            data = schedule_sheet.get_all_values()
            await cache.set(cache_key, data, ttl=300)  # 5 минут
            logger.info(f"✅ Загружено {len(data)} строк расписания для {group_name}, подгруппа {subgroup}")
            return data
        except Exception as e:
            logger.error(f"❌ Ошибка получения расписания: {e}")
            return []

    async def save_attendance_batch(self, group_name: str, subgroup: int, student_number: str, updates: Dict[str, str]):
        """Сохранить отметки посещаемости batch-запросом"""
        if group_name not in self.spreadsheets:
            logger.error(f"❌ Группа {group_name} не подключена")
            return False

        try:
            spreadsheet = self.spreadsheets[group_name]
            schedule_sheet = spreadsheet.worksheet(f"{subgroup} подгруппа")

            header = schedule_sheet.row_values(1)
            student_col = next((i + 1 for i, c in enumerate(header) if str(c).strip() == str(student_number)), None)
            if not student_col:
                logger.error(f"❌ Колонка студента {student_number} не найдена")
                return False

            batch_updates = [
                {
                    'range': gspread.utils.rowcol_to_a1(int(row_num), student_col),
                    'values': [[emoji]]
                }
                for row_num, emoji in updates.items()
            ]

            if batch_updates:
                schedule_sheet.batch_update(batch_updates)
                logger.info(f"✅ Сохранено {len(batch_updates)} отметок")

                # Очищаем кэш расписания
                await cache.delete(f"schedule:{group_name}:subgroup_{subgroup}")
                return True
            return False
        except Exception as e:
            logger.error(f"❌ Ошибка сохранения отметок: {e}")
            return False

    async def sync_sqlite_to_gs(self, attendance_repo, student_repo) -> int:
        try:
            unsynced = await attendance_repo.get_unsynced()
            if not unsynced:
                logger.info("📦 Нет несинхронизированных отметок")
                return 0

            by_student = {}
            for att in unsynced:
                key = (att.group_name, att.subgroup, att.student_id)
                if key not in by_student:
                    by_student[key] = []
                by_student[key].append((att.id, str(att.row_num), att.status))

            synced_count = 0
            for (group_name, subgroup, student_id), items in by_student.items():
                if group_name not in self.spreadsheets:
                    logger.error(f"❌ Группа {group_name} не подключена")
                    continue

                student = await student_repo.get_by_id(student_id)
                if not student:
                    logger.warning(f"⚠️ Студент с ID={student_id} не найден")
                    continue

                updates = {row_num: status for _, row_num, status in items}
                success = await self.save_attendance_batch(
                    group_name=group_name,
                    subgroup=subgroup,
                    student_number=student.student_number,
                    updates=updates
                )

                if success:
                    for att_id, _, _ in items:
                        await attendance_repo.mark_synced_by_id(att_id)
                    synced_count += len(items)

            logger.info(f"✅ Воркер: Успешно синхронизировано {synced_count} записей")
            return synced_count
        except Exception as e:
            logger.error(f"❌ Ошибка воркера синхронизации: {e}")
            return 0

    def clear_cache(self):
        """Очистить весь кэш — только для обратной совместимости"""
        # Не используется — очищайте через aiocache
        pass


# ===========================
# ГЛОБАЛЬНЫЙ ЭКЗЕМПЛЯР
# ===========================
gs_client: Optional[GoogleSheetsClient] = None


def get_gs_client() -> GoogleSheetsClient:
    global gs_client
    if gs_client is None:
        gs_client = GoogleSheetsClient()
    return gs_client