# -*- coding: utf-8 -*- """ 定时检查 YouTube 频道是否有直播存档视频，有就删除 优化重点： - 新增开关：DELETE_ONLY_RECENT_48H（只删最近48小时内的存档） - 慢速删除：每删一个随机等待 5~20 秒，每 5 个长休息 1~5 分钟 - 每日上限：默认每天最多删 30 个，避免风控/配额耗尽 - 限流保护：遇到 quota/rate 错误自动暂停 30 分钟 - 支持 uploads playlist + search fallback """ # 网址 获取频道ID # https://www.tunepocket.com/youtube-channel-id-finder # 申请OATH2 # https://console.cloud.google.com/apis/library/youtube.googleapis.com # 启用 # YouTube Data API v3 import os import pickle import time import random from datetime import datetime, timedelta, timezone from googleapiclient.discovery import build from google_auth_oauthlib.flow import InstalledAppFlow from google.auth.transport.requests import Request import logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) # ===================== 配置 ===================== CLIENT_SECRETS_FILE = "client_secrets.json" SCOPES = [ "https://www.googleapis.com/auth/youtube", "https://www.googleapis.com/auth/youtube.force-ssl", "https://www.googleapis.com/auth/youtubepartner" ] API_SERVICE_NAME = "youtube" API_VERSION = "v3" CHECK_INTERVAL_SECONDS = 1800 # 检查间隔：30 分钟 # CHANNEL_ID = "UC-8V3pDMlM2B75fvbYRBsNg" #vlog CHANNEL_ID = "UCXFErW9DeE8EXrInlnyCnVw" #food # 安全删除参数 MAX_DAILY_DELETES = 30 # 每天最多删 30 个（可调 10~50） SHORT_DELAY_MIN = 5 # 每删一个最小等待秒 SHORT_DELAY_MAX = 20 # 每删一个最大等待秒 LONG_DELAY_AFTER = 5 # 每删几个后长休息 LONG_DELAY_MIN = 60 # 长休息最小秒 LONG_DELAY_MAX = 300 # 长休息最大秒 LIMIT_PAUSE_SECONDS = 1800 # 配额/限流时暂停 30 分钟 # 新增开关：是否只删除最近 48 小时内的直播存档 DELETE_ONLY_RECENT_48H = True # True = 只删 48 小时内；False = 删除所有 # ================================================ CREDENTIALS_FILE = "youtube_token.pickle" def get_authenticated_service(): credentials = None if os.path.exists(CREDENTIALS_FILE): with open(CREDENTIALS_FILE, "rb") as token: credentials = pickle.load(token) if not credentials or not credentials.valid: if credentials and credentials.expired and credentials.refresh_token: credentials.refresh(Request()) else: flow = InstalledAppFlow.from_client_secrets_file(CLIENT_SECRETS_FILE, SCOPES) credentials = flow.run_local_server(port=0) with open(CREDENTIALS_FILE, "wb") as token: pickle.dump(credentials, token) return build(API_SERVICE_NAME, API_VERSION, credentials=credentials) def delete_live_archive_videos(youtube): try: deleted_count = 0 uploads_playlist_id = None # 尝试获取 uploads playlist try: channels_response = youtube.channels().list( part="contentDetails", id=CHANNEL_ID ).execute() if channels_response.get("items"): related_playlists = channels_response["items"][0].get("contentDetails", {}).get("relatedPlaylists", {}) uploads_playlist_id = related_playlists.get("uploads") if uploads_playlist_id: logger.info(f"找到 uploads playlist ID: {uploads_playlist_id}") else: logger.warning("未找到 uploads playlist - 尝试 search fallback") except Exception as e: logger.warning(f"获取 uploads playlist 失败: {e}") video_ids = [] # 优先用 uploads playlist if uploads_playlist_id: next_page_token = None while True: playlist_response = youtube.playlistItems().list( part="snippet", playlistId=uploads_playlist_id, maxResults=50, pageToken=next_page_token ).execute() for item in playlist_response.get("items", []): video_ids.append(item["snippet"]["resourceId"]["videoId"]) next_page_token = playlist_response.get("nextPageToken") if not next_page_token: break time.sleep(2) logger.info(f"从 uploads playlist 发现 {len(video_ids)} 个视频 ID") # fallback: search 已结束直播 else: next_page_token = None while True: search_response = youtube.search().list( part="id,snippet", channelId=CHANNEL_ID, eventType="completed", type="video", maxResults=50, pageToken=next_page_token, order="date" ).execute() for item in search_response.get("items", []): if item["id"]["kind"] == "youtube#video": video_ids.append(item["id"]["videoId"]) next_page_token = search_response.get("nextPageToken") if not next_page_token: break time.sleep(2) logger.info(f"从 search 发现 {len(video_ids)} 个潜在直播存档视频 ID") if not video_ids: logger.info("未发现任何视频 ID，本次检查结束") return # 计算 48 小时前的 UTC 时间（如果开关开启） cutoff_time = None if DELETE_ONLY_RECENT_48H: cutoff_time = datetime.now(timezone.utc) - timedelta(hours=48) logger.info(f"只删除结束时间晚于 {cutoff_time.isoformat()} 的存档") # 批量检查 + 慢速删除 batch_size = 20 for i in range(0, len(video_ids), batch_size): if deleted_count >= MAX_DAILY_DELETES: logger.info(f"达到每日上限 {MAX_DAILY_DELETES} 个，停止本次检查") break batch = video_ids[i:i + batch_size] try: videos_response = youtube.videos().list( part="id,snippet,liveStreamingDetails", id=",".join(batch) ).execute() except Exception as e: logger.error(f"批量获取视频详情失败: {e}") continue for video in videos_response.get("items", []): if deleted_count >= MAX_DAILY_DELETES: break video_id = video["id"] title = video["snippet"].get("title", "无标题") live_details = video.get("liveStreamingDetails", {}) actual_end_time_str = live_details.get("actualEndTime") if not actual_end_time_str: continue # 解析结束时间 try: end_time = datetime.fromisoformat(actual_end_time_str.replace("Z", "+00:00")) except: continue # 如果开关开启，只删 48 小时内的 if DELETE_ONLY_RECENT_48H and end_time < cutoff_time: continue # 慢速删除 try: delay = random.uniform(SHORT_DELAY_MIN, SHORT_DELAY_MAX) logger.info(f"找到直播存档: {title} | {video_id} | 结束于 {actual_end_time_str}") logger.info(f"等待 {delay:.1f} 秒后删除...") time.sleep(delay) youtube.videos().delete(id=video_id).execute() logger.info(f"成功删除: {title} | {video_id}") deleted_count += 1 if deleted_count % LONG_DELAY_AFTER == 0: long_delay = random.uniform(LONG_DELAY_MIN, LONG_DELAY_MAX) logger.info(f"已删 {deleted_count} 个，长休息 {long_delay/60:.1f} 分钟...") time.sleep(long_delay) except Exception as e: err_str = str(e) logger.error(f"删除失败 {video_id}: {err_str}") if any(word in err_str.lower() for word in ["quota", "rate", "limit", "exceeded"]): logger.warning(f"触发限流/配额，暂停 {LIMIT_PAUSE_SECONDS//60} 分钟...") time.sleep(LIMIT_PAUSE_SECONDS) time.sleep(3) # 批次间额外间隔 if deleted_count > 0: logger.info(f"本次检查删除了 {deleted_count} 个直播存档") else: logger.info("本次未发现或未删除任何直播存档") except Exception as e: logger.exception("检查/删除过程中发生严重错误") def main(): youtube = get_authenticated_service() logger.info("YouTube 直播存档自动删除任务启动...") logger.info(f"监控频道: {CHANNEL_ID}") logger.info(f"检查间隔: {CHECK_INTERVAL_SECONDS} 秒") logger.info(f"每日删除上限: {MAX_DAILY_DELETES} 个") logger.info(f"只删除最近 48 小时内存档: {'是' if DELETE_ONLY_RECENT_48H else '否'}\n") while True: try: delete_live_archive_videos(youtube) except Exception as e: logger.exception("主循环异常") time.sleep(CHECK_INTERVAL_SECONDS) if __name__ == "__main__": main()