115 lines
4.1 KiB
Python
115 lines
4.1 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
|
|
def _bridge_app():
|
|
return sys.modules["services.asterisk_bridge_service.app"]
|
|
|
|
|
|
def fetch_recording_file(remote_path: str, file_name: str | None) -> tuple[Path, bool]:
|
|
bridge = _bridge_app()
|
|
source = Path(remote_path)
|
|
if source.is_absolute() and source.exists():
|
|
return source.resolve(), False
|
|
|
|
if not bridge._sftp_enabled():
|
|
raise RuntimeError("SFTP is not configured and remote_path is not a local file")
|
|
|
|
try:
|
|
import paramiko # type: ignore
|
|
except ImportError as exc: # pragma: no cover - depends on runtime package set
|
|
raise RuntimeError("paramiko is required for SFTP recording transfer") from exc
|
|
|
|
safe_name = file_name or Path(remote_path).name or "asterisk-call.wav"
|
|
suffix = Path(safe_name).suffix or ".wav"
|
|
handle = tempfile.NamedTemporaryFile(prefix="mvpcc-ast-", suffix=suffix, delete=False)
|
|
handle.close()
|
|
target = Path(handle.name)
|
|
|
|
transport = paramiko.Transport((bridge._sftp_host(), bridge._sftp_port()))
|
|
try:
|
|
transport.connect(username=bridge._sftp_username(), password=bridge._sftp_password())
|
|
sftp = paramiko.SFTPClient.from_transport(transport)
|
|
try:
|
|
sftp.get(remote_path, str(target))
|
|
finally:
|
|
sftp.close()
|
|
finally:
|
|
transport.close()
|
|
return target.resolve(), True
|
|
|
|
|
|
def recording_search_directories() -> list[str]:
|
|
base = os.getenv("ASTERISK_SFTP_BASE_PATH", "/var/spool/asterisk/monitor").rstrip("/")
|
|
return [f"{base}/mvpcc", base]
|
|
|
|
|
|
def guess_mime_type(file_name: str) -> str:
|
|
lower = file_name.lower()
|
|
if lower.endswith(".mp3"):
|
|
return "audio/mpeg"
|
|
if lower.endswith(".ogg"):
|
|
return "audio/ogg"
|
|
return "audio/wav"
|
|
|
|
|
|
def find_remote_recording_for_call(call_id: str) -> tuple[str, str, str, datetime] | None:
|
|
bridge = _bridge_app()
|
|
suffixes = (".wav", ".mp3", ".ogg")
|
|
candidates: list[tuple[float, str, str]] = []
|
|
|
|
if bridge._sftp_enabled():
|
|
try:
|
|
import paramiko # type: ignore
|
|
except ImportError:
|
|
return None
|
|
|
|
transport = paramiko.Transport((bridge._sftp_host(), bridge._sftp_port()))
|
|
try:
|
|
transport.connect(username=bridge._sftp_username(), password=bridge._sftp_password())
|
|
sftp = paramiko.SFTPClient.from_transport(transport)
|
|
try:
|
|
for directory in recording_search_directories():
|
|
try:
|
|
names = sftp.listdir(directory)
|
|
except Exception: # noqa: BLE001
|
|
continue
|
|
for name in names:
|
|
lower = name.lower()
|
|
if call_id not in name or not lower.endswith(suffixes):
|
|
continue
|
|
full_path = f"{directory}/{name}"
|
|
try:
|
|
st = sftp.stat(full_path)
|
|
except Exception: # noqa: BLE001
|
|
continue
|
|
candidates.append((float(st.st_mtime), full_path, name))
|
|
finally:
|
|
sftp.close()
|
|
finally:
|
|
transport.close()
|
|
else:
|
|
for directory in recording_search_directories():
|
|
root = Path(directory)
|
|
if not root.exists():
|
|
continue
|
|
for file_path in root.glob(f"*{call_id}*"):
|
|
if not file_path.is_file():
|
|
continue
|
|
lower = file_path.name.lower()
|
|
if not lower.endswith(suffixes):
|
|
continue
|
|
candidates.append((file_path.stat().st_mtime, str(file_path), file_path.name))
|
|
|
|
if not candidates:
|
|
return None
|
|
candidates.sort(key=lambda item: item[0], reverse=True)
|
|
mtime, full_path, file_name = candidates[0]
|
|
mtime_dt = datetime.fromtimestamp(mtime, tz=timezone.utc)
|
|
return full_path, file_name, guess_mime_type(file_name), mtime_dt
|