import json import os import hashlib import requests import time import traceback from datetime import datetime from dotenv import load_dotenv load_dotenv() # ================= КОНФИГУРАЦИЯ ================= URL_PARSER_ROOT = "https://fazziclay.com/api/v1/vstu_schedule_parser_v2/parser.json" BASE_URL_FILES = "https://fazziclay.com/api/v1/vstu_schedule_parser_v2/parsed/" FILE_RESULT = "result.json" FILE_CACHE = "raw_cache.json" # Настройки для Docker / Окружения POLL_INTERVAL = int(os.environ.get("POLL_INTERVAL", 600)) # По умолчанию 10 минут GEMINI_API_KEY = os.environ.get("GEMINI_API_KEY") PROXY_URL = os.environ.get("PROXY_URL") # Например: socks5://user:pass@127.0.0.1:1080 # ================================================= def log(msg): """Логирование с принудительным сбросом буфера (важно для Docker).""" print(f"[{datetime.now().strftime('%d.%m.%Y %H:%M:%S')}] {msg}", flush=True) def get_raw_hash(raw_list): normalized = "|".join(sorted([str(i).strip() for i in raw_list])) return hashlib.sha1(normalized.encode('utf-8')).hexdigest() def load_json(filename, default): if os.path.exists(filename): with open(filename, 'r', encoding='utf-8') as f: try: return json.load(f) except: return default return default def save_json(filename, data): with open(filename, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) def ask_gemini(unknown_raws): """Отправляет неизвестные записи в Gemini API для разбора.""" if not GEMINI_API_KEY: log("[!] GEMINI_API_KEY не задан! ИИ-парсинг пропущен.") return {} proxies = {} if PROXY_URL: # requests автоматически перенаправит трафик через proxy proxies = {"http": PROXY_URL, "https": PROXY_URL} results = {} items = list(unknown_raws.items()) batch_size = 40 # Разбиваем на батчи, чтобы не перегрузить лимиты вывода ИИ for i in range(0, len(items), batch_size): batch = dict(items[i:i+batch_size]) log(f"[*] Отправка батча в Gemini ({i+1}-{min(i+batch_size, len(items))} из {len(items)})...") prompt = f""" Ты парсер академического расписания. Я даю тебе JSON, где ключи - хэши, а значения - массив сырых строк расписания. Твоя задача вернуть JSON, где ключи - те же хэши, а значения - объекты с ключами: - "subject": строка, название предмета (или "Не указан") - "location": строка, аудитории через запятую (или "Не указана") - "teacher": строка, ФИО преподавателя(ей) через запятую (или "Не указан") Верни ТОЛЬКО валидный JSON без форматирования markdown. Входные данные: {json.dumps(batch, ensure_ascii=False)} """ url = f"https://generativelanguage.googleapis.com/v1beta/models/gemini-1.5-flash:generateContent?key={GEMINI_API_KEY}" payload = { "contents": [{"parts":[{"text": prompt}]}], "generationConfig": { "response_mime_type": "application/json" # Заставляем Gemini вернуть строго JSON } } try: resp = requests.post(url, json=payload, proxies=proxies, timeout=60) resp.raise_for_status() data = resp.json() text_response = data['candidates'][0]['content']['parts'][0]['text'] parsed_batch = json.loads(text_response) results.update(parsed_batch) log(f"[+] Батч успешно обработан ИИ.") except Exception as e: log(f"[!] Ошибка API Gemini в батче: {e}") # Не останавливаем процесс, просто вернем те результаты, которые успели распарситься. # Ошибочные записи отправятся при следующем цикле через 10 минут. return results def fetch_and_build(cache): """Основной этап скачивания файлов V2 и построения структуры.""" log("[*] Загрузка списка файлов парсера V2...") parser_data = requests.get(URL_PARSER_ROOT, timeout=30).json() final_groups = {} excels_list =[] unknown_raws = {} for file_info in parser_data.get("all_files",[]): file_url = f"{BASE_URL_FILES}{requests.utils.quote(file_info['json_represent'])}" log(f"[*] Обработка файла: {file_info['json_represent']}") try: faculty_data = requests.get(file_url, timeout=30).json() except Exception as e: log(f"[!] Ошибка скачивания {file_url}: {e}") continue # В-третьих: Добавляем копию excel без 'sheets' excel_meta = {k: v for k, v in faculty_data.items() if k != 'sheets'} excels_list.append(excel_meta) # Получаем data_source_hash из файла excel file_ds_hash = faculty_data.get("data_source_hash", faculty_data.get("excel", {}).get("data_source_hash", "UNKNOWN")) sheets = faculty_data.get("sheets", {}) for sheet_data in sheets.values(): groups = sheet_data.get("groups", {}) for group_id, group_data in groups.items(): if group_data["name"] not in final_groups: final_groups[group_data["name"]] = { "name": group_data["name"], "facultet": faculty_data.get('excel', {}).get('facultet', 'Неизвестно'), "position": group_data.get("position"), "position_human": group_data.get("position_human"), "slots": {}, "data_source_hash": file_ds_hash # ИЗМЕНЕНИЕ: Берем хэш из метаданных экселя } slots = group_data.get("slots", {}) for slot_key, pair_value in slots.items(): if not isinstance(pair_value, dict): continue if slot_key not in final_groups[group_data["name"]]["slots"]: final_groups[group_data["name"]]["slots"][slot_key] = {} for pair_key, pair_data in pair_value.items(): if not (isinstance(pair_key, str) and '-' in pair_key): continue events = pair_data if isinstance(pair_data, list) else [pair_data] for i, event in enumerate(events): current_pair_id = pair_key if i == 0 else f"{pair_key}_{i}" if not isinstance(event, dict): continue raw_list = event.get("raw",[]) r_hash = get_raw_hash(raw_list) if r_hash in cache: res = cache[r_hash] locs =[l.strip() for l in res.get("location", "").split(",")] if res.get("location") and res.get("location") not in ["Не указана", "Не указан"] else [] leads =[l.strip() for l in res.get("teacher", "").split(",")] if res.get("teacher") and res.get("teacher") not in ["Не указана", "Не указан"] else[] final_groups[group_data["name"]]["slots"][slot_key][current_pair_id] = { "discipline_name": res.get("subject", "Не указан"), "locations": locs, "leads": leads, "is_solid": event.get("is_solid", True), "is_flow": event.get("is_flow", False), "raw": raw_list, "weekday": event.get("weekday"), "weeknum": event.get("weeknum"), "excel_range": event.get("excel_range"), "excel_pos": event.get("excel_pos") } else: unknown_raws[r_hash] = raw_list return final_groups, excels_list, unknown_raws def job_iteration(): """Выполнение одной итерации обновления расписания.""" log("--- Начало итерации обновления расписания ---") cache = load_json(FILE_CACHE, {}) # ПЕРВЫЙ ПРОХОД: собираем данные и ищем неизвестные строки final_groups, excels_list, unknown_raws = fetch_and_build(cache) # Если есть неизвестные строки — отправляем в ИИ if unknown_raws: log(f"[*] Найдено {len(unknown_raws)} новых уникальных записей. Обращаемся к Gemini API...") ai_data = ask_gemini(unknown_raws) if ai_data: cache.update(ai_data) save_json(FILE_CACHE, cache) log(f"[+] Кэш обновлен, добавлено {len(ai_data)} записей.") # ВТОРОЙ ПРОХОД: пересобираем расписание уже с новыми кэшированными данными ИИ log("[*] Пересборка структуры с учетом новых данных от ИИ...") final_groups, excels_list, _ = fetch_and_build(cache) # Формируем финальный результат output = { "actual_at": int(datetime.now().timestamp()), "groups": final_groups, "excels": excels_list } save_json(FILE_RESULT, output) log(f"[+] Итерация завершена. Файл {FILE_RESULT} успешно обновлен.") def main(): log("=== Система Совместимости V1-V2 (Служба) запущена ===") if PROXY_URL: log(f"[*] Настроен прокси-сервер: {PROXY_URL}") if not GEMINI_API_KEY: log("[!] ВНИМАНИЕ: GEMINI_API_KEY не задан. Интеллектуальный парсинг работать не будет!") # Бесконечный цикл - требование №1 while True: try: job_iteration() except Exception as e: log(f"[!!!] Критическая ошибка на верхнем уровне: {e}") traceback.print_exc() log(f"[*] Ожидание {POLL_INTERVAL} секунд перед следующей итерацией...\n") time.sleep(POLL_INTERVAL) if __name__ == "__main__": main()