百度360必应搜狗淘宝本站头条
当前位置:网站首页 > 技术文章 > 正文

comfyui-关于自定义插件-音频转文本

itomcoil 2024-12-23 11:08 23 浏览

效果图


实现方式:

工作流说明:

视频下载:通过上传url从网上下载视频

路径分支:下载后的视频分两个路径完成流转。

一个是视频转图片

一个是音频转文本

主要的节点类,如下图目录结构:


源码分析

音频转文本.py

import os
import torch
import whisper
import subprocess
import json
from datetime import timedelta
import time
import sys

class AudioTranscriber:
    def __init__(self):
        print("\n[初始化] 音频转文本节点被创建")
        self.execution_count = 0
        
        # 创建输出目录
        self.output_dir = "transcribed_text"
        os.makedirs(self.output_dir, exist_ok=True)
        
        # 检查CUDA可用性
        self.device = "cuda" if torch.cuda.is_available() else "cpu"
        print(f"[初始化] 使用设备: {self.device}")
        
        # 加载whisper模型
        self.model = None
        self.current_model_size = None

    def save_text_file(self, text, video_name):
        """保存转录文本到文件"""
        try:
            # 生成文件名
            timestamp = time.strftime("%Y%m%d_%H%M%S")
            filename = os.path.join(self.output_dir, f"{video_name}_{timestamp}.txt")
            
            # 使用UTF-8编码保存文件
            with open(filename, 'w', encoding='utf-8') as f:
                f.write(text)
            
            print(f"[保存] 文本已保存至: {filename}")
            return filename
        except Exception as e:
            print(f"[错误] 保存文件失败: {str(e)}")
            raise ValueError(f"保存文件失败: {str(e)}")

    @classmethod
    def INPUT_TYPES(s):
        return {
            "required": {
                "视频路径_音频": ("STRING", {"forceInput": True, "multiline": False}),
                "模型大小": (["tiny", "base", "small", "medium", "large"], {
                    "default": "base"
                }),
            },
            "optional": {
                "语言": (["auto", "Chinese", "English", "Japanese", "Korean"], {
                    "default": "Chinese"
                }),
                "时间戳": ("BOOLEAN", {"default": True}),
            }
        }

    RETURN_TYPES = ("STRING",)
    RETURN_NAMES = ("文本文件路径",)
    FUNCTION = "transcribe_audio"
    CATEGORY = "视频转文章"
    OUTPUT_NODE = True

    def verify_video_file(self, video_path):
        """验证视频文件"""
        if not video_path:
            raise ValueError("视频路径为空")
        if not os.path.exists(video_path):
            raise ValueError(f"视频文件不存在: {video_path}")
        if not os.path.isfile(video_path):
            raise ValueError(f"不是文件: {video_path}")
        try:
            size = os.path.getsize(video_path)
            print(f"[验证] 视频文件大小: {size} 字节")
            with open(video_path, 'rb') as f:
                print("[验证] 视频文件可访问")
        except Exception as e:
            raise ValueError(f"视频文件访问失败: {str(e)}")

    def extract_audio(self, video_path):
        """从视频中提取音频"""
        try:
            # 生成音频文件名
            video_name = os.path.splitext(os.path.basename(video_path))[0]
            # timestamp = time.strftime("%Y%m%d_%H%M%S")
            audio_path = os.path.join(self.output_dir, f"{video_name}.wav")
            
            print(f"[音频] 输出路径: {audio_path}")
            
            # 构建FFmpeg命令
            command = [
                'ffmpeg', '-i', video_path,
                '-vn',  # 不处理视频
                '-acodec', 'pcm_s16le',  # 设置音频编码
                '-ar', '16000',  # 设置采样率
                '-ac', '1',  # 设置为单声道
                '-y',  # 覆盖已存在的文件
                audio_path
            ]
            
            print("[音频] 执行FFmpeg命令...")
            print(f"[音频] 命令: {' '.join(command)}")
            
            # 修改subprocess调用,指定编码
            startupinfo = None
            if os.name == 'nt':  # Windows系统
                startupinfo = subprocess.STARTUPINFO()
                startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
            
            # 执行命令
            process = subprocess.Popen(
                command,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                startupinfo=startupinfo,
                encoding='utf-8',
                errors='replace'  # 处理无法解码的字符
            )
            
            # 获取输出
            stdout, stderr = process.communicate()
            
            # 检查结果
            if process.returncode == 0 and os.path.exists(audio_path):
                size = os.path.getsize(audio_path)
                print(f"[音频] 提取成功,大小: {size} 字节")
                if stderr:
                    print(f"[警告] FFmpeg输出: {stderr}")
                return audio_path
            else:
                print(f"[错误] FFmpeg返回码: {process.returncode}")
                print(f"[错误] FFmpeg输出: {stderr}")
                raise ValueError("音频提取失败")
                
        except Exception as e:
            print(f"[错误] 音频提取失败: {str(e)}")
            print(f"[错误] 异常类型: {type(e)}")
            print(f"[错误] 堆栈信息: ", sys.exc_info())
            raise ValueError(f"音频提取失败: {str(e)}")

    def load_model(self, model_size):
        """加载模型并处理设备选择"""
        try:
            print(f"[模型] 开始加载 {model_size} 模型...")
            model = whisper.load_model(model_size)
            
            # 如果有GPU,将模型移到GPU
            if self.device == "cuda":
                print("[模型] 将模型移至GPU")
                model = model.to(self.device)
                # 使用半精度以提高性能
                if torch.cuda.is_available():
                    model = model.half()
            else:
                print("[模型] 在CPU上运行模型")
                model = model.float()  # 确保在CPU上使用FP32
            
            print(f"[模型] 加载完成,使用设备: {self.device}")
            return model
            
        except Exception as e:
            print(f"[错误] 模型加载失败: {str(e)}")
            raise ValueError(f"模型加载失败: {str(e)}")

    def transcribe_audio(self, 视频路径_音频, 模型大小, 语言="Chinese", 时间戳=True):
        self.execution_count += 1
        try:
            print("\n========== 音频转文本开始 ==========")
            print(f"[执行] 第 {self.execution_count} 次执行")
            print(f"[时间] {time.strftime('%Y-%m-%d %H:%M:%S')}")
            print(f"[设备] 使用 {self.device.upper()}")
            
            # 验证输入
            print("\n[输入参数]")
            print(f"视频路径: {视频路径_音频}")
            print(f"模型大小: {模型大小}")
            print(f"语言: {语言}")
            print(f"时间戳: {时间戳}")
            
            if not 视频路径_音频:
                raise ValueError("未接收到视频路径")

            # 标准化路径
            video_path = os.path.abspath(视频路径_音频)
            print(f"[路径] 标准化后: {video_path}")
            
            # 验证视频文件
            print("\n[视频验证]")
            self.verify_video_file(video_path)
            
            # 提取音频
            print("\n[音频提取]")
            audio_path = self.extract_audio(video_path)
            
            # 加载模型
            print("\n[模型加载]")
            if self.model is None or self.current_model_size != 模型大小:
                self.model = self.load_model(模型大小)
                self.current_model_size = 模型大小
            else:
                print("[模型] 使用已加载的模型")
            
            # 转录音频
            print("\n[转录]")
            transcribe_options = {
                "language": 语言 if 语言 != "auto" else None,
                "task": "transcribe",
                "fp16": torch.cuda.is_available() and self.device == "cuda"
            }
            
            # 添加性能监控
            start_time = time.time()
            result = self.model.transcribe(audio_path, **transcribe_options)
            end_time = time.time()
            
            print(f"[性能] 转录耗时: {end_time - start_time:.2f} 秒")
            if self.device == "cuda":
                print(f"[性能] GPU内存使用: {torch.cuda.max_memory_allocated() / 1024**2:.2f} MB")
            
            # 处理结果
            print("\n[处理结果]")
            if 时间戳:
                text_content = []
                for segment in result["segments"]:
                    start = str(timedelta(seconds=round(segment["start"])))
                    end = str(timedelta(seconds=round(segment["end"])))
                    text_content.append(f"[{start} -> {end}] {segment['text']}")
                text_content = "\n".join(text_content)
            else:
                text_content = result["text"]
            
            # 保存文本到文件
            video_name = os.path.splitext(os.path.basename(video_path))[0]
            text_file_path = self.save_text_file(text_content, video_name)
            
            print("\n[完成]")
            print(f"音频文件: {audio_path}")
            print(f"文本文件: {text_file_path}")
            print("========== 音频转文本完成 ==========\n")
            
            return (text_file_path,)  # 返回文本文件路径

        except Exception as e:
            print("\n[异常信息]")
            print(f"类型: {type(e)}")
            print(f"描述: {str(e)}")
            print("========== 音频转文本异常终止 ==========\n")
            raise ValueError(f"音频转录失败: {str(e)}")

    @classmethod
    def IS_CHANGED(cls, 视频路径_音频, *args):
        """确保节点每次都执行"""
        print(f"[IS_CHANGED] 输入: {视频路径_音频}")
        import random
        return random.random()
 

文本预览.py

# -*- coding: utf-8 -*-
import os
import sys
import time
import json

class TextPreview:
    def __init__(self):
        print("\n[初始化] 文本预览节点被创建")
        print(f"[初始化] 时间: {time.strftime('%Y-%m-%d %H:%M:%S')}")
        print(f"[初始化] 工作目录: {os.getcwd()}")
        self.execution_count = 0

    @classmethod
    def INPUT_TYPES(s):
        return {
            "required": {
                "文本文件路径": ("STRING", {"forceInput": True}),
            },
            "optional": {
                "预览长度": ("INT", {
                    "default": 1000,
                    "min": 100,
                    "max": 5000,
                    "step": 500
                }),
            }
        }

    RETURN_TYPES = ()
    FUNCTION = "preview_text"
    CATEGORY = "视频转文章"
    OUTPUT_NODE = True

    def format_text(self, text, length):
        """格式化文本内容"""
        try:
            print(f"\n[格式化] 开始格式化文本")
            print(f"[格式化] 原始文本长度: {len(text)}")
            print(f"[格式化] 目标预览长度: {length}")
            
            # 截取指定长度
            preview = text[:length]
            
            # 添加基本信息
            info = [
                "=== 文本预览 ===",
                f"总字数: {len(text)}",
                f"预览字数: {min(length, len(text))}",
                f"文件: {os.path.basename(self.current_file)}",
                "=" * 50,
                preview,
                "=" * 50
            ]
            
            if len(text) > length:
                info.append("(更多内容请查看原文件)")
            
            formatted = "\n".join(info)
            print(f"[格式化] 完成,格式化后长度: {len(formatted)}")
            return formatted
            
        except Exception as e:
            print(f"[错误] 格式化失败: {str(e)}")
            print(f"[错误] 异常类型: {type(e)}")
            print(f"[错误] 堆栈信息: ", sys.exc_info())
            return f"格式化失败: {str(e)}"

    def preview_text(self, 文本文件路径, 预览长度=1000):
        self.execution_count += 1
        try:
            print("\n========== 文本预览开始 ==========")
            print(f"[执行] 第 {self.execution_count} 次执行")
            print(f"[时间] {time.strftime('%Y-%m-%d %H:%M:%S')}")
            print(f"[输入] 文件路径: {文本文件路径}")
            print(f"[输入] 预览长度: {预览长度}")
            
            # 验证文件
            if not os.path.exists(文本文件路径):
                raise ValueError(f"文件不存在: {文本文件路径}")
            
            # 保存当前文件路径
            self.current_file = 文本文件路径
            print(f"[状态] 当前处理文件: {os.path.basename(self.current_file)}")
            
            # 读取文件
            print("[流程] 开始读取文件...")
            with open(文本文件路径, 'r', encoding='utf-8') as f:
                text = f.read()
            file_size = os.path.getsize(文本文件路径)
            print(f"[文件] 大小: {file_size} 字节")
            print(f"[文件] 字符数: {len(text)}")
            
            # 格式化文本
            print("[流程] 开始格式化文本...")
            formatted_text = self.format_text(text, 预览长度)
            print("[流程] 格式化完成")
            
            print("\n[预览输出]")
            print("-" * 50)
            print(formatted_text)
            print("-" * 50)
            
            print("========== 文本预览完成 ==========\n")
            
            # 返回预览内容
            return {
                "ui": {
                    "value": formatted_text,
                    "type": "text",
                    "multiline": True,
                    "scroll": True,
                    "height": "500px"
                }
            }

        except Exception as e:
            error_msg = f"预览失败: {str(e)}"
            print("\n[异常信息]")
            print(f"[错误] 类型: {type(e)}")
            print(f"[错误] 描述: {str(e)}")
            print(f"[错误] 堆栈: ", sys.exc_info())
            print("========== 文本预览异常终止 ==========\n")
            return {
                "ui": {
                    "value": error_msg,
                    "type": "text",
                    "multiline": True,
                    "color": "red"
                }
            }

    @classmethod
    def IS_CHANGED(cls, 文本文件路径, *args):
        """确保节点每次都执行"""
        print(f"[IS_CHANGED] 被调用: {文本文件路径}")
        import random
        return random.random()

有不清楚的,在评论里留言,有时间我就会关注的。

分享我的bilibili视频:

【双色球自动采集发送到微信coze工作流使用教程】 https://www.bilibili.com/video/BV1YVqCYCEjV/?share_source=copy_web&vd_source=4f96fddc82e951a8caff5c703881b8da

相关推荐

PS小技巧 调整命令,让人物肤色变得更加白皙 #后期修图

我们来看一下如何去将人物的皮肤变得更加的白皙。·首先选中图层,Ctrl键加J键复制一层。·打开这里的属性面板,选择快速操作删除背景,这样就会将人物进行单独的抠取。·接下来在上方去添加一个黑白调整图层,...

把人物肤色提亮的方法和技巧

PS后期调白肤色提亮照片的方法。一白遮百丑,所以对于Photoshop后期来说把人物肤色调白是一项非常重要的任务。就拿这张素材图片来说,这张素材图片人脸的肤色主要偏红、偏黄,也不够白皙,该怎样对它进行...

《Photoshop教程》把美女图片调成清爽色彩及润肤技巧

关注PS精品教程,每天不断更新~~室内人物图片一般会偏暗,人物脸部、肤色及背景会出现一些杂点。处理之前需要认真的给人物磨皮及美白,然后再整体润色。最终效果原图一、用修补工具及图章工具简单去除大一点的黑...

PS后期对皮肤进行美白的技巧

PS后期进行皮肤美白的技巧。PS后期对皮肤进行美白的技巧:·打开素材图片之后直接复制原图。·接下来直接点击上方的图像,选择应用图像命令。·在通道这里直接选择红通道,混合这里直接选择柔光,然后点击确定。...

493 [PS调色]调模特通透肤色

效果对比:效果图吧:1、光位图:2、拍摄参数:·快门:160;光圈:8;ISO:1003、步骤分解图:用曲线调整图层调出基本色调。用可选颜色调整图层调整红色、黄色、白色和灰色4种颜色的混合比例。用色彩...

先选肤色再涂面部,卡戴珊的摄影师透露:为明星拍完照后怎么修图

据英国媒体12月17日报道,真人秀明星金·卡戴珊终于承认,她把女儿小北P进了家族的圣诞贺卡,怪不得粉丝们都表示这张贺卡照得非常失败。上周,这位39岁的女星遭到了一些粉丝针对这张照片的批评,她于当地时间...

如何在PS中运用曲线复制另一张照片的色调

怎样把另一张作品的外观感觉,套用到自己的照片上?单靠肉眼来猜,可能很不容易,而来自BenSecret的教学,关键是在PS使用了两个工具,让你可以准确比较两张照片的曝光、色调与饱和度,方便你调整及复制...

PS在LAB模式下调出水嫩肤色的美女

本PS教程主要使用Photoshop使用LAB模式调出水嫩肤色的美女,教程调色比较独特。作者比较注重图片高光部分的颜色,增加质感及肤色调红润等都是在高光区域完成。尤其在Lab模式下,用高光选区调色后图...

在Photoshop图像后期处理中如何将人物皮肤处理得白皙通透

我们在人像后期处理中,需要将人物皮肤处理的白皙通透,处理方法很多,大多数都喜欢使用曲线、磨皮等进行调整,可以达到亮但是不透,最终效果往往不是很好,今天就教大家一种如何将任务皮肤处理得白皙通透,希望能帮...

PS调色自学教程:宝宝照片快速调通透,简单实用!

PS调色自学教程:宝宝照片快速调通透。·首先复制图层,然后选择进入ACR滤镜,选择曲线锁定照片的亮部,也就高光位置,其他部位补亮一点,尤其是阴影的部位补亮多一些,让画面的层次均匀一点。·然后回到基本项...

【干货】如何利用PS进行人物美化

人物图像美化在Photoshop中非常常用,Photoshop作为一款功能强大的图像处理软件,不仅可以对人像进行基本的调色、美化和修复等处理,还可以改变人物的线条和幅度,如调整脸部器官和脸型的大小、调...

教大家一种可以快速把肤色处理均匀的方法@抖音短视频

快速把肤色处理均匀的方法。今天教大家一种可以快速把肤色处理均匀的方法。像这张照片整体肤色走紫红色,但是局部偏黄缘处理起来非常的麻烦。其实我们只需要新建空白图层,图层混合模式更改为颜色,再选择画笔工具把...

PS调色教程 利用RAW调出干净通透的肤色

要么不发,要么干货。后期教程来噜~用RAW调出干净通透的肤色。这次终于不会原片比PS后好看了吧。如果你依然这么觉得,请不要残忍的告诉我这个事实,泪谢TAT)附送拍摄花絮,感谢各位的支持更多风格请关注m...

photoshop后期皮肤变白的技巧

PS后期皮肤变白的技巧。1.PS后期让皮肤变白的方法有很多种,接下来教你一种非常简单容易上手的方法。2.打开素材图片之后,直接在小太极下拉框的位置添加一个纯色调整图层,颜色设置一个纯白色,点击...

Photoshop调出人物的淡雅粉嫩肤色教程

本教程主要使用Photoshop调出人物的淡雅粉嫩肤色教程,最终的效果非常的通透迷人,下面让我们一起来学习.出自:86ps效果图:原图:1、打开原图复制一层。2、用Topaz滤镜磨皮(点此下载)。3、...