From 21024f18054330737e38c351e09dc3edd878e107 Mon Sep 17 00:00:00 2001 From: zik Date: Thu, 29 May 2025 20:28:02 +0300 Subject: [PATCH] =?UTF-8?q?=D1=80=D0=BE=D0=B7=D0=BA=D0=BB=D0=B0=D0=B4=20?= =?UTF-8?q?=D0=BF=D0=B0=D1=80=D1=81=D0=B8=D1=82=D1=8C=D1=81=D1=8F=20=D0=B2?= =?UTF-8?q?=20=D0=BE=D0=B1=D0=B8=D0=B4=D0=B2=D1=96=20=D1=81=D1=82=D0=BE?= =?UTF-8?q?=D1=80=D0=BE=D0=BD=D0=B8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- db.py | 115 +++++++++++++++++++++++---------------------------- parser.py | 56 ++++++++++++------------- scheduler.py | 26 +++++------- 3 files changed, 88 insertions(+), 109 deletions(-) diff --git a/db.py b/db.py index 6a373c8..8318546 100644 --- a/db.py +++ b/db.py @@ -7,16 +7,15 @@ DB_PATH = 'schedules.db' def init_db(): """Створює таблиці trains, stations та schedules, якщо їх ще нема.""" with sqlite3.connect(DB_PATH) as con: - # Таблиця поїздів з номером, днями курсування і маршрутом con.execute(''' CREATE TABLE IF NOT EXISTS trains ( id INTEGER PRIMARY KEY AUTOINCREMENT, - train_number TEXT UNIQUE NOT NULL, + train_number TEXT NOT NULL, days TEXT NOT NULL, - route TEXT + route TEXT NOT NULL, + UNIQUE(train_number, route) ); ''') - # Таблиця станцій con.execute(''' CREATE TABLE IF NOT EXISTS stations ( id INTEGER PRIMARY KEY AUTOINCREMENT, @@ -24,7 +23,6 @@ def init_db(): km REAL ); ''') - # Таблиця розкладу: зв'язок поїзда зі станцією та часи con.execute(''' CREATE TABLE IF NOT EXISTS schedules ( train_id INTEGER NOT NULL, @@ -41,90 +39,79 @@ def init_db(): con.commit() -def save_schedule(direction: str, entries: List[Dict]): +def save_schedule(entries: List[Dict]): """ - Зберігає повний розклад для заданого напряму. - direction не використовується напряму — маршрут береться з поля 'route' у entries. + Зберігає повний розклад без видалення попередніх записів. entries — список словників з полями: - { - 'train_number': str, - 'days': '1111111', - 'route': str, - 'times': [ - {'station': str, 'arrival': str, 'departure': str}, ... - ] - } + 'train_number', 'days', 'route', 'times'. """ today = date.today().isoformat() with sqlite3.connect(DB_PATH) as con: - # Видаляємо старі записи за сьогоднішню дату - con.execute('DELETE FROM schedules WHERE travel_date = ?', (today,)) - + train_ids = [] + # Додаємо або оновлюємо поїзди for e in entries: - train_number = e['train_number'] + tn = e['train_number'] days = e['days'] - # Беріть маршрут із поля entry['route'] route = e.get('route', '') - # Додаємо або оновлюємо поїзд con.execute(''' INSERT INTO trains (train_number, days, route) VALUES (?, ?, ?) - ON CONFLICT(train_number) DO UPDATE SET - days = excluded.days, - route = excluded.route - ''', (train_number, days, route)) - train_id = con.execute( - 'SELECT id FROM trains WHERE train_number = ?', (train_number,) + ON CONFLICT(train_number, route) DO UPDATE SET days = excluded.days + ''', (tn, days, route)) + tid = con.execute( + 'SELECT id FROM trains WHERE train_number = ? AND route = ?', + (tn, route) ).fetchone()[0] + train_ids.append(tid) + # Додаємо або оновлюємо станції та розклад + for idx, e in enumerate(entries): + tid = train_ids[idx] for t in e['times']: - station = t['station'] - km = t.get('km') # може бути None - # Додаємо або оновлюємо станцію + st = t['station'] + km = t.get('km') con.execute(''' INSERT INTO stations (name, km) VALUES (?, ?) ON CONFLICT(name) DO UPDATE SET km = COALESCE(excluded.km, stations.km) - ''', (station, km)) - station_id = con.execute( - 'SELECT id FROM stations WHERE name = ?', (station,) + ''', (st, km)) + sid = con.execute( + 'SELECT id FROM stations WHERE name = ?', (st,) ).fetchone()[0] - - arrival = t['arrival'] - departure = t['departure'] - # Вставляємо або замінюємо розклад поїзда на станції + arr = t['arrival'] + dep = t['departure'] con.execute(''' INSERT OR REPLACE INTO schedules (train_id, station_id, arrival_time, departure_time, travel_date) VALUES (?, ?, ?, ?, ?) - ''', (train_id, station_id, arrival, departure, today)) + ''', (tid, sid, arr, dep, today)) con.commit() -def get_schedule(direction: str = None, travel_date: Optional[str] = None) -> List[Dict]: - """ - Повертає розклад поїздів за датою. - direction поки ігнорується; можна додати фільтрацію за route. - """ - travel_date = travel_date or date.today().isoformat() +def get_schedule(route: Optional[str] = None, travel_date: Optional[str] = None) -> List[Dict]: + """Повертає розклад поїздів. Якщо вказано route, фільтрує за ним.""" + from datetime import date as _date + travel_date = travel_date or _date.today().isoformat() with sqlite3.connect(DB_PATH) as con: - rows = con.execute(''' - SELECT tr.train_number, tr.route, st.name, sc.arrival_time, sc.departure_time - FROM schedules sc - JOIN trains tr ON sc.train_id = tr.id - JOIN stations st ON sc.station_id = st.id - WHERE sc.travel_date = ? - ORDER BY tr.train_number, st.id - ''', (travel_date,)).fetchall() - + if route: + rows = con.execute(''' + SELECT tr.train_number, tr.route, st.name, sc.arrival_time, sc.departure_time + FROM schedules sc + JOIN trains tr ON sc.train_id = tr.id + JOIN stations st ON sc.station_id = st.id + WHERE sc.travel_date = ? AND tr.route = ? + ORDER BY tr.train_number, st.id + ''', (travel_date, route)).fetchall() + else: + rows = con.execute(''' + SELECT tr.train_number, tr.route, st.name, sc.arrival_time, sc.departure_time + FROM schedules sc + JOIN trains tr ON sc.train_id = tr.id + JOIN stations st ON sc.station_id = st.id + WHERE sc.travel_date = ? + ORDER BY tr.train_number, st.id + ''', (travel_date,)).fetchall() schedule: Dict[tuple, List[Dict]] = {} - for num, route, station, arrival, departure in rows: - schedule.setdefault((num, route), []).append({ - 'station': station, - 'arrival': arrival, - 'departure': departure - }) - return [ - {'train_number': num, 'route': route, 'times': times} - for (num, route), times in schedule.items() - ] + for num, rt, station, arrival, departure in rows: + schedule.setdefault((num, rt), []).append({'station': station, 'arrival': arrival, 'departure': departure}) + return [{'train_number': num, 'route': rt, 'times': times} for (num, rt), times in schedule.items()] diff --git a/parser.py b/parser.py index c14608a..4b1ff17 100644 --- a/parser.py +++ b/parser.py @@ -18,18 +18,16 @@ def parse_days(text: str) -> str: if 'щоденно' in text: return '1111111' if text.startswith('крім'): - days_part = text.split('крім', 1)[1] - days = [d.strip(' .') for d in days_part.split(',')] - mask = [1] * 7 + days = [d.strip(' .') for d in text.split('крім', 1)[1].split(',')] + mask = [1]*7 for d in days: idx = DAY_INDEX.get(d) if idx is not None: mask[idx] = 0 return ''.join(str(b) for b in mask) if text.startswith('по'): - days_part = text.split('по', 1)[1] - days = [d.strip(' .') for d in days_part.split(',')] - mask = [0] * 7 + days = [d.strip(' .') for d in text.split('по', 1)[1].split(',')] + mask = [0]*7 for d in days: idx = DAY_INDEX.get(d) if idx is not None: @@ -37,10 +35,13 @@ def parse_days(text: str) -> str: return ''.join(str(b) for b in mask) return '1111111' - -def fetch_schedule(use_local: bool = False) -> List[Dict]: +def fetch_schedule(tab: int = 1, use_local: bool = False) -> List[Dict]: """ - Повертає список поїздів зі станціями, часами і маршрутом для кожного потяга. + Парсить вкладку розкладу: + tab=1 — Київ→Ніжин, + tab=2 — Ніжин→Київ. + Повертає список поїздів з полями: + 'train_number', 'days', 'route', 'times' (список словників station/arrival/departure). """ # Завантаження HTML if use_local: @@ -52,35 +53,34 @@ def fetch_schedule(use_local: bool = False) -> List[Dict]: html = resp.text soup = BeautifulSoup(html, 'html.parser') + prefix = f'div#tabs-trains{tab}' - # Таблиця з часами руху - times_table = soup.select_one('div#tabs-trains1 table.td_center') + # Таблиця з розкладом + times_table = soup.select_one(f'{prefix} table.td_center') if not times_table: - raise RuntimeError('Не знайдено таблицю розкладу') + raise RuntimeError(f'Не знайдено таблицю розкладу для tab={tab}') - # Маршрути: кожен відповідає одному потягу + # Парсимо маршрути (по одному на потяг) route_tags = times_table.select('td.course') routes = [tag.get_text(strip=True) for tag in route_tags] # Список станцій (35) station_tags = soup.select( - 'div#tabs-trains1 table.left tr.on a.et, ' - 'div#tabs-trains1 table.left tr.onx a.et' + f'{prefix} table.left tr.on a.et, ' + f'{prefix} table.left tr.onx a.et' ) stations = [a.get_text(strip=True) for a in station_tags] - # Заголовок з номерами потягів - rows = times_table.find_all('tr') - header_row = next(r for r in rows if r.find('td', class_='on_right_t')) - train_cells = header_row.find_all('td', class_='on_right_t') + # Заголовок з номерами потягів і днями курсування + trs = times_table.find_all('tr') + header_row = next(r for r in trs if r.find('td', class_='on_right_t')) + cells = header_row.find_all('td', class_='on_right_t') entries: List[Dict] = [] - for idx, cell in enumerate(train_cells): - text = cell.get_text(separator='|', strip=True) - parts = text.split('|') + for idx, cell in enumerate(cells): + parts = cell.get_text(separator='|', strip=True).split('|') num = parts[0].rstrip(',').strip() - days_text = parts[1].strip() if len(parts) > 1 else 'щоденно' - days = parse_days(days_text) + days = parse_days(parts[1] if len(parts) > 1 else 'щоденно') route = routes[idx] if idx < len(routes) else '' entries.append({ 'train_number': num, @@ -90,15 +90,13 @@ def fetch_schedule(use_local: bool = False) -> List[Dict]: }) # Рядки з часами руху - time_rows = [r for r in rows if r.find('td', class_='q0') or r.find('td', class_='q1')] - - # Збирання часу для кожного поїзда та станції + time_rows = [r for r in trs if r.find('td', class_='q0') or r.find('td', class_='q1')] for idx, entry in enumerate(entries): base = idx * 3 for si, row in enumerate(time_rows): cells = row.find_all('td') - arrival = cells[base + 1].get_text(strip=True) if base + 1 < len(cells) else '' - departure = cells[base + 2].get_text(strip=True) if base + 2 < len(cells) else '' + arrival = cells[base+1].get_text(strip=True) if base+1 < len(cells) else '' + departure = cells[base+2].get_text(strip=True) if base+2 < len(cells) else '' entry['times'].append({ 'station': stations[si], 'arrival': arrival, diff --git a/scheduler.py b/scheduler.py index 1e525b3..fb82fbd 100644 --- a/scheduler.py +++ b/scheduler.py @@ -3,31 +3,25 @@ from parser import fetch_schedule from db import init_db, save_schedule from datetime import date - def daily_job(): - # Ініціалізуємо базу даних (створюємо таблиці, якщо їх нема) + # Створюємо таблиці, якщо їх нема init_db() - # Отримуємо повний розклад з сайту (Kyiv→Nizhyn) - entries = fetch_schedule(use_local=False) - - # Діагностика: виведемо кількість поїздів та список зупинок - print(f"DEBUG: знайдено поїздів: {len(entries)}") - for e in entries: - print(f"DEBUG: Поїзд {e['train_number']} днів {e['days']}, станцій: {len(e['times'])}") - - # Зберігаємо розклад у БД (додає поїзди, станції та часи) - save_schedule('kyiv_nizhyn', entries) - print(f"{date.today().isoformat()}: збережено розклад для {len(entries)} поїздів.") + # Парсимо обидва напрямки: tab=1 (Київ→Ніжин), tab=2 (Ніжин→Київ) + for tab in (1, 2): + entries = fetch_schedule(tab=tab, use_local=False) + print(f"DEBUG: Напрямок tab={tab}, знайдено {len(entries)} поїздів") + save_schedule(entries) + print(f"DEBUG: Збережено {len(entries)} поїздів для tab={tab}") + print(f"{date.today().isoformat()}: Розклад обох напрямків оновлено.") if __name__ == '__main__': - # Налаштування планувальника: щодня о 05:00 (Europe/Kyiv) sched = BlockingScheduler(timezone='Europe/Kyiv') sched.add_job(daily_job, 'cron', hour=5, minute=0) - # Одноразове оновлення та діагностика при старті + # Одноразове оновлення при старті daily_job() - # Запуск планувальника + # Старт планувальника sched.start()