Spaces:
No application file
No application file
File size: 4,139 Bytes
0d63e34 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 | import os
import logging
from pathlib import Path
from typing import Dict, Union, List
from watermark import embed_watermark, extract_watermark
# Configurar logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def batch_embed(
input_dir: str,
output_dir: str,
message: bytes,
password: str,
strength: float = 1e-4
) -> Dict[str, Union[int, str]]:
"""
Recorre WAV/MP3 en input_dir, inserta watermark en cada uno, guarda en output_dir.
Retorna dict {filename: bits_insertados | error_message}.
"""
try:
# Validaciones
if not os.path.exists(input_dir):
raise FileNotFoundError(f"Directorio de entrada no existe: {input_dir}")
if not message:
raise ValueError("Mensaje no puede estar vacío")
if not password:
raise ValueError("Contraseña requerida")
# Crear directorio de salida
os.makedirs(output_dir, exist_ok=True)
results = {}
input_path = Path(input_dir)
output_path = Path(output_dir)
# Buscar archivos de audio
audio_files = []
for ext in [".wav", ".mp3", ".WAV", ".MP3"]:
audio_files.extend(input_path.glob(f"*{ext}"))
if not audio_files:
logger.warning(f"No se encontraron archivos de audio en {input_dir}")
return {"warning": "No se encontraron archivos de audio"}
logger.info(f"Procesando {len(audio_files)} archivos de audio")
for audio_file in audio_files:
try:
output_file = output_path / audio_file.name
bits_inserted = embed_watermark(
str(audio_file),
str(output_file),
message,
password,
strength=strength
)
results[audio_file.name] = bits_inserted
logger.info(f"✅ {audio_file.name}: {bits_inserted} bits insertados")
except Exception as e:
error_msg = f"Error: {str(e)}"
results[audio_file.name] = error_msg
logger.error(f"❌ {audio_file.name}: {error_msg}")
return results
except Exception as e:
logger.error(f"Error en batch_embed: {str(e)}")
return {"error": f"Error crítico en batch_embed: {str(e)}"}
def batch_extract(
input_dir: str,
password: str
) -> Dict[str, Union[bytes, str]]:
"""
Extrae watermark de cada archivo en input_dir.
Retorna {filename: payload_bytes | error_message}.
"""
try:
# Validaciones
if not os.path.exists(input_dir):
raise FileNotFoundError(f"Directorio de entrada no existe: {input_dir}")
if not password:
raise ValueError("Contraseña requerida")
results = {}
input_path = Path(input_dir)
# Buscar archivos de audio
audio_files = []
for ext in [".wav", ".mp3", ".WAV", ".MP3"]:
audio_files.extend(input_path.glob(f"*{ext}"))
if not audio_files:
logger.warning(f"No se encontraron archivos de audio en {input_dir}")
return {"warning": "No se encontraron archivos de audio"}
logger.info(f"Procesando {len(audio_files)} archivos de audio")
for audio_file in audio_files:
try:
payload = extract_watermark(str(audio_file), password)
results[audio_file.name] = payload
logger.info(f"✅ {audio_file.name}: payload extraído")
except Exception as e:
error_msg = f"Error: {str(e)}"
results[audio_file.name] = error_msg
logger.error(f"❌ {audio_file.name}: {error_msg}")
return results
except Exception as e:
logger.error(f"Error en batch_extract: {str(e)}")
return {"error": f"Error crítico en batch_extract: {str(e)}"} |