Usando la IA para extraer vídeos cortos (highlights) de partidos de padel grabados
Ejemplo de toma de la pista desde el fondo
apt install git
apt install python3-pip
pip install -r requirements.txt --break-system-packages
python3 -m pip install flask --break-system-packages
apt install -y ffmpeg git" + python3 -m pip install openai-whisper --break-system-packages
Con todo esto, hemos instalado Python para poder ejecutar el código, flask para generar las páginas web de información, ffmpeg para poder editar video, Whisper para transcribir el audio del mismo.
Lo de break system packages es para que se cargue en el contenedor sin tener que instalar un entorno temporal de trabajo.
lanzo el comando "python3 app.py" y se muetra la app, pero sólo es accesible desde 127.0.0.1, así que no puedo entrar porque el contenedor no tiene GUI
Me toca modificar esta línea de app.py mediante "nano app.py"
if __name__ == '__main__':
app.run(host='0.0.0.0', port=5000, debug=True)
Al poner 0.0.0.0 se da permiso para que otro equipo en la misma red de mi contenedor acceda.
La web carga, pero me falla al cargar archivos
Con ayuda de Gemini detecto que me fallan fuentes, así que desde CLI ejecuto:
apt install -y fonts-dejavu-core fonts-dejavu
En la siguiente prueba el programa dice que el uso de Google Api está obsoleta. Así que modifico app.py y video_processor.py para que utilice mis propias llm en mi servidor de ollama.
El flujo de trabajo de la app es el siguiente:
Carga el video en /uploads
Usa Whisper para analizar el audio y transcribirlo
Le pasa el audio transcrito a una llm que tengas (antes se lo pasaba a gemini, pero eso ya no funciona ni dándole tu API
Le pide a la LLM que analice el vídeo y saque 5clips de jugadas destacadas
Luego con los tiempos de los vídeos seleccionados, corta los mismos incluyendo el sonido y genera una web para poder verlos y descargarlos
Pego aquí el código de app.py final y de video_processor.py (los puedes pegar dentro de nano)
app.py
from flask import Flask, render_template, request, redirect, url_for, flash, send_from_directory
import os
from werkzeug.utils import secure_filename
import sqlite3
from datetime import datetime
from video_processor import VideoProcessor
import threading
from dotenv import load_dotenv
def adapt_datetime(ts):
return ts.isoformat()
def convert_datetime(val):
return datetime.fromisoformat(val)
# Register the adapter and converter
sqlite3.register_adapter(datetime, adapt_datetime)
sqlite3.register_converter("datetime", convert_datetime)
load_dotenv()
app = Flask(__name__)
app.secret_key = os.getenv('FLASK_SECRET_KEY', 'dev') # Change this in production
app.config['UPLOAD_FOLDER'] = os.path.abspath('uploads')
app.config['CLIPS_FOLDER'] = os.path.abspath('clips')
app.config['TRANSCRIPTS_FOLDER'] = os.path.abspath('transcripts')
app.config['DATABASE'] = os.path.abspath('videos.db')
app.config['MAX_CONTENT_LENGTH'] = 5 * 1024 * 1024 * 1024 # 5000MB max file size
app.config['ALLOWED_EXTENSIONS'] = {'mp4', 'avi', 'mov', 'mkv'}
# Create required directories if they don't exist
os.makedirs(app.config['UPLOAD_FOLDER'], exist_ok=True)
os.makedirs(app.config['CLIPS_FOLDER'], exist_ok=True)
os.makedirs(app.config['TRANSCRIPTS_FOLDER'], exist_ok=True)
# Database initialization
def init_db():
conn = sqlite3.connect(app.config['DATABASE'], detect_types=sqlite3.PARSE_DECLTYPES)
c = conn.cursor()
c.execute('''CREATE TABLE IF NOT EXISTS videos
(id INTEGER PRIMARY KEY AUTOINCREMENT,
filename TEXT,
upload_date DATETIME,
status TEXT)''')
c.execute('''CREATE TABLE IF NOT EXISTS clips
(id INTEGER PRIMARY KEY AUTOINCREMENT,
video_id INTEGER,
clip_path TEXT,
start_time REAL,
end_time REAL,
description TEXT,
FOREIGN KEY(video_id) REFERENCES videos(id))''')
conn.commit()
conn.close()
init_db()
@app.route('/')
def index():
return render_template('upload.html')
# Add after init_db()
@app.route('/upload', methods=['POST'])
def upload_video():
if 'video' not in request.files:
return redirect(url_for('index'))
file = request.files['video']
if file.filename == '':
return redirect(url_for('index'))
if not allowed_file(file.filename):
flash('File type not allowed')
return redirect(url_for('index'))
filename = secure_filename(file.filename)
filepath = os.path.join(app.config['UPLOAD_FOLDER'], filename)
file.save(filepath)
# Store in database
conn = sqlite3.connect(app.config['DATABASE'], detect_types=sqlite3.PARSE_DECLTYPES)
c = conn.cursor()
c.execute('INSERT INTO videos (filename, upload_date, status) VALUES (?, ?, ?)',
(filename, datetime.now(), 'uploaded'))
video_id = c.lastrowid
conn.commit()
conn.close()
# Start background processing
# processor = VideoProcessor()
threading.Thread(target=process_video_background, args=(video_id, filepath)).start()
return redirect(url_for('status', video_id=video_id))
def allowed_file(filename):
return '.' in filename and filename.rsplit('.', 1)[1].lower() in app.config['ALLOWED_EXTENSIONS']
def process_video_background(video_id, filepath):
conn = sqlite3.connect(app.config['DATABASE'], detect_types=sqlite3.PARSE_DECLTYPES)
c = conn.cursor()
def update_status(status):
print(f"[STATUS] {status}", flush=True)
c.execute('UPDATE videos SET status=? WHERE id=?', (status, video_id))
conn.commit()
try:
update_status('initializing')
try:
print("[INFO] Cargando VideoProcessor...", flush=True)
processor = VideoProcessor()
print("[INFO] VideoProcessor listo.", flush=True)
except Exception as e:
print(f"[ERROR] Error al instanciar VideoProcessor: {e}", flush=True)
update_status(f'error: Failed to initialize video processor: {str(e)}')
return
update_status('transcribing')
try:
print(f"[INFO] Procesando video {filepath}...", flush=True)
clips = processor.process_video(filepath)
print(f"[INFO] Clips devueltos: {len(clips)}", flush=True)
except Exception as e:
print(f"[ERROR] Error en process_video: {e}", flush=True)
update_status(f'error: Failed to process video: {str(e)}')
return
# Store clips in database
for clip in clips:
try:
c.execute('''INSERT INTO clips
(video_id, clip_path, start_time, end_time, description)
VALUES (?, ?, ?, ?, ?)''',
(video_id, clip['path'], clip['start'],
clip['end'], clip['description']))
except Exception as e:
update_status(f'error: Failed to save clip: {str(e)}')
return
update_status('processed')
conn.commit()
except Exception as e:
update_status(f'error: Unexpected error: {str(e)}')
finally:
conn.close()
@app.route('/status/<int:video_id>')
def status(video_id):
conn = sqlite3.connect(app.config['DATABASE'])
c = conn.cursor()
c.execute('SELECT v.status, v.filename FROM videos v WHERE v.id=?', (video_id,))
result = c.fetchone()
conn.close()
if not result:
return "Vídeo no encontrado", 404
current_status, filename = result
# Si ya terminó, redirigir de inmediato al dashboard
if current_status == 'processed':
return redirect(url_for('dashboard', video_id=video_id))
# Si sigue procesando, mostrar pantalla que se autorefresca cada 3 segundos
return f"""
<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<title>Procesando {filename}</title>
<meta http-equiv="refresh" content="3">
<style>
body {{ font-family: system-ui, sans-serif; display: flex; justify-content: center; align-items: center; height: 90vh; margin: 0; background: #0f172a; color: #f8fafc; }}
.card {{ text-align: center; padding: 2rem; background: #1e293b; border-radius: 12px; box-shadow: 0 4px 6px -1px rgba(0,0,0,0.5); }}
.spinner {{ margin: 1.5rem auto; width: 40px; height: 40px; border: 4px solid #334155; border-top-color: #38bdf8; border-radius: 50%; animation: spin 1s infinite linear; }}
@keyframes spin {{ to {{ transform: rotate(360deg); }} }}
</style>
</head>
<body>
<div class="card">
<h2>Analizando jugadas...</h2>
<p><strong>Archivo:</strong> {filename}</p>
<div class="spinner"></div>
<p>Estado actual: <code>{current_status}</code></p>
<small>La página se actualizará automáticamente al terminar.</small>
</div>
</body>
</html>
"""
@app.route('/clips/<path:filename>')
def serve_clip(filename):
return send_from_directory(app.config['CLIPS_FOLDER'], filename)
# Add after status route
@app.route('/dashboard/<int:video_id>')
def dashboard(video_id):
conn = sqlite3.connect(app.config['DATABASE'])
c = conn.cursor()
# Get video info
c.execute('SELECT * FROM videos WHERE id=?', (video_id,))
video = c.fetchone()
if not video:
flash('Video not found', 'error')
return redirect(url_for('index'))
# Get clips
c.execute('SELECT * FROM clips WHERE video_id=?', (video_id,))
clips = c.fetchall()
# Convert clip paths to URLs
formatted_clips = []
for clip in clips:
clip_filename = os.path.basename(clip[2])
clip_url = url_for('serve_clip', filename=clip_filename)
formatted_clips.append({
'id': clip[0],
'video_id': clip[1],
'url': clip_url,
'start_time': clip[3],
'end_time': clip[4],
'description': clip[5]
})
conn.close()
return render_template('dashboard.html',
video=video,
clips=formatted_clips)
if __name__ == '__main__':
app.run(host='0.0.0.0', port=5000, debug=True)
*******************************fin del fichero********************
video_processor.py
# filepath: c:\Users\Hariom kumar\Desktop\videocutterAI\video_processor.py
import os
import time
import whisper
import ffmpeg
import sqlite3
from datetime import datetime
from dotenv import load_dotenv
import ollama
load_dotenv()
class VideoProcessor:
def __init__(self, clips_folder='clips'):
# 1. Configuración de Ollama remoto
self.ollama_host = os.getenv('OLLAMA_HOST', 'http://192.168.1.206:11434')
self.ollama_model = os.getenv('OLLAMA_MODEL', 'llama3.2:3b')
try:
print(f"Conectando con Ollama en {self.ollama_host} (modelo: {self.ollama_model})...")
self.ollama_client = ollama.Client(host=self.ollama_host)
except Exception as e:
raise Exception(f"No se pudo inicializar el cliente de Ollama: {str(e)}")
# 2. Carga del modelo Whisper para CPU
max_retries = 3
retry_count = 0
last_error = None
while retry_count < max_retries:
try:
print(f"Cargando modelo Whisper (intento {retry_count + 1}/{max_retries})...")
self.transcription_model = whisper.load_model(
name="tiny",
device="cpu",
download_root=os.path.join(os.path.expanduser("~"), ".cache", "whisper")
)
print("Modelo Whisper cargado en memoria.")
break
except Exception as e:
last_error = str(e)
retry_count += 1
print(f"Error cargando Whisper: {str(e)}")
if retry_count < max_retries:
print("Reintentando en 5 segundos...")
time.sleep(5)
if retry_count == max_retries:
raise Exception(f"Fallo al cargar Whisper tras {max_retries} intentos: {last_error}")
self.clips_folder = clips_folder
os.makedirs(self.clips_folder, exist_ok=True)
def process_video(self, video_path):
print("Iniciando transcripción y análisis temporal del vídeo...")
transcript = self.transcribe_video(video_path)
print("Analizando los puntos y momentos destacados con Ollama...")
highlights = self.analyze_highlights(transcript)
print(f"Generando clips recortados ({len(highlights)} detectados)...")
clips = self.generate_clips(video_path, highlights)
return clips
def transcribe_video(self, video_path):
options = {
'fp16': False,
'language': 'es',
'task': 'transcribe',
'beam_size': 1,
'best_of': 1
}
result = self.transcription_model.transcribe(video_path, **options)
# Extraemos marcas temporales y duración de actividad sonora/vocal
structured_log = []
for seg in result.get('segments', []):
start = round(seg['start'], 1)
end = round(seg['end'], 1)
dur = round(end - start, 1)
text = seg['text'].strip()
mins = int(start // 60)
secs = int(start % 60)
timestamp_str = f"{mins:02d}:{secs:02d}"
structured_log.append(f"[{timestamp_str}] (Duración: {dur}s) {text}")
return "\n".join(structured_log)
def analyze_highlights(self, transcript):
if not transcript.strip():
print("Aviso: Whisper no detectó audio ni actividad vocal suficiente.")
return []
prompt = f"""Eres un analista experto de partidos de pádel. A continuación tienes la lista cronológica de secuencias de juego con sus marcas de tiempo [MM:SS], duración del intercambio y audio/comentarios detectados:
{transcript}
TAREA:
Identifica los 5 puntos más largos, disputados o intensos del partido. Si un intercambio largo abarca varias marcas consecutivas, incluye el rango completo desde que empieza el saque hasta que termina el punto.
Devuelve ÚNICAMENTE los 5 puntos, uno por línea, siguiendo estrictamente este formato con tiempo de inicio y fin:
[MM:SS - MM:SS] Descripción breve del punto
Ejemplo de respuesta esperada:
[02:15 - 03:10] Punto largo de 55 segundos con peloteo continuo y remate final
[08:40 - 09:15] Peloteo disputado en la red con varias voleas y celebración
"""
try:
print(f"Enviando transcripción a {self.ollama_model} en {self.ollama_host}...", flush=True)
response = self.ollama_client.chat(
model=self.ollama_model,
messages=[{'role': 'user', 'content': prompt}],
options={'temperature': 0.2}
)
response_text = response['message']['content']
print(f"Respuesta recibida de Ollama:\n{response_text}\n", flush=True)
return self.parse_gemini_response(response_text)
except Exception as e:
print(f"Error consultando Ollama: {str(e)}")
return []
def parse_gemini_response(self, response_text):
import re
highlights = []
lines = [line.strip() for line in response_text.split('\n') if line.strip()]
for line in lines:
# Buscar cualquier patrón de tiempo del tipo MM:SS o MM:SS - MM:SS / MM:SS a MM:SS
# Ejemplos: [01:30 - 02:15], [14:27 a 14:49], (14:27 - 14:49) o simplemente [01:30]
time_matches = re.findall(r'(\d{1,2}:\d{2})', line)
if not time_matches:
continue
try:
# Limpiar la descripción quitando corchetes y marcas de tiempo
desc = re.sub(r'\[.*?\]|\(.*?\)', '', line).strip()
if not desc:
desc = "Punto destacado del partido"
# CASO 1: Ollama devolvió dos marcas de tiempo (Inicio y Fin)
if len(time_matches) >= 2:
m1, s1 = map(float, time_matches[0].split(':'))
m2, s2 = map(float, time_matches[1].split(':'))
start_time = m1 * 60 + s1
end_time = m2 * 60 + s2
# CASO 2: Ollama solo dio una marca de tiempo
else:
m, s = map(float, time_matches[0].split(':'))
center_time = m * 60 + s
# Intentar leer si en el texto puso los segundos reales (ej: "de 25.9 segundos")
sec_match = re.search(r'(\d+(?:\.\d+)?)\s*(?:segundos|s)', line, re.IGNORECASE)
if sec_match:
dur = float(sec_match.group(1))
start_time = max(0, center_time - (dur / 2))
end_time = center_time + (dur / 2)
else:
# Si no especifica, asignar un punto estándar de 30 segundos
start_time = max(0, center_time - 15)
end_time = center_time + 15
# Aplicar margen de seguridad (3 segundos antes del saque y 3 tras el punto)
real_start = max(0, start_time - 3)
real_end = end_time + 3
# REGLA DE ORO: En pádel ningún punto destacado dura menos de 15 segundos
# Si la duración calculada es inferior a 15s, ampliar a 25s
if (real_end - real_start) < 15:
real_end = real_start + 25
highlights.append({
'start': real_start,
'end': real_end,
'description': desc
})
except Exception as e:
print(f"Error parseando línea: {line} -> {str(e)}")
continue
return highlights[:5]
def generate_clips(self, video_path, highlights):
if not highlights:
print("No se encontraron jugadas destacadas para recortar.")
return []
font_path = '/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf'
use_font = os.path.exists(font_path)
clips = []
for i, highlight in enumerate(highlights, 1):
try:
output_path = os.path.join(self.clips_folder, f"clip_{i}_{datetime.now().timestamp():.0f}.mp4")
print(f"Generando clip {i}/{len(highlights)}: {highlight['description'][:50]}...")
start = float(highlight['start'])
duration = float(highlight['end'] - highlight['start'])
# Leemos directamente del vídeo original aplicando el seek
in_file = ffmpeg.input(video_path, ss=start, t=duration)
# Procesamos el stream de vídeo (con o sin rótulo)
if use_font:
subtitle_text = highlight.get('description', f'Punto {i}')
v_stream = in_file.video.drawtext(
text=subtitle_text,
fontfile=font_path,
fontsize=22,
fontcolor='white',
borderw=2,
x='(w-text_w)/2',
y='h-50'
)
else:
v_stream = in_file.video
# Mapeamos explícitamente el stream de audio
a_stream = in_file.audio
# Renderizamos en un único paso con codecs estándar para navegadores
(
ffmpeg.output(
v_stream,
a_stream,
output_path,
vcodec='libx264',
acodec='aac',
audio_bitrate='192k',
ac=2, # Forzar 2 canales estéreo
pix_fmt='yuv420p', # Máxima compatibilidad HTML5
preset='fast',
movflags='+faststart' # Permite reproducción web sin descargar el archivo completo
)
.overwrite_output()
.run(quiet=True)
)
clips.append({
'path': output_path,
'start': highlight['start'],
'end': highlight['end'],
'description': highlight['description']
})
except Exception as e:
print(f"Error procesando el clip {i}: {str(e)}")
continue
return clips
*******************************Fin del fichero ************************
Nota: Ojo que en la línea
self.ollama_host = os.getenv('OLLAMA_HOST', 'http://192.168.1.206:11434')
Yo he incluido a fuego la dirección de m servidor de ollama con varios llm, en este caso suando llama3.2:3b
En tu caso tendrás que redirigirlo a tu servidor de ollama, que es quien realiza el trabajo de analizar lo transcrito.
Espero que esta modificación del proyecto os pueda servir para editar fácilmente vuestros vídeos de pádel.
Modificando el prompt se podría editar para fútbol u otros deportes, incluso para detectar palabras específicas dichas durante el partido.
Saludos a todos
