get_transcript_by_yt_api

app.py

@@ -155,7 +155,7 @@ def check_open_ai_access(open_ai_api_key):
     client = OpenAI(api_key=open_ai_api_key)
     try:
         response = client.chat.completions.create(
-            model="gpt-3.5-turbo",
+            model="gpt-4o",
             messages=[
                 {"role": "user", "content": "This is a test."},
             ],
@@ -399,10 +399,18 @@ def get_transcript_by_yt_api(video_id):
     
     for language in languages:
         try:
-            transcript = YouTubeTranscriptApi.get_transcript(video_id, languages=[language])
+            yt_api_transcript = YouTubeTranscriptApi.get_transcript(video_id, languages=[language])
             print("===transcript===")
-            print(transcript)
+            print(yt_api_transcript)
+
+            transcript = ""
+            for entry in yt_api_transcript:
+                transcript_part = (f"{entry['start']:.0f}s: {entry['text']}")
+                print(transcript_part)
+                original_transcript += f"{transcript_part} \n"
             print("===transcript===")
+
+            transcript = convert_transcription_to_json(original_transcript)
             return transcript  # 成功獲取字幕，直接返回結果
         except NoTranscriptFound:
             continue  # 當前語言的字幕沒有找到，繼續嘗試下一個語言
@@ -415,7 +423,7 @@ def generate_transcription_by_gemini(video_id):
     video_url = f"https://www.youtube.com/watch?v={video_id}"
     
     # 初始化 Gemini Pro Vision 模型
-    model = vertexai.generative_models.GenerativeModel("gemini-2.0-flash-exp")
+    model = vertexai.generative_models.GenerativeModel("gemini-2.5-flash-preview-05-20")
     
     # 建立影片部分
     video_part = Part.from_uri(
@@ -424,7 +432,7 @@ def generate_transcription_by_gemini(video_id):
     )
 
     # 設定提示詞
-    prompt = "給我包含時間軸的逐字稿，只需要給我有講話的時間軸跟內容，其他時間軸不需要"
+    prompt = "給我包含時間軸的完整逐字稿，包含時間軸跟原文內容，一句話一行"
 
     # 生成逐字稿
     original_transcription = ""
@@ -434,7 +442,7 @@ def generate_transcription_by_gemini(video_id):
             generation_config=vertexai.generative_models.GenerationConfig(
                 temperature=1.0,
                 top_p=0.95,
-                max_output_tokens=8192,
+                max_output_tokens=65535,
                 candidate_count=1
             ),
             stream=False
@@ -462,7 +470,7 @@ def generate_transcription_by_gemini(video_id):
 
 def convert_transcription_to_json(original_transcription):
     """
-    將原始逐字稿轉換成指定的 JSON 格式
+    將原始逐字稿轉換成指定的 JSON 格式，支援長文本分段處理
     
     Args:
         original_transcription (str): 原始逐字稿文本
@@ -470,63 +478,104 @@ def convert_transcription_to_json(original_transcription):
     Returns:
         list: 包含逐字稿段落的列表，每個段落包含 text, start, end, duration
     """
-    
+    if not original_transcription:
+        print("原始逐字稿為空")
+        return None
+        
     # 使用 Vertex AI 來處理轉換
-    model = vertexai.generative_models.GenerativeModel("gemini-2.0-flash-exp")
+    model = vertexai.generative_models.GenerativeModel("gemini-2.5-flash-preview-05-20")
     
-    prompt = f"""
-    請將以下逐字稿轉換成 JSON 格式:
-    {original_transcription}
+    # 設定每段最大字數
+    # 考慮到：
+    # 1. Gemini 輸出限制為 65,535 tokens
+    # 2. 需要預留空間給系統提示詞
+    # 3. JSON 格式會增加額外字符
+    # 4. 中文一個字約等於 2-3 個 tokens
+    MAX_CHUNK_SIZE = 15000
     
-    轉換規則:
-    1. 每個段落需包含 text, start, end, duration
-    2. 時間格式需轉換為秒數(例如 1:02 轉為 62 秒)
-    3. duration 為 end - start 的差值
-    4. 回傳格式為 JSON array
+    # 分段處理
+    chunks = []
+    current_chunk = []
+    current_size = 0
     
-    範例輸出格式:
-    [
-        {{
-            "text": "在一片無人的森林裡",
-            "start": 1,
-            "end": 2,
-            "duration": 1
-        }},
-        {{
-            "text": "你撿到一張羊皮紙",
-            "start": 2,
-            "end": 4,
-            "duration": 2
-        }}
-    ]
+    # 按行分割文本
+    lines = original_transcription.split('\n')
     
-    請直接返回 JSON 格式，不要加入任何說明文字。
-    """
+    for line in lines:
+        line = line.strip()
+        if not line:
+            continue
+            
+        # 如果這一行加入後會超過限制，就先處理當前chunk
+        if current_size + len(line) > MAX_CHUNK_SIZE and current_chunk:
+            chunks.append('\n'.join(current_chunk))
+            current_chunk = []
+            current_size = 0
+            
+        current_chunk.append(line)
+        current_size += len(line)
     
-    try:
-        response = model.generate_content(prompt)
-        json_str = response.text
-
-        print("===json_str===")
-        print(json_str)
-        print("===json_str===")
+    # 處理最後一個chunk
+    if current_chunk:
+        chunks.append('\n'.join(current_chunk))
+    
+    # 用於儲存所有處理結果
+    all_results = []
+    
+    # 處理每個chunk
+    for i, chunk in enumerate(chunks):
+        print(f"===chunk: {i+1}===")
         
-        # 移除可能的 markdown 標記
-        json_str = json_str.replace("```json", "").replace("```", "").strip()
+        prompt = f"""
+        請將以下逐字稿轉換成 JSON 格式:
+        {chunk}
         
-        # 解析 JSON
-        transcript_json = json.loads(json_str)
+        轉換規則:
+        1. 每個段落需包含 text, start, end, duration
+        2. 時間格式需轉換為秒數(例如 1:02 轉為 62 秒)
+        3. duration 為 end - start 的差值
+        4. 回傳格式為 JSON array
+        5. 合理的合併句子，不要有不合理的斷句，一句話至少要有完整的主詞、謂詞
+        6. 每句話盡量在 10~15 個字左右，但要以完整語意為主
+        7. 如果遇到 [Music] 這類的標記，可以直接忽略不計
+        8. 這是第 {i+1}/{len(chunks)} 段，請確保時間軸的連續性
         
-        # 驗證格式
-        for entry in transcript_json:
-            if not all(k in entry for k in ["text", "start", "end", "duration"]):
-                raise ValueError("JSON 格式錯誤：缺少必要欄位")
-                
-        return transcript_json
+        請直接返回 JSON 格式，不要加入任何說明文字或 markdown 標記。
+        """
         
-    except Exception as e:
-        print(f"轉換逐字稿時發生錯誤：{str(e)}")
+        try:
+            response = model.generate_content(prompt)
+            json_str = response.text
+            
+            print(f"===json_str for chunk {i+1}===")
+            print(json_str)
+            print(f"===json_str for chunk {i+1}===")
+            
+            # 移除可能的 markdown 標記
+            json_str = json_str.replace("```json", "").replace("```", "").strip()
+            
+            # 解析 JSON
+            chunk_result = json.loads(json_str)
+            
+            # 驗證格式
+            for entry in chunk_result:
+                if not all(k in entry for k in ["text", "start", "end", "duration"]):
+                    raise ValueError(f"JSON 格式錯誤：缺少必要欄位，在第 {i+1} 段")
+                    
+            all_results.extend(chunk_result)
+                
+        except Exception as e:
+            print(f"處理第 {i+1} 段時發生錯誤：{str(e)}")
+            continue
+    
+    # 如果沒有任何有效結果，返回 None
+    if not all_results:
         return None
+        
+    # 按時間排序
+    all_results.sort(key=lambda x: x["start"])
+    
+    return all_results
 
 def generate_transcription_by_whisper(video_id):
     youtube_url = f'https://www.youtube.com/watch?v={video_id}'
@@ -607,9 +656,11 @@ def process_transcript_and_screenshots_on_gcs(video_id):
     if not exists:
         print("==== video transcript is not exists ====")
         try:
-            transcript = generate_transcription_by_gemini(video_id)
+            transcript = get_transcript_by_yt_api(video_id)
+            # transcript = generate_transcription_by_gemini(video_id)
         except Exception as e:
             print(f"generate_transcription_by_gemini Error generating transcription: {str(e)}")
+            transcript = generate_transcription_by_gemini(video_id)
             # transcript = generate_transcription_by_whisper(video_id)
 
         upload_transcript_to_gcs(video_id, transcript)
@@ -640,6 +691,10 @@ def process_transcript_and_screenshots_on_gcs(video_id):
                     is_new_transcript = True
                 except Exception as e:
                     print(f"Error processing screenshot: {str(e)}")
+                    # 如果影片有下載成功，但是截圖失敗，則將 img_file_id 設為空字串
+                    entry['img_file_id'] = ""
+                    print(f"截圖空白")
+                    is_new_transcript = True
             else:
                 entry['img_file_id'] = ""
                 print(f"截圖空白")