Создание аналога PLAUD Note из диктофона кулона и Raspberry Pi 4

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

Я так и не выработал привычку записывать что-то в блокноте, с переменным успехом записываю что-то в календарь или в заметки в ТГ, но это не то, нужен список задач, отдельно записанные рассуждения или записи в дневник. Давно наблюдаю за устройствами Plaud, но цена на них кусается, да и тянет меня на создание таких сервисов своими руками.

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

В качестве хардварной составляющей были выбраны.

1. Диктофон-кулон с озона.

2. Raspberry Pi 4.

Распознавание речи: Yandex Speech Kit

Обработка текста и генерация команд для трекера: Deepseek API

Трекер: YouGile

Ну и написание скриптов, ибо сам я такое не сделаю: Gemini

Ниже схема работы сервиса.

[ Диктофон (ON/OFF) ]
         │  (Запись в /RECORD)

         ▼

[ USB-подключение к Малинке ]

         ▼

[ watcher.py (каждые 3 сек) ] ──(монтирование /mnt/recorder)──┐

         ▼                                                    │

(Перенос через shutil.move)                                   │

         ▼                                                    │

[ Размонтирование / Зарядка ]                                 │

         ▼                                                    │

[ processor.py ] 

         ├──► [ Yandex SpeechKit ] (Текст)

         │         ▼

         ├──► [ DeepSeek API ] (JSON: проект, тип, дедлайн)

         ▼

[ YouGile API ] ──► Создание карточки + Архивация аудио

В итоге вся реализация заняла два часа, с условием того, что все сервисы я использовал ранее и у меня был настроен SpeechKit и Deepseek api с пополненным счётом(кстати вполне себе удобно пополняется через Alipay).

В процессе реализации было пару моментов с отладкой распознавания подключения диктофона, по классике с правами на работу с файлами в Linux, ну и самое обидное, на диктофоне не создавались записи, оказалось, что запись стартует примерно через 3 секунды.

В итоге получаем готовый сервис, записанные задачи, которые я могу при наличии свободного времени просматривать и разгребать. Теперь главное не забывать подключать диктофон раз в деньь на работе к малинке. Посмотрим, изменит ли это мою жизнь, если решение приживётся, добавлю туда отправку записей в ТГ и создание событий в гугл календаре.

Все скрипты ниже, не все они оптимальны, тот же опрос факта подключения флешки раз в 3 секунды с точки зрения автоматизации дикий костыль, но на текущий момент задачи это выполняет. Буду рад дополнениям, идеям, да и критике пожалуй)

Подготовка окружения

# Обновляем пакеты и ставим системные утилиты (ffmpeg нужен, если в будущем будете конвертировать форматы, uhubctl — для питания USB)
sudo apt update && sudo apt install -y 
    python3-pip 
    python3-venv 
    ffmpeg 
    uhubctl 
    git 
    lsblk

#Создание структуры папок проекта
mkdir -p $HOME/voice-assistant/scripts
mkdir -p $HOME/voice-assistant/inbox/raw
mkdir -p $HOME/voice-assistant/archive/audio
mkdir -p $HOME/voice-assistant/logs

#Создание и настройка виртуального окружения (venv)

cd $HOME/voice-assistant
python3 -m venv venv
$HOME/voice-assistant/venv/bin/pip install --upgrade pip
$HOME/voice-assistant/venv/bin/pip install requests python-dotenv


#Создание файла конфигурации .env
cat << 'EOF' > $HOME/voice-assistant/.env
YOUGILE_API_TOKEN=
YANDEX_STT_API_KEY=
DEEPSEEK_API_KEY=
DEEPSEEK_BASE_URL=https://api.deepseek.com/v1

YOUGILE_COLUMN_HOME_TASKS=
YOUGILE_COLUMN_HOME_IDEAS=
YOUGILE_COLUMN_HOME_THOUGHTS=
YOUGILE_COLUMN_WORK_TASKS=
YOUGILE_COLUMN_WORK_IDEAS=
YOUGILE_COLUMN_WORK_THOUGHTS=
EOF

# Защищаем файл, чтобы секреты не светились
chmod 600 $HOME/voice-assistant/.env

Шаблон файла переменных окружения (.env)

Создайте в корне проекта ($HOME/voice-assistant/.env) файл со следующим содержимым и подставьте свои реальные данные:

# --- API Ключи ---
# Токен бота или аккаунта YouGile (получается в настройках компании/пользователя)
YOUGILE_API_TOKEN=your_yougile_api_token_here

# API-ключ Yandex SpeechKit (для распознавания речи в текст)
YANDEX_STT_API_KEY=your_yandex_speech_key_here

# API-ключ DeepSeek (или совместимого OpenAI-провайдера)
DEEPSEEK_API_KEY=your_deepseek_api_key_here
DEEPSEEK_BASE_URL=https://api.deepseek.com/v1

# --- ID колонок / досок YouGile ---
# Идентификаторы досок и колонок, куда бот будет раскладывать карточки в зависимости от типа и проекта
YOUGILE_COLUMN_HOME_TASKS=uuid_column_home_tasks_here
YOUGILE_COLUMN_HOME_IDEAS=uuid_column_home_ideas_here
YOUGILE_COLUMN_HOME_THOUGHTS=uuid_column_home_thoughts_here

YOUGILE_COLUMN_WORK_TASKS=uuid_column_work_tasks_here
YOUGILE_COLUMN_WORK_IDEAS=uuid_column_work_ideas_here
YOUGILE_COLUMN_WORK_THOUGHTS=uuid_column_work_thoughts_here

Фоновый наблюдатель ($HOME/voice-assistant/scripts/watcher.py)

Демон, который крутится от root, мониторит шину USB, монтирует диктофон, безопасно переносит файлы через shutil. move и вызывает обработчик:

import os
import time
import shutil
import subprocess

MOUNT_POINT = "/mnt/recorder"
BASE_DIR = "/home/rurmanov/voice-assistant"
INBOX_RAW = os.path.join(BASE_DIR, "inbox/raw")
LOG_FILE = os.path.join(BASE_DIR, "logs/usb_ingest.log")
PROCESSOR_SCRIPT = os.path.join(BASE_DIR, "scripts/processor.py")
PYTHON_BIN = os.path.join(BASE_DIR, "venv/bin/python")

def log(msg):
    ts = time.strftime("%Y-%m-%d %H:%M:%S")
    line = f"{ts} [WATCHER] {msg}"
    print(line)
    os.makedirs(os.path.dirname(LOG_FILE), exist_ok=True)
    with open(LOG_FILE, "a", encoding="utf-8") as f:
        f.write(line + "n")

def find_recorder_device():
    try:
        output = subprocess.check_output(["lsblk", "-o", "NAME,LABEL,FSTYPE", "-n", "-r"], text=True)
        for line in output.splitlines():
            parts = line.split()
            if len(parts) >= 2 and parts[1] == "VOICE":
                return f"/dev/{parts[0]}"
    except Exception:
        pass
    return None

log("Стабильный наблюдатель запущен (режим root)...")

while True:
    try:
        dev = find_recorder_device()
        if dev:
            log(f"Обнаружен диктофон на {dev}")
            subprocess.run(["umount", dev], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
            
            os.makedirs(MOUNT_POINT, exist_ok=True)
            res = subprocess.run(["mount", "-o", "rw,uid=1000,gid=1000,iocharset=utf8", dev, MOUNT_POINT])
            
            count = 0
            if res.returncode == 0:
                rec_dir = os.path.join(MOUNT_POINT, "RECORD")
                if os.path.isdir(rec_dir):
                    files = [f for f in os.listdir(rec_dir) if f.lower().endswith(('.wav', '.mp3'))]
                    count = len(files)
                    log(f"Найдено новых записей в RECORD: {count}")
                    
                    for f in files:
                        src = os.path.join(rec_dir, f)
                        dst = os.path.join(INBOX_RAW, f)
                        try:
                            shutil.move(src, dst)
                            log(f"Успешно перенесен файл: {f} -> inbox/raw/")
                        except Exception as ex:
                            log(f"Ошибка переноса файла {f}: {ex}")
                    
                    if count > 0:
                        subprocess.run(["sync"])
                
                subprocess.run(["umount", MOUNT_POINT], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
                log("Диктофон размонтирован.")
                
                if count > 0:
                    log("Запуск процессора обработки от имени пользователя rurmanov...")
                    proc_res = subprocess.run(["sudo", "-u", "rurmanov", PYTHON_BIN, PROCESSOR_SCRIPT], capture_output=True, text=True)
                    log(f"Результат процессора: код {proc_res.returncode}")
                    if proc_res.stdout:
                        log(f"Вывод процессора: {proc_res.stdout.strip()}")
                    if proc_res.stderr:
                        log(f"Ошибки процессора: {proc_res.stderr.strip()}")
            
            while find_recorder_device() is not None:
                time.sleep(3)
            log("Диктофон отключен пользователем.")
            
    except Exception as e:
        log(f"Критическая ошибка в цикле наблюдателя: {e}")
        
    time.sleep(3)

Systemd-служба (/etc/systemd/system/voice-watcher.service)

Конфигурация демона автозапуска для Linux:

[Unit]
Description=USB Voice Recorder Background Watcher
After=network.target local-fs.target

[Service]
Type=simple
User=root
ExecStart=/home/rurmanov/voice-assistant/venv/bin/python /home/rurmanov/voice-assistant/scripts/watcher.py
Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target

Команды для активации службы:

sudo systemctl daemon-reload
sudo systemctl enable --now voice-watcher.service

Код скрипта обработки ($HOME/voice-assistant/scripts/processor.py)

import os
import time
import shutil
import logging
import requests
from dotenv import load_dotenv

# Загружаем окружение
load_dotenv(dotenv_path="/home/rurmanov/voice-assistant/.env")

BASE_DIR = "/home/rurmanov/voice-assistant"
INBOX_DIR = os.path.join(BASE_DIR, "inbox/raw")
ARCHIVE_DIR = os.path.join(BASE_DIR, "archive/audio")
LOG_FILE = os.path.join(BASE_DIR, "logs/processor.log")

# Настройка логирования
os.makedirs(os.path.dirname(LOG_FILE), exist_ok=True)
logging.basicConfig(
    filename=LOG_FILE,
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    encoding="utf-8"
)

# Читаем ключи и настройки
YOUGILE_TOKEN = os.getenv("YOUGILE_API_TOKEN")
YANDEX_KEY = os.getenv("YANDEX_STT_API_KEY")
DEEPSEEK_KEY = os.getenv("DEEPSEEK_API_KEY")
DEEPSEEK_BASE_URL = os.getenv("DEEPSEEK_BASE_URL", "https://api.deepseek.com/v1")

# Маппинг колонок YouGile из .env
COLUMN_MAPPING = {
    "home": {
        "tasks": os.getenv("YOUGILE_COLUMN_HOME_TASKS"),
        "ideas": os.getenv("YOUGILE_COLUMN_HOME_IDEAS"),
        "thoughts": os.getenv("YOUGILE_COLUMN_HOME_THOUGHTS"),
    },
    "work": {
        "tasks": os.getenv("YOUGILE_COLUMN_WORK_TASKS"),
        "ideas": os.getenv("YOUGILE_COLUMN_WORK_IDEAS"),
        "thoughts": os.getenv("YOUGILE_COLUMN_WORK_THOUGHTS"),
    }
}

def transcribe_yandex(audio_path):
    """Отправка аудио в Yandex SpeechKit STT"""
    url = f"https://stt.api.cloud.yandex.net/speech/v1/stt:recognize?lang=ru-RU"
    headers = {"Authorization": f"Api-Key {YANDEX_KEY}"}
    
    with open(audio_path, "rb") as f:
        audio_data = f.read()
        
    response = requests.post(url, headers=headers, data=audio_data)
    if response.status_code == 200:
        result = response.json()
        text = result.get("result", "").strip()
        logging.info(f"Распознанный текст: «{text}»")
        return text
    else:
        logging.error(f"Ошибка Yandex STT: {response.status_code} - {response.text}")
        return None

def analyze_with_deepseek(text):
    """Анализ текста через DeepSeek LLM с возвратом структурированного JSON"""
    import json
    url = f"{DEEPSEEK_BASE_URL}/chat/completions"
    headers = {
        "Authorization": f"Bearer {DEEPSEEK_KEY}",
        "Content-Type": "application/json"
    }
    
    prompt = f"""
Ты — умный ассистент для разбора голосовых заметок. Проанализируй текст и разбей его на одну или несколько задач/идей.
Верни строго валидный JSON-массив объектов без текста вокруг (без markdown-оформления ```json ... ```), следующей структуры:
[
  {{
    "project": "home" или "work",
    "type": "tasks" или "ideas" или "thoughts",
    "title": "Краткая суть задачи",
    "color": "task-red" или null,
    "due": "YYYY-MM-DD" или null
  }}
]

Текст заметки: "{text}"
"""

    payload = {
        "model": "deepseek-chat",
        "messages": [{"role": "user", "content": prompt}],
        "temperature": 0.1
    }
    
    try:
        response = requests.post(url, headers=headers, json=payload, timeout=30)
        if response.status_code == 200:
            content = response.json()["choices"][0]["message"]["content"].strip()
            # Убираем markdown обертку если LLM её всё же добавила
            if content.startswith("```"):
                content = content.split("```")[1]
                if content.startswith("json"):
                    content = content[4:]
            return json.loads(content.strip())
        else:
            logging.error(f"Ошибка DeepSeek API: {response.text}")
            return []
    except Exception as e:
        logging.error(f"Ошибка парсинга ответа DeepSeek: {e}")
        return []

def create_yougile_card(task_data):
    """Создание карточки в YouGile через REST API"""
    project = task_data.get("project", "home")
    task_type = task_data.get("type", "thoughts")
    
    column_id = COLUMN_MAPPING.get(project, {}).get(task_type)
    if not column_id:
        logging.warning(f"Не найден ID колонки для project={project}, type={task_type}. Используем дефолтную.")
        column_id = COLUMN_MAPPING["home"]["thoughts"]
        
    url = "https://ru.yougile.com-api-v2proxy or official/api-v2/tasks" # (или ваш актуальный эндпоинт YouGile API v2)
    url = "https://api.yougile.com/api-v2/tasks"
    
    headers = {
        "Authorization": f"Bearer {YOUGILE_TOKEN}",
        "Content-Type": "application/json"
    }
    
    payload = {
        "columnId": column_id,
        "title": task_data.get("title", "Заметка без названия")
    }
    
    if task_data.get("color"):
        payload["color"] = task_data["color"]
    if task_data.get("due"):
        payload["deadline"] = {"date": task_data["due"]}
        
    response = requests.post(url, headers=headers, json=payload)
    if response.status_code in [200, 201]:
        res_json = response.json()
        card_id = res_json.get("id")
        logging.info(f"Создана карточка YouGile [{project}/{task_type}]: {task_data['title']} (ID: {card_id})")
    else:
        logging.error(f"Ошибка создания карточки в YouGile: {response.status_code} - {response.text}")

def main():
    files = [f for f in os.listdir(INBOX_DIR) if f.lower().endswith(('.wav', '.mp3', '.WAV'))]
    if not files:
        logging.info("Новых аудиофайлов в inbox/raw не найдено.")
        print("Новых аудиофайлов в inbox/raw не найдено.")
        return

    for filename in sorted(files):
        file_path = os.path.join(INBOX_DIR, filename)
        logging.info(f"Начинаю обработку файла: {filename}")
        print(f"Обработка: {filename}")
        
        # 1. Распознавание
        text = transcribe_yandex(file_path)
        if not text:
            continue
            
        # 2. Анализ нейронкой
        tasks = analyze_with_deepseek(text)
        
        # Если LLM ничего не развернула в список, создаем просто сырую заметку
        if not tasks:
            tasks = [{"project": "home", "type": "thoughts", "title": text, "color": None, "due": None}]
            
        # 3. Пушим в YouGile
        for task in tasks:
            create_yougile_card(task)
            
        # 4. Архивация исходника
        timestamp = time.strftime("%Y%m%d_%H%M%S")
        archive_name = f"{timestamp}_{filename}"
        archive_path = os.path.join(ARCHIVE_DIR, archive_name)
        
        shutil.move(file_path, archive_path)
        logging.info(f"Файл {filename} успешно обработан и перемещён в {archive_path}")

if __name__ == "__main__":
    main()