Update app.py

app.py

@@ -1,127 +1,223 @@
-import os, csv, requests, gradio as gr
-from huggingface_hub import hf_hub_download, upload_file
-import logging
-
-# 設定日誌等級
-logging.basicConfig(level=logging.INFO)
-
-# -- 環境變數 & 路徑 --
-REPO_ID         = os.getenv("REPO_ID", "YuShen1124/youtube-tags-tool")
-HF_TOKEN        = os.getenv("HF_TOKEN")
-YOUTUBE_API_KEY = os.getenv("AIzaSyCvgpJMhcrDao6Li4qMUbFcifFSobbcIZw")
-DATA_DIR        = "./data"
-CSV_FILE        = os.path.join(DATA_DIR, "tags_data.csv")
-COUNTER_FILE    = os.path.join(DATA_DIR, "counter.txt")
-
-os.makedirs(DATA_DIR, exist_ok=True)
-
-# 初始化遠端檔案，若不存在則本地建立
-def init_file(remote, local):
-    try:
-        hf_hub_download(repo_id=REPO_ID, filename=remote, local_dir=DATA_DIR, token=HF_TOKEN)
-        logging.info(f"Downloaded {remote} from Hub")
-    except Exception:
-        open(local, "a", encoding="utf-8").close()
-        logging.info(f"Initialized empty {local}")
-
-init_file("tags_data.csv", CSV_FILE)
-init_file("counter.txt", COUNTER_FILE)
-
-# 讀取 CSV
+import os
+
+import csv
+
+import requests
+
+import gradio as gr
+
+# 請替換為你的 YouTube API Key
+
+YOUTUBE_API_KEY = "AIzaSyCvgpJMhcrDao6Li4qMUbFcifFSobbcIZw"
+
+# CSV 檔案名稱，用來儲存累計標籤與使用次數
+
+CSV_FILE = "tags_data.csv"
+
+# 執行次數計數檔案名稱
+
+COUNTER_FILE = "counter.txt"
+
 def read_csv_data():
+
     data = {}
-    try:
-        with open(CSV_FILE, encoding='utf-8', newline='') as f:
-            for tag, count in csv.reader(f):
-                data[tag] = int(count)
-    except Exception as e:
-        logging.error(f"Error reading CSV: {e}")
+
+    if os.path.exists(CSV_FILE):
+
+        with open(CSV_FILE, mode="r", newline="", encoding="utf-8") as f:
+
+            reader = csv.reader(f)
+
+            for row in reader:
+
+                if len(row) == 2:
+
+                    tag, count = row
+
+                    data[tag] = int(count)
+
     return data
 
-# 寫入並上傳 CSV
 def write_csv_data(data):
-    try:
-        with open(CSV_FILE, 'w', encoding='utf-8', newline='') as f:
-            csv.writer(f).writerows(data.items())
-        upload_file(path_or_fileobj=CSV_FILE, path_in_repo="tags_data.csv",
-                    repo_id=REPO_ID, token=HF_TOKEN)
-        logging.info("CSV uploaded to Hub")
-    except Exception as e:
-        logging.error(f"Error writing/uploading CSV: {e}")
-
-# 呼叫 YouTube API 取得 tags，附加 timeout 與錯誤處理
+
+    with open(CSV_FILE, mode="w", newline="", encoding="utf-8") as f:
+
+        writer = csv.writer(f)
+
+        for tag, count in data.items():
+
+            writer.writerow([tag, count])
+
 def get_video_tags(keyword):
-    tags_count = {}
-    try:
-        resp = requests.get(
-            "https://www.googleapis.com/youtube/v3/search",
-            params={'part':'snippet','q':keyword,'type':'video','maxResults':10,'key':YOUTUBE_API_KEY},
-            timeout=10
+
+    search_url = (
+
+        f"https://www.googleapis.com/youtube/v3/search?part=snippet&q={keyword}"
+
+        f"&type=video&maxResults=10&key={YOUTUBE_API_KEY}"
+
+    )
+
+    search_response = requests.get(search_url).json()
+
+    video_tags = {}
+
+    # 如果沒有取得影片結果，印出回應內容作除錯
+
+    if "items" not in search_response or not search_response["items"]:
+
+        print("搜尋回應無影片結果：", search_response)
+
+    for item in search_response.get("items", []):
+
+        video_id = item["id"]["videoId"]
+
+        video_url = (
+
+            f"https://www.googleapis.com/youtube/v3/videos?part=snippet&id={video_id}"
+
+            f"&key={YOUTUBE_API_KEY}"
+
         )
-        resp.raise_for_status()
-        items = resp.json().get('items', [])
-    except Exception as e:
-        logging.error(f"Search API error: {e}")
-        return tags_count
-
-    for item in items:
-        vid = item['id'].get('videoId')
-        if not vid: continue
-        try:
-            info = requests.get(
-                "https://www.googleapis.com/youtube/v3/videos",
-                params={'part':'snippet','id':vid,'key':YOUTUBE_API_KEY},
-                timeout=10
-            )
-            info.raise_for_status()
-            for v in info.json().get('items', []):
-                for tag in v['snippet'].get('tags', []):
-                    tags_count[tag] = tags_count.get(tag, 0) + 1
-        except Exception as e:
-            logging.warning(f"Video API error for {vid}: {e}")
-    return tags_count
-
-# 更新並分析標籤
+
+        video_response = requests.get(video_url).json()
+
+        print("影片回傳資料：", video_response)  # 除錯：印出影片資料回應
+
+        for video_item in video_response.get("items", []):
+
+            # 嘗試取得 tags 欄位
+
+            tags = video_item["snippet"].get("tags", [])
+
+            print("取得的 tags：", tags)  # 除錯：印出取得的 tags
+
+            for tag in tags:
+
+                video_tags[tag] = video_tags.get(tag, 0) + 1
+
+    return video_tags
+
 def update_and_analyze_tags(keyword):
+
     if not keyword.strip():
-        return "未輸入關鍵字", "", ""
+
+        return "您未輸入關鍵字", "您未輸入關鍵字", "目前執行次數：不增加"
+
+    
+
     new_tags = get_video_tags(keyword)
+
     if not new_tags:
-        logging.info("No tags found for keyword")
-    agg = read_csv_data()
-    for t, c in new_tags.items(): agg[t] = agg.get(t, 0) + c
-    write_csv_data(agg)
-
-    # 計數器讀取、更新與上傳
-    try:
-        cnt = int(open(COUNTER_FILE, encoding='utf-8').read().strip() or 0) + 1
-    except:
-        cnt = 1
-    with open(COUNTER_FILE, 'w', encoding='utf-8') as f:
-        f.write(str(cnt))
-    try:
-        upload_file(path_or_fileobj=COUNTER_FILE, path_in_repo="counter.txt",
-                    repo_id=REPO_ID, token=HF_TOKEN)
-        logging.info("Counter uploaded")
-    except Exception as e:
-        logging.error(f"Error uploading counter: {e}")
-
-    sorted_tags = sorted(agg.items(), key=lambda x: x[1], reverse=True)
-    full = "\n".join(f"{t}: {c}" for t, c in sorted_tags)
-    copy = ",".join(t for t, _ in sorted_tags)
-    return full or "未找到標籤。", copy or "", f"執行次數：{cnt}"
-
-# Gradio 介面
+
+        print("查無標籤資料，請確認搜尋關鍵字是否正確或影片是否有公開標籤")
+
+    aggregated = read_csv_data()
+
+    for tag, count in new_tags.items():
+
+        aggregated[tag] = aggregated.get(tag, 0) + count
+
+    write_csv_data(aggregated)
+
+    
+
+    sorted_tags = sorted(aggregated.items(), key=lambda x: x[1], reverse=True)
+
+    full_output = "\n".join(f"{tag}: {count}" for tag, count in sorted_tags)
+
+    copy_output = ",".join(tag for tag, count in sorted_tags)
+
+    
+
+    if os.path.exists(COUNTER_FILE):
+
+        with open(COUNTER_FILE, "r", encoding="utf-8") as f:
+
+            try:
+
+                usage_count = int(f.read())
+
+            except Exception as e:
+
+                print("讀取 counter.txt 時發生例外：", e)
+
+                usage_count = 0
+
+    else:
+
+        usage_count = 0
+
+    usage_count += 1
+
+    with open(COUNTER_FILE, "w", encoding="utf-8") as f:
+
+        f.write(str(usage_count))
+
+    usage_output = f"執行次數：{usage_count}"
+
+    
+
+    # 如果 full_output 或 copy_output 為空，則回傳「未找到相關標籤。」
+
+    return (
+
+        full_output if full_output else "未找到相關標籤。",
+
+        copy_output if copy_output else "未找到相關標籤。",
+
+        usage_output
+
+    )
+
 with gr.Blocks() as demo:
+
     gr.Markdown("## YouTube 關鍵字工具")
+
+    with gr.Row():
+
+        keyword_input = gr.Textbox(lines=1, placeholder="輸入關鍵字，例如：遊戲", label="關鍵字")
+
+        confirm_button = gr.Button("確認")
+
+    with gr.Row():
+
+        full_output_box = gr.Textbox(label="累計結果 (標籤: 次數)", interactive=False)
+
     with gr.Row():
-        inp = gr.Textbox(lines=1, placeholder="輸入關鍵字，例如：遊戲", label="關鍵字")
-        btn = gr.Button("確認")
-    out1 = gr.Textbox(label="累計結果 (標籤: 次數)", interactive=False)
-    out2 = gr.Textbox(label="標籤清單", interactive=False, show_copy_button=True)
-    out3 = gr.Textbox(label="執行次數", interactive=False)
-    btn.click(update_and_analyze_tags, inputs=inp, outputs=[out1, out2, out3])
-    inp.submit(update_and_analyze_tags, inputs=inp, outputs=[out1, out2, out3])
-
-if __name__ == "__main__":
-    demo.launch(server_name="0.0.0.0", server_port=int(os.getenv("PORT", 7860)))
+
+        copy_output_box = gr.Textbox(label="僅標籤 (以逗號分隔，一鍵複製)", interactive=False, show_copy_button=True)
+
+    with gr.Row():
+
+        usage_box = gr.Textbox(label="執行次數", interactive=False)
+
+    
+
+    gr.Markdown(
+
+        "使用說明：\n"
+
+        "- 請先輸入關鍵字，再點選【確認】按鈕或按 Enter 鍵觸發查詢。\n"
+
+        "- 若未輸入內容，則不會更新資料與執行次數。\n"
+
+        "- 上方輸出框顯示累計結果（每行格式：標籤: 次數）。\n"
+
+        "- 中間輸出框顯示所有標籤，以半形逗號隔開，點擊複製按鈕即可複製全部標籤。\n"
+
+        "- 下方輸出框顯示該工具累計的執行次數。"
+
+    )
+
+    
+
+    confirm_button.click(update_and_analyze_tags, inputs=keyword_input,
+
+                         outputs=[full_output_box, copy_output_box, usage_box])
+
+    keyword_input.submit(update_and_analyze_tags, inputs=keyword_input,
+
+                         outputs=[full_output_box, copy_output_box, usage_box])
+
+demo.launch()