Как превратить микро-скрипт для автоматизации в коммерческий продукт: библиотеки, чистый ООП и немного 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 install
from 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 # Конфигурация воркера
Какие еще рутинные задачи вы хотели бы автоматизировать с помощью этого архитектурного шаблона?
