Модуль 9: Робота з базами даних

import sqlite3 # Підключення (створює файл якщо не існує) conn = sqlite3.connect('osint.db') cursor = conn.cursor() # Створення таблиці cursor.execute(''' CREATE TABLE IF NOT EXISTS events ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp TEXT NOT NULL, event_type TEXT NOT NULL, sector TEXT, lat REAL, lon REAL, source TEXT, description TEXT, created_at TEXT DEFAULT CURRENT_TIMESTAMP ) ''') # Зберігаємо зміни conn.commit() print("Таблицю events створено")

# Один запис cursor.execute(''' INSERT INTO events (timestamp, event_type, sector, lat, lon, source) VALUES (?, ?, ?, ?, ?, ?) ''', ('2025-01-15 08:23', 'movement', 'A', 48.4567, 35.0234, 'radio')) # Багато записів events_data = [ ('2025-01-15 09:12', 'fire', 'B', 48.5123, 35.1456, 'radar'), ('2025-01-15 10:45', 'contact', 'A', 48.4700, 35.0400, 'visual'), ('2025-01-15 11:30', 'movement', 'C', 49.1234, 36.5678, 'radio'), ] cursor.executemany(''' INSERT INTO events (timestamp, event_type, sector, lat, lon, source) VALUES (?, ?, ?, ?, ?, ?) ''', events_data) conn.commit() print(f"Додано записів: {cursor.rowcount}")

# Усі записи cursor.execute('SELECT * FROM events') rows = cursor.fetchall() for row in rows: print(row) # З умовою cursor.execute('SELECT * FROM events WHERE sector = ?', ('A',)) sector_a = cursor.fetchall() # Конкретні колонки cursor.execute('SELECT timestamp, event_type, sector FROM events WHERE event_type = ?', ('fire',)) fires = cursor.fetchall() # Сортування та ліміт cursor.execute('SELECT * FROM events ORDER BY timestamp DESC LIMIT 5') recent = cursor.fetchall()

# Оновити один запис cursor.execute(''' UPDATE events SET description = ? WHERE id = ? ''', ('Підтверджено візуально', 1)) # Оновити по умові cursor.execute(''' UPDATE events SET source = 'confirmed' WHERE sector = ? AND event_type = ? ''', ('A', 'contact')) conn.commit()

# Видалити по id cursor.execute('DELETE FROM events WHERE id = ?', (5,)) # Видалити по умові cursor.execute('DELETE FROM events WHERE event_type = ? AND sector = ?', ('test', 'X')) conn.commit()

# Кількість по типах cursor.execute(''' SELECT event_type, COUNT(*) as count FROM events GROUP BY event_type ORDER BY count DESC ''') for row in cursor.fetchall(): print(f"{row[0]}: {row[1]}") # Статистика по секторах cursor.execute(''' SELECT sector, COUNT(*) as total, AVG(lat) as avg_lat, AVG(lon) as avg_lon FROM events GROUP BY sector ''')

# Пошук по частині тексту cursor.execute(''' SELECT * FROM events WHERE description LIKE ? ''', ('%підтвердж%',)) # % — будь-які символи # _ — один символ

# Події за період cursor.execute(''' SELECT * FROM events WHERE timestamp BETWEEN ? AND ? ORDER BY timestamp ''', ('2025-01-15 09:00', '2025-01-15 12:00')) # Події за останню годину (якщо timestamp у форматі ISO) cursor.execute(''' SELECT * FROM events WHERE timestamp > datetime('now', '-1 hour') ''')

# Створення індексів cursor.execute('CREATE INDEX IF NOT EXISTS idx_timestamp ON events(timestamp)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_sector ON events(sector)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_event_type ON events(event_type)') # Складений індекс (для запитів з кількома умовами) cursor.execute('CREATE INDEX IF NOT EXISTS idx_sector_type ON events(sector, event_type)') conn.commit()

import pandas as pd import sqlite3 conn = sqlite3.connect('osint.db') # Читання в DataFrame df = pd.read_sql_query('SELECT * FROM events', conn) print(df.head()) # Читання з параметрами df_sector_a = pd.read_sql_query( 'SELECT * FROM events WHERE sector = ?', conn, params=['A'] ) # Запис DataFrame у базу new_events = pd.DataFrame({ 'timestamp': ['2025-01-16 08:00', '2025-01-16 09:00'], 'event_type': ['movement', 'fire'], 'sector': ['B', 'C'], 'lat': [48.52, 49.13], 'lon': [35.15, 36.57], 'source': ['radio', 'radar'] }) # if_exists: 'fail', 'replace', 'append' new_events.to_sql('events', conn, if_exists='append', index=False) conn.commit()

# Таблиця позивних cursor.execute(''' CREATE TABLE IF NOT EXISTS callsigns ( id INTEGER PRIMARY KEY, callsign TEXT UNIQUE NOT NULL, unit_type TEXT, first_seen TEXT, last_seen TEXT, total_messages INTEGER DEFAULT 0 ) ''') # Таблиця повідомлень з посиланням на позивні cursor.execute(''' CREATE TABLE IF NOT EXISTS messages ( id INTEGER PRIMARY KEY AUTOINCREMENT, callsign_id INTEGER, timestamp TEXT, content TEXT, FOREIGN KEY (callsign_id) REFERENCES callsigns(id) ) ''') conn.commit()

# Повідомлення з інформацією про позивні cursor.execute(''' SELECT m.timestamp, c.callsign, c.unit_type, m.content FROM messages m JOIN callsigns c ON m.callsign_id = c.id WHERE c.unit_type = ? ORDER BY m.timestamp DESC ''', ('infantry',)) for row in cursor.fetchall(): print(f"{row[0]} | {row[1]} ({row[2]}): {row[3]}")

import sqlite3 import pandas as pd from datetime import datetime from pathlib import Path class OSINTDatabase: def __init__(self, db_path='osint.db'): self.db_path = db_path self.conn = sqlite3.connect(db_path) self.conn.row_factory = sqlite3.Row # Доступ по іменах колонок self._init_tables() def _init_tables(self): """Ініціалізація таблиць""" cursor = self.conn.cursor() cursor.execute(''' CREATE TABLE IF NOT EXISTS events ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp TEXT NOT NULL, event_type TEXT NOT NULL, sector TEXT, lat REAL, lon REAL, source TEXT, description TEXT, created_at TEXT DEFAULT CURRENT_TIMESTAMP ) ''') # Індекси cursor.execute('CREATE INDEX IF NOT EXISTS idx_ts ON events(timestamp)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_type ON events(event_type)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_sector ON events(sector)') self.conn.commit() def add_event(self, timestamp, event_type, sector=None, lat=None, lon=None, source=None, description=None): """Додає одну подію""" cursor = self.conn.cursor() cursor.execute(''' INSERT INTO events (timestamp, event_type, sector, lat, lon, source, description) VALUES (?, ?, ?, ?, ?, ?, ?) ''', (timestamp, event_type, sector, lat, lon, source, description)) self.conn.commit() return cursor.lastrowid def add_events_from_df(self, df): """Імпорт з DataFrame""" df.to_sql('events', self.conn, if_exists='append', index=False) self.conn.commit() return len(df) def get_events(self, event_type=None, sector=None, start_time=None, end_time=None, limit=100): """Отримує події з фільтрами""" query = 'SELECT * FROM events WHERE 1=1' params = [] if event_type: query += ' AND event_type = ?' params.append(event_type) if sector: query += ' AND sector = ?' params.append(sector) if start_time: query += ' AND timestamp >= ?' params.append(start_time) if end_time: query += ' AND timestamp <= ?' params.append(end_time) query += ' ORDER BY timestamp DESC LIMIT ?' params.append(limit) return pd.read_sql_query(query, self.conn, params=params) def get_stats(self): """Статистика по базі""" cursor = self.conn.cursor() stats = {} cursor.execute('SELECT COUNT(*) FROM events') stats['total'] = cursor.fetchone()[0] cursor.execute(''' SELECT event_type, COUNT(*) FROM events GROUP BY event_type ''') stats['by_type'] = dict(cursor.fetchall()) cursor.execute(''' SELECT sector, COUNT(*) FROM events GROUP BY sector ''') stats['by_sector'] = dict(cursor.fetchall()) return stats def search(self, text): """Пошук по опису""" return pd.read_sql_query( 'SELECT * FROM events WHERE description LIKE ?', self.conn, params=[f'%{text}%'] ) def close(self): self.conn.close() # --- ТЕСТ --- if __name__ == '__main__': db = OSINTDatabase('test_osint.db') # Додаємо події db.add_event('2025-01-15 08:23', 'movement', 'A', 48.4567, 35.0234, 'radio') db.add_event('2025-01-15 09:12', 'fire', 'B', 48.5123, 35.1456, 'radar') db.add_event('2025-01-15 10:45', 'contact', 'A', 48.4700, 35.0400, 'visual', 'Підтверджено спостерігачем') # Статистика print("=== СТАТИСТИКА ===") stats = db.get_stats() print(f"Усього: {stats['total']}") print(f"По типах: {stats['by_type']}") # Фільтрація print("\n=== СЕКТОР A ===") df = db.get_events(sector='A') print(df) # Пошук print("\n=== ПОШУК 'підтверд' ===") results = db.search('підтверд') print(results) db.close()