disinfolib 1.0.0

Streaming library for disinformation detection in multilingual text streams

disinfolib

Потокова Python-бібліотека для виявлення й моніторингу дезінформації у багатомовних текстових потоках.

from disinfolib import TextAnalyzer

analyzer = TextAnalyzer()
result = analyzer.analyze("Вороги знищать усіх! Небезпека для кожного!")

print(result.score)              # 0.72
print(result.is_disinformation)  # True
print(result.technique_names)    # ["appeal_to_fear", "loaded_language"]

---

Зміст

1. Навіщо бібліотека, а не сервер
2. Як встановити
3. Quickstart
4. Аналіз одного тексту
5. Потоковий моніторинг
6. Джерела даних
7. Збереження результатів
8. Моніторинг та алерти
9. Prometheus-метрики
10. Конфігурація
11. Підтримувані мови та техніки
12. Власні детектори та джерела
13. Системні вимоги
14. Приклади

---

Навіщо бібліотека, а не сервер

Більшість інструментів аналізу тексту — це окремі HTTP-сервіси. Вони вимагають розгортання, підтримки, мережі між вашим кодом і сервісом. disinfolib — це бібліотека: ви просто викликаєте функцію Python.

Аспект	Окремий сервер	disinfolib (бібліотека)
Запуск	python server.py + curl	import disinfolib
Інтеграція	HTTP-запити з обробкою помилок	Виклик методу Python
Затримка	Мережа + HTTP overhead	Пряма функція (~мс)
Розгортання	Docker/Kubernetes	pip install disinfolib
Масштабування	Окремий сервер	У вашому процесі
Залежності	Запущений сервер	Тільки pip-пакети

Якщо вам все ж потрібен HTTP API — examples/rest_api_server.py показує як обернути бібліотеку у FastAPI за ~100 рядків коду.

---

Як встановити

Крок 1. Переконайтеся що Python встановлено

Відкрийте термінал (cmd або PowerShell на Windows, Terminal на macOS/Linux) і виконайте:

python --version

Має вивести Python 3.11.x або вище. Якщо Python не встановлено — завантажте з python.org.

Крок 2. Створіть віртуальне середовище (рекомендовано)

Що таке віртуальне середовище? Це ізольована папка де встановлюються пакети для вашого проекту — вони не конфліктують з системними пакетами.

# Створити середовище (виконати один раз)
python -m venv venv

# Активувати (Windows)
venv\Scripts\activate

# Активувати (macOS/Linux)
source venv/bin/activate

Після активації ви побачите (venv) перед рядком введення — це означає що середовище активне.

Крок 3. Встановіть бібліотеку

З публічного PyPI:

# Базовий — rule-based детектор, без ML
pip install disinfolib

# З RSS-підтримкою
pip install "disinfolib[rss]"

# З ML-детектором (завантажує PyTorch + mDeBERTa ~2-3 ГБ)
pip install "disinfolib[ml]"

# З Telegram
pip install "disinfolib[telegram]"

# Все разом
pip install "disinfolib[all]"

З приватного репозиторію GitHub:

Якщо бібліотека розповсюджується через приватний GitHub-репозиторій, pip встановлює її безпосередньо з Git. Для доступу потрібен персональний токен (PAT).

# Базове
pip install "git+https://ВАШ_ТОКЕН@github.com/SKPdeveloper/disinfolib.git"

# З ML-детектором
pip install "disinfolib[ml] @ git+https://ВАШ_ТОКЕН@github.com/SKPdeveloper/disinfolib.git"

# Конкретна версія
pip install "git+https://ВАШ_ТОКЕН@github.com/SKPdeveloper/disinfolib.git@v1.0.0"

Щоб не зберігати токен у коді — використовуйте змінну середовища:

# Зберегти токен (один раз)
export GITHUB_TOKEN=ВАШ_ТОКЕН          # macOS/Linux
$env:GITHUB_TOKEN = "ВАШ_ТОКЕН"        # Windows PowerShell

# Встановити
pip install "git+https://$[e-mail]/SKPdeveloper/disinfolib.git"

Перевірка встановлення

python -c "import disinfolib; print(disinfolib.__version__)"

Має вивести: 1.0.0

Перевірте які детектори доступні:

python -c "from disinfolib import TextAnalyzer; a = TextAnalyzer(); print(a.detector_names)"

Якщо встановлено тільки базову версію — виведе ['rule_based'].
Якщо встановлено з [ml] — виведе ['rule_based', 'ml'].

---

Quickstart

Мінімальний приклад — скопіюйте в будь-який .py файл:

from disinfolib import TextAnalyzer

# Створюємо аналізатор
analyzer = TextAnalyzer()

# Аналізуємо текст
result = analyzer.analyze("Вороги знищать усіх! Небезпека для кожного!")

# Результат
print(f"Score: {result.score:.2f}")
print(f"Дезінформація: {result.is_disinformation}")
print(f"Мова: {result.language}")
print(f"Техніки: {result.technique_names}")

Запустіть:

python my_script.py

Очікуваний вивід:

Score: 0.72
Дезінформація: True
Мова: uk
Техніки: ['appeal_to_fear', 'loaded_language']

---

Аналіз одного тексту

Базовий аналіз

from disinfolib import TextAnalyzer

analyzer = TextAnalyzer()
result = analyzer.analyze("Текст для аналізу")

Що повертає result

Поле	Тип	Опис	Приклад
result.score	float	Оцінка від 0.0 до 1.0	0.72
result.is_disinformation	bool	True якщо score >= threshold	True
result.language	str	Виявлена мова (ISO 639-1)	"uk"
result.techniques	List	Техніки пропаганди	[PropagandaTechnique.APPEAL_TO_FEAR]
result.technique_names	List[str]	Назви технік	["appeal_to_fear"]
result.found_markers	List[str]	Знайдені маркери	["небезпека", "загроза"]
result.confidence	float	Впевненість (0.0–1.0)	0.68
result.detector_scores	Dict	Score кожного детектора	{"rule_based": 0.85}

Вказати мову явно

# Якщо мова відома заздалегідь — зазначте її для швидшого і точнішого аналізу
result = analyzer.analyze(text, language="uk")

Аналіз кількох текстів

texts = [
    "Перший текст для аналізу",
    "Другий текст — зовсім інший",
    "Вороги знищать усіх!",
]

results = analyzer.analyze_batch(texts)
for text, result in zip(texts, results):
    print(f"[{result.score:.2f}] {text[:40]}")

Порогове значення

За замовчуванням threshold = 0.35. Тексти з score >= 0.35 вважаються дезінформацією.

from disinfolib import TextAnalyzer, DisinfoConfig

# Підвищений поріг — менше спрацьовувань
config = DisinfoConfig(threshold=0.50)
analyzer = TextAnalyzer(config=config)

Коли підвищувати поріг? Якщо отримуєте забагато хибних спрацьовувань (нейтральні тексти класифікуються як дезінформація).

Коли знижувати поріг? Якщо хочете виявляти навіть слабко маніпулятивні тексти.

---

Потоковий моніторинг

DisinfoBroker — основний клас для роботи з потоками даних. Ви підключаєте джерела, і він безперервно аналізує всі вхідні повідомлення.

Синхронний потік

from disinfolib import DisinfoBroker

broker = DisinfoBroker()
broker.add_source("rss", urls=[
    "https://www.pravda.com.ua/rss/",
    "https://rss.unian.net/site/news_ukr.rss",
])

# Ітеруємо по результатам аналізу
for event in broker.stream():
    if event.is_disinformation:
        print(f"[{event.score:.2f}] {event.title}")
        print(f"  Джерело: {event.source_name}")
        print(f"  URL: {event.url}")

broker.stream() — це Python-генератор. Він блокує виконання і повертає один результат за раз.

Підписки на події

Замість того щоб перебирати результати у циклі, можна підписатися на конкретні події:

broker = DisinfoBroker()
broker.add_source("rss", urls=["https://example.com/rss"])

@broker.on("detection")   # Тільки коли виявлено дезінформацію
def on_detect(event):
    send_telegram_alert(event.title, event.score)

@broker.on("result")      # Кожне повідомлення (незалежно від результату)
def on_all(event):
    log_to_file(event)

@broker.on("error")       # Помилки (неробочий URL, тощо)
def on_error(exc):
    print(f"Помилка: {exc}")

broker.start()    # Запускає обробку у фоновому потоці
broker.wait()     # Чекає до завершення (або Ctrl+C)

Асинхронний потік (для FastAPI, aiohttp тощо)

import asyncio
from disinfolib import DisinfoBroker

async def monitor():
    broker = DisinfoBroker()
    broker.add_source("rss", urls=["https://example.com/rss"])

    async for event in broker.astream():
        if event.is_disinformation:
            await send_notification(event)

asyncio.run(monitor())

---

Джерела даних

RSS / Atom стрічки

from disinfolib import RSSSource

# Одноразовий прохід (one_shot=True)
source = RSSSource(
    urls=["https://example.com/rss"],
    one_shot=True,        # Прочитати і зупинитися
    language_hint="uk",  # Підказка мови, якщо всі статті однією мовою
)

# Постійний моніторинг (polling кожні 5 хвилин)
source = RSSSource(
    urls=["https://example.com/rss"],
    one_shot=False,
    poll_interval=300,  # секунди
)

broker.add_source(source)

# Або коротко:
broker.add_source("rss", urls=["https://example.com/rss"])

Що вміє RSSSource:

• Автоматично прибирає HTML-теги з контенту
• Дедублікує записи (SHA-256 хеш від url+title+content)
• Підтримує RSS 2.0 та Atom

Файлові джерела (для тестування та replay)

from disinfolib import FileSource

# JSONL файл (один JSON-об'єкт на рядок)
broker.add_source(FileSource("data/messages.jsonl"))

# CSV файл
broker.add_source(FileSource("data/news.csv"))

# Зі списку записів у пам'яті
records = [
    {"id": "1", "text": "Перший текст", "language": "uk"},
    {"id": "2", "text": "Другий текст", "language": "uk"},
]
broker.add_source(FileSource.from_records(records, source_id="my_data"))

# Нескінченний repeat (для тестів навантаження)
broker.add_source(FileSource.from_records(records, repeat=True))

Підтримувані формати: .jsonl, .json, .csv, .txt, .zip (архів із підтримуваними файлами)

Telegram

# Потрібно: pip install "disinfolib[telegram]"
from disinfolib import TelegramSource

source = TelegramSource(
    api_id=12345678,           # З my.telegram.org
    api_hash="abc123...",      # З my.telegram.org
    channels=["@channel1", "@channel2"],
    mode="realtime",           # або "history" для читання архіву
)
broker.add_source(source)

Apache Kafka

# Потрібно: pip install "disinfolib[kafka]"
from disinfolib import KafkaSource

source = KafkaSource(
    bootstrap_servers=["localhost:9092"],
    topics=["news.raw", "social.posts"],
    group_id="disinfolib",
)
broker.add_source(source)

Кілька джерел одночасно

broker = DisinfoBroker()
broker.add_source("rss", urls=["https://pravda.com.ua/rss/"])
broker.add_source("rss", urls=["https://unian.net/rss/"])
broker.add_source(FileSource("historical_data.jsonl"))

# Всі джерела працюють паралельно
for event in broker.stream():
    print(f"[{event.source_name}] {event.title}")

---

Збереження результатів

SQLite (рекомендовано)

from disinfolib import DisinfoBroker, SQLiteStorage

broker = DisinfoBroker()
broker.add_source("rss", urls=["https://example.com/rss"])
broker.add_storage(SQLiteStorage("results.db"))

for event in broker.stream():
    pass  # Результати автоматично записуються в БД

Чому SQLite, а не PostgreSQL? SQLite — це файл, не потрібен окремий сервер. Для більшості задач аналізу новин достатньо. Для великих навантажень (>100 записів/сек) — реалізуйте власний BaseStorage з PostgreSQL.

Що зберігається: message_id, collected_at, source_id, source_name, title, url, score, is_disinformation, techniques (JSON), found_markers (JSON), language, text (перші 500 символів).

# Запити до БД
storage = SQLiteStorage("results.db")

# Кількість записів
print(storage.count())

# Останні 20 результатів
rows = storage.query(limit=20)

# Тільки дезінформація
rows = storage.query(only_disinformation=True)

# Статистика
stats = storage.stats()
print(stats["disinformation_rate"])  # 0.142
print(stats["by_source"])            # {"rss_pravda": {...}}

CSV

from disinfolib import CSVStorage

# Файл відкривається у режимі append — нові рядки додаються
storage = CSVStorage("results.csv")
broker.add_storage(storage)

Файл зберігається в кодуванні UTF-8 з BOM — Excel відкриває правильно без проблем з кирилицею.

У пам'яті (для тестів)

from disinfolib import InMemoryStorage

storage = InMemoryStorage(max_size=1000)  # Зберігає останні 1000 записів
broker.add_storage(storage)

# Після обробки
print(storage.count())
print(storage.disinformation_rate)
results = storage.results  # List[StreamResult]

---

Моніторинг та алерти

Статистика потоку

from disinfolib import StatsTracker

tracker = StatsTracker()
broker.add_monitor(tracker)

# Після обробки кількох повідомлень:
stats = tracker.get_stats()
print(f"Оброблено: {stats.total_processed}")
print(f"Виявлено дезінформації: {stats.disinformation_count}")
print(f"Частка: {stats.disinformation_rate:.1%}")
print(f"Середній score: {stats.avg_score:.2f}")

# По джерелах
for source, data in stats.by_source.items():
    print(f"  {source}: {data['total']} повідомлень, rate={data['rate']:.2%}")

# По техніках
for technique, count in sorted(stats.by_technique.items(), key=lambda x: -x[1]):
    print(f"  {technique}: {count}")

Алерти при перевищенні порогів

from disinfolib import AlertManager, AlertRule

manager = AlertManager(check_every=10)  # Перевіряти кожні 10 повідомлень

# Правило: більше 30% дезінформації
manager.add_rule(AlertRule(
    name="high_rate",
    condition=lambda stats: stats.disinformation_rate > 0.30,
    handler=lambda alert: print(f"УВАГА: {alert.message}"),
    severity="warning",
    cooldown=600.0,              # Не спрацьовувати частіше ніж раз на 10 хв
    message="Частка дезінформації перевищила 30%",
))

# Правило з динамічним повідомленням
manager.add_rule(AlertRule(
    name="critical_rate",
    condition=lambda stats: stats.disinformation_rate > 0.60,
    handler=lambda alert: send_telegram(f"Критично: {alert.message}"),
    severity="critical",
    cooldown=300.0,
    message=lambda stats: f"Критична частка: {stats.disinformation_rate:.1%} (з {stats.total_processed} повідомлень)",
))

broker.add_monitor(manager)

# Після роботи
for alert in manager.history:
    print(f"[{alert.severity}] {alert.rule_name}: {alert.message}")

---

Prometheus-метрики

from disinfolib import DisinfoMetrics

metrics = DisinfoMetrics()
broker.add_monitor(metrics)

# Отримати текст у форматі Prometheus:
print(metrics.to_prometheus_text())

Приклад виводу:

# HELP disinfolib_messages_processed_total Total messages processed by disinfolib
# TYPE disinfolib_messages_processed_total counter
disinfolib_messages_processed_total 1500.0
# HELP disinfolib_disinformation_detected_total Total messages classified as disinformation
# TYPE disinfolib_disinformation_detected_total counter
disinfolib_disinformation_detected_total 213.0
# TYPE disinfolib_disinformation_rate gauge
disinfolib_disinformation_rate 0.142
disinfolib_messages_by_source_total{source="rss_pravda"} 800
disinfolib_messages_by_source_total{source="rss_unian"} 700

Інтеграція з FastAPI

from fastapi import FastAPI, Response
from disinfolib import DisinfoMetrics

metrics = DisinfoMetrics()
app = FastAPI()

@app.get("/metrics")
def get_metrics():
    return Response(
        content=metrics.to_prometheus_text(),
        media_type="text/plain; version=0.0.4",
    )

---

Конфігурація

Через код

from disinfolib import DisinfoConfig, TextAnalyzer

config = DisinfoConfig(
    threshold=0.35,                        # Поріг класифікації (за замовч. 0.35)
    languages=["uk", "ru", "en"],          # Підтримувані мови
    auto_detect_language=True,             # Автодетекція через langdetect
    detectors=["rule_based", "ml", "ai"], # За замовч. всі три активні
    stream_buffer_size=100,                # Розмір черги між джерелами та аналізатором
    log_level="WARNING",                   # "DEBUG", "INFO", "WARNING", "ERROR"
)

analyzer = TextAnalyzer(config=config)

Детектори: як вони працюють разом

За замовчуванням увімкнені всі три детектори. Кожен аналізує текст незалежно, результати об'єднуються зваженим усередненням.

Детектор	Вага	Що робить	Що потрібно
rule_based	1.0	Шукає маркери за YAML-словниками (uk/ru/en), 23 техніки	Нічого
ml	1.3	Zero-shot класифікація через mDeBERTa-v3 — розуміє контекст	pip install disinfolib[ml]
ai	1.5	Аналіз через Gemini/OpenAI — найточніший	ai_api_key у конфігурації

Детектор що повернув 0.0 — абстентується: він не входить у знаменник і не знижує загальний score. Це означає що відсутність ML або AI не штрафує результат — вони просто не беруть участі у підрахунку.

Якщо ml не встановлено або ai_api_key не вказано — відповідний детектор тихо пропускається без помилок і warnings.

Через YAML-файл

Створіть файл disinfolib.yaml:

threshold: 0.40
languages:
  - uk
  - en
auto_detect_language: true
detectors:
  - rule_based
  - ml
  - ai
stream_buffer_size: 200
log_level: WARNING

Завантажте:

config = DisinfoConfig.from_yaml("disinfolib.yaml")

Через змінні середовища

Всі параметри можна задати через змінні середовища з префіксом DISINFOLIB_:

export DISINFOLIB_THRESHOLD=0.40
export DISINFOLIB_LOG_LEVEL=DEBUG
export DISINFOLIB_AI_API_KEY=your_key_here

config = DisinfoConfig.from_env()

---

Підтримувані мови та техніки

Мови

Код	Мова	Маркерів	Детектор
uk	Українська	330+	Rule-based + ML
ru	Російська	330+	Rule-based + ML
en	Англійська	150+	Rule-based + ML
інші	—	—	ML (mDeBERTa, multilingual)

Техніки пропаганди (23 шт.)

Код	Назва	Приклад маркеру
appeal_to_fear	Апеляція до страху	"небезпека для кожного"
loaded_language	Емоційно забарвлена мова	"жахливий", "кошмарний"
name_calling	Наклеювання ярликів	образливі характеристики
dehumanization	Дегуманізація	відмова в людяності
exaggeration	Перебільшення	"знищать усіх"
whataboutism	Вотабаутизм	"а що щодо..."
black_white	Чорно-біле мислення	"або з нами, або проти нас"
bandwagon	Апеляція до більшості	"всі знають що..."
appeal_to_authority	Апеляція до авторитету	—
false_causality	Хибна причинність	—
glittering_generalities	Блискучі узагальнення	"свобода", "справедливість" (без контексту)
absolute_judgment	Абсолютні судження	"завжди", "ніколи", "всі"
slogans	Гасла	короткі повторювані фрази
flag_waving	Апеляція до патріотизму	—
doubt_seeding	Сіяння сумнівів	"кажуть що...", "деякі вважають..."
repetition	Повторення	—
state_propaganda	Держпропаганда	офіційна риторика
manipulative_pattern	Маніпулятивний шаблон	—
emotional_punctuation	Емоційна пунктуація	!!!, ...
caps_abuse	Зловживання капслоком	УВАГА ВСІМ
cult_of_personality	Культ особистості	—
defeat_euphemisms	Евфемізми поразки	"відступ на заздалегідь підготовлені позиції"
silencing_dissent	Замовчування інакодумства	—

Детальний опис кожної техніки: docs/techniques.md

---

Власні детектори та джерела

Власний детектор

from disinfolib import BaseDetector
from disinfolib.analysis.detectors.base import DetectorResult

class MyDetector(BaseDetector):
    name = "my_detector"
    weight = 0.3   # Вага в ensemble

    def detect(self, text: str, language: str) -> DetectorResult:
        score = 0.8 if "небезпека" in text.lower() else 0.0
        return DetectorResult(score=score, techniques=[], found_markers=[])

analyzer = TextAnalyzer()
analyzer.add_detector(MyDetector())

Власне джерело

from disinfolib.sources.base import BaseSource
from disinfolib.models.message import RawMessage
from typing import Iterator

class MySource(BaseSource):
    source_type = "custom"

    def __iter__(self) -> Iterator[RawMessage]:
        for item in fetch_my_data():
            yield self._make_message(text=item["content"], title=item["title"])

broker.add_source(MySource(source_id="my_api"))

Повна документація: docs/extending.md

---

Системні вимоги

Компонент	Вимога
Python	3.11 або вище
ОС	Windows, macOS, Linux
RAM (базовий)	~50 МБ
RAM (з ML)	~2 ГБ (модель mDeBERTa)
Диск (базовий)	~10 МБ
Диск (з ML)	~3 ГБ (PyTorch + модель)

Залежності за варіантом встановлення

Варіант	Додаткові залежності	Коли використовувати
pip install disinfolib	—	Rule-based аналіз, мінімальні ресурси
disinfolib[rss]	feedparser, httpx	Моніторинг RSS/Atom стрічок
disinfolib[ml]	transformers, torch (~3 ГБ)	Точніший аналіз через ML
disinfolib[telegram]	telethon	Моніторинг Telegram-каналів
disinfolib[kafka]	kafka-python	Отримання даних з Apache Kafka
disinfolib[ai]	httpx	Аналіз через Gemini/OpenRouter API
disinfolib[all]	всі вище	Максимальний функціонал

---

Приклади

Файл	Що демонструє
examples/basic_analysis.py	Аналіз одного тексту, batch аналіз
examples/stream_monitoring.py	Потоковий моніторинг з кількох джерел
examples/ml_analysis.py	ML та AI детектори
examples/rest_api_server.py	FastAPI сервер поверх disinfolib

Запуск прикладу:

# Переконайтеся що venv активоване
python examples/basic_analysis.py

---

Оригінальність та академічна база

disinfolib — це оригінальна розробка, написана з нуля. Бібліотека не є форком, копією або похідним твором жодного існуючого відкритого проекту.

Академічні джерела технік

Класифікація пропагандистських технік у бібліотеці ґрунтується на загальновідомих академічних та аналітичних框架:

Джерело	Рік	Що використано
Institute for Propaganda Analysis (IPA)	1937	Базова типологія 7 технік: name-calling, glittering generalities, transfer, testimonial, plain folks, card stacking, bandwagon
Jacques Ellul, "Propagandes"	1962	Класифікація інтеграційної та агітаційної пропаганди, концепція соціологічної пропаганди
NATO Strategic Communications Centre of Excellence	2014–2024	Сучасна таксономія інформаційних операцій, визначення деhumanzation та whataboutism
EU DisinfoLab / EUvsDisinfo	2015–2024	Операційні визначення дезінформації, відмінність від пропаганди
Renée DiResta, "Computational Propaganda"	2018	Автоматизоване поширення наративів

Ці джерела є публічними академічними та аналітичними матеріалами. Бібліотека реалізує власну інтерпретацію цих концепцій у вигляді програмного коду — мовні патерни, алгоритми, архітектура та реалізація є оригінальними.

Використані відкриті компоненти

Бібліотека використовує сторонні пакети згідно з їхніми ліцензіями:

Компонент	Ліцензія	Використання
PyTorch	BSD-3	Основа для ML-детектора
Hugging Face Transformers	Apache 2.0	Zero-shot класифікація
mDeBERTa-v3-base-mnli-xnli (Moritz Laurer)	MIT	NLI модель для детекції технік
Pydantic	MIT	Конфігурація та моделі даних
feedparser	MIT	Парсинг RSS
langdetect	Apache 2.0	Визначення мови тексту
trafilatura	Apache 2.0	Витяг тексту з веб-сторінок

Жоден з цих компонентів не є частиною кодової бази disinfolib — вони є залежностями, що встановлюються окремо.

---

Ліцензія

MIT License — вільне використання, модифікація та розповсюдження з обов'язковим збереженням тексту ліцензії.

Дивіться файл LICENSE.