7f03949593
- Клиент Novofon DATA API (JSON-RPC, get.calls_report, чанки по 90 дней) - faster-whisper (модель turbo) на CUDA с разделением стереоканалов - Маркировка спикеров: левый канал = caller, правый = callee - Directus: коллекции calls, calls_74951284933, contacts - CLI: process, process-file, export, daemon (polling 5 мин) - Docker Compose для Directus + PostgreSQL - Архив email-парсинга в old/
88 lines
2.7 KiB
Python
88 lines
2.7 KiB
Python
import os
|
|
import shutil
|
|
import subprocess
|
|
import json
|
|
from typing import Optional
|
|
|
|
|
|
def _find_ffmpeg() -> str:
|
|
path = shutil.which("ffmpeg")
|
|
if path:
|
|
return path
|
|
winget_path = os.path.join(
|
|
os.environ.get("LOCALAPPDATA", ""),
|
|
"Microsoft", "WinGet", "Links", "ffmpeg.exe"
|
|
)
|
|
if os.path.exists(winget_path):
|
|
return winget_path
|
|
alt_paths = [
|
|
r"C:\ffmpeg\bin\ffmpeg.exe",
|
|
r"C:\Program Files\ffmpeg\bin\ffmpeg.exe",
|
|
]
|
|
for p in alt_paths:
|
|
if os.path.exists(p):
|
|
return p
|
|
raise RuntimeError("FFmpeg not found. Install it via: winget install ffmpeg")
|
|
|
|
|
|
def get_audio_channels(input_path: str) -> int:
|
|
ffmpeg_path = _find_ffmpeg()
|
|
cmd = [
|
|
ffmpeg_path, "-i", input_path,
|
|
"-f", "null", "-"
|
|
]
|
|
result = subprocess.run(cmd, capture_output=True, text=True)
|
|
stderr = result.stderr
|
|
for line in stderr.split("\n"):
|
|
if "Audio:" in line:
|
|
if "stereo" in line or "2 channels" in line:
|
|
return 2
|
|
elif "mono" in line or "1 channel" in line:
|
|
return 1
|
|
import re
|
|
m = re.search(r"(\d+)\s+channels?", line)
|
|
if m:
|
|
return int(m.group(1))
|
|
return 2
|
|
|
|
|
|
def convert_to_wav(input_path: str, output_dir: str, sample_rate: int = 16000) -> Optional[str]:
|
|
os.makedirs(output_dir, exist_ok=True)
|
|
base = os.path.splitext(os.path.basename(input_path))[0]
|
|
output_path = os.path.join(output_dir, f"{base}_16khz.wav")
|
|
|
|
channels = get_audio_channels(input_path)
|
|
ac = "2" if channels == 2 else "1"
|
|
|
|
ffmpeg_path = _find_ffmpeg()
|
|
cmd = [
|
|
ffmpeg_path, "-y",
|
|
"-i", input_path,
|
|
"-acodec", "pcm_s16le",
|
|
"-ac", ac,
|
|
"-ar", str(sample_rate),
|
|
output_path
|
|
]
|
|
result = subprocess.run(cmd, capture_output=True, text=True)
|
|
if result.returncode != 0:
|
|
raise RuntimeError(f"FFmpeg error: {result.stderr}")
|
|
return output_path, channels
|
|
|
|
|
|
def split_stereo_channels(wav_path: str, output_dir: str) -> tuple[Optional[str], Optional[str]]:
|
|
base = os.path.splitext(os.path.basename(wav_path))[0]
|
|
left_path = os.path.join(output_dir, f"{base}_left.wav")
|
|
right_path = os.path.join(output_dir, f"{base}_right.wav")
|
|
ffmpeg_path = _find_ffmpeg()
|
|
|
|
cmd = [
|
|
ffmpeg_path, "-y",
|
|
"-i", wav_path,
|
|
"-filter_complex", "[0:a]pan=mono|c0=c0[left];[0:a]pan=mono|c0=c1[right]",
|
|
"-map", "[left]", left_path,
|
|
"-map", "[right]", right_path,
|
|
]
|
|
result = subprocess.run(cmd, capture_output=True, text=True)
|
|
if result.returncode != 0:
|
|
raise RuntimeError(f"FFmpeg split error: {result.stderr}")
|
|
return left_path, right_path
|