#!/usr/bin/env python3 # -*- coding: utf-8 -*- import datetime import base64 import html import json import os import re import sys import time import uuid import urllib.parse import urllib.request WORKSPACE = os.path.dirname(os.path.abspath(__file__)) OUTPUT_PATH = os.path.join(WORKSPACE, "ai-learning-videos.json") LIFE_DATA_PATH = os.path.join(WORKSPACE, "life-data.json") ROADMAP_URL = "https://agent.codepost.site/" ROADMAP_STAGE_CACHE_PATH = os.path.join(WORKSPACE, "ai-learning-stages-cache.json") MIN_DAILY_VIDEOS = 3 TARGET_DAILY_VIDEOS = 5 RECENT_WINDOW_DAYS = 183 STAGES = [ { "name": "Stage 0：环境准备 — Foundations", "goal": "按 AI Agent 学习路线图完成 Python/CLI/Git/API/JSON 基础，跑通第一个本地或云端 AI 调用。", "queries": [ "Codex 安装 配置 入门 教程", "Claude Code 安装 配置 入门 教程", "Gemini CLI 安装 配置 入门 教程", "Ollama 本地大模型 安装 入门 教程", "AI Agent 环境准备 Python API JSON 教程", "Python API JSON 新手 实操 教程 AI", "Git 命令行 入门 实操 教程", ], "track": "AI Agent Foundations", }, { "name": "Stage 1：LLM 基础 — Language Model Basics", "goal": "理解 Token、模型供应商、本地模型和云端 API，完成一次跨模型调用和错误处理。", "queries": [ "LLM 基础 Token API 调用 教程", "OpenAI API 入门 Python 教程", "大模型 API 错误处理 重试 教程", "Ollama OpenAI API 对比 教程", "本地大模型 云端大模型 入门 教程", ], "track": "LLM 基础", }, { "name": "Stage 2：Prompt Engineering — 提示词工程", "goal": "学习 System Prompt、Few-shot、CoT 和 Context Engineering 入门，产出可复用提示词模板。", "queries": [ "Prompt Engineering 提示词工程 实操 教程", "System Prompt Few shot Chain of Thought 教程", "ChatGPT 提示词工程 新手 实战", "Context Engineering 入门 教程", "AI 提示词 模板 实操 教程", ], "track": "Prompt Engineering", }, { "name": "Track A1：CLI Agent 入门", "goal": "按网站 Track A 路线熟悉 Claude Code、Codex、Gemini CLI 等现成 CLI Agent，不写复杂代码先提升工作流。", "queries": [ "Claude Code 入门 教程 实操", "Codex CLI 入门 教程 实操", "Gemini CLI 入门 教程", "AI CLI Agent 入门 教程", "Claude Code Codex 工作台 使用教程", ], "track": "CLI Agent 入门", }, { "name": "Track A2：CLI 工作流 — Slash 命令/MCP/子代理", "goal": "把 CLI Agent 用到真实工作流：命令、上下文文件、MCP、子代理和自动化任务。", "queries": [ "Claude Code Slash 命令 MCP 教程", "Codex MCP 工作流 教程", "AI Agent MCP 入门 实操 教程", "Claude Code 子代理 subagent 教程", "CLI Agent 自动化 工作流 教程", ], "track": "CLI Agent 工作流", }, { "name": "Stage 5：Claude Code 生态 — Skills/CLAUDE.md", "goal": "学习网站强调的 Claude Code 生态能力：CLAUDE.md、Skills、项目上下文和可复用工作规范。", "queries": [ "Claude Code CLAUDE.md 教程", "Claude Code Skills 教程 实操", "Claude Code 项目上下文 教程", "AI 编程助手 项目规范 CLAUDE.md", "Claude Code 工作流 实战 教程", ], "track": "Claude Code 生态", }, { "name": "Track A3：CLI 生产级 — GitHub Actions/成本追踪/团队协作", "goal": "把 CLI Agent 从个人练习推进到可复用、可协作、可控成本的生产工作流。", "queries": [ "Claude Code GitHub Actions 自动化 教程", "AI Agent 成本追踪 教程", "Claude Code 团队协作 工作流", "Codex GitHub Actions 自动化 教程", "AI CLI Agent 生产级 工作流 教程", ], "track": "CLI Agent 生产级", }, { "name": "Stage 7.5：高阶概念 — Context/Harness Engineering", "goal": "理解网站的三层演进：Prompt Engineering、Context Engineering、Harness Engineering，并能迁移到自己的工作台流程。", "queries": [ "Context Engineering 入门 教程", "Harness Engineering AI Agent 教程", "AI Agent 上下文工程 实操", "Prompt Context Harness Engineering", "Claude Code 上下文管理 教程", ], "track": "Agent 高阶概念", }, { "name": "Stage 8：Agent Interface — Agent 接口", "goal": "学习 Agent 作为产品入口的形态：聊天、CLI、浏览器、自动化面板，并设计一个自己的最小 Agent 工作台入口。", "queries": [ "AI Agent Interface 交互设计 教程", "AI Agent 产品设计 工作台 教程", "Agent CLI Web UI 入门 教程", "AI Agent 自动化面板 教程", "从0到1 AI Agent 应用 教程", ], "track": "Agent 产品接口", }, ] MONETIZATION_QUERIES = [ "AI Agent 提效 案例 实操", "Claude Code 真实项目 案例", "Codex 自动化 工作流 案例", ] BEGINNER_FALLBACK_VIDEOS = [] LAST_BILIBILI_ERRORS = [] RECENT_VALIDATED_SEED_VIDEOS = [ ("【最新版Codex安装Git】保姆级教程和讲解 做到零风险操作Codex", "https://www.bilibili.com/video/BV14WGw6UEYu/", "AI Agent Foundations", "B站公开搜索已验证，最近半年内发布"), ("【2026最新】Codex 接入 Gemini 3.5 Flash｜第三方 API 配置实战｜macOS + Windows 双版本｜RouterLink 教程", "https://www.bilibili.com/video/BV1F8GR6sE6t/", "AI Agent Foundations", "B站公开搜索已验证，最近半年内发布"), ("Codex的安装并且如何配置转接第三方API完整使用教程#codex #ai #安装#大模型#API", "https://www.bilibili.com/video/BV1RNjb6JEZf/", "AI Agent Foundations", "B站公开搜索已验证，最近半年内发布"), ("最强 AI 组合诞生！Kimi K3 接入 Codex 教程，3 分钟搞定｜CC Switch", "https://www.bilibili.com/video/BV1gWKY6tE4U/", "CLI Agent 入门", "B站公开搜索已验证，最近半年内发布"), ("[中配]Claude Code 免费且无限制 OmniRoute 设置指南 - AI For You", "https://www.bilibili.com/video/BV1gF396AEBU/", "CLI Agent 入门", "B站公开搜索已验证，最近半年内发布"), ("Claude Code接入飞书教程, 手机远程操纵电脑写代码", "https://www.bilibili.com/video/BV15TPrzeELm/", "CLI Agent 入门", "B站公开搜索已验证，最近半年内发布"), ("Codex零基础速成系列-15分钟从入门到实战", "https://www.bilibili.com/video/BV1nDuv6tEhs/", "CLI Agent 入门", "B站搜索结果显示为最近发布"), ("【零基础玩转AI：Claude Code CLI实操指南】", "https://www.bilibili.com/video/BV1eWu261E1U/", "CLI Agent 入门", "B站搜索结果显示为最近发布"), ("【最新版】小白速通 Codex 教程（含 DeepSeek 接入，无需 ChatGPT 订阅）", "https://www.bilibili.com/video/BV1Zgud6LEoh/", "CLI Agent 入门", "B站搜索结果显示为最近发布"), ("【2026最新版】全网最适合初学者的Ollama教程，免费开源，带你从0到1吃透Ollama！", "https://www.bilibili.com/video/BV1KVd6BwEhs/", "AI Agent Foundations", "B站搜索结果显示为最近半年内发布"), ] PRACTICAL_KEYWORDS = ( "教程", "实操", "案例", "工作流", "自动化", "从0到1", "从零到一", "保姆级", "模板", "脚本", "复盘", "教学", "实战", "演示", "手把手", ) TOOL_KEYWORDS = ( "ChatGPT", "豆包", "Kimi", "Coze", "扣子", "剪映", "Canva", "即梦", "可灵", "Excel", "飞书", "多维表格", "PPT", "Word", "Codex", "Claude Code", "Gemini CLI", "Ollama", "MCP", "GitHub Actions", "LangGraph", "OpenAI API", "JSON", "Python", "CLI", "Agent", ) LOW_VALUE_KEYWORDS = ( "焦虑", "割韭菜", "暴富", "月入百万", "躺赚", "收徒", "训练营", "私域课", "卖课", "加盟", "骗局", "资讯", "新闻", "发布会", "热门项目推荐", "项目推荐", "神器", "效率起飞", "绕过", "破解", "限制解除", "道德限制", "越狱", "jailbreak", ) PLATFORM_SPECS = { "YouTube": { "search_sites": ("youtube.com/watch", "youtu.be"), "url_patterns": ("youtube.com/watch", "youtu.be/"), "note": "YouTube 公开视频搜索", }, "小红书": { "search_sites": ("xiaohongshu.com/explore", "xiaohongshu.com/discovery/item", "xhslink.com"), "url_patterns": ("xiaohongshu.com/explore", "xiaohongshu.com/discovery/item", "xhslink.com"), "note": "小红书公开网页搜索", }, "抖音": { "search_sites": ("douyin.com/video", "douyin.com/note", "iesdouyin.com"), "url_patterns": ("douyin.com/video", "douyin.com/note", "iesdouyin.com"), "note": "抖音公开网页搜索", }, } STAGE_KEYWORDS = ( "Python", "CLI", "Git", "API", "JSON", "Ollama", "LLM", "Token", "OpenAI SDK", "Anthropic", "Prompt Engineering", "System Prompt", "Few-shot", "Chain-of-Thought", "Context Engineering", "Function Calling", "ReAct", "Tool Use", "Schema", "RAG", "Vector", "Embedding", "LangGraph", "CrewAI", "MCP", "Claude Code", "Codex", "Gemini CLI", "GitHub Actions", "Agent Interface", "Browser Agent", "Memory", "Observability", "Evaluation", "Prompt", "Agent", ) TRADITIONAL_HINTS = { "環境": "环境", "準備": "准备", "基礎": "基础", "學": "学", "什麼": "什么", "動手": "动手", "調用": "调用", "供應商": "供应商", "雲端": "云端", "取捨": "取舍", "提示詞": "提示词", "設計": "设计", "推理鏈": "推理链", "練習": "练习", "內嵌": "内嵌", "構建": "构建", "選擇": "选择", "轉折點": "转折点", "循環": "循环", "記憶": "记忆", "檢索": "检索", "觀測性": "观测性", "生產": "生产", "級": "级", "開發": "开发", "錯誤": "错误", "處理": "处理", "項目": "项目", "文檔": "文档", "實戰": "实战", "第一個": "第一个", "對比": "对比", "本地": "本地", } STAGE_BEGINNER_BLOCKLIST = ( "Cursor", "Coze", "扣子", "RAG", "LangChain", "部署", "模型训练", "微调", "Sora", "即梦", "可灵", "短片", "电影", "影视", "漫剧", "漫画", "数字人", "全套", "全36集", "全30集", "AI视频", "视频提示词", "分镜", "故事板", "剧本", "GPT-Image", "ComfyUI", "AI绘画", "绘画", "生图", "图片", "图像", "海报", "Midjourney", "识图", "插件", "国赛", "竞赛", "数学建模", "论文", "科研", "提示词赚钱", "新媒体", "小红书", "抖音", "直播", "电商", "运营", "起号", "涨粉", "音乐", "Suno", "图片", "文生图", "图生图", "无水印", "下载", "系统课", "通关课", "必修课", "训练营", "马士兵", "邪修", "番茄", "签约", "签一篇", "过签", "结算", "战绩", "就业", "天花板", "最全", "出图", "参考图", "论文", "essay", "白嫖", "百万年薪", "终极", "小说", "完整版", "资源", "GPT5.6", "GPT 5.6", "ChatGPT 5.6", "今日头条", "情感赛道", "Seedance", "生成视频", "生图", "制图", "绘图", "广告", "设计师", "skill", "Skill", "违规", "欺骗", "进阶", "高级", "自动化测试", "Playwright", "Copilot CLI", "OpenCode", ) def now_dt(): return datetime.datetime.now(datetime.timezone(datetime.timedelta(hours=8))) def now_str(): return now_dt().strftime("%Y-%m-%d %H:%M:%S") def browser_headers(referer="https://search.bilibili.com/"): return { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36", "Referer": referer, "Origin": "https://search.bilibili.com", "Accept": "application/json, text/plain, */*", "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", "Cookie": f"buvid3={uuid.uuid4().hex.upper()}infoc; b_nut={int(time.time())}; CURRENT_FNVAL=4048;", } def fetch_json(url, timeout=15, referer="https://search.bilibili.com/"): req = urllib.request.Request(url, headers=browser_headers(referer)) with urllib.request.urlopen(req, timeout=timeout) as resp: return json.loads(resp.read().decode("utf-8", "ignore")) def fetch_text(url, timeout=20, referer=None): headers = browser_headers(referer or url) headers["Accept"] = "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8" req = urllib.request.Request(url, headers=headers) with urllib.request.urlopen(req, timeout=timeout) as resp: return resp.read().decode("utf-8", "ignore") def clean_title(title): title = re.sub(r"]*>|", "", title or "", flags=re.I) title = re.sub(r"<[^>]+>", "", title) return html.unescape(re.sub(r"\s+", " ", title)).strip() def strip_html(text): text = re.sub(r"]*>.*?", " ", text or "", flags=re.I | re.S) text = re.sub(r"]*>.*?", " ", text, flags=re.I | re.S) text = re.sub(r"<[^>]+>", " ", text) return html.unescape(re.sub(r"\s+", " ", text)).strip() def to_simplified_hint(text): out = text or "" for trad, simple in TRADITIONAL_HINTS.items(): out = out.replace(trad, simple) return out def stage_title_from_html(raw_title, stage_num): title = re.sub(r"]*>.*?", "", raw_title or "", flags=re.I | re.S) title = to_simplified_hint(strip_html(title)) title = re.sub(r"\s+", " ", title).strip(" -—") if re.match(r"^(Stage|Track)\b", title, flags=re.I): return title return f"Stage {stage_num}：{title}" if stage_num is not None else title def compact_keyword_text(text): text = to_simplified_hint(strip_html(text)) text = re.sub(r"[^\w\u4e00-\u9fff\-\s]+", " ", text) return re.sub(r"\s+", " ", text).strip() def extract_stage_keywords(text, limit=8): compact = compact_keyword_text(text) found = [] low = compact.lower() for keyword in STAGE_KEYWORDS: if keyword.lower() in low and keyword not in found: found.append(keyword) cjk_terms = re.findall(r"[\u4e00-\u9fffA-Za-z][\u4e00-\u9fffA-Za-z0-9\-\+]{2,}", compact) stop = {"Stage", "Track", "Foundations", "Language", "Model", "Basics", "Agent", "什么", "动手做", "完整文档", "分钟从零", "共修"} for term in cjk_terms: if term in stop or term.lower() in {s.lower() for s in found}: continue if any(term in existing or existing in term for existing in found): continue if re.search(r"[\u4e00-\u9fff]", term) or term in STAGE_KEYWORDS: found.append(term) if len(found) >= limit: break return found[:limit] def build_queries_from_stage(title, body_text, keywords): title_core = re.sub(r"Stage\s*[\d.]+[:：]?", "", title or "", flags=re.I) title_core = re.sub(r"Track\s*[A-Z]\d*[:：]?", "", title_core, flags=re.I) title_core = re.sub(r"\b(A\+B|Track A|Track B|共修)\b", " ", title_core, flags=re.I) title_core = re.sub(r"\s+", " ", title_core).strip(" -—") keywords = [k for k in keywords if k] core_terms = " ".join(keywords[:4]) queries = [] for q in ( f"{title_core} AI Agent 教程 实操", f"{title_core} 入门 教程", f"{core_terms} 教程 实战", f"{core_terms} AI Agent 入门", ): q = re.sub(r"\s+", " ", q).strip() if len(q) >= 6 and q not in queries: queries.append(q) for keyword in keywords[:5]: for suffix in ("教程 实操", "入门 教程"): q = f"{keyword} {suffix}" if q not in queries: queries.append(q) if not queries: queries = ["AI Agent 入门 教程", "AI Agent 实操 教程"] return queries[:8] def extract_roadmap_stages(page_html): markers = list(re.finditer(r']*data-stage="([^"]+)"[^>]*data-track="([^"]+)"[^>]*>', page_html or "", flags=re.I)) stages = [] for idx, marker in enumerate(markers): start = marker.start() end = markers[idx + 1].start() if idx + 1 < len(markers) else (page_html or "").find("\n\n", marker.end()) if end <= start: end = len(page_html or "") block = (page_html or "")[start:end] stage_num = marker.group(1) track = marker.group(2) title_match = re.search(r"]*>(.*?)", block, flags=re.I | re.S) title = stage_title_from_html(title_match.group(1) if title_match else f"Stage {stage_num}", stage_num) paragraphs = re.findall(r"]*>(.*?)

", block, flags=re.I | re.S) exercise_desc = re.findall(r']*>(.*?)', block, flags=re.I | re.S) resource_desc = re.findall(r']*>(.*?)', block, flags=re.I | re.S) body_text = compact_keyword_text(" ".join(paragraphs + exercise_desc + resource_desc)) keywords = extract_stage_keywords(f"{title} {body_text}") stages.append({ "name": title, "goal": body_text[:260] or f"按路线图完成 {title} 阶段的学习与练习。", "queries": build_queries_from_stage(title, body_text, keywords), "track": title.replace("Stage ", "").split("：", 1)[-1].strip() or "AI Agent 学习路线图", "stageId": stage_num, "roadmapTrack": track, "keywords": keywords, "sourceUrl": ROADMAP_URL + "#stages", }) return stages def load_cached_stages(): try: with open(ROADMAP_STAGE_CACHE_PATH, "r", encoding="utf-8-sig") as f: data = json.load(f) stages = data.get("stages") if isinstance(data, dict) else None return stages if isinstance(stages, list) and stages else None except Exception: return None def load_live_stages(notes): try: page = fetch_text(ROADMAP_URL, timeout=25, referer=ROADMAP_URL) stages = extract_roadmap_stages(page) if len(stages) < 3: raise ValueError(f"只解析到 {len(stages)} 个 Stage") cache = {"updatedAt": now_str(), "sourceUrl": ROADMAP_URL + "#stages", "stages": stages} with open(ROADMAP_STAGE_CACHE_PATH, "w", encoding="utf-8") as f: json.dump(cache, f, ensure_ascii=False, indent=2) notes.append(f"已先抓取 {ROADMAP_URL}#stages，并解析 {len(stages)} 个 Stage；本次关键词来自当前阶段网页内容。") return stages, "live" except Exception as e: cached = load_cached_stages() if cached: notes.append(f"路线图网页抓取失败，改用上次缓存的 Stage：{e}") return cached, "cache" notes.append(f"路线图网页抓取失败且无缓存，改用内置 Stage 兜底：{e}") return STAGES, "fallback" def normalize_number(value): if value in (None, "", "-"): return None if isinstance(value, (int, float)): return int(value) text = str(value).strip().lower() try: if text.endswith("万"): return int(float(text[:-1]) * 10000) if text.endswith("亿"): return int(float(text[:-1]) * 100000000) return int(float(text)) except Exception: return None def load_completed_count(): try: with open(LIFE_DATA_PATH, "r", encoding="utf-8-sig") as f: data = json.load(f) history = ((data.get("aiLearn") or {}).get("history") or []) return len([h for h in history if isinstance(h, dict) and h.get("done")]) except Exception: return 0 def get_completed_count_from_args(): for arg in sys.argv[1:]: if arg.startswith("--completed-count="): try: return max(0, int(arg.split("=", 1)[1])) except Exception: return None return None def get_refresh_offset_from_args(): for arg in sys.argv[1:]: if arg.startswith("--refresh-offset="): try: return max(0, int(arg.split("=", 1)[1])) except Exception: return 0 return 0 def stage_for_count(completed_count, stages=None): stages = stages or STAGES idx = min(len(stages) - 1, max(0, int(completed_count or 0) // 7)) return idx, stages[idx] def get_published_dt(pubdate): if not pubdate: return None try: return datetime.datetime.fromtimestamp(int(pubdate), tz=datetime.timezone(datetime.timedelta(hours=8))) except Exception: return None def is_recent_enough(pubdate): published = get_published_dt(pubdate) if not published: return False return 0 <= (now_dt() - published).days <= RECENT_WINDOW_DAYS def format_published_at(pubdate): published = get_published_dt(pubdate) return published.strftime("%Y-%m-%d") if published else None def get_freshness_label(pubdate, is_classic=False): if is_classic: return "经典教程补足" published = get_published_dt(pubdate) if not published: return "发布时间未知" days = max(0, (now_dt() - published).days) if days <= 7: return "近 7 天" if days <= 30: return "近 30 天" if days <= RECENT_WINDOW_DAYS: return "最近半年" return "超过半年" def practical_score(title, tags="", pubdate=None): text = f"{title} {tags}" score = 0 score += sum(4 for k in PRACTICAL_KEYWORDS if k.lower() in text.lower()) score += sum(4 for k in ("tutorial", "guide", "setup", "install", "beginner", "basics", "hands-on", "demo", "walkthrough") if k in text.lower()) score += sum(3 for k in TOOL_KEYWORDS if k.lower() in text.lower()) score -= sum(6 for k in LOW_VALUE_KEYWORDS if k in text) if is_recent_enough(pubdate): published = get_published_dt(pubdate) days = (now_dt() - published).days if published else RECENT_WINDOW_DAYS score += max(4, 12 - min(days, RECENT_WINDOW_DAYS) // 20) if any(k in text.lower() for k in ("入门", "新手", "小白", "零基础", "0基础", "beginner", "basics", "from scratch")): score += 4 return score def is_stage_appropriate(title, stage_index): text = title or "" if stage_index == 0: if len(re.sub(r"\s+", "", text)) < 8: return False if any(k.lower() in text.lower() for k in ("mcp", "subagent", "subagents", "多agent", "多智能体", "生产环境", "生产级", "扩展实战", "adapter", "server", "langgraph", "crew", "绕过", "破解", "道德限制", "越狱", "jailbreak", "无法使用的问题")): return False if any(k.lower() in text.lower() for k in STAGE_BEGINNER_BLOCKLIST): return False if re.search(r"全\d+[集讲]|[1-9]\d{2,}\s*集", text): return False has_tool = any(k.lower() in text.lower() for k in ("python", "cli", "git", "api", "json", "ollama", "codex", "claude code", "gemini cli", "agent", "llm", "大模型")) has_simple_task = any(k.lower() in text.lower() for k in ("新手", "入门", "基础", "安装", "环境", "教程", "实操", "命令行", "使用", "指南", "从零", "0基础", "tutorial", "guide", "setup", "install", "beginner", "basics")) return has_tool and has_simple_task if stage_index <= 1: if any(k.lower() in text.lower() for k in ("部署", "模型训练", "微调", "langchain", "rag", "多智能体")): return False return True def title_matches_stage_keywords(title, stage): keywords = [k for k in (stage.get("keywords") or []) if len(str(k).strip()) >= 2] if not keywords: return True weak = {"AI", "Agent", "CLI", "教程", "入门", "实操", "基础", "学什么"} meaningful = [k for k in keywords if k not in weak] if not meaningful: meaningful = keywords text = (title or "").lower() return any(str(k).lower() in text for k in meaningful) def infer_difficulty(title, stage_index): text = title or "" if stage_index <= 1 or any(k in text for k in ("入门", "新手", "小白", "零基础", "保姆级")): return "入门" if any(k in text for k in ("进阶", "高级", "复杂", "部署", "开发")): return "进阶" return "初级" def infer_track(title, default_track): text = title or "" if any(k.lower() in text.lower() for k in ("python", "git", "json", "api", "ollama", "环境", "命令行")): return "AI Agent Foundations" if any(k.lower() in text.lower() for k in ("llm", "token", "openai api", "模型调用", "错误处理", "重试")): return "LLM 基础" if any(k.lower() in text.lower() for k in ("prompt", "提示词", "system prompt", "few-shot", "cot", "chain of thought")): return "Prompt Engineering" if any(k.lower() in text.lower() for k in ("claude code", "codex", "gemini cli", "cli agent")): return "CLI Agent 入门" if any(k.lower() in text.lower() for k in ("mcp", "slash", "子代理", "subagent", "工作流")): return "CLI Agent 工作流" if any(k in text for k in ("CLAUDE.md", "Skills", "技能", "项目上下文")): return "Claude Code 生态" if any(k.lower() in text.lower() for k in ("github actions", "成本", "团队协作", "生产级")): return "CLI Agent 生产级" if any(k.lower() in text.lower() for k in ("context engineering", "harness engineering", "上下文工程")): return "Agent 高阶概念" if any(k.lower() in text.lower() for k in ("agent interface", "web ui", "产品设计", "工作台", "自动化面板")): return "Agent 产品接口" if any(k in text for k in ("办公", "周报", "会议", "PPT", "Word", "飞书")): return "AI办公自动化" if any(k in text for k in ("小红书", "文案", "脚本", "短视频", "标题")): return "AI内容生产" if any(k in text for k in ("图片", "海报", "商品图", "主图", "电商", "Canva", "即梦")): return "AI设计/电商素材" if any(k in text for k in ("Excel", "表格", "数据", "多维表格")): return "AI数据整理" if any(k in text for k in ("Coze", "扣子", "工作流", "自动化", "低代码")): return "AI低代码/自动化工作流" if any(k in text for k in ("变现", "副业", "接单", "案例拆解")): return "AI变现案例" return default_track def build_learning_fields(title, stage, priority): track = infer_track(title, stage["track"]) if track == "AI Agent Foundations": task = "跟着视频完成一个最小环境检查：命令行、Python/Git/API/JSON 或 Ollama 任一项跑通。" output = "一份 AI Agent 学习环境检查清单" elif track == "LLM 基础": task = "复现一次模型调用或本地/云端模型对比，并记录输入、输出、报错处理。" output = "一份 LLM 调用实验记录" elif track == "Prompt Engineering": task = "整理 3 个 System Prompt/Few-shot 模板，并用同一任务测试效果差异。" output = "一套可复用提示词模板" elif track == "CLI Agent 入门": task = "用 Codex、Claude Code 或 Gemini CLI 完成一个真实小任务，并记录命令和结果。" output = "一个 CLI Agent 小任务记录" elif track == "CLI Agent 工作流": task = "把一个重复任务拆成输入、上下文、命令、输出四步，尝试接入 MCP 或 Slash 命令。" output = "一个可复用 CLI Agent 工作流" elif track == "Claude Code 生态": task = "为一个项目写一份 CLAUDE.md/工作规范，并测试它是否能约束 AI 输出。" output = "一份项目级 AI 协作规范" elif track == "CLI Agent 生产级": task = "设计一个可协作的自动化流程，记录触发条件、成本控制和失败处理。" output = "一份生产级 CLI Agent 流程草案" elif track == "Agent 高阶概念": task = "把 Prompt、Context、Harness 三层分别映射到你自己的一个工作流。" output = "一张 Agent 能力分层拆解表" elif track == "Agent 产品接口": task = "画出一个最小 Agent 入口：用户输入、工具调用、结果展示和保存记录。" output = "一个最小 Agent 工作台入口方案" elif track == "AI办公自动化": task = "跟着视频做一份可复用的办公提示词或自动化流程。" output = "一个可复制的周报/会议纪要/邮件处理模板" elif track == "AI内容生产": task = "用视频里的方法生成 5 个标题和 1 段短视频脚本。" output = "一套小红书标题模板或短视频脚本草稿" elif track == "AI设计/电商素材": task = "按视频流程做一张商品图、封面或海报。" output = "一张可发布的商品主图或内容封面" elif track == "AI数据整理": task = "用示例表格复现一次清洗、分类或摘要流程。" output = "一份整理后的表格和操作提示词" elif track == "AI低代码/自动化工作流": task = "搭一个最小可用工作流，只保留输入、处理、输出三步。" output = "一个可运行的简单 AI 工作流" elif track == "AI变现案例": task = "拆解视频中的交付物、客户、流程和报价方式。" output = "一份 AI 变现案例复盘表" else: task = "把视频中的提示词方法复现一遍，并保存可复用模板。" output = "一个可复用的 AI 提示词模板" return { "track": track, "whyWatch": "标题包含明确实操/教程/工具线索，适合按步骤跟做，不是纯概念内容。", "practicalValue": f"看完应能掌握：{task.replace('跟着视频', '').replace('按视频流程', '').replace('用视频里的方法', '').strip()}", "practiceTask": task, "outputExample": output, "priority": priority, } def collect_bilibili(stage_index, stage): global LAST_BILIBILI_ERRORS LAST_BILIBILI_ERRORS = [] beginner_monetization = ["ChatGPT 办公 变现 案例", "AI 提示词 变现 案例", "Kimi AI 办公 效率 案例"] queries = list(stage["queries"]) + (beginner_monetization if stage_index == 0 else MONETIZATION_QUERIES) results = [] seen = set() for query in queries: for order in ("pubdate", "totalrank"): for page in (1, 2, 3): url = "https://api.bilibili.com/x/web-interface/search/type?" + urllib.parse.urlencode( {"search_type": "video", "keyword": query, "page": page, "page_size": 30, "order": order} ) try: data = fetch_json(url, referer="https://search.bilibili.com/all?keyword=" + urllib.parse.quote(query)) except Exception as e: err = f"{type(e).__name__}: {e}" if err not in LAST_BILIBILI_ERRORS: LAST_BILIBILI_ERRORS.append(err) continue if data.get("code") != 0: continue for item in (data.get("data") or {}).get("result") or []: title = clean_title(item.get("title")) bvid = item.get("bvid") or "" video_url = f"https://www.bilibili.com/video/{bvid}/" if bvid else str(item.get("arcurl") or "").replace("http://", "https://") if not title or "bilibili.com/video" not in video_url or video_url in seen: continue if not is_stage_appropriate(title, stage_index): continue if not title_matches_stage_keywords(title, stage): continue if not is_recent_enough(item.get("pubdate")): continue tag_text = str(item.get("tag") or "") score = practical_score(title, tag_text, item.get("pubdate")) if score < 4: continue plays = normalize_number(item.get("play")) likes = normalize_number(item.get("like")) rec = { "id": bvid or re.sub(r"\W+", "", video_url)[-20:], "platform": "B站", "title": title, "url": video_url, "difficulty": infer_difficulty(title, stage_index), "plays": plays, "playCount": plays, "likes": likes, "likeCount": likes, "fans": None, "followerCount": None, "author": item.get("author") or "", "duration": item.get("duration") or None, "tags": [t for t in re.split(r"[,，#\s]+", tag_text) if t][:5], "capturedAt": now_str(), "sourceQuery": query, "pubdate": item.get("pubdate") or None, "publishedAt": format_published_at(item.get("pubdate")), "freshnessLabel": get_freshness_label(item.get("pubdate")), "isClassic": False, "score": score, "completed": False, "completedAt": None, "notes": "B站公开搜索接口抓取；粉丝数未稳定获取时保留为 null。", } results.append(rec) seen.add(video_url) results.sort(key=lambda v: (v.get("score") or 0, v.get("plays") or 0, v.get("likes") or 0), reverse=True) return results def decode_possible_bing_url(url): url = html.unescape(url or "").replace("\\/", "/") parsed = urllib.parse.urlparse(url) query = urllib.parse.parse_qs(parsed.query) for key in ("u", "url", "r"): value = (query.get(key) or [None])[0] if not value: continue value = urllib.parse.unquote(value) if value.startswith("a1"): encoded = value[2:] try: padding = "=" * (-len(encoded) % 4) value = base64.urlsafe_b64decode((encoded + padding).encode("ascii")).decode("utf-8", "ignore") except Exception: pass if value.startswith("http"): return value encoded_url = re.search(r"https?%3A%2F%2F[^&\"'<>]+", url, flags=re.I) if encoded_url: return urllib.parse.unquote(encoded_url.group(0)) return url def normalize_platform_url(url, platform): url = decode_possible_bing_url(url) url = urllib.parse.unquote(html.unescape(url or "")).replace("\\/", "/") if not url.startswith("http"): return "" parsed = urllib.parse.urlparse(url) host = parsed.netloc.lower() path = parsed.path query = urllib.parse.parse_qs(parsed.query) if platform == "YouTube": if "youtube.com" in host and path == "/watch" and query.get("v"): return "https://www.youtube.com/watch?v=" + query["v"][0] if "youtu.be" in host and path.strip("/"): return "https://youtu.be/" + path.strip("/").split("/")[0] if platform == "小红书": if "xhslink.com" in host: return urllib.parse.urlunparse(("https", parsed.netloc, parsed.path, "", parsed.query, "")) if "xiaohongshu.com" in host and ("/explore/" in path or "/discovery/item/" in path): return urllib.parse.urlunparse(("https", parsed.netloc, parsed.path, "", "", "")) if platform == "抖音": if "iesdouyin.com" in host: return urllib.parse.urlunparse(("https", parsed.netloc, parsed.path, "", parsed.query, "")) if "douyin.com" in host and ("/video/" in path or "/note/" in path): return urllib.parse.urlunparse(("https", parsed.netloc, parsed.path, "", "", "")) return "" def parse_search_anchors(page_html): anchors = [] for href, title in re.findall(r']*href=["\']([^"\']+)["\'][^>]*>(.*?)', page_html or "", flags=re.I | re.S): clean = clean_title(title) if clean: anchors.append((href, clean)) return anchors def collect_web_platform(stage_index, stage, platform, max_results=4): spec = PLATFORM_SPECS[platform] results = [] seen = set() queries = list(stage.get("queries") or [])[:4] for query in queries: if len(results) >= max_results: break site = spec["search_sites"][0] search_query = f"site:{site} {query}" url = "https://www.bing.com/search?" + urllib.parse.urlencode({"q": search_query, "setlang": "zh-CN"}) try: page = fetch_text(url, timeout=18, referer="https://www.bing.com/") except Exception: continue for href, title in parse_search_anchors(page): real_url = normalize_platform_url(href, platform) if not real_url or real_url in seen: continue if any(blocked in real_url.lower() for blocked in ("kuaishou.com", "gifshow.com")): continue if platform == "YouTube" and "youtube.com/watch" not in real_url and "youtu.be/" not in real_url: continue if platform != "YouTube" and not any(pattern in real_url for pattern in spec["url_patterns"]): continue if not is_stage_appropriate(title, stage_index): continue if not title_matches_stage_keywords(title, stage): continue score = practical_score(title, query) if score < 3: continue keywords = list(stage.get("keywords") or [])[:4] rec = { "id": re.sub(r"\W+", "", real_url)[-24:], "platform": platform, "title": title, "url": real_url, "difficulty": infer_difficulty(title, stage_index), "plays": None, "playCount": None, "likes": None, "likeCount": None, "fans": None, "followerCount": None, "author": None, "duration": None, "tags": keywords or ["AI学习", "Agent", "教程"], "capturedAt": now_str(), "sourceQuery": search_query, "pubdate": None, "publishedAt": None, "freshnessLabel": "公开搜索结果", "isClassic": False, "score": score, "completed": False, "completedAt": None, "notes": f"{spec['note']}抓取；公开视频指标未稳定获取时保留为 null。", } results.append(rec) seen.add(real_url) if len(results) >= max_results: break results.sort(key=lambda v: (v.get("score") or 0, len(v.get("title", ""))), reverse=True) return results def dedupe(videos): out = [] seen_urls = set() seen_titles = set() allowed_platforms = {"B站", "YouTube", "小红书", "抖音"} for video in videos: if video.get("platform") not in allowed_platforms: continue url = str(video.get("url") or "") if "kuaishou.com" in url.lower() or "gifshow.com" in url.lower(): continue title = clean_title(video.get("title")) key_title = re.sub(r"\W+", "", title.lower())[:34] if not url or not title or url in seen_urls or key_title in seen_titles: continue video["title"] = title out.append(video) seen_urls.add(url) seen_titles.add(key_title) return out def rotate_items(items, refresh_offset, step=8): if not items: return [] start = (max(0, int(refresh_offset or 0)) * step) % len(items) return items[start:] + items[:start] def assign_priorities(videos, stage, refresh_offset=0): videos = sorted(videos, key=lambda v: (v.get("score") or 0, v.get("plays") or 0, v.get("likes") or 0), reverse=True) fresh_videos = [v for v in videos if not v.get("isClassic")] classic_videos = [v for v in videos if v.get("isClassic")] monetization = [v for v in fresh_videos if any(k in v.get("title", "") for k in ("变现", "副业", "接单", "案例", "复盘"))] normal = [v for v in fresh_videos if v not in monetization] normal = rotate_items(normal, refresh_offset, step=8) monetization = rotate_items(monetization, refresh_offset, step=2) classic_videos = rotate_items(classic_videos, refresh_offset, step=4) selected = [] for platform in ("B站", "YouTube", "小红书", "抖音"): pick = next((v for v in normal if v.get("platform") == platform and v not in selected), None) if pick: selected.append(pick) selected.extend([v for v in normal[:3] if v not in selected]) selected.extend([v for v in normal[3:8] if v not in selected]) selected.extend(monetization[:2]) for v in fresh_videos + classic_videos: if len(selected) >= TARGET_DAILY_VIDEOS: break if v not in selected: selected.append(v) selected = selected[:TARGET_DAILY_VIDEOS] for idx, video in enumerate(selected): if video.get("isClassic"): priority = "经典教程" else: priority = "今日必看" if idx < 3 else ("可选拓展" if idx < 8 else "变现案例") video.update(build_learning_fields(video.get("title", ""), stage, priority)) return selected def fallback_beginner_videos(stage): captured = now_str() videos = [] for title, url, track in BEGINNER_FALLBACK_VIDEOS: videos.append({ "id": url.rstrip("/").split("/")[-1], "platform": "B站", "title": title, "url": url, "difficulty": "入门", "plays": None, "playCount": None, "likes": None, "likeCount": None, "fans": None, "followerCount": None, "author": None, "duration": None, "tags": ["AI学习", "提示词", "新手入门"], "capturedAt": captured, "sourceQuery": "public search fallback", "pubdate": None, "publishedAt": None, "freshnessLabel": "经典教程补足", "isClassic": True, "score": 1, "completed": False, "completedAt": None, "notes": "经典教程补足：当最近半年内可验证实操视频不足时使用；公开视频指标未稳定获取时保留为 null。", "track": track, }) return videos def recent_validated_seed_videos(stage): captured = now_str() videos = [] for title, url, track, freshness_note in RECENT_VALIDATED_SEED_VIDEOS: if not is_stage_appropriate(title, 0): continue if not title_matches_stage_keywords(title, stage): continue videos.append({ "id": url.rstrip("/").split("/")[-1], "platform": "B站", "title": title, "url": url, "difficulty": "入门", "plays": None, "playCount": None, "likes": None, "likeCount": None, "fans": None, "followerCount": None, "author": None, "duration": None, "tags": ["AI学习", "提示词", "新手入门"], "capturedAt": captured, "sourceQuery": "validated recent search seed", "pubdate": None, "publishedAt": None, "freshnessLabel": "最近半年（搜索验证）", "isClassic": False, "validatedRecent": True, "score": 18, "completed": False, "completedAt": None, "notes": f"{freshness_note}；公开视频指标未稳定获取时保留为 null。", "track": track, }) return videos def main(): arg_count = get_completed_count_from_args() refresh_offset = get_refresh_offset_from_args() completed_count = load_completed_count() if arg_count is None else arg_count notes = [] live_stages, stage_source = load_live_stages(notes) stage_index, stage = stage_for_count(completed_count, live_stages) videos = [] try: bili_videos = collect_bilibili(stage_index, stage) videos.extend(bili_videos) notes.append(f"B站按当前 Stage 关键词搜索，写入前候选 {len(bili_videos)} 条。") if LAST_BILIBILI_ERRORS: notes.append("B站接口部分请求失败：" + "；".join(LAST_BILIBILI_ERRORS[:3])) except Exception as e: notes.append(f"B站公开搜索抓取失败：{e}") for platform in ("YouTube", "小红书", "抖音"): try: platform_videos = collect_web_platform(stage_index, stage, platform, max_results=4) videos.extend(platform_videos) if platform_videos: notes.append(f"{platform} 按当前 Stage 关键词搜索，拿到 {len(platform_videos)} 条真实平台链接。") else: notes.append(f"{platform} 已按当前 Stage 关键词搜索，但公开搜索结果未返回可验证的视频/笔记链接。") except Exception as e: notes.append(f"{platform} 公开搜索抓取失败：{e}") videos = dedupe(videos) if stage_index == 0 and len(videos) < MIN_DAILY_VIDEOS: notes.append("当前 Stage 可验证公开视频少于 3 条时，只补入与当前 Stage 关键词匹配的已验证 B站新手实操视频；不再为了凑 10 条放宽主题。") videos = dedupe(videos + recent_validated_seed_videos(stage)) if stage_index == 0 and len(videos) < MIN_DAILY_VIDEOS: notes.append("当前 Stage 相关的最近半年公开视频仍不足 3 条，已用主题匹配的经典教程补足并在卡片标注“经典教程补足”。") videos = dedupe(videos + fallback_beginner_videos(stage)) videos = assign_priorities(videos, stage, refresh_offset=refresh_offset) if len(videos) < MIN_DAILY_VIDEOS: notes.append(f"全平台当前 Stage 可验证实操视频不足 {MIN_DAILY_VIDEOS} 条，本次仅保留 {len(videos)} 条真实且贴合的链接。") notes.append(f"本模块不再固定 10 条；优先保证当前 Stage 贴合度，目标每天 {MIN_DAILY_VIDEOS}-{TARGET_DAILY_VIDEOS} 条。") notes.append("YouTube/小红书/抖音公开搜索稳定性较差，未拿到可验证真实视频或笔记链接时不写入；不推荐快手。") notes.append(f"常规推荐只保留最近 {RECENT_WINDOW_DAYS} 天内的视频；已过滤明显卖课、标题党、纯资讯、纯概念或无实操指向的内容；播放量/点赞/作者拿不到时保留 null。") data = { "updatedAt": now_str(), "source": "agent.codepost.site live Stage + public video search", "track": "AI Agent 学习路线图（agent.codepost.site/#stages）", "stageSource": stage_source, "stageSourceUrl": ROADMAP_URL + "#stages", "stageKeywords": stage.get("keywords") or [], "stageQueries": stage.get("queries") or [], "stage": stage["name"], "stageIndex": stage_index, "completedCountBasis": completed_count, "refreshOffset": refresh_offset, "goal": stage["goal"], "minDailyVideos": MIN_DAILY_VIDEOS, "targetDailyVideos": TARGET_DAILY_VIDEOS, "notes": "；".join(notes), "videos": videos, } with open(OUTPUT_PATH, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) print(json.dumps({"ok": len(videos) > 0, "count": len(videos), "path": OUTPUT_PATH, "stage": stage["name"], "stageSource": stage_source}, ensure_ascii=False)) if __name__ == "__main__": main()