Как превратить микро-скрипт для автоматизации в коммерческий продукт: библиотеки, чистый ООП и немного C++ для скорости
Всем привет! Снова отыскала интересную статью на Medium, интересно услышать ваше мнение :-)
Я создал этот инструмент, потому что устал от одних и тех же скучных кликов каждую неделю. Мне нужен был инструмент, который: отслеживает папку, извлекает данные из PDF, обогащает их, отправляет отчеты и, в идеале, позволяет выставлять кому-то счет за сэкономленное время. Два выходных, несколько библиотек и пачка кофе – и у меня был продукт, за который люди действительно платили.
Ниже я покажу точный технологический стек, архитектуру, методы монетизации и паттерны кода, которые я использовал. Вас ждет практический код, ООП-структура и один небольшой трюк с C++, когда чистого Python уже не хватало.
1. Выбирайте маленькую, но болезненную задачу
Большинство проектов по автоматизации умирают, потому что пытаются решить слишком много. Вместо этого выберите одну повторяющуюся «боль» с измеримым ROI. Моя проблема была такой:
Клиент ежедневно присылает счета (инвойсы) в виде разрозненных PDF.
Я вручную открываю их, извлекаю поставщика, дату, сумму и заношу в Google Sheet.
Трата: ≈ 20 минут в день, или ≈ 6 часов в месяц.
Цель: Свести это к нулю человеко-минут и предложить как платную услугу.
2. Быстрый MVP: Наблюдатель за файлами + PDF-экстрактор
Начните с малого: отслеживайте папку, обнаруживайте новый PDF, извлекайте текст. Используем watchdog и PyMuPDF (fitz).
pip install watchdog pymupdf
# file_watcher.py
import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import fitz # pymupdf
class PDFHandler(FileSystemEventHandler):
"""Обработчик событий файловой системы."""
def on_created(self, event):
if event.src_path.lower().endswith(".pdf"):
print(f"[+] Новый PDF: {event.src_path}")
text = extract_text(event.src_path)
print(text[:200], "...\n") # быстрый превью
def extract_text(path: str) -> str:
"""Извлечение текстового слоя из PDF."""
doc = fitz.open(path)
pages = [page.get_text() for page in doc]
doc.close()
return "\n".join(pages)
if __name__ == "__main__":
observer = Observer()
handler = PDFHandler()
# Следим за папкой 'inbox'
observer.schedule(handler, path="./inbox", recursive=False)
observer.start()
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
observer.stop()
observer.join()Этот единственный скрипт сократил мою ежедневную работу до 5 минут – в основном на проверку
3. Повышаем отказоустойчивость: OCR + текстовый фолбэк
Некоторые PDF – это просто сканированные изображения. Добавим фолбэк с помощью pytesseract.
pip install pytesseract pillow
# Внимание: tesseract должен быть установлен в системе (apt / brew / choco)from PIL import Image
import pytesseract
import fitz
def extract_text_with_ocr(path: str) -> str:
"""Извлечение текста: сначала текстовый слой, затем OCR для сканов."""
doc = fitz.open(path)
aggregated = []
for page in doc:
text = page.get_text()
if text.strip():
# Если есть текстовый слой, используем его
aggregated.append(text)
else:
# Иначе делаем скриншот страницы и прогоняем через Tesseract
pix = page.get_pixmap(dpi=200)
img = Image.frombytes("RGB", [pix.width, pix.height], pix.samples)
aggregated.append(pytesseract.image_to_string(img))
doc.close()
return "\n".join(aggregated)Такой гибридный подход (текстовый слой → OCR) сделал инструмент надежным для 95% входящих счетов.
4. Архитектура: ООП для модульности (Пайплайн)
Если вы хотите превратить это в продукт, пайплайн должен быть модульным. Каждый шаг – это класс: Loader → Parser → Enricher → Sink. Это позволяет легко менять хранилище (Google Sheets, СУБД, вебхук) без переписывания логики.
# pipeline.py
from abc import ABC, abstractmethod
from typing import Dict
class Step(ABC):
"""Абстрактный класс для шага обработки."""
@abstractmethod
def run(self, data: Dict) -> Dict:
pass
class Loader(Step):
"""Загружает данные (извлекает текст)."""
def __init__(self, path): self.path = path
def run(self, data):
data['text'] = extract_text_with_ocr(self.path)
return data
class Parser(Step):
"""Извлекает ключевые поля (сумма, дата, поставщик)."""
def run(self, data):
text = data['text']
# Здесь будет реальная логика парсинга (regex/NLP)
data['vendor'] = find_vendor(text)
data['amount'] = find_amount(text)
return data
class Sink(Step):
"""Выгружает данные в целевую систему (например, Google Sheets)."""
def run(self, data):
push_to_google_sheet(data)
return data
class Pipeline:
"""Сборщик и исполнитель пайплайна."""
def __init__(self, steps: list[Step]):
self.steps = steps
def execute(self, initial: Dict):
data = initial
for step in self.steps:
data = step.run(data)
return dataЭтот паттерн масштабируем: легко добавить ClassifierStep для определения языка или TranslatorStep для неанглоязычных документов.
5. Извлечение и обогащение: Regex, а затем ML
Начинайте с детерминированного парсинга (regex). Если счета сложные, переходите к ML-моделям (например, spacy или layout-parser). Пример для извлечения суммы:
import re
# Регулярное выражение для поиска сумм с валютой (USD/EUR/$, с запятыми/точками)
AMOUNT_RE = re.compile(r"(?<!\d)(?:USD|EUR|\$)?\s?([\d]{1,3}(?:,\d{3})*(?:\.\d{2})?)\b")
def find_amount(text: str) -> float | None:
text = text.replace("\n", " ")
m = AMOUNT_RE.search(text)
if m:
# Удаляем разделители тысяч, чтобы корректно преобразовать в float
s = m.group(1).replace(',', '')
return float(s)
return NoneДля повышения надежности используйте spacy + кастомный NER или библиотеки вроде layout-parser для пространственного обнаружения полей.
6. Веб-автоматизация: Playwright для загрузки
Если счета находятся за веб-дашбордами, автоматизируйте их загрузку с помощью Playwright.
pip install playwright
playwright installfrom playwright.sync_api import sync_playwright
def login_and_download(url, user, password, download_path):
"""Автоматически логинится и скачивает файл."""
with sync_playwright() as p:
browser = p.chromium.launch(headless=True)
page = browser.new_page()
page.goto(url)
# Автоматическое заполнение полей
page.fill('#username', user)
page.fill('#password', password)
page.click('#login')
# Ожидание элемента и скачивание
page.wait_for_selector('a.download')
with page.expect_download() as download_info:
page.click('a.download')
download = download_info.value
download.save_as(download_path)
browser.close()Это позволяет вашему сервису самостоятельно собирать исходные PDF – критически важно для подписки, где система должна сама получать документы клиента каждое утро.
7. Распространение: CLI с Typer / Click
Для удобства использования оберните функциональность в CLI-интерфейс с помощью Typer (или Click). Это позволяет не-разработчикам запускать его локально или упрощает развертывание на серверах.
pip install typer# cli.py
import typer
from pipeline import Pipeline, Loader, Parser, Sink
app = typer.Typer()
@app.command()
def process(path: str):
"""Обрабатывает один файл по указанному пути."""
steps = [Loader(path), Parser(), Sink()]
p = Pipeline(steps)
p.execute({})
typer.echo(f"Файл {path} успешно обработан!")
if __name__ == "__main__":
app()Опубликуйте проект на PyPI (с setup.py или pyproject.toml) или упакуйте в Docker-образ.
8. Оптимизация: Ускорение с C++ (pybind11)
Для тяжелой обработки изображений или масштабного пре-процессинга для OCR Python может стать узким местом. У меня был шаг (кастомное преобразование изображения), который должен был обрабатывать тысячи страниц в день.
Я переписал этот единственный шаг на C++ и сделал его доступным из Python через pybind11 (можно использовать и Cython).
Суть подхода:
Написать тяжелую функцию на C++.
Обернуть ее с помощью pybind11 (для Python-интерфейса).
Импортировать скомпилированный модуль в Python как обычную библиотеку.
Этот микро-рефакторинг сократил время выполнения шага со ≈ 120 мс/страница до ≈ 10 мс/страница. Знайте, где замедляет Python, и изолируйте это место.
9. Масштабирование: Воркеры с Celery + Redis
При росте числа пользователей обрабатывайте задачи в очередях воркеров, чтобы не блокировать основной поток.
pip install celery redis# tasks.py
from celery import Celery
from pipeline import Pipeline, Loader, Parser, Sink
# Настройка Celery с Redis в качестве брокера
app = Celery('tasks', broker='redis://localhost:6379/0')
@app.task
def process_file(path):
"""Задача Celery для асинхронной обработки."""
steps = [Loader(path), Parser(), Sink()]
Pipeline(steps).execute({})10. Монетизация: От фриланса до SaaS
Как я превратил это в деньги:
Фриланс-контракты (ранняя выручка): Предложил автоматизацию обработки счетов нескольким местным клиентам. Быстрые победы, минимальная поддержка.
Цена за документ (Volume pricing): Тарификация за обработанный счет (например, $0.10–$0.50). Отлично для клиентов с большим объемом.
Ежемесячная подписка (SaaS): Хостинг сервиса, автоматический инжест (через Playwright или SFTP) и оплата за удобство + SLA.
White-label / Enterprise: Интеграция в существующую бухгалтерскую платформу.
Ключевые тактики, которые помогли конвертировать лиды:
Двухнедельный бесплатный пробный период: Обработка первых 50 счетов клиента бесплатно.
Прозрачный отчет о точности: Показываем, что было спарсено машиной, а что пришлось править вручную.
Human-in-the-loop (человек в контуре): Предлагаем ручную коррекцию для результатов с низкой уверенностью (дополнительный доход).
11. Скелет репозитория (стартовая точка)
Используйте эту структуру в качестве основы для любого серьезного проекта по автоматизации:
invoice-automator/
├─ pyproject.toml
├─ README.md
├─ src/
│ ├─ automator/
│ │ ├─ __init__.py
│ │ ├─ cli.py # Точка входа CLI
│ │ ├─ pipeline.py # Основная логика ООП-пайплайна
│ │ ├─ loaders.py # Классы Loader (файл, загрузка, Playwright)
│ │ ├─ parsers.py # Regex / NLP парсеры
│ │ ├─ ocr.py # OCR-утилиты и фоллбэк
│ │ ├─ enrichers.py # Нормализация валют, поиск поставщиков
│ │ ├─ sinks.py # Google Sheets / DB / Webhook
│ │ └─ utils.py
│ └─ tests/
│ ├─ test_parsers.py
│ └─ test_pipeline.py
├─ docker/
│ ├─ Dockerfile
│ └─ prod-compose.yml
└─ infra/
└─ celery_worker.yml # Конфигурация воркераКакие еще рутинные задачи вы хотели бы автоматизировать с помощью этого архитектурного шаблона?