583 lines
19 KiB
Bash
583 lines
19 KiB
Bash
#!/bin/bash
|
||
set -e
|
||
|
||
# --- НАСТРОЙКИ (ПРОВЕРЬТЕ ПЕРЕД ЗАПУСКОМ) ---
|
||
ASR_HOST="192.168.254.34"
|
||
AUDIO_PORT="8080"
|
||
CMD_PORT="8003"
|
||
DB_HOST="192.168.254.35"
|
||
DB_PORT="5432"
|
||
DB_NAME="asr_db"
|
||
DB_USER="asr_user"
|
||
DB_PASSWORD="poluyan77"
|
||
EMOTION_HOST="192.168.254.37"
|
||
EMOTION_PORT="8001"
|
||
# --- КОНЕЦ НАСТРОЕК ---
|
||
|
||
echo "=== 1. Установка базовых пакетов ==="
|
||
apt update
|
||
apt install -y python3 python3-pip python3-venv ffmpeg nginx
|
||
|
||
echo "=== 2. Создание проекта и виртуального окружения ==="
|
||
mkdir -p /root/asr_service
|
||
cd /root/asr_service
|
||
python3 -m venv venv
|
||
source venv/bin/activate
|
||
|
||
echo "=== 3. Установка Python-пакетов ==="
|
||
pip install torch==2.5.1 torchaudio==2.5.1 --index-url https://download.pytorch.org/whl/cpu
|
||
pip install kairos-asr[cpu] pydub numpy soundfile psycopg2-binary requests
|
||
|
||
echo "=== 4. Создание необходимых папок ==="
|
||
mkdir -p /opt/asr/incoming_audio
|
||
mkdir -p /opt/asr/output
|
||
mkdir -p /opt/asr/processed
|
||
mkdir -p /var/log/asr
|
||
|
||
echo "=== 5. Создание модуля извлечения имени (name_extractor.py) ==="
|
||
cat > /root/asr_service/name_extractor.py << 'EOF'
|
||
import re
|
||
|
||
def extract_name(text):
|
||
patterns = [
|
||
r'Добрый день,\s*([А-ЯЁ][а-яё]+)\s*,\s*слушаю',
|
||
r'Здравствуйте,\s*([А-ЯЁ][а-яё]+)\s*,\s*слушаю',
|
||
r'Добрый вечер,\s*([А-ЯЁ][а-яё]+)\s*,\s*слушаю',
|
||
]
|
||
for pattern in patterns:
|
||
match = re.search(pattern, text)
|
||
if match:
|
||
return match.group(1)
|
||
return None
|
||
|
||
def get_third_operator_segment(segments):
|
||
operator_segments = [seg for seg in segments if seg[3] == 'оператор']
|
||
if len(operator_segments) >= 3:
|
||
return operator_segments[2][2]
|
||
return None
|
||
EOF
|
||
|
||
echo "=== 6. Создание модуля парсинга имени файла (filename_parser.py) ==="
|
||
cat > /root/asr_service/filename_parser.py << 'EOF'
|
||
from datetime import datetime
|
||
|
||
def parse_filename(filename):
|
||
name = filename.replace('.wav', '').replace('.flac', '')
|
||
parts = name.split('-')
|
||
if len(parts) >= 6 and parts[0] == 'in':
|
||
incoming = parts[1]
|
||
outgoing = parts[2]
|
||
date_str = parts[3]
|
||
time_str = parts[4]
|
||
timestamp = parts[5] if len(parts) > 5 else None
|
||
|
||
call_date = f"{date_str[:4]}-{date_str[4:6]}-{date_str[6:8]}" if len(date_str) == 8 else None
|
||
call_time = f"{time_str[:2]}:{time_str[2:4]}:{time_str[4:6]}" if len(time_str) == 6 else None
|
||
|
||
return {
|
||
'incoming_number': incoming,
|
||
'outgoing_number': outgoing,
|
||
'call_date': call_date,
|
||
'call_time': call_time,
|
||
'call_timestamp': float(timestamp) if timestamp else None
|
||
}
|
||
return None
|
||
EOF
|
||
|
||
echo "=== 7. Создание модуля тегирования (tagging_service.py) ==="
|
||
cat > /root/asr_service/tagging_service.py << 'EOF'
|
||
import psycopg2
|
||
from psycopg2.extras import RealDictCursor
|
||
|
||
DB_CONFIG = {
|
||
"host": "192.168.254.35",
|
||
"port": 5432,
|
||
"database": "asr_db",
|
||
"user": "asr_user",
|
||
"password": "poluyan77"
|
||
}
|
||
|
||
def get_current_tag_version():
|
||
conn = psycopg2.connect(**DB_CONFIG, cursor_factory=RealDictCursor)
|
||
cursor = conn.cursor()
|
||
cursor.execute("SELECT version FROM tag_version WHERE id = 1")
|
||
row = cursor.fetchone()
|
||
cursor.close()
|
||
conn.close()
|
||
return row['version'] if row else 1
|
||
|
||
def get_rules():
|
||
conn = psycopg2.connect(**DB_CONFIG, cursor_factory=RealDictCursor)
|
||
cursor = conn.cursor()
|
||
cursor.execute("""
|
||
SELECT tr.tag_name, tk.keyword
|
||
FROM tag_rules tr
|
||
JOIN tag_keywords tk ON tk.tag_id = tr.id
|
||
WHERE tr.is_active = TRUE
|
||
""")
|
||
rows = cursor.fetchall()
|
||
cursor.close()
|
||
conn.close()
|
||
|
||
rules = {}
|
||
for row in rows:
|
||
tag = row['tag_name']
|
||
keyword = row['keyword']
|
||
if tag not in rules:
|
||
rules[tag] = []
|
||
rules[tag].append(keyword)
|
||
return rules
|
||
|
||
def tag_text(text):
|
||
rules = get_rules()
|
||
found_tags = set()
|
||
text_lower = text.lower()
|
||
|
||
for tag, keywords in rules.items():
|
||
for keyword in keywords:
|
||
if keyword in text_lower:
|
||
found_tags.add(tag)
|
||
break
|
||
return list(found_tags)
|
||
|
||
def tag_call(call_id):
|
||
conn = psycopg2.connect(**DB_CONFIG)
|
||
cursor = conn.cursor()
|
||
|
||
cursor.execute("SELECT full_text FROM transcriptions WHERE call_id = %s", (call_id,))
|
||
row = cursor.fetchone()
|
||
|
||
if not row or not row[0]:
|
||
cursor.close()
|
||
conn.close()
|
||
return []
|
||
|
||
full_text = row[0]
|
||
tags = tag_text(full_text)
|
||
current_version = get_current_tag_version()
|
||
|
||
cursor.execute("DELETE FROM auto_tags WHERE call_id = %s", (call_id,))
|
||
|
||
for tag in tags:
|
||
cursor.execute("INSERT INTO auto_tags (call_id, tag_name, confidence) VALUES (%s, %s, 0.8)", (call_id, tag))
|
||
|
||
cursor.execute("UPDATE calls SET tag_version = %s WHERE id = %s", (current_version, call_id))
|
||
|
||
conn.commit()
|
||
cursor.close()
|
||
conn.close()
|
||
return tags
|
||
EOF
|
||
|
||
echo "=== 8. Создание основного скрипта обработки (transcribe_auto.py) ==="
|
||
cat > /root/asr_service/transcribe_auto.py << 'EOF'
|
||
#!/usr/bin/env python3
|
||
import warnings
|
||
warnings.filterwarnings("ignore")
|
||
import logging
|
||
logging.getLogger().setLevel(logging.ERROR)
|
||
|
||
import os
|
||
import sys
|
||
import json
|
||
import tempfile
|
||
import shutil
|
||
import fcntl
|
||
from datetime import datetime
|
||
from pydub import AudioSegment
|
||
from pydub.effects import normalize
|
||
from kairos_asr import KairosASR
|
||
import psycopg2
|
||
from psycopg2.extras import execute_values
|
||
from name_extractor import extract_name, get_third_operator_segment
|
||
from filename_parser import parse_filename
|
||
from tagging_service import tag_call
|
||
|
||
# ========== НАСТРОЙКИ ==========
|
||
INPUT_DIR = "/opt/asr/incoming_audio"
|
||
OUTPUT_DIR = "/opt/asr/output"
|
||
PROCESSED_DIR = "/opt/asr/processed"
|
||
LOG_FILE = "/var/log/asr/processor.log"
|
||
|
||
DB_CONFIG = {
|
||
"host": "192.168.254.35",
|
||
"port": 5432,
|
||
"database": "asr_db",
|
||
"user": "asr_user",
|
||
"password": "poluyan77"
|
||
}
|
||
# ================================
|
||
|
||
os.makedirs(INPUT_DIR, exist_ok=True)
|
||
os.makedirs(OUTPUT_DIR, exist_ok=True)
|
||
os.makedirs(PROCESSED_DIR, exist_ok=True)
|
||
|
||
# Кэш имён операторов
|
||
_operator_names_cache = None
|
||
_operator_names_cache_time = None
|
||
|
||
def log(msg):
|
||
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
with open(LOG_FILE, "a") as f:
|
||
f.write(f"[{timestamp}] {msg}\n")
|
||
print(msg)
|
||
|
||
def format_time(seconds):
|
||
minutes = int(seconds // 60)
|
||
secs = seconds % 60
|
||
return f"{minutes:02d}:{secs:06.3f}"
|
||
|
||
def load_operator_names(force=False):
|
||
global _operator_names_cache, _operator_names_cache_time
|
||
|
||
if not force and _operator_names_cache_time is not None:
|
||
age = (datetime.now() - _operator_names_cache_time).total_seconds()
|
||
if age < 86400:
|
||
return _operator_names_cache
|
||
|
||
try:
|
||
conn = psycopg2.connect(**DB_CONFIG)
|
||
cursor = conn.cursor()
|
||
cursor.execute("SELECT name FROM operator_names ORDER BY name")
|
||
rows = cursor.fetchall()
|
||
cursor.close()
|
||
conn.close()
|
||
|
||
_operator_names_cache = set(row[0].lower() for row in rows)
|
||
_operator_names_cache_time = datetime.now()
|
||
log(f"📇 Загружено имён операторов: {len(_operator_names_cache)}")
|
||
return _operator_names_cache
|
||
except Exception as e:
|
||
log(f"❌ Ошибка загрузки имён: {e}")
|
||
return set()
|
||
|
||
def extract_name_from_third_segment(segments):
|
||
operator_segments = [seg for seg in segments if seg[3] == 'оператор']
|
||
if len(operator_segments) < 3:
|
||
return None
|
||
|
||
third_phrase = operator_segments[2][2]
|
||
if not third_phrase:
|
||
return None
|
||
|
||
names_set = load_operator_names()
|
||
words = third_phrase.lower().split()
|
||
|
||
for word in words:
|
||
clean_word = word.strip('.,!?;:()"\'')
|
||
if clean_word in names_set:
|
||
return clean_word.capitalize()
|
||
return None
|
||
|
||
def analyze_emotion_text(text, max_retries=3, timeout=10):
|
||
import requests
|
||
import time
|
||
|
||
for attempt in range(max_retries):
|
||
try:
|
||
response = requests.post(f'http://192.168.254.37:8001/analyze', data={'text': text}, timeout=timeout)
|
||
if response.status_code == 200:
|
||
result = response.json()
|
||
return result.get('emotion'), result.get('confidence')
|
||
else:
|
||
log(f"⚠️ Эмоции: HTTP {response.status_code}, попытка {attempt + 1}/{max_retries}")
|
||
except requests.exceptions.Timeout:
|
||
log(f"⚠️ Эмоции: таймаут, попытка {attempt + 1}/{max_retries}")
|
||
except requests.exceptions.ConnectionError:
|
||
log(f"⚠️ Эмоции: нет соединения, попытка {attempt + 1}/{max_retries}")
|
||
except Exception as e:
|
||
log(f"⚠️ Эмоции: ошибка {e}, попытка {attempt + 1}/{max_retries}")
|
||
|
||
if attempt < max_retries - 1:
|
||
time.sleep(2)
|
||
|
||
log("❌ Эмоции: не удалось получить ответ")
|
||
return None, None
|
||
|
||
def save_to_db(call_data, segments):
|
||
conn = None
|
||
try:
|
||
conn = psycopg2.connect(**DB_CONFIG)
|
||
cursor = conn.cursor()
|
||
|
||
cursor.execute("""
|
||
INSERT INTO calls
|
||
(incoming_number, outgoing_number, call_date, call_time, call_timestamp,
|
||
audio_filename, audio_filepath, audio_duration_sec, audio_size_bytes, status)
|
||
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, 'completed')
|
||
RETURNING id
|
||
""", (
|
||
call_data.get('incoming_number'),
|
||
call_data.get('outgoing_number'),
|
||
call_data.get('call_date'),
|
||
call_data.get('call_time'),
|
||
datetime.fromtimestamp(call_data['call_timestamp']) if call_data.get('call_timestamp') else None,
|
||
call_data['audio_filename'],
|
||
call_data['audio_filepath'],
|
||
call_data['audio_duration_sec'],
|
||
call_data['audio_size_bytes']
|
||
))
|
||
call_id = cursor.fetchone()[0]
|
||
|
||
full_text = ' '.join([seg[2] for seg in segments])
|
||
cursor.execute("INSERT INTO transcriptions (call_id, full_text) VALUES (%s, %s) RETURNING id", (call_id, full_text))
|
||
transcript_id = cursor.fetchone()[0]
|
||
|
||
segments_data = [(transcript_id, idx, seg[3], seg[0], seg[1], seg[2]) for idx, seg in enumerate(segments)]
|
||
execute_values(cursor, """
|
||
INSERT INTO transcription_segments
|
||
(transcript_id, segment_index, speaker, start_time_sec, end_time_sec, text)
|
||
VALUES %s
|
||
""", segments_data)
|
||
|
||
operator_name = extract_name_from_third_segment(segments)
|
||
if operator_name:
|
||
cursor.execute("UPDATE calls SET operator_name = %s WHERE id = %s", (operator_name, call_id))
|
||
log(f"👤 Извлечено имя оператора: {operator_name}")
|
||
|
||
log("🔍 Запускаем тегирование...")
|
||
tags = tag_call(call_id)
|
||
if tags:
|
||
log(f"🏷️ Найдены теги: {', '.join(tags)}")
|
||
|
||
emotion, confidence = analyze_emotion_text(full_text)
|
||
if emotion:
|
||
cursor.execute("INSERT INTO auto_tags (call_id, tag_name, confidence) VALUES (%s, %s, %s)", (call_id, f"эмоция_{emotion}", confidence))
|
||
log(f"🎭 Эмоция звонка: {emotion} ({confidence:.2f})")
|
||
|
||
conn.commit()
|
||
log(f"💾 Сохранено в БД: call_id={call_id}, segments={len(segments)}")
|
||
return call_id
|
||
|
||
except Exception as e:
|
||
log(f"❌ Ошибка БД: {e}")
|
||
if conn:
|
||
conn.rollback()
|
||
return None
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
def process_file(filepath):
|
||
log(f"🔄 Обработка: {filepath}")
|
||
|
||
filename = os.path.basename(filepath)
|
||
file_size = os.path.getsize(filepath)
|
||
|
||
metadata = parse_filename(filename)
|
||
|
||
audio = AudioSegment.from_wav(filepath)
|
||
duration = len(audio) / 1000.0
|
||
log(f"📊 Длительность: {duration:.1f} сек, размер: {file_size/1024/1024:.2f} МБ")
|
||
|
||
audio_asr = audio.set_frame_rate(16000).set_sample_width(2)
|
||
audio_asr = normalize(audio_asr, headroom=0.5)
|
||
|
||
if audio_asr.channels < 2:
|
||
left_audio = audio_asr
|
||
right_audio = None
|
||
else:
|
||
mono = audio_asr.split_to_mono()
|
||
left_audio = mono[0]
|
||
right_audio = mono[1]
|
||
|
||
asr = KairosASR(device="cpu")
|
||
all_segments = []
|
||
|
||
if left_audio:
|
||
with tempfile.NamedTemporaryFile(suffix='.wav', delete=False) as tmp:
|
||
left_audio.export(tmp.name, format='wav')
|
||
result = asr.transcribe(wav_file=tmp.name)
|
||
for sent in result.sentences:
|
||
all_segments.append((sent.start, sent.end, sent.text, "абонент"))
|
||
os.unlink(tmp.name)
|
||
|
||
if right_audio:
|
||
with tempfile.NamedTemporaryFile(suffix='.wav', delete=False) as tmp:
|
||
right_audio.export(tmp.name, format='wav')
|
||
result = asr.transcribe(wav_file=tmp.name)
|
||
for sent in result.sentences:
|
||
all_segments.append((sent.start, sent.end, sent.text, "оператор"))
|
||
os.unlink(tmp.name)
|
||
|
||
all_segments.sort(key=lambda x: x[0])
|
||
|
||
output_file = os.path.join(OUTPUT_DIR, filename.replace('.wav', '.json'))
|
||
with open(output_file, 'w', encoding='utf-8') as f:
|
||
json.dump({"file": filepath, "segments": [[s, e, t, sp] for s, e, t, sp in all_segments]}, f, ensure_ascii=False, indent=2)
|
||
|
||
text_file = os.path.join(OUTPUT_DIR, filename.replace('.wav', '.txt'))
|
||
with open(text_file, 'w', encoding='utf-8') as f:
|
||
for start, end, text, speaker in all_segments:
|
||
f.write(f"[{format_time(start)} → {format_time(end)}] {speaker}: {text}\n")
|
||
|
||
call_data = {
|
||
'audio_filename': filename,
|
||
'audio_filepath': filepath,
|
||
'audio_duration_sec': duration,
|
||
'audio_size_bytes': file_size,
|
||
}
|
||
if metadata:
|
||
call_data.update(metadata)
|
||
|
||
call_id = save_to_db(call_data, all_segments)
|
||
|
||
flac_filename = filename.replace('.wav', '.flac')
|
||
flac_path = os.path.join(PROCESSED_DIR, flac_filename)
|
||
audio.export(flac_path, format='flac')
|
||
os.remove(filepath)
|
||
|
||
flac_size = os.path.getsize(flac_path) / 1024 / 1024
|
||
log(f"✅ Готово: {flac_filename} ({flac_size:.2f} МБ), call_id={call_id}")
|
||
return True
|
||
|
||
if __name__ == "__main__":
|
||
lock_fd = open('/tmp/asr_processor.lock', 'w')
|
||
try:
|
||
fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||
except BlockingIOError:
|
||
print("Скрипт уже запущен, выхожу")
|
||
sys.exit(0)
|
||
|
||
log("="*60)
|
||
log("ASR Processor started")
|
||
|
||
files = [f for f in os.listdir(INPUT_DIR) if f.endswith('.wav')]
|
||
|
||
if not files:
|
||
log("Нет новых файлов")
|
||
else:
|
||
log(f"Найдено файлов: {len(files)}")
|
||
for filename in files:
|
||
filepath = os.path.join(INPUT_DIR, filename)
|
||
try:
|
||
process_file(filepath)
|
||
except Exception as e:
|
||
log(f"❌ Ошибка при обработке {filename}: {e}")
|
||
|
||
log("ASR Processor finished")
|
||
fcntl.flock(lock_fd, fcntl.LOCK_UN)
|
||
lock_fd.close()
|
||
EOF
|
||
|
||
chmod +x /root/asr_service/transcribe_auto.py
|
||
|
||
echo "=== 9. Создание командного сервера (command_server.py) ==="
|
||
cat > /root/asr_service/command_server.py << 'EOF'
|
||
#!/usr/bin/env python3
|
||
import sys
|
||
import os
|
||
import json
|
||
import http.server
|
||
|
||
sys.path.insert(0, '/root/asr_service')
|
||
|
||
def load_operator_names():
|
||
import psycopg2
|
||
DB_CONFIG = {"host": "192.168.254.35", "port": 5432, "database": "asr_db", "user": "asr_user", "password": "poluyan77"}
|
||
try:
|
||
conn = psycopg2.connect(**DB_CONFIG)
|
||
cursor = conn.cursor()
|
||
cursor.execute("SELECT name FROM operator_names ORDER BY name")
|
||
rows = cursor.fetchall()
|
||
cursor.close()
|
||
conn.close()
|
||
return [row[0] for row in rows]
|
||
except Exception as e:
|
||
print(f"Ошибка: {e}")
|
||
return []
|
||
|
||
def reload_operator_names():
|
||
global _operator_names_cache
|
||
_operator_names_cache = load_operator_names()
|
||
return _operator_names_cache
|
||
|
||
_operator_names_cache = load_operator_names()
|
||
|
||
class CommandHandler(http.server.SimpleHTTPRequestHandler):
|
||
def do_POST(self):
|
||
if self.path == '/reload-names':
|
||
try:
|
||
names = reload_operator_names()
|
||
self.send_response(200)
|
||
self.send_header('Content-type', 'application/json')
|
||
self.end_headers()
|
||
self.wfile.write(json.dumps({"status": "ok", "count": len(names)}).encode())
|
||
print(f"🔄 Кэш имён операторов обновлён, загружено {len(names)} имён")
|
||
except Exception as e:
|
||
self.send_response(500)
|
||
self.end_headers()
|
||
self.wfile.write(json.dumps({"status": "error", "error": str(e)}).encode())
|
||
else:
|
||
self.send_response(404)
|
||
self.end_headers()
|
||
|
||
def log_message(self, format, *args):
|
||
pass
|
||
|
||
def start_command_server(port=8003):
|
||
server = http.server.HTTPServer(('0.0.0.0', port), CommandHandler)
|
||
print(f"🚀 Командный сервер запущен на порту {port}")
|
||
server.serve_forever()
|
||
|
||
if __name__ == "__main__":
|
||
start_command_server()
|
||
EOF
|
||
|
||
chmod +x /root/asr_service/command_server.py
|
||
|
||
echo "=== 10. Настройка Nginx для аудио ==="
|
||
cat > /etc/nginx/sites-available/audio-server << EOF
|
||
server {
|
||
listen $AUDIO_PORT;
|
||
server_name $ASR_HOST;
|
||
|
||
location /audio/ {
|
||
alias /opt/asr/processed/;
|
||
add_header Access-Control-Allow-Origin *;
|
||
add_header Access-Control-Allow-Methods "GET, HEAD, OPTIONS";
|
||
add_header Access-Control-Allow-Headers "Range";
|
||
add_header Content-Type audio/flac;
|
||
}
|
||
}
|
||
EOF
|
||
|
||
ln -sf /etc/nginx/sites-available/audio-server /etc/nginx/sites-enabled/
|
||
rm -f /etc/nginx/sites-enabled/default
|
||
nginx -t
|
||
systemctl restart nginx
|
||
|
||
echo "=== 11. Создание systemd сервиса для командного сервера ==="
|
||
cat > /etc/systemd/system/asr-command.service << EOF
|
||
[Unit]
|
||
Description=ASR Command Server for operator names cache
|
||
After=network.target
|
||
|
||
[Service]
|
||
Type=simple
|
||
User=root
|
||
WorkingDirectory=/root/asr_service
|
||
Environment="PATH=/root/asr_service/venv/bin"
|
||
ExecStart=/root/asr_service/venv/bin/python /root/asr_service/command_server.py
|
||
Restart=always
|
||
RestartSec=10
|
||
|
||
[Install]
|
||
WantedBy=multi-user.target
|
||
EOF
|
||
|
||
echo "=== 12. Настройка cron для запуска воркера ==="
|
||
cat > /etc/cron.d/asr-processor << EOF
|
||
*/5 * * * * root cd /root/asr_service && /root/asr_service/venv/bin/python /root/asr_service/transcribe_auto.py >> /var/log/asr/cron.log 2>&1
|
||
EOF
|
||
|
||
echo "=== 13. Запуск сервисов ==="
|
||
systemctl daemon-reload
|
||
systemctl enable asr-command
|
||
systemctl start asr-command
|
||
systemctl restart nginx
|
||
|
||
echo "=== ASR ВМ готова ==="
|
||
echo "Аудио порт: http://$ASR_HOST:$AUDIO_PORT"
|
||
echo "Командный порт: $CMD_PORT"
|
||
echo "Cron настроен на запуск каждые 5 минут" |