借助AI打造高效短视频工作流，每日提取2万+爆款视频文案

pip install faster-whisper

from faster whisper import whisperModel
model size ="large-v3"
# Run on GPU with Fp16
model = WhisperModel(model size, device="cuda", compute type="float16")
# or run on GpU with INT8# model = WhisperModel(model size, device="cuda", compute type="int8 float16")
# or run on cpu with INT8
# model = WhisperModel(model size, device="cpu", compute type="int8")
segments,info = model.transcribe("audio.mp3", beam size=5)
print("Detected language '%s’with probability %f" % (info.language, info.language probability))
for segment in segments:
	print("[%.2fs ->%.2fs]%s"%(segment.start, segment.end, segment.text))

conda create -n sensevoice python=3.8
conda activate sensevoice

pip install -r requirements.txt

export SENSEVOICE DEVICE=cuda:0
fastapi run --port 50000

def whisper(file path):
	#在 GPU 上使用 FP16 运行 faster-whisper-large-v3 模型
	model = whisperModel("E: fastwhisper modelllfaster-whisper-large-v3", device="cuda", compute type="float16"
	segments,info = model.transcribe(file path, beam size=5)
	print("Detected language '%s'with probability %f" % (info.language, info.language probability))
	texts =[]
	for segment in segments:
		texts.append(segment.text)
		print(segment.text)
	full_transcript ='\n'.join(texts)
	return full_transcript

import sqlite3
from datetime import datetime
import requests
import os
from tqdm import tqdm
import subprocess
from datetime import datetime
from concurrent.futures import ThreadPoolExecutor, as completed

def get video url(link):
	url = "http:/localhost:8666"
	response_content_body = requests.request("posT", url, data={'url': link})
	res_data=response_content_body.json()
	
	url = res data['video url']
	return url
import shutil
def convert_to_mp3(video path, output dir='downloads )
	"""
	使用 ffmpeg 将视频转换为 mp3
	:param video_path:视频文件路径
	:param output_dir:输出目录
	:return:mp3 文件路径Hu He Go
	"""
	# 确保输出目录存在
	os.makedirs(output dir,exist ok=True)

	#生成输出文件名
	video_filename =os.path.basename(video path)
	mp3_filename = os.path.splitext(video filename)[0] +'.mp3
	mp3_path =os.path.join(output dir, mp3 filename)\
	
	# 检查 MP3 文件是否已存在
	if os.path.exists(mp3_path):
		print(f"MP3 文件已存在:{mp3_path}")
		return mp3 path
	ffmpeg_dir = r"E:\ffmpeg-6.1.1-full build\bin"
	os.environ["PATH"]= ffmpeg dir + os.pathsep + os.environ["PATH"]
	
	# 构建 ffmpeg 命令
	command =[
		ffmpeg_path,
		'-i', video_path,
		'-vn'， #不包含视频
		'-acodec'，'libmp3lame', # 使用 MP3 编码器
		'-q:a'，'2', # 音频质量，2 是较好的质量
		mp3_path
	]
	try:
		# 执行 ffmpeg 命令
		print(f"执行命令:{''.join(command)}")result = subprocess.run(comand, check=True, stdout=subproces,PIpE, stderr=subprocess,PIpE, text=True, encoding='utf-8’, errors-'replace')
		print(f"命令输出:\n{result.stdout}\n{result.stderr}")
		print(f"已成功将视频转换为 MP3:{mp3_path}")
		return mp3_path
	except subprocess.CalledProcessError as e:
		print(f"转换失败:fe}")
		print(f"错误输出:\nfe.stdout}\nfe.stderr}")
		return None
	except FileNotFoundError:
		print("错误:尽管在系统路径中找到了 ffmpeg，但无法执行。请检查文件权限。")
		return None
		
def download_video(video info, output dir='downloads ):
	"""
	根据给定的URL下载视频并保存到指定路径，同时显示下载进度
	如果文件已存在，则直接返回文件路径
	
	:param video info:包含视频信息的字典
	:param output dir:保存视频的目录，默认为'downloads
	:return:保存的文件路径
	"""
	url = video info['url']
	title = video info['title']
	
	# 创建安全的文件名，并限制长度
	safe_title = "".join([c for c in title if c.isalnum()or c in(’','',' ')]).rstrip()
	safe_title = safe title[:10] # 限制标题长度为10个字符
	output_path = os.path.join(output dir, f"fsafe title}.mp4")
	
	# 如果文件名仍然太长，使用哈希值作为文件名
	if len(output path) > 255:
		import hashlib
		hash_object = hashlib.md5(title.encode())
		safe_title = hash object.hexdigest()
		output_path = os.path.join(output_dir, f"{safe_title}.mp4")
		
	# 确保输出路径是绝对路径
	output_path = os.path.abspath(output_path)
	
	# 检查文件是否已存在
	if os.path.exists(output_path):
		print(f"文件'{output_path}'已存在，跳过下载。")
		return output_path
	headers={
		'User-Agent': 'Mozilla/5.8 (Windows NT 10.8; Win64; x64)AppleWebKit/537.36 (KHTML, like Gecko) chrome/91.0.4472.124 safari/537.36'
	}
	
	response = requests.get(url,stream=True, headers=headers)
	response.raise for status() #如果请求不成功则抛出异常
	
	# 获取文件大小
	total_size =int(response.headers.get('content-length',0))
	
	#确保输出目录存在
	output_dir = os.path.dirname(output_path)
	if output_dir:
		os.makedirs(output dir,exist ok=True)
		
	# 使用tgdm创建进度条
	with open(output path,wb')as file,tqdm(
		desc=os.path.basename(output_path),
		total=total_size,
		unit='iB',
		unit_scale=True,
		unit_divisor=1024,
	) as progress_bar:
		for data in response.iter content(chunk_size=1024):
			size = file.write(data)
			progress_bar.update(size)
		return output_path

# 新增函数:从数据库读取aweme id和link
def get_video links from db(db path, limit=800):
	"""
	从sQLite数据库中读取aweme id和link
	:param db_path:数据库文件路径
	:param limit:限制读取的记录数,默认为10
	:return:包含aweme id和link的列表	
	"""

	conn=sqlite3.connect(db path)
	cursor =conn.cursor()
	cursor.execute("SELEcT aweme_id, link FRoM videos WHERE video_path IS NULL LIMIT ?",(limit,))
	results = cursor.fetchall()
	conn.close()
	return results
	
def update_video _paths(db_path, aweme id, video path, mp3_path):
	"""
	更新数据库中指定aweme id的视频和音频路径
	:param db_path:数据库文件路径
	:param aweme_id:视频的aweme id
	:param video_path:下载的视频文件路径
	:param mp3_path:生成的MP3文件路径
	"""

	conn =sglite3.connect(db path)
	cursor = conn.cursor()
	cursor.execute("UpDATE videos SET video_path =?, mp3_path = ? WHERE aweme_id = ?",(video_path, mp3_path, aweme_id))
	conn.commit()
	conn.close()
def process video(db_path, aweme id, link):
	try:
		link = link.replace('douyin.com','iesdouyin.com')print(f"处理链接:{link}")
		video_url= get_video_url(link)
		
		if video url:
			downloaded file = download video(video url)
			# 转换视频为 MP3
			mp3 file = convert to mp3(downloaded file)
			if mp3 file:
				print(f"MP3 文件已生成:{mp3_file}")
				print("开始调用:"，datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
				# 更新数据库中的视频和音频路径
				update video paths(db path, aweme id, downloaded file, mp3 file)
		else:
			print("无法获取视频信息")
			
		print("_"*50) # 分隔不同链接的处理结果
	except Exception as e:
		print(f"处理链接 {link} 时发生错误:{e}")
		
# 修改主程序
def main(max workers=5):
	db path ="douyin data.db"
	video_links = get_video_links from db(db path)
	with ThreadPoolExecutor(max workers=max workers) as executor:
		futures = [executor.submit(process_video, db path, aweme_id, link) for aweme id, link in video_links]
	
		for future in as completed(futures):
			future.result() #这里可以处理每个任务的结果，如果需要的话
			
if __name__ == "__main__"
	main(max workers=10)#设10个并发线程

import requests
import os
from tqdm import tgdm
from faster whisper import whisperModel
from datetime import datetime
import sqlite3
from concurrent.futures import ThreadPoolExecutor
import concurrent.futures
from datetime import datetime

def call_asr_api(file path, language="auto"):
	# API端点
	url ="http://localhost:8000/transcribe/
	# 准备文件
	with open(file path,'rb')as audio file:
		files =f'audio file':(os.path.basename(file path), audio file)}
		#发送POST请求
		response =requests.post(url,files=files)
	# 检查响应
	if response.status code == 200:
		return response.json()["transcription"]
	else:
		return f"错误:{response.status code,{response.text}"
		
def process_video(aweme id,mp3 path):
	print(f"处理视频 ID:{aweme id}")
	if os.path.exists(mp3 path):
		try:
			full text = call asr api(mp3 path)
			print("转录文本为:"，full text)
			transcript_time = datetime.now().strftime("%Y-%m-%d %H:%M:%")
			return aweme_id, full_text,transcript_time
		except Exception as e:
			print(f"处理视频 ID {aweme_id} 时出错:{str(e)}")
			return aweme id, None, None
else:
	print(f"MP3 文件不存在:{mp3_path}")
	return aweme_id, None, None
	
def process_videos from db(max workers=5):
	db_path='douyin data.db'
	conn =sqlite3.connect(db path)
	cursor = conn.cursor()
	try:
		cursor,execute("sElEcT aweme id, mp3 path FRoM videos WHERE (mp3 path is not null or mp3 path!='') and ( transcript Is NulL oR transcript = '')")
		videos =cursor.fetchall()
		
	with ThreadPoolExecutor(max workers=max workers)as executor:
		future_to_video = fexecutor,submit(process video, aweme id, mp3 path): (aweme id, mp3 path) for aweme id, mp3 path in videos}
		
		for future in concurrent.futures.as completed(future to video):
			aweme_id, full_text,transcript_time = future.result()
			if full_text is not None:
				cursor.execute("""
					UPDATE videos
					SET transcript =?,transcript_time = ?
					WHERE aweme id =?
				""",(full_text, transcript_time, aweme_id))
				conn.commit()
				print(f"视频 ID faweme id} 转录完成并已更新数据库")
			print("=”* 50)
			
	finally:
		conn.close()
def main():
	max_workers=5	# 设置并发线程数
	print("开始转录:"，datetime.now().strftime("%-%m-%d %H:%M:%S"))
	process_videos_from_db(max workers)
	print("结束转录:"，datetime.now().strftime("%-%m-%d %H:%M:%S"))

if __name__ == "__main__"
	main()