Основы автоматизации: работа с файлами и API

Мы уже научились проектировать правила детекции и снижать количество ложных срабатываний. Однако данных становится так много, что ручная проверка каждого кейса через веб-интерфейс превращается в «бутылочное горлышко». Чтобы освободить время для анализа сложных схем, нужно делегировать рутину коду.

Зачем антифрод-исследователю Python

Автоматизация не заменяет аналитика, а расширяет его возможности. Ваша задача — научиться связывать базы данных, файлы и внешние сервисы в единый поток.

Типичный рабочий цикл состоит из трех шагов:

1. Извлечение: взять список ID из SQL-выгрузки или CSV.
2. Обогащение: проверить IP, email или ID устройства через внешние базы.
3. Сохранение: записать результат для загрузки в антифрод-систему.

Работа с файловой системой

Для глубокого анализа мы используем Pandas, но для быстрой обработки логов или конфигураций важно освоить базовое чтение файла и запись файла встроенными средствами Python. Это быстрее и потребляет меньше памяти.

Для работы с файлами используйте контекстный менеджер with. Он автоматически закрывает файл, даже если в коде произойдет ошибка.

# Чтение списка ID из текстового файла
with open('suspicious_ids.txt', 'r', encoding='utf-8') as file:
    # Убираем лишние пробелы и переносы строк
    fraud_list = [line.strip() for line in file.readlines()]

# Запись результатов в CSV
with open('report_2026.csv', 'w', encoding='utf-8') as output:
    output.write("user_id,status\n")
    for user_id in fraud_list:
        output.write(f"{user_id},verified\n")

Автоматизация через API

Автоматизация через API (Application Programming Interface) — это способ заставить скрипт общаться с другими сервисами напрямую, без браузера.

В антифроде API используют для:

• проверки репутации IP через черные списки;
• запроса данных из блокчейн-эксплореров;
• проверки данных в базах утечек.

Мы будем работать с библиотекой httpx. Она поддерживает современные стандарты HTTP и работает стабильнее старых аналогов.

Как показано в Схеме 1, скрипт становится посредником: он собирает данные из разных точек и объединяет их в отчет.

Формат JSON

Большинство API возвращают данные в формате JSON. В Python они превращаются в обычные словари. Важно извлекать только нужные поля, так как ответ может содержать десятки лишних параметров.

Пример ответа API:

{
  "ip": "195.133.x.x",
  "risk_score": 85,
  "details": {
    "is_proxy": true,
    "provider": "DigitalOcean"
  }
}

Здесь нас интересуют только risk_score и is_proxy.

Практикум: запрос к API

Объединим чтение файла и запрос к внешнему сервису.

import httpx
import time

# Читаем адреса
with open('ips.txt', 'r') as f:
    ips = [line.strip() for line in f]

results = []

# Создаем клиент для запросов
with httpx.Client() as client:
    for ip in ips:
        try:
            response = client.get(f"https://api.antifraud-service.ru/v1/check/{ip}")
            
            if response.status_code == 200:
                data = response.json()
                results.append(f"{ip}: {data['risk_score']}")
            
            # Пауза, чтобы сервис не заблокировал нас за спам (Rate Limiting)
            time.sleep(0.5) 
            
        except Exception as e:
            print(f"Ошибка на IP {ip}: {e}")

# Сохраняем результат
with open('check_results.txt', 'w') as f:
    f.write("\n".join(results))

Три критические ошибки автоматизации

1. Игнорирование ошибок: без блока try-except любая сетевая заминка прервет работу скрипта, и вы потеряете прогресс.
2. Превышение лимитов (Rate Limiting): слишком частые запросы приведут к блокировке вашего IP или ключа. Всегда используйте time.sleep().
3. Отсутствие логов: если скрипт обрабатывает 10 000 строк, вы должны видеть прогресс в консоли, иначе непонятно, работает он или завис.

Мы освоили базу: чтение файлов, работу с API и сохранение данных. В следующей теме мы перейдем к созданию полноценных конвейеров (пайплайнов) для одновременной обработки данных из множества источников.