90 lines
4.8 KiB
Python
90 lines
4.8 KiB
Python
"""Bounded local media analysis. No HTTP, shell interpolation, or model-generated file paths.
|
|
Scene boundaries: PySceneDetect. Beats: librosa. ASR: FunASR sentence timestamps,
|
|
as used by FunClip. Silence rough-cut uses an independently implemented RMS gate.
|
|
"""
|
|
import contextlib
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
for key in ('OMP_NUM_THREADS', 'OPENBLAS_NUM_THREADS', 'MKL_NUM_THREADS', 'NUMBA_NUM_THREADS'):
|
|
os.environ[key] = '2'
|
|
|
|
def capabilities():
|
|
import shutil
|
|
return {'scenes': importlib.util.find_spec('scenedetect') is not None,
|
|
'beats': importlib.util.find_spec('librosa') is not None,
|
|
'asr': importlib.util.find_spec('funasr') is not None and all(Path(os.environ.get(k, '/nonexistent')).is_dir() for k in ('SMART_EDIT_ASR_MODEL','SMART_EDIT_VAD_MODEL','SMART_EDIT_PUNC_MODEL')),
|
|
'ffmpeg': shutil.which('ffmpeg') is not None,
|
|
'ffprobe': shutil.which('ffprobe') is not None}
|
|
|
|
def active_regions(wave, sr, threshold_db=-35):
|
|
import numpy as np
|
|
hop=max(1,round(sr*.02))
|
|
energy=np.array([np.sqrt(np.mean(wave[i:i+hop]**2)) for i in range(0,len(wave),hop)])
|
|
active=energy > 10 ** (threshold_db/20)
|
|
intervals=[]; start=None
|
|
for i,keep in enumerate(active):
|
|
if keep and start is None:start=i*hop/sr
|
|
if start is not None and (not keep or i==len(active)-1):
|
|
end=(i+1 if keep else i)*hop/sr
|
|
intervals.append([max(0,start-.2),min(len(wave)/sr,end+.2)]);start=None
|
|
merged=[]
|
|
for start,end in intervals:
|
|
if merged and start-merged[-1][1]<.3:merged[-1][1]=end
|
|
else:merged.append([start,end])
|
|
return [[round(a*1e6),round(b*1e6)] for a,b in merged if b-a>=.2]
|
|
|
|
def analyze(data):
|
|
result={'scenes':[], 'active':[], 'subtitles':[], 'beats':[], 'warnings':[]}
|
|
duration=data.get('duration',0)
|
|
if data.get('video'):
|
|
import cv2
|
|
cv2.setNumThreads(2)
|
|
from scenedetect import detect,AdaptiveDetector
|
|
result['scenes']=[[round(a.get_seconds()*1e6),round(b.get_seconds()*1e6)] for a,b in detect(data['video'],AdaptiveDetector(min_scene_len=15),start_in_scene=True)] or [[0,duration]]
|
|
if data.get('audio'):
|
|
import soundfile as sf
|
|
import numpy as np
|
|
wave,sr=sf.read(data['audio'],dtype='float32')
|
|
if wave.ndim>1:wave=wave.mean(axis=1)
|
|
result['active']=active_regions(wave,sr,data.get('thresholdDb',-35))
|
|
if data.get('beats'):
|
|
import librosa
|
|
tempo,beats=librosa.beat.beat_track(y=wave,sr=sr,units='time')
|
|
result['beats']=[round(float(b)*1e6) for b in beats]
|
|
if len(result['beats'])<3:
|
|
result['beats']=[];result['warnings'].append('未检测到稳定音乐节拍，将保留普通切镜。')
|
|
if data.get('transcribe'):
|
|
if not capabilities()['asr']:raise RuntimeError('本地语音识别尚未配置；请安装 ASR 依赖及三个模型，或关闭自动转写并导入 SRT 字幕。')
|
|
import torch
|
|
from funasr import AutoModel
|
|
torch.set_num_threads(2)
|
|
model=AutoModel(model=os.environ['SMART_EDIT_ASR_MODEL'],vad_model=os.environ['SMART_EDIT_VAD_MODEL'],punc_model=os.environ['SMART_EDIT_PUNC_MODEL'],device='cpu',disable_update=True,trust_remote_code=False)
|
|
outputs=model.generate(input=data['audio'],sentence_timestamp=True,batch_size_s=30)
|
|
for output in outputs:
|
|
for sentence in output.get('sentence_info',[]):
|
|
start=sentence.get('start');end=sentence.get('end')
|
|
if start is None or end is None:
|
|
times=sentence.get('timestamp') or []
|
|
if times:start,end=times[0][0],times[-1][1]
|
|
text=str(sentence.get('text','')).strip()
|
|
if text and start is not None and end is not None and end>start:
|
|
result['subtitles'].append({'start':round(start*1000),'end':round(end*1000),'text':text})
|
|
if not result['subtitles']:result['warnings'].append('语音识别没有返回有效语句，将按画面镜头处理；不会生成虚构字幕。')
|
|
elif data.get('transcribe'):result['warnings'].append('素材没有音轨，跳过语音转写。')
|
|
return result
|
|
|
|
if __name__=='__main__':
|
|
if sys.argv[1:] == ['--check']:
|
|
print(json.dumps(capabilities()));sys.exit(0)
|
|
request_path,output_path=sys.argv[1:3]
|
|
try:
|
|
data=json.loads(Path(request_path).read_text())
|
|
with contextlib.redirect_stdout(sys.stderr):result=analyze(data)
|
|
Path(output_path).write_text(json.dumps(result,ensure_ascii=False))
|
|
except Exception as error:
|
|
Path(output_path).write_text(json.dumps({'error':str(error)},ensure_ascii=False));sys.exit(1)
|