自建 YouTube 字幕问答系统

由于直接使用 Google Gemini YouTube 接口存在较高的延迟问题，因此本文介绍了一套自建 YouTube 视频问答系统的方案，通过调用第三方 supadata API 获取视频字幕，并将字幕进行存储与展示，实现视频问答功能。下面将从获取字幕、Java 代码实现、业务流程解析以及已知问题几个方面进行介绍。

---

1. 获取字幕

系统通过调用 supadata API 来获取 YouTube 视频的字幕，API 调用方式如下：

supadata API 调用示例

get https://api.supadata.ai/v1/youtube/transcript?videoId=kIo2BAubO6k

成功返回时的 JSON 格式如下：

{
  "lang": "zh",
  "availableLangs": ["zh"],
  "content": [
    {
      "lang": "zh",
      "text": "去持有某一家公司",
      "offset": 1115600,
      "duration": 1266
    },
    {
      "lang": "zh",
      "text": "或者去人为的干预这个公司的决策",
      "offset": 1116866,
      "duration": 2200
    }
  ]
}

返回结果中，content 数组中的每个对象都包含了字幕的文本内容、字幕对应的起始时间 (offset) 以及持续时间 (duration)。

---

2. Java 代码实现

2.1 YoutubeVideoSubtitleService 类

该类主要负责获取视频字幕，并将字幕文本存入数据库缓存。代码中使用了线程锁避免并发问题，同时支持两种获取方式：

1. 通过 supadata API 获取字幕
2. 如果获取不到，则调用备用接口 transcriptWithGemini （目前尚未实现）

完整代码如下：

package com.litongjava.llm.service;

import java.util.List;
import java.util.concurrent.locks.Lock;

import com.google.common.util.concurrent.Striped;
import com.litongjava.db.activerecord.Db;
import com.litongjava.db.activerecord.Row;
import com.litongjava.kit.PgObjectUtils;
import com.litongjava.llm.consts.AgentTableNames;
import com.litongjava.model.http.response.ResponseVo;
import com.litongjava.supadata.SubTitleContent;
import com.litongjava.supadata.SubTitleResponse;
import com.litongjava.supadata.SupadataClient;
import com.litongjava.tio.utils.hutool.StrUtil;
import com.litongjava.tio.utils.json.FastJson2Utils;
import com.litongjava.tio.utils.snowflake.SnowflakeIdUtils;
import com.litongjava.tio.utils.video.VideoTimeUtils;

public class YoutubeVideoSubtitleService {
  public static final Striped<Lock> locks = Striped.lock(1024);

  public String get(String videoId) {
    if(StrUtil.isBlank(videoId)) {
      return null;
    }
    Lock lock = locks.get(videoId);
    lock.lock();
    try {
      String sql = "select text_subtitle from %s where video_id=?";
      sql = String.format(sql, AgentTableNames.youtube_video_subtitle);
      String textSubTitle = Db.queryStr(sql, videoId);
      if (textSubTitle != null) {
        return textSubTitle;
      }
      ResponseVo responseVo = SupadataClient.get(videoId);
      SubTitleResponse subTitle = null;
      if (responseVo.isOk()) {
        subTitle = FastJson2Utils.parse(responseVo.getBodyString(), SubTitleResponse.class);
      } else {
        return null;
      }

      List<SubTitleContent> content = subTitle.getContent();
      if (content == null) {
        return null;
      }
      StringBuffer stringBuffer = new StringBuffer();
      for (SubTitleContent subTitleContent : content) {
        long offset = subTitleContent.getOffset();
        long duration = subTitleContent.getDuration();
        // 计算结束时间 = 开始时间 + 持续时间
        long endTime = offset + duration;
        String startStr = VideoTimeUtils.formatTime(offset);
        String endStr = VideoTimeUtils.formatTime(endTime);
        String text = subTitleContent.getText();
        stringBuffer.append(startStr).append("-").append(endStr).append(" ").append(text).append("  \r\n");
      }
      textSubTitle = stringBuffer.toString();
      Row row = Row.by("id", SnowflakeIdUtils.id()).set("video_id", videoId).set("text_subtitle", stringBuffer.toString())
          //
          .set("supadata_subtitle", PgObjectUtils.json(responseVo.getBodyString()));
      Db.save(AgentTableNames.youtube_video_subtitle, row);
      return textSubTitle;
    } finally {
      lock.unlock();
    }
  }

  public String transcriptWithGemini(String url, String videoId) {
    Lock lock = locks.get(videoId);
    lock.lock();
    try {
      String sql = "select text_subtitle from %s where video_id=?";
      sql = String.format(sql, AgentTableNames.youtube_video_subtitle);
      String textSubTitle = Db.queryStr(sql, videoId);
      if (textSubTitle != null) {
        return textSubTitle;
      }
      textSubTitle = transcriptWithGemini(url);
      if (textSubTitle == null) {
        return null;
      }
      Row row = Row.by("id", SnowflakeIdUtils.id()).set("video_id", videoId).set("text_subtitle", textSubTitle);
      Db.save(AgentTableNames.youtube_video_subtitle, row);
      return textSubTitle;
    } finally {
      lock.unlock();
    }
  }

  private String transcriptWithGemini(String url) {
    // TODO Auto-generated method stub
    return null;
  }
}

业务流程说明

• 线程锁保护： 使用 Guava 的 Striped<Lock> 实现对每个 videoId 级别的并发控制，确保同一视频在处理时不会出现并发问题。
• 缓存处理： 首先从数据库中查询缓存的字幕，如果存在则直接返回，避免重复调用 API。
• 调用 supadata API： 若缓存不存在，则通过 SupadataClient.get(videoId) 请求第三方接口获取字幕数据。
• 字幕处理： 遍历返回的 content 数组，将每段字幕的开始时间、结束时间及文本进行格式化，并拼接成最终的字幕文本。
• 存储数据： 将最终的字幕文本以及原始接口返回的数据存入数据库，便于后续直接读取。
• 备用方案： 若 supadata 获取不到字幕，则调用 transcriptWithGemini 方法（目前未实现）尝试通过其他方式获取字幕。

---

2.2 YoutubeService 类

该类主要负责处理前端请求，并将获取到的字幕内容通过 SSE（Server-Sent Events）实时发送到客户端，同时更新聊天记录。完整代码如下：

package com.litongjava.llm.service;

import java.util.List;

import com.jfinal.kit.Kv;
import com.litongjava.jfinal.aop.Aop;
import com.litongjava.llm.consts.AiChatEventName;
import com.litongjava.openai.chat.ChatMessage;
import com.litongjava.openai.chat.ChatMessageArgs;
import com.litongjava.tio.core.ChannelContext;
import com.litongjava.tio.core.Tio;
import com.litongjava.tio.http.common.sse.SsePacket;
import com.litongjava.tio.utils.json.JsonUtils;
import com.litongjava.tio.utils.youtube.YouTubeIdUtil;

public class YoutubeService {

  private YoutubeVideoSubtitleService youtubeVideoSubtitleService = Aop.get(YoutubeVideoSubtitleService.class);

  public void youtube(ChannelContext channelContext, ChatMessageArgs chatSendArgs, List<ChatMessage> historyMessage) {
    String message = null;
    if (channelContext != null) {
      if (chatSendArgs != null && chatSendArgs.getUrl() != null) {
        String url = chatSendArgs.getUrl();
        message = "First, let me get the YouTube video sub title. It will take a few minutes " + url + ".  ";

        Kv by = Kv.by("content", message);
        SsePacket ssePacket = new SsePacket(AiChatEventName.reasoning, JsonUtils.toJson(by));
        Tio.send(channelContext, ssePacket);

        String videoId = YouTubeIdUtil.extractVideoId(url);
        String subTitle = youtubeVideoSubtitleService.get(videoId);
        if (subTitle == null) {
          message = "Sorry, No transcript is available for this video, let me downlaod video.  It will take a few minutes.  ";
          by = Kv.by("content", message);
          ssePacket = new SsePacket(AiChatEventName.reasoning, JsonUtils.toJson(by));
          Tio.send(channelContext, ssePacket);

          subTitle = youtubeVideoSubtitleService.transcriptWithGemini(url, videoId);
        }

        if (subTitle != null) {
          by = Kv.by("content", subTitle);
          ssePacket = new SsePacket(AiChatEventName.reasoning, JsonUtils.toJson(by));
          Tio.send(channelContext, ssePacket);

          historyMessage.add(0, new ChatMessage("user", subTitle));
        } else {
          message = "Sorry, No transcript is available for this video, please try again later.";
          by = Kv.by("content", message);
          ssePacket = new SsePacket(AiChatEventName.reasoning, JsonUtils.toJson(by));
          Tio.send(channelContext, ssePacket);
        }

      } else {
        message = "First, let me review the YouTube video. It will take a few minutes .";
        Kv by = Kv.by("content", message);
        SsePacket ssePacket = new SsePacket(AiChatEventName.reasoning, JsonUtils.toJson(by));
        Tio.send(channelContext, ssePacket);
      }
    }
  }
}

业务流程说明

• 请求处理： 当接收到前端请求时，先判断 chatSendArgs 是否包含视频 URL。
• 初步反馈： 系统先通过 SSE 向客户端发送提示信息，告知正在获取视频字幕。
• 视频 ID 提取： 通过工具类 YouTubeIdUtil.extractVideoId(url) 从 URL 中提取视频 ID。
• 获取字幕： 调用 YoutubeVideoSubtitleService.get(videoId) 尝试获取缓存或新获取字幕；若为空则调用备用方案 transcriptWithGemini(url, videoId)。
• 结果返回： 获取到字幕后，系统将字幕内容通过 SSE 发送到客户端，并更新聊天记录；如果字幕仍为空，则返回错误提示信息。

---

2.3 第二次问答重载 YouTube 字幕的代码

在第二次问答时，系统会根据历史记录中的参数重新加载字幕信息。代码如下：

          String role = record.getStr("role");
          String content = record.getStr("content");
          String args = record.getString("args");
          if (args != null) {
            ChatMessageArgs historyArgs = JsonUtils.parse(args, ChatMessageArgs.class);

            String url = historyArgs.getUrl();
            if (StrUtil.isNotBlank(url)) {
              if (ApiChatSendType.youtube.equals(historyArgs.getType())) {
                String extractVideoId = YouTubeIdUtil.extractVideoId(url);
                if (extractVideoId != null) {
                  String subTitle = youtubeVideoSubtitleService.get(extractVideoId);
                  if (subTitle != null) {
                    historyMessage.add(new ChatMessage(role, subTitle));

                  }
                }
              }
            }
            String[] urls = historyArgs.getUrls();
            if (urls != null && urls.length > 0) {
              if (ApiChatSendType.general.equals(historyArgs.getType())) {
                String htmlContent = webPageService.readHtmlPage(urls);
                historyMessage.add(new ChatMessage(role, htmlContent));
              }
            }

            if (StrUtil.isNotBlank(content)) {
              historyMessage.add(new ChatMessage(role, content));
            }
          } else {
            historyMessage.add(new ChatMessage(role, content));
          }

业务流程说明

• 参数解析： 根据数据库中存储的历史记录，解析出角色、内容和参数（args）。
• 视频 URL 处理： 若参数中包含 URL，则根据 URL 判断消息类型，如果是 YouTube 类型，则重新提取视频 ID，并调用 get 方法加载字幕。
• 通用消息处理： 若参数中包含多个 URL，并且类型为通用类型，则调用 webPageService.readHtmlPage(urls) 获取网页内容并添加到聊天记录中。
• 聊天记录更新： 最后根据是否存在 content，更新聊天历史记录，保证问答过程中的连续性。

---

3. 业务流程总结

整个系统的业务流程可以总结为以下几个步骤：

1. 视频字幕获取：

   • 根据视频 URL 提取视频 ID。
   • 首先尝试从数据库缓存中获取字幕内容，避免重复请求。
   • 若缓存不存在，则通过调用 supadata API 获取字幕，并将字幕内容格式化为带有时间戳的文本。
   • 将获取到的字幕内容及原始返回数据存入数据库。

2. 前端交互：

   • 使用 SSE（Server-Sent Events）方式向客户端发送实时处理状态和最终结果。
   • 在首次请求和后续重载问答时，均会将字幕内容加入聊天记录，实现问答历史追踪。

3. 备用字幕获取：

   • 当 supadata API 无法返回字幕时，系统提供了备用方法 transcriptWithGemini（目前尚未实现），以期支持其他方式获取字幕。

---

4. 已知问题

目前存在一个已知问题：

• 部分视频的第三方 URL 无法返回字幕
  有些视频由于版权或其他原因，调用第三方接口时可能无法获取到字幕。对此，系统在获取不到字幕时会反馈错误信息，提示用户稍后再试或者尝试其他方案。

---

5. 结论

本文详细介绍了如何通过自建服务来实现 YouTube 视频问答系统，主要流程包括调用 supadata API 获取字幕、格式化与存储字幕内容以及通过 SSE 实时返回给客户端。同时也对代码中的业务流程做了详细说明，并指出了已知的第三方字幕获取问题。未来可进一步完善备用字幕接口 transcriptWithGemini，提高对无法返回字幕视频的支持能力。