YouTube Learning Skill

def generate_notes_and_card_prompts(
    youtube_url: str, max_cards: int = 5
) -> tuple[str, dict | None]:
    """两步生成：先笔记（需要视频），再图卡 prompt（纯文本）"""
    client = init_client()

    # Step 1: Streaming 模式分析视频生成笔记
    # 传入 YouTube URL，Gemini 原生支持视频输入
    notes_markdown = _generate_notes(client, youtube_url)

    # Step 2: 根据笔记生成图卡 prompt（不需要视频）
    # 纯文本输入，速度快且稳定
    card_prompts = _generate_card_prompts(client, notes_markdown, max_cards)

    return notes_markdown, card_prompts

# Step 1: Streaming 模式避免长视频超时
for chunk in client.models.generate_content_stream(
    model=MODEL,
    contents=[types.Content(role="user", parts=[
        types.Part(file_data=types.FileData(
            file_uri=youtube_url, mime_type="video/mp4"  # Gemini 原生 YouTube 支持
        )),
        types.Part(text=prompt)
    ])],
    config=types.GenerateContentConfig(temperature=0.7),
):
    chunks.append(chunk.candidates[0].content.parts[0].text)

def generate_cards_from_prompts(card_prompts: dict, output_dir: str) -> list[str]:
    """从图卡 prompt 并行生成图片"""
    from concurrent.futures import ThreadPoolExecutor, as_completed

    cards = card_prompts.get("cards", [])
    results: dict[int, str | None] = {}

    # 所有图卡并行生成，max_workers = 图卡数量
    with ThreadPoolExecutor(max_workers=len(cards)) as executor:
        futures = {
            executor.submit(_gen_one, i, card): i
            for i, card in enumerate(cards, 1)
        }
        for future in as_completed(futures):
            idx, path = future.result()
            results[idx] = path

    # 按顺序返回成功的图片路径
    return [results[i] for i in sorted(results) if results[i]]

def generate_image_from_prompt(client, prompt, output_path, filename, max_retries=2):
    """单张图卡生成，支持自动重试"""
    for attempt in range(max_retries + 1):
        try:
            if attempt > 0:
                time.sleep(3)  # 重试间隔
            response = client.models.generate_content(
                model="gemini-3-pro-image-preview",
                contents=[IMAGE_GEN_PREFIX + prompt],  # 统一风格前缀
                config=types.GenerateContentConfig(
                    response_modalities=["image", "text"],
                    image_config=types.ImageConfig(aspect_ratio="16:9", image_size="2K")
                )
            )

# 防重复处理
processing_tasks: set[str] = set()

def handle_card_action(data: P2CardActionTrigger) -> P2CardActionTriggerResponse:
    action_value = action.get("value", {})

    if action_value.get("action") == "start_learning":
        video_url = action_value.get("video_url")

        # 防重入：同一视频不重复处理
        if video_url in processing_tasks:
            resp.toast = CallBackToast(type="warning", content="该视频正在处理中...")
            return resp

        processing_tasks.add(video_url)

        # 后台线程执行流水线，回调立即返回
        thread = threading.Thread(
            target=process_video,
            args=(video_url, video_title, callback_token, chat_id),
            daemon=True
        )
        thread.start()

        # 立即返回「处理中」卡片
        card = CallBackCard(type="raw", data=generate_processing_card(video_title))
        resp.card = card
        return resp

def process_video(video_url, video_title, callback_token, chat_id):
    try:
        # Step 1: Gemini 生成笔记
        notes_markdown, card_prompts = generate_notes_and_card_prompts(video_url)
        # Step 2: 并行生成图卡
        image_paths = generate_cards_from_prompts(card_prompts, str(output_dir))
        # Step 3: 同步飞书文档
        doc_url = sync_to_feishu(notes_path, video_title, video_url, image_paths)
        # 更新卡片为完成状态
        update_card_via_api(callback_token, generate_completed_card(...))
    except Exception as e:
        update_card_via_api(callback_token, generate_error_card(video_title, str(e)))
    finally:
        processing_tasks.discard(video_url)  # 清理防重入标记

def load_prompts() -> tuple[str, str]:
    """读取用户背景和输出格式配置"""
    user_profile_path = PROJECT_DIR / "references" / "user_profile.md"
    output_format_path = PROJECT_DIR / "references" / "output_format.md"

    user_profile = user_profile_path.read_text(encoding='utf-8') if user_profile_path.exists() else ""
    output_format = output_format_path.read_text(encoding='utf-8') if output_format_path.exists() else ""
    return user_profile, output_format

def _generate_notes(client, youtube_url) -> str:
    user_profile, output_format = load_prompts()

    prompt = f"""你是一个专业的视频学习助手。请仔细观看这个 YouTube 视频，生成学习笔记。

## 视频信息
链接: {youtube_url}

## 用户背景
{user_profile}

## 输出格式要求
{output_format}

请严格按照输出格式要求生成学习笔记。注意：
3. Insight 部分必须结合用户背景来写
4. 详细内容部分要保留所有有价值的信息
6. 链接字段必须使用上面提供的实际 YouTube 链接
"""

def get_oauth_tokens(client_id: str, client_secret: str) -> dict:
    """执行 OAuth 流程：本地服务器接收回调"""
    auth_url = f"{GOOGLE_AUTH_URL}?" + urlencode({
        "client_id": client_id, "redirect_uri": REDIRECT_URI,
        "response_type": "code", "scope": " ".join(SCOPES),
        "access_type": "offline", "prompt": "consent"
    })

    # 启动本地服务器接收回调
    server = HTTPServer(('localhost', 8888), OAuthCallbackHandler)
    server_thread = threading.Thread(target=server.handle_request)
    server_thread.start()
    webbrowser.open(auth_url)  # 自动打开浏览器

    server_thread.join(timeout=120)
    server.server_close()

    # 用授权码换取 Token（含 refresh_token）
    resp = httpx.post(GOOGLE_TOKEN_URL, data={
        "client_id": client_id, "client_secret": client_secret,
        "code": OAuthCallbackHandler.auth_code,
        "grant_type": "authorization_code", "redirect_uri": REDIRECT_URI
    })
    return resp.json()

# RSS Monitor 自动刷新 Token
def _get_youtube_access_token() -> str | None:
    tokens = json.loads(tokens_file.read_text())
    resp = httpx.post("https://oauth2.googleapis.com/token", data={
        "client_id": client_id, "client_secret": client_secret,
        "refresh_token": tokens["refresh_token"],
        "grant_type": "refresh_token"
    })
    return resp.json().get("access_token")