from playwright.sync_api import sync_playwright
from pathlib import Path
from datetime import datetime, timedelta
from logging.handlers import RotatingFileHandler
import json
import threading
import signal
import sys
import os
import logging
import urllib.request
import urllib.parse
import ssl
from flask import Flask, jsonify, Response, request
os.environ.setdefault("DISPLAY", ":99")
BASE_DIR = Path(__file__).resolve().parent
SESSION_FILE = BASE_DIR / "session.json"
SCHEDULE_FILE = BASE_DIR / "schedule.json"
LOG_FILE = BASE_DIR / "app.log"
CREDENTIALS_FILE = BASE_DIR / ".journal_credentials"
LOGIN_URL = "https://journal.top-academy.ru/ru/auth/login/index"
SCHEDULE_URL = "https://journal.top-academy.ru/ru/main/schedule/page"
API_BASE = "https://journal.top-academy.ru/api/v2/schedule/operations/get-month"
LOG_FORMAT = "%(asctime)s | %(levelname)-7s | %(message)s"
LOG_DATE = "%Y-%m-%d %H:%M:%S"
log = logging.getLogger("schedule")
log.setLevel(logging.DEBUG)
_console = logging.StreamHandler(sys.stdout)
_console.setLevel(logging.INFO)
_console.setFormatter(logging.Formatter(LOG_FORMAT, datefmt=LOG_DATE))
log.addHandler(_console)
_file = RotatingFileHandler(
str(LOG_FILE), maxBytes=5 * 1024 * 1024, backupCount=5, encoding="utf-8",
)
_file.setLevel(logging.DEBUG)
_file.setFormatter(logging.Formatter(LOG_FORMAT, datefmt=LOG_DATE))
log.addHandler(_file)
app = Flask(__name__)
fetch_status = {
"running": False,
"message": "",
"last_error": "",
"last_update": "",
}
status_lock = threading.Lock()
browser_lock = threading.Lock()
shutdown_event = threading.Event()
activity_log = []
activity_lock = threading.Lock()
def log_event(level, text):
"""Append a human-readable event shown on the site (кириллица)."""
try:
with activity_lock:
stamp = datetime.now().strftime("%d.%m %H:%M:%S")
activity_log.append({"time": stamp, "level": level, "text": text})
del activity_log[:-60]
except Exception as e:
log.error("Failed to log event: %s", e)
def set_status(**kwargs):
try:
with status_lock:
fetch_status.update(kwargs)
log.debug("Status updated: %s", kwargs)
except Exception as e:
log.error("Failed to set status: %s", e)
def safe_json_load(path):
try:
if not path.exists():
return []
with open(path, "r", encoding="utf-8") as f:
data = json.load(f)
log.debug("Loaded %d items from %s", len(data), path.name)
return data
except json.JSONDecodeError as e:
log.error("Corrupt JSON in %s: %s", path.name, e)
backup = path.with_suffix(".json.bak")
try:
path.rename(backup)
except Exception:
pass
return []
except Exception as e:
log.error("Failed to read %s: %s", path.name, e)
return []
def safe_json_save(path, data):
try:
tmp = path.with_suffix(".json.tmp")
with open(tmp, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
tmp.replace(path)
log.debug("Saved %d items to %s", len(data), path.name)
except Exception as e:
log.error("Failed to save %s: %s", path.name, e)
raise
def get_schedule():
log.info("Starting schedule fetch...")
if not browser_lock.acquire(timeout=5):
raise RuntimeError("Another update is already running.")
try:
return _get_schedule_inner()
finally:
browser_lock.release()
log.info("Lock released.")
def _merge_with_previous(new_items):
"""Merge fresh lessons with the last known cache.
Fresh items win for the same (date, lesson); older items that are absent
from the fresh fetch are kept as a fallback so already-published days are
not lost when the journal temporarily stops returning them.
"""
old_items = safe_json_load(SCHEDULE_FILE) or []
old_keys = {(x.get("date"), x.get("lesson")) for x in old_items}
new_keys = {(x.get("date"), x.get("lesson")) for x in new_items}
dropped = sorted({k for k in old_keys - new_keys})
added = sorted({k for k in new_keys - old_keys})
kept = [x for x in old_items if (x.get("date"), x.get("lesson")) not in new_keys]
merged = new_items + kept
if dropped:
dates_dropped = sorted({d for d, _ in dropped})
log_event("warning", f"Журнал вернул меньше: убраны даты {', '.join(dates_dropped)}, оставлены из кэша ({len(dropped)} ур.)")
if added:
dates_added = sorted({d for d, _ in added})
log_event("ok", f"Обнаружены новые занятия: {', '.join(dates_added)}")
return sorted(merged, key=lambda x: (x.get("date", ""), x.get("lesson", 0)))
def _load_cookies():
if not SESSION_FILE.exists():
return {}
try:
with open(SESSION_FILE, "r", encoding="utf-8") as f:
state = json.load(f)
cookies = {}
for c in state.get("cookies", []):
if "top-academy.ru" in c.get("domain", ""):
cookies[c["name"]] = c["value"]
return cookies
except Exception as e:
log.error("Failed to load cookies: %s", e)
return {}
def _fetch_schedule_direct():
cookies = _load_cookies()
if not cookies:
return None
cookie_header = "; ".join(f"{k}={v}" for k, v in cookies.items())
now = datetime.now()
months = []
for offset in range(-1, 2):
d = now + timedelta(days=offset * 30)
months.append(d.strftime("%Y-%m"))
all_data = []
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
for month in sorted(set(months)):
url = f"{API_BASE}?month={month}"
try:
req = urllib.request.Request(url)
req.add_header("Cookie", cookie_header)
req.add_header("User-Agent", "Mozilla/5.0")
req.add_header("Accept", "application/json")
req.add_header("Referer", SCHEDULE_URL)
resp = urllib.request.urlopen(req, timeout=15, context=ctx)
data = json.loads(resp.read().decode("utf-8"))
if isinstance(data, list) and len(data) > 0:
all_data.extend(data)
log.info("API month %s: %d lessons.", month, len(data))
except urllib.error.HTTPError as e:
if e.code in (401, 403):
log.warning("Auth failed for month %s (HTTP %d).", month, e.code)
return None
except Exception as e:
log.warning("Request error for month %s: %s", month, e)
if not all_data:
return None
return sorted(all_data, key=lambda x: (x.get("date", ""), x.get("lesson", 0)))
def _fetch_schedule_browser():
log.info("Falling back to browser...")
with sync_playwright() as p:
browser = p.chromium.launch(headless=True, args=["--no-sandbox", "--disable-blink-features=AutomationControlled"])
try:
if SESSION_FILE.exists():
context = browser.new_context(
storage_state=str(SESSION_FILE),
user_agent="Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36"
)
set_status(message="Updating via browser...")
else:
set_status(message="No session available.")
browser.close()
return None
schedule_data = []
seen = set()
page = context.pages[0] if context.pages else context.new_page()
def handle_response(response):
try:
if "/api/v2/schedule/operations/get-month" not in response.url:
return
if response.status != 200:
return
data = response.json()
if isinstance(data, list):
for item in data:
key = (item.get("date"), item.get("lesson"))
if key not in seen:
seen.add(key)
schedule_data.append(item)
except Exception:
pass
page.on("response", handle_response)
page.goto(SCHEDULE_URL)
page.wait_for_load_state("networkidle")
page.wait_for_timeout(5000)
finally:
browser.close()
if not schedule_data:
return None
return sorted(schedule_data, key=lambda x: (x.get("date", ""), x.get("lesson", 0)))
def _web_relogin():
"""Renew the journal session using protected perms-600 credentials file."""
if not CREDENTIALS_FILE.exists():
log.warning("No credentials file; relogin skipped.")
return False
try:
with open(CREDENTIALS_FILE, "r", encoding="utf-8") as f:
lines = f.read().splitlines()
if len(lines) < 2:
log.warning("Credentials file malformed.")
return False
username, password = lines[0].strip(), lines[1].strip()
if not username or not password:
return False
except Exception as e:
log.error("Failed to read credentials: %s", e)
return False
try:
log.info("Relogin: opening login page...")
with sync_playwright() as p:
browser = p.chromium.launch(
headless=True,
args=["--no-sandbox", "--disable-blink-features=AutomationControlled"],
)
context = browser.new_context()
page = context.new_page()
page.goto(LOGIN_URL)
page.wait_for_load_state("networkidle")
page.wait_for_timeout(2000)
try:
page.fill('input[name="username"]', username)
page.fill('input[name="password"]', password)
page.wait_for_timeout(500)
page.click('button[type="submit"], input[type="submit"]')
page.wait_for_load_state("networkidle")
page.wait_for_timeout(4000)
except Exception as e:
log.warning("Relogin form step error: %s", e)
if "login" not in page.url:
context.storage_state(path=str(SESSION_FILE))
log.info("Relogin successful. Session renewed and saved.")
return True
else:
log.warning("Relogin failed: still on login page.")
return False
except Exception as e:
log.error("Relogin error: %s", e)
return False
def _retry_after_relogin():
log_event("warning", "Сессия журнала устарела — пытаюсь войти заново (автоматически)...")
if not _web_relogin():
log_event("error", "Не удалось заново войти в журнал. Проверь аккаунт.")
return None
log_event("info", "Вход в журнал выполнен заново. Повторяю загрузку расписания...")
result = _fetch_schedule_direct()
if result:
merged = _merge_with_previous(result)
safe_json_save(SCHEDULE_FILE, merged)
set_status(
last_update=datetime.now().strftime("%H:%M:%S"),
message=f"Готово: {len(merged)} занятий (переподключение).",
)
log.info("Fetch complete (relogin+http): %d lessons.", len(merged))
log_event("ok", f"Расписание обновлено после повторного входа: {len(merged)} занятий (данные журнала: {len(result)}).")
return merged
result = _fetch_schedule_browser()
if result:
merged = _merge_with_previous(result)
safe_json_save(SCHEDULE_FILE, merged)
set_status(
last_update=datetime.now().strftime("%H:%M:%S"),
message=f"Готово: {len(merged)} занятий (браузер).",
)
log.info("Fetch complete (relogin+browser): %d lessons.", len(merged))
log_event("ok", f"Расписание обновлено после повторного входа (браузер): {len(merged)} занятий (данные журнала: {len(result)}).")
return merged
log_event("error", "После повторного входа расписание так и не получилось загрузить.")
return None
def _get_schedule_inner():
set_status(message="Получаю расписание из журнала...")
log_event("info", "Начинаю обновление расписания...")
result = _fetch_schedule_direct()
if result:
merged = _merge_with_previous(result)
safe_json_save(SCHEDULE_FILE, merged)
set_status(
last_update=datetime.now().strftime("%H:%M:%S"),
message=f"Готово: {len(merged)} занятий.",
)
log.info("Fetch complete (HTTP): %d lessons.", len(merged))
log_event("ok", f"Журнал ответил (напрямую): {len(result)} занятий. Показано: {len(merged)}.")
return merged
log.info("Direct fetch failed, trying browser...")
log_event("warning", "Прямое соединение с журналом не сработало — открываю журнал в браузере...")
if not SESSION_FILE.exists():
log.info("No session file; going straight to relogin...")
return _retry_after_relogin()
result = _fetch_schedule_browser()
if result:
merged = _merge_with_previous(result)
safe_json_save(SCHEDULE_FILE, merged)
set_status(
last_update=datetime.now().strftime("%H:%M:%S"),
message=f"Готово: {len(merged)} занятий (браузер).",
)
log.info("Fetch complete (browser): %d lessons.", len(merged))
log_event("ok", f"Журнал открылся в браузере: {len(result)} занятий. Показано: {len(merged)}.")
return merged
log_event("error", "Браузер тоже не смог получить расписание из журнала.")
log.warning("Session likely expired; attempting relogin...")
return _retry_after_relogin()
def run_update():
log.info("Update thread started.")
try:
set_status(running=True, last_error="")
for attempt in range(1, 4):
try:
log.info("Attempt %d/3", attempt)
get_schedule()
break
except Exception as e:
log.warning("Attempt %d failed: %s", attempt, e, exc_info=True)
log_event("error", f"Попытка {attempt} не удалась: {e}")
if attempt < 3:
set_status(message=f"Попытка {attempt} не удалась, повтор...")
except Exception as e:
set_status(message=f"Ошибка: {e}", last_error=str(e))
log.error("All update attempts failed.", exc_info=True)
log_event("error", f"Обновление окончательно не удалось: {e}")
finally:
set_status(running=False)
log.info("Update thread finished.")
@app.route("/")
def index():
try:
return Response(INDEX_HTML, mimetype="text/html")
except Exception as e:
log.error("Failed to serve index: %s", e)
return "Internal Server Error", 500
@app.route("/api/schedule")
def api_schedule():
try:
data = safe_json_load(SCHEDULE_FILE)
return jsonify(sorted(data, key=lambda x: (x.get("date", ""), x.get("lesson", 0))))
except Exception as e:
log.error("API /schedule error: %s", e)
return jsonify([])
@app.route("/api/update", methods=["POST"])
def api_update():
try:
with status_lock:
if fetch_status["running"]:
return jsonify({"ok": False, "message": "Already updating"})
thread = threading.Thread(target=run_update, daemon=True)
thread.start()
log.info("Update triggered by user.")
return jsonify({"ok": True, "message": "Update started"})
except Exception as e:
log.error("API /update error: %s", e)
return jsonify({"ok": False, "message": str(e)})
@app.route("/api/status")
def api_status():
try:
with status_lock:
return jsonify(dict(fetch_status))
except Exception as e:
return jsonify({"running": False, "message": "", "last_error": ""})
@app.route("/api/logs")
def api_logs():
try:
with activity_lock:
return jsonify(list(activity_log[-30:]))
except Exception:
return jsonify([])
@app.errorhandler(404)
def not_found(e):
return jsonify({"error": "Not found"}), 404
@app.errorhandler(500)
def server_error(e):
log.error("500 error: %s", e)
return jsonify({"error": "Internal server error"}), 500
INDEX_HTML = r"""
TRIRON — Расписание