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()
|