get_sheet_data

app.py

@@ -1,4 +1,5 @@
 import urllib.parse
+import re
 
 import gradio as gr
 from starlette.middleware.base import BaseHTTPMiddleware
@@ -69,6 +70,7 @@ from googleapiclient.http import MediaIoBaseUpload
 
 from educational_material import EducationalMaterial
 from storage_service import GoogleCloudStorage
+from sheet_service import SheetService
 from google.oauth2.service_account import Credentials
 import vertexai
 from vertexai.generative_models import GenerativeModel, Part
@@ -93,6 +95,7 @@ if is_env_local:
         GCS_KEY = json.dumps(config["GOOGLE_APPLICATION_CREDENTIALS_JSON"])
         DRIVE_KEY = json.dumps(config["GOOGLE_APPLICATION_CREDENTIALS_JSON"])
         GBQ_KEY = json.dumps(config["GOOGLE_APPLICATION_CREDENTIALS_JSON"])
+        SHEET_KEY = json.dumps(config["GOOGLE_APPLICATION_CREDENTIALS_JSON"])
         OPEN_AI_KEY = config["OPEN_AI_KEY"]
         OPEN_AI_ASSISTANT_ID_GPT4_BOT1 = config["OPEN_AI_ASSISTANT_ID_GPT4_BOT1"]
         OPEN_AI_ASSISTANT_ID_GPT3_BOT1 = config["OPEN_AI_ASSISTANT_ID_GPT3_BOT1"]
@@ -142,6 +145,7 @@ GBQ_CLIENT = bigquery.Client.from_service_account_info(json.loads(GBQ_KEY))
 GROQ_CLIENT = Groq(api_key=GROQ_API_KEY)
 GCS_SERVICE = GoogleCloudStorage(GCS_KEY)
 GCS_CLIENT = GCS_SERVICE.client
+SHEET_SERVICE = SheetService(SHEET_KEY)
 PERPLEXITY_CLIENT = OpenAI(api_key=PERPLEXITY_API_KEY, base_url="https://api.perplexity.ai")
 
 # check open ai access
@@ -2150,6 +2154,12 @@ def summary_add_markdown_version(video_id):
 
 
 # LLM 強制重刷
+def get_sheet_data(sheet_url, range_name):
+    data = SHEET_SERVICE.get_sheet_data_by_url(sheet_url, range_name)
+    flattened_data = SHEET_SERVICE.flatten_column_data(data)
+    return flattened_data
+
+
 def refresh_video_LLM_all_content(video_ids):
     # 輸入影片 id，以 , 逗號分隔 或是 \n 換行
     video_id_list = video_ids.replace('\n', ',').split(',')
@@ -3647,8 +3657,22 @@ def create_app():
                 with gr.Row():
                     gr.Markdown("## 清單影片：重新生成所有內容")
                 with gr.Row():
-                    refresh_video_ids = gr.Textbox(label="輸入影片 id，以 , 逗號分隔")
-                    refresh_btn = gr.Button("refresh", variant="primary")
+                    # tab refresh_video_ids & by sheets
+                    with gr.Tab("refresh_video_ids"):
+                        refresh_video_ids = gr.Textbox(label="輸入影片 id，以 , 逗號分隔")
+                        refresh_btn = gr.Button("refresh", variant="primary")
+                    with gr.Tab("by sheets"):
+                        sheet_url = gr.Textbox(label="輸入 Google Sheets 的 URL")
+                        sheet_get_value_btn = gr.Button("取得 ids", variant="primary")
+                        sheet_get_value_result = gr.Textbox(label="ids", interactive=False)
+                        sheet_refresh_btn = gr.Button("refresh by sheets", variant="primary")
+                
+                        sheet_get_value_btn.click(
+                            get_sheet_data,
+                            inputs=[sheet_url, gr.Textbox(value="D:D", visible=True)], # 將範圍修改為 D 欄
+                            outputs=[sheet_get_value_result]
+                        )
+                
                 with gr.Row():
                     refresh_result = gr.JSON()
     

sheet_service.py

@@ -0,0 +1,269 @@
+import google.oauth2.credentials
+import googleapiclient.discovery
+from google.oauth2 import service_account
+import json
+from urllib.parse import urlparse, parse_qs
+import logging # 建議使用 logging 而非 print
+
+# 設定基本的 logging
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+
+class SheetService:
+    """
+    一個用於與 Google Sheets API 互動的服務類別。
+    """
+    SCOPES = ['https://www.googleapis.com/auth/spreadsheets.readonly']
+
+    def __init__(self, service_account_key_string: str, api_service_name: str = 'sheets', api_version: str = 'v4'):
+        """
+        初始化 SheetService。
+
+        Args:
+            service_account_key_string (str): 包含 Google 服務帳戶憑證資訊的 JSON 字串。
+                                             通常是從 JSON 金鑰檔案讀取的內容。
+            api_service_name (str): 要使用的 Google API 服務名稱。預設為 'sheets'。
+            api_version (str): 要使用的 Google API 版本。預設為 'v4'。
+        """
+        try:
+            credentials_info = json.loads(service_account_key_string)
+            self.credentials = service_account.Credentials.from_service_account_info(
+                credentials_info, scopes=self.SCOPES
+            )
+            self.service = googleapiclient.discovery.build(
+                api_service_name, api_version, credentials=self.credentials
+            )
+            self.sheet = self.service.spreadsheets()
+            logging.info("成功連接 Google Sheets API")
+        except json.JSONDecodeError as e:
+            logging.error(f"解析憑證 JSON 字串時發生錯誤: {e}")
+            self.service = None
+            self.sheet = None
+        except Exception as e:
+            logging.error(f"連接 Google Sheets API 時發生錯誤: {e}")
+            self.service = None
+            self.sheet = None
+
+    def get_sheet_id_by_url(self, sheet_url: str) -> str | None:
+        """
+        從 Google Sheets URL 中提取試算表 ID。
+        """
+        parsed_url = urlparse(sheet_url)
+        path_parts = parsed_url.path.split('/')
+        try:
+            # Google Sheet URL 格式通常是 /spreadsheets/d/SPREADSHEET_ID/edit...
+            if 'd' in path_parts:
+                id_index = path_parts.index('d') + 1
+                if id_index < len(path_parts):
+                    spreadsheet_id = path_parts[id_index]
+                    # 進行一些基本檢查，確保它看起來像一個 ID
+                    if len(spreadsheet_id) > 30: # Google Sheet ID 通常很長
+                        return spreadsheet_id
+        except ValueError:
+            pass # 'd' 不在路徑中
+
+        logging.warning(f"無法從 URL 中提取有效的 Spreadsheet ID: {sheet_url}")
+        return None
+
+    def get_sheet_gid_by_url(self, sheet_url: str) -> int | None:
+        """
+        從 Google Sheets URL 中提取 gid (工作表分頁 ID)。
+        返回整數型別的 gid 或 None。
+        """
+        parsed_url = urlparse(sheet_url)
+        query_params = parse_qs(parsed_url.query)
+        fragment_params = parse_qs(parsed_url.fragment) # gid 也可能在 # 後面
+
+        gid_str = None
+        if 'gid' in query_params:
+            gid_str = query_params['gid'][0]
+        elif 'gid' in fragment_params:
+            gid_str = fragment_params['gid'][0]
+
+        if gid_str:
+            try:
+                return int(gid_str)
+            except ValueError:
+                logging.warning(f"URL 中的 gid 不是有效的整數: {gid_str}")
+                return None
+        else:
+            # logging.info(f"URL 中未找到 gid 參數，將嘗試使用第一個工作表: {sheet_url}")
+            # 如果 URL 沒有 gid，通常表示是第一個工作表，其 gid 通常是 0
+            # 但我們在這裡返回 None，讓後續邏輯決定如何處理
+            return None
+
+    def get_sheet_name_by_gid(self, spreadsheet_id: str, gid: int | None) -> str | None:
+        """
+        使用 spreadsheetId 和 gid 獲取工作表名稱 (title)。
+        如果 gid 為 None，則返回第一個工作表的名稱。
+
+        Args:
+            spreadsheet_id (str): Google 試算表的 ID。
+            gid (int | None): 目標工作表分頁的 ID。如果為 None，則獲取第一個工作表。
+
+        Returns:
+            str | None: 工作表的名稱 (title)，如果找不到或發生錯誤則返回 None。
+        """
+        if not self.service:
+            logging.error("Sheet API 服務未成功初始化。")
+            return None
+        try:
+            # 使用 spreadsheets.get 獲取試算表的中繼資料
+            # fields 參數限制只返回我們需要的 sheets.properties (包含 title 和 sheetId)
+            sheet_metadata = self.service.spreadsheets().get(
+                spreadsheetId=spreadsheet_id,
+                fields='sheets(properties(sheetId,title))'
+            ).execute()
+
+            sheets = sheet_metadata.get('sheets', [])
+            if not sheets:
+                logging.warning(f"試算表 {spreadsheet_id} 中沒有找到任何工作表。")
+                return None
+
+            if gid is not None:
+                # 如果提供了 gid，尋找匹配的工作表
+                for sheet in sheets:
+                    properties = sheet.get('properties', {})
+                    if properties.get('sheetId') == gid:
+                        sheet_title = properties.get('title')
+                        if sheet_title:
+                            logging.info(f"找到 gid={gid} 對應的工作表名稱: '{sheet_title}'")
+                            return sheet_title
+                        else:
+                            logging.warning(f"找到 gid={gid} 但缺少 title 屬性。")
+                            return None
+                # 如果遍歷完畢沒有找到匹配的 gid
+                logging.warning(f"在試算表 {spreadsheet_id} 中未找到 gid={gid} 的工作表。")
+                return None
+            else:
+                # 如果 gid 為 None，返回第一個工作表的名稱
+                first_sheet_properties = sheets[0].get('properties', {})
+                first_sheet_title = first_sheet_properties.get('title')
+                first_sheet_gid = first_sheet_properties.get('sheetId', '未知')
+                if first_sheet_title:
+                    logging.info(f"未提供 gid，使用第一個工作表 (gid={first_sheet_gid}): '{first_sheet_title}'")
+                    return first_sheet_title
+                else:
+                    logging.warning(f"第一個工作表 (gid={first_sheet_gid}) 缺少 title 屬性。")
+                    return None
+
+        except googleapiclient.errors.HttpError as error:
+            logging.error(f"獲取工作表名稱時發生 API 錯誤: {error}")
+            return None
+        except Exception as e:
+            logging.error(f"獲取工作表名稱時發生未知錯誤: {e}")
+            return None
+
+    def get_sheet_data_by_url(self, sheet_url: str, read_range: str | None = None) -> list | None:
+        """
+        通過 Google Sheets URL 自動獲取 Spreadsheet ID 和工作表名稱，並讀取數據。
+        如果 URL 中包含 gid，則讀取對應的工作表；否則讀取第一個工作表。
+        默認讀取整個工作表的數據。
+
+        Args:
+            sheet_url (str): Google 試算表的完整 URL。
+            read_range (str | None): 可選。指定要讀取的儲存格範圍 (例如 'A1:C10')。
+                                     如果提供，則只讀取此範圍；否則讀取整個工作表。
+
+        Returns:
+            list | None: 包含讀取到的資料的列表 (list of lists)，如果發生錯誤則返回 None。
+        """
+        spreadsheet_id = self.get_sheet_id_by_url(sheet_url)
+        if not spreadsheet_id:
+            logging.error("無法從 URL 獲取 Spreadsheet ID。")
+            return None
+
+        gid = self.get_sheet_gid_by_url(sheet_url)
+        # 無論 gid 是否為 None，都嘗試獲取工作表名稱
+        sheet_name = self.get_sheet_name_by_gid(spreadsheet_id, gid)
+
+        if not sheet_name:
+            logging.error(f"無法根據 URL ({sheet_url}) 確定要讀取的工作表名稱。")
+            return None
+
+        # 組合 range_name
+        if read_range:
+            # 如果使用者指定了範圍，將其與工作表名稱結合
+            # 需要確保工作表名稱不包含特殊字符，或者正確引用
+            # 簡單起見，如果名稱包含空格或特殊符號，用單引號括起來
+            if ' ' in sheet_name or '!' in sheet_name or ':' in sheet_name:
+                 range_name = f"'{sheet_name}'!{read_range}"
+            else:
+                 range_name = f"{sheet_name}!{read_range}"
+        else:
+            # 如果未指定範圍，則讀取整個工作表
+            # 只需要提供工作表名稱即可
+             if ' ' in sheet_name or '!' in sheet_name or ':' in sheet_name:
+                 range_name = f"'{sheet_name}'"
+             else:
+                 range_name = sheet_name
+
+
+        logging.info(f"準備從試算表 '{spreadsheet_id}' 的 '{range_name}' 範圍讀取數據。")
+        # 使用現有的 get_sheet_value 方法讀取數據
+        return self.get_sheet_value(spreadsheet_id, range_name)
+
+    def get_sheet_value(self, spreadsheet_id: str, range_name: str) -> list | None:
+        """
+        從指定的試算表和範圍讀取資料。
+
+        Args:
+            spreadsheet_id (str): Google 試算表的 ID。
+            range_name (str): 要讀取的範圍，例如 'Sheet1!A1:B2' 或僅 'Sheet1' (讀取整個工作表)。
+
+        Returns:
+            list | None: 包含讀取到的資料的列表 (list of lists)，如果發生錯誤則返回 None。
+        """
+        if not self.sheet:
+            logging.error("Sheet API 服務未成功初始化。")
+            return None
+
+        try:
+            logging.info(f"正在讀取 Spreadsheet ID: {spreadsheet_id}, Range: {range_name}")
+
+            result = self.sheet.values().get(
+                spreadsheetId=spreadsheet_id,
+                range=range_name
+            ).execute()
+            values = result.get('values', [])
+            logging.info(f"成功從 {spreadsheet_id} 的 {range_name} 讀取 {len(values)} 列資料。")
+            # 如果 values 是 None 或空列表，直接返回
+            if not values:
+                logging.warning(f"在 {spreadsheet_id} 的 {range_name} 範圍內未找到任何資料。")
+                return [] # 返回空列表而不是 None，以便後續處理
+            return values
+        except googleapiclient.errors.HttpError as error:
+            # 更詳細地記錄錯誤信息
+            error_details = error.resp.get('content', '{}')
+            try:
+                error_json = json.loads(error_details)
+                error_message = error_json.get('error', {}).get('message', str(error))
+            except json.JSONDecodeError:
+                error_message = str(error)
+            logging.error(f"讀取試算表時發生 API 錯誤 (ID: {spreadsheet_id}, Range: {range_name}): {error_message}")
+            return None
+        except Exception as e:
+            logging.error(f"讀取試算表時發生未知錯誤 (ID: {spreadsheet_id}, Range: {range_name}): {e}")
+            return None
+
+    @staticmethod
+    def flatten_column_data(data: list[list[str]]) -> list[str]:
+        """
+        將從 Google Sheets API 獲取的單欄數據（列表的列表）扁平化為單一列表。
+
+        例如，將 [['A'], ['B'], ['C']] 轉換為 ['A', 'B', 'C']。
+        此方法會跳過空的內部列表，並假設每個非空內部列表只取第一個元素。
+
+        Args:
+            data (list[list[str]]): 從 API 獲取的原始數據，通常是 list of lists。
+
+        Returns:
+            list[str]: 包含所有第一欄元素的單一列表。如果輸入為 None 或空列表，
+                       則返回空列表。
+        """
+        if not data:
+            return []
+        # 使用列表推導式，提取每個子列表的第一個元素
+        # 添加 if sublist and sublist[0] is not None 確保子列表非空且第一個元素存在
+        # 並將其轉換為字串 str() 以確保類型一致性
+        flattened = [str(sublist[0]) for sublist in data if sublist and sublist[0] is not None]
+        return flattened