Урок 7: Автоматизація збору — API / скрапінг / пайплайни та гігієна даних

Навігація курсу: Хаб: Уроки

Попередній: Урок-06-Теорія-графів | Наступний: Урок-08-Воркшоп

Слайди (Презентаційний текст)

Слайд 1 — Заголовок

Автоматизація збору
API · Скрапінг · Пайплайни · Гігієна даних

Слайд 2 — Навіщо автоматизація?

Без автоматизації: аналітик вручну переглядає 50 джерел → 3-4 години
З автоматизацією: скрипт перевіряє 50 джерел → 5 хвилин

Але автоматизація без гігієни = збір сміття:

Дублікати, хибні спрацювання, застарілі дані
Юридичні ризики якщо ігнорувати rate limits і умови сервісів
Баги = неправильні рішення

Ціль: якісний, відтворюваний, аудитований збір даних

Слайд 3 — Три способи збору

Метод	Коли використовувати	Обмеження
API	Офіційний доступ до структурованих даних	Вимагає ключ, ліміти запитів
Скрапінг	Немає API, дані на сайті	Може порушувати Terms of Service
RSS/Feeds	Моніторинг оновлень	Тільки те що публікується у feed

Слайд 4 — API: найкращий варіант

Переваги API:

Структуровані дані (JSON/XML) — не потрібно парсити HTML
Офіційно дозволений доступ
Стабільний інтерфейс

Типові OSINT API (2025-2026):

Perplexity API    → AI-пошук з цитатами
Exa AI API        → семантичний пошук
Tavily API        → пошук для агентів
Apify API         → 55+ скраперів соцмереж
Jina AI API       → читання URL → Markdown
BreachDirectory   → пошук за email у витоках
Shodan API        → пошук підключених пристроїв

Слайд 5 — Python для OSINT: ключові бібліотеки

# HTTP-запити
import requests
 
# Парсинг HTML
from bs4 import BeautifulSoup
 
# Обробка даних
import pandas as pd
 
# Rate limiting (обов'язково!)
import time
from ratelimit import limits, sleep_and_retry
 
# Асинхронний збір
import asyncio
import aiohttp

Повний список: github.com/topics/osint-python

Слайд 6 — Базовий пайплайн збору

# Простий OSINT-пайплайн
class OSINTPipeline:
    def collect(self, query):
        """Фаза 1: Збір з джерел"""
        results = []
        results += self.search_web(query)    # Brave/Google
        results += self.search_social(query) # соцмережі
        return results
    
    def deduplicate(self, results):
        """Фаза 2: Дедуплікація"""
        seen = set()
        return [r for r in results if r['url'] not in seen
                and not seen.add(r['url'])]
    
    def enrich(self, results):
        """Фаза 3: Збагачення"""
        for r in results:
            r['summary'] = self.ai_summarize(r['content'])
        return results
    
    def store(self, results):
        """Фаза 4: Зберігання"""
        pd.DataFrame(results).to_csv('osint_results.csv')

Слайд 7 — Rate Limiting: критично важливо

Чому: API і сайти блокують агресивні запити

from ratelimit import limits, sleep_and_retry
import time
 
@sleep_and_retry
@limits(calls=10, period=60)  # 10 запитів за хвилину
def api_call(url):
    return requests.get(url)
 
# Або простіше:
def safe_request(url):
    time.sleep(1)  # 1 секунда між запитами
    return requests.get(url)

Правила:

Brave Search: 1с між запитами
Jina AI: 2с між запитами
При 429 (Too Many Requests): чекай 5с, потім 15с, потім зупинись

Слайд 8 — Гігієна даних

5 принципів чистих даних:

Нормалізація — однаковий формат (UTF-8, ISO 8601 для дат)
Дедуплікація — одне джерело, одна версія
Атрибуція — кожна запис зберігає URL і дату збору
Версіювання — зберігати оригінальний HTML, не тільки витяжку
Аудит — лог що, коли і звідки зібрано

# Мінімальна структура запису
record = {
    "content": "...",
    "source_url": "https://...",
    "collected_at": "2026-04-14T10:30:00Z",
    "collector": "brave_search",
    "query": "Іван Петренко TechUA"
}

Слайд 9 — Апify: 55+ скраперів

Apify — хмарна платформа акторів-скраперів:

Платформа	Apify актор
LinkedIn	apify/linkedin-scraper
Instagram профіль	apify/instagram-scraper
Instagram теги	apify/instagram-tagged-scraper
Facebook	apify/facebook-pages-scraper
TikTok	clockworks/tiktok-profile-scraper
YouTube	streamers/youtube-channel-scraper
Google Maps	compass/crawler-google-places
Контакти	vdrmota/contact-info-scraper

# Запуск Apify актора
bash scripts/run-actor.sh "apify/instagram-scraper" \
  '{"usernames": ["target_handle"]}'

Слайд 10 — Bright Data MCP (2025-2026)

Bright Data — вирішує найскладніші проблеми:

CAPTCHA обхід
JS-рендеринг
Authwall (Facebook, LinkedIn за логіном)
Геообмежений контент
Yandex пошук (важливо для СНД)

# Через MCP-протокол
bash scripts/brightdata.sh scrape "https://facebook.com/profile"
bash scripts/brightdata.sh search-yandex "Іван Петренко"

Ціна: від $10/міс, але відкриває доступ до інакше недоступних даних

Слайд 11 — Зберігання і формати

Формат	Коли використовувати
JSON	Структуровані дані, API-відповіді
CSV	Табличні дані, сумісність з Excel
SQLite	Локальна реляційна база для малих проектів
Neo4j	Граф-база для зв’язаних даних
Markdown	Нотатки, звіти, Obsidian
JSONL	Потокова обробка великих наборів

Слайд 12 — Etика і легальність збору

Що зазвичай дозволено:

Читання публічно доступних сторінок (як браузер)
Використання офіційних API згідно з Terms of Service
Архівні дані (Wayback Machine)

Сірі зони:

Скрапінг сайтів де Terms of Service забороняє автоматизований доступ
Збір даних за логіном (навіть якщо акаунт публічний)

Явно заборонено:

Bypass авторизації
Завантаження захищених авторським правом матеріалів
Обробка персональних даних EU-осіб без законних підстав (GDPR)

Нотатки доповідача

До слайду 2 — Навіщо автоматизація

Покажіть “до і після”: вручну перевірити 20 джерел → 2 години. Той самий скрипт → 3 хвилини. Але підкресліть: автоматизація лише прискорює збір, аналіз і судження залишаються за людиною.

До слайду 5 — Python

Не потрібно бути розробником. Достатньо вміти запустити готовий скрипт, змінити параметри і прочитати результат. Покажіть Jupyter Notebook як найзручніше середовище для OSINT-аналітиків.

До слайду 7 — Rate Limiting

Підкресліть: ігнорування rate limits = блокування вашого IP. Після блокування — вся автоматизація зупиняється. Правильний підхід: краще повільно і стабільно, ніж швидко і заблоковано.

До слайду 12 — Etика

Це важливий слайд. GDPR у Європі, CCPA в Каліфорнії — реальні закони з реальними штрафами. Правило: якщо збираєш дані про громадян ЄС для комерційних цілей — потрібен юрист. Для власних досліджень і журналістики — більше свободи, але не необмежена.

Домашнє завдання

Завдання 1 — Перший скрипт (обов’язкове)

Встановіть Python і бібліотеки: pip install requests beautifulsoup4 pandas

Напишіть скрипт який:

Робить пошук через Jina AI API (безкоштовний ключ)
Зберігає результати у CSV з полями: title, url, snippet, collected_at
Виводить кількість унікальних доменів у результатах

# Стартовий шаблон
import requests, csv, datetime
 
JINA_KEY = "ваш_ключ"
query = "OSINT 2026 tools"
url = f"https://s.jina.ai/{query}"
headers = {"Authorization": f"Bearer {JINA_KEY}", "Accept": "application/json"}
response = requests.get(url, headers=headers)
# Далі ваш код...

Завдання 2 — Дедуплікація (обов’язкове)

Зберіть результати пошуку одного запиту з трьох джерел (Brave, Perplexity, Google).
Скопіюйте заголовки і URL у CSV.
Напишіть скрипт що видаляє дублікати за URL і доменом.
Порівняйте: скільки унікальних результатів дає кожне джерело?

Завдання 3 — RSS-моніторинг (бажане)

Налаштуйте RSS-читач (Feedly, Inoreader або локальний) з 5 джерелами у вашій темі.
Додайте Telegram-канал через RSS-конвертер (rss.app або RSS.hub).
Налаштуйте keyword alert на 3 ключових слова.

Матеріали для студентів

Ресурси

Python для OSINT:

Apify і BrightData:

Apify Console — керування акторами
Bright Data MCP

Стартовий шаблон пайплайну (Python)

"""
OSINT Data Collection Pipeline
Базовий шаблон для студентів курсу AI for OSINT 2026
"""
import requests
import json
import csv
import time
from datetime import datetime, timezone
 
class SimpleOSINTPipeline:
    def __init__(self, output_file="osint_results.csv"):
        self.output_file = output_file
        self.results = []
        
    def search_jina(self, query, api_key):
        """Пошук через Jina AI"""
        url = f"https://s.jina.ai/{query}"
        headers = {
            "Authorization": f"Bearer {api_key}",
            "Accept": "application/json"
        }
        time.sleep(2)  # Rate limiting: 2с між запитами Jina
        response = requests.get(url, headers=headers)
        if response.status_code == 200:
            return response.json().get('data', [])
        return []
    
    def add_result(self, content, source_url, query, collector):
        """Додати результат з атрибуцією"""
        self.results.append({
            "content": content,
            "source_url": source_url,
            "query": query,
            "collector": collector,
            "collected_at": datetime.now(timezone.utc).isoformat()
        })
    
    def deduplicate(self):
        """Видалити дублікати за URL"""
        seen_urls = set()
        unique = []
        for r in self.results:
            if r['source_url'] not in seen_urls:
                seen_urls.add(r['source_url'])
                unique.append(r)
        self.results = unique
        return len(self.results)
    
    def save(self):
        """Зберегти у CSV"""
        if not self.results:
            print("Немає результатів для збереження")
            return
        keys = self.results[0].keys()
        with open(self.output_file, 'w', newline='', encoding='utf-8') as f:
            writer = csv.DictWriter(f, fieldnames=keys)
            writer.writeheader()
            writer.writerows(self.results)
        print(f"Збережено {len(self.results)} записів у {self.output_file}")
 
# Використання:
# pipeline = SimpleOSINTPipeline()
# results = pipeline.search_jina("Іван Петренко TechUA", "ваш_jina_ключ")
# for r in results:
#     pipeline.add_result(r.get('content',''), r.get('url',''), query, 'jina')
# pipeline.deduplicate()
# pipeline.save()

SHATTERED: Ingest + Search + Export як центр автоматизованого pipeline

Слайд — SHATTERED як кінцева ланка автоматизованого збору

Pipeline збору → SHATTERED:

Зовнішній збір                  SHATTERED
─────────────────                ─────────────────────────────
snscrape, Apify, jina     →      /api/ingest/upload
                                 /api/ingest/ingest-path
                                 /api/ingest/upload/batch
                                         ↓
                                 Автоматична обробка:
                                 OCR → Parse → NER → Embed
                                         ↓
                                 Search / Entities / Summary
                                 Claims / Contradictions / ACH

Слайд — Batch інгест: автоматизація завантаження

Завантажити папку з файлами:

# Вся папка рекурсивно (PDF, DOCX, JPG, TXT, HTML)
curl -X POST http://localhost:8100/api/ingest/ingest-path \
  -H "Content-Type: application/json" \
  -d '{
    "path": "/data/osint_collection/",
    "project_id": "proj_abc",
    "recursive": true,
    "ocr_mode": "auto",
    "priority": "batch"
  }'

Batch upload кількох файлів одночасно:

curl -X POST http://localhost:8100/api/ingest/upload/batch \
  -F "files=@report1.pdf" \
  -F "files=@report2.pdf" \
  -F "files=@screenshot.jpg" \
  -F "project_id=proj_abc"

Дедупліація вбудована: SHATTERED перевіряє checksum — якщо файл вже є, не завантажує повторно.

Слайд — Python pipeline з SHATTERED API

import httpx
import glob
from pathlib import Path
 
BASE = "http://localhost:8100/api"
 
def ingest_folder_to_shattered(folder_path: str, project_name: str):
    """Завантажує папку документів у SHATTERED"""
    
    # 1. Створи проєкт
    proj = httpx.post(f"{BASE}/projects", 
                      json={"name": project_name}).json()
    project_id = proj["id"]
    print(f"Project: {project_id}")
    
    # 2. Batch інгест
    result = httpx.post(f"{BASE}/ingest/ingest-path",
        json={
            "path": str(Path(folder_path).resolve()),
            "project_id": project_id,
            "recursive": True,
            "ocr_mode": "auto"
        }, timeout=30.0).json()
    
    batch_id = result.get("batch_id")
    print(f"Batch: {batch_id}, Jobs queued: {result.get('job_count')}")
    
    # 3. Моніторинг прогресу
    import time
    while True:
        status = httpx.get(f"{BASE}/ingest/batch/{batch_id}").json()
        done = status["completed_count"]
        total = status["total_count"]
        print(f"Progress: {done}/{total}", end="\r")
        if done >= total:
            break
        time.sleep(3)
    
    print(f"\nIngest complete! {total} documents processed.")
    return project_id
 
# Запуск
pid = ingest_folder_to_shattered("/data/investigation/", "Розслідування 2025-03")

Слайд — SHATTERED Export: вивід результатів

Після аналізу — автоматичний export:

# JSON для подальшої обробки
curl -X POST http://localhost:8100/api/export \
  -d '{"project_id": "proj_abc", "format": "json",
       "include": ["entities", "claims", "contradictions"]}' \
  --output results.json
 
# CSV (entities)
curl "http://localhost:8100/api/entities/items?project_id=proj_abc" \
  -H "Accept: text/csv" --output entities.csv
 
# PDF звіт розслідування
curl -X POST http://localhost:8100/api/reports \
  -d '{"type": "investigation_summary", "project_id": "proj_abc",
       "format": "pdf"}' --output report.pdf

Packets — архів всього розслідування:

curl -X POST http://localhost:8100/api/packets \
  -d '{"project_id": "proj_abc", "include_documents": true,
       "version": "1.0", "notes": "Final package for review"}' \
  --output investigation_packet.zip

Нотатки до SHATTERED-блоку (Урок 7)

Ключовий меседж: SHATTERED — це центр куди стікаються дані з усіх зовнішніх джерел. Python pipeline → SHATTERED API → аналіз. Показати live: запусти ingest_folder_to_shattered() з реальними файлами і покажи як документи з’являються у SHATTERED UI в реальному часі.

← Урок-06-Теорія-графів | Наступний урок: Урок-08-Воркшоп →

AI для OSINT 2026

Провідник

Урок-07-Автоматизація-збору