09-YouTube主动发现、候选确认与入库衔接

文档性质：内容系统主动扩源补充文档
用途：把 YouTube 自动发现、候选确认、为什么推荐、与现有入库主链的衔接关系一次讲清楚
适用范围：YouTube 候选发现、频道监控补充、Playlist / Series 候选判断、自动化扩源策略讨论
边界说明：

当前系统现实仍以 ./02-当前系统现实.md 为准
新资源确认后的入库主链仍以 ./03-新资源入库流程.md 为准
这份文档解决的是”怎么自动找源、怎么解释为什么值得确认、确认后怎么接回入库主链”
这份文档不是在证明 yomiya-service 现网已经有完整候选池、推荐池和自动化调度
这份文档不替代内容价值的最终人工判断，只替代”纯靠个人感觉找资源”的上游过程 默认读者：产品 / Backend / Admin / Content Ops / iOS
最后更新：2026-04-16

1. 这份文档解决什么问题

Yomiya 现在已经有两件事：

有正式内容处理主链（transcription_tasks -> content_processing -> news）
有人工补样本和人工预筛选流程

但还缺一件很重要的事：

系统如何主动发现值得做成内容的 YouTube 来源，而不是一直靠人凭感觉去挑。

这份文档就专门回答 4 个问题：

为什么现在需要一套 YouTube 主动发现补充层
系统应该按什么策略自动发现候选，而不是按主管偏好拍脑袋
候选确认后，怎么接回 03-新资源入库流程.md
YouTube 的 video / channel / playlist，怎么映射到 Yomiya 里大家已经在讨论的 item / channel / collection 理解层

2. 一句话原则

把”找什么内容”从个人主观挑选，改成”用户需要 + 数据证据 + 人工确认”的组合机制。

也就是：

系统负责主动找（通过 YouTube Data API 搜索 + 用户导入信号聚合）
系统负责解释为什么找到了它（基于用户导入次数、ASO 需求匹配、字幕可用性等证据）
人工负责最后确认（在 admin 后台审核候选池）
确认后继续走现有入库主链（接回 transcription_tasks）

这套机制不是要替代人，而是要替代下面这些低质量动作：

凭个人经验零散刷 YouTube
凭主管个人喜好决定最近应该加什么
看到一个好视频就临时记一下，但没有系统复用
找到好频道后，没有长期监控和持续复用能力

3. 为什么现在必须补这层

如果没有这层，内容扩源会一直有 4 个问题：

主观偏差大
- 更容易加入”我觉得不错”的内容，而不是”用户真的需要”的内容
复用差
- 今天刷到一个好视频，明天可能就忘了它来自哪个频道、哪个 playlist、为什么值得长期跟
效率低
- 每次扩源都像从头重新开始，没有候选池、没有证据链、没有复盘
无法积累系统性判断
- 你知道哪些内容好，但系统不知道
- 系统也无法反过来越跑越准

所以这套方案的意义不是”多抓一点 YouTube”，而是：

把扩源从一次次人工狩猎，变成一个可积累、可解释、可复用的候选运营系统。

4. 为什么选择这些信号源（基于已有能力扩展）

4.1 用户导入信号 - 已有能力，只需加统计层

现状：yomiya-service 已有 ContentImportApplication，用户导入 YouTube URL 后会：

调用 youtube-gateway 获取 metadata / transcript
创建 transcription_tasks
进入 content_processing

为什么要加统计层：

当前只记录”导入了一个视频”，但没记录”这个频道被反复导入”
如果 7 天内 4 个不同用户都导入了同一个频道的视频，说明这个频道值得长期监控
这是最强信号，因为代表真实用户主动行为

怎么做：

新增 youtube_source_signals 表，按天聚合
用户每次导入时，异步写入信号：(candidate_source_id, signal_type='user_import', signal_date=today, signal_count++)
Admin 后台可以看到”最近 30 天被导入 8 次，来自 5 个不同用户”

4.2 ASO 需求信号 - 已有数据，只需转成 query family

现状：本地 aso/ 目录已有 ASO 报告，持续出现的需求词如：

日语听力、日语新闻、JLPT、NHK、日语阅读

为什么要转成 query family：

ASO 词是”用户在应用商店搜什么”，不能直接拿去搜 YouTube
需要转成”学习者意图词”，例如：
- 日语听力 → japanese listening practice (英文主搜索) + 日本語リスニング (日文补漏)
- JLPT → jlpt n3 listening + 日本語能力試験聴解

为什么英文为主、日文为辅：

目标内容仍是日语内容，但搜索语种要拆开
很多面向全球学习者的优质频道用英文标题做 SEO
英文 learner-intent query 更容易命中”适合学习”的内容结构（listening practice / shadowing / news for learners）
日文 query 做补漏，避免漏掉原生日语高质量来源（如 やさしい日本語）

怎么做：

新增 youtube_query_families 表，每个 family 代表一类需求
新增 youtube_search_seeds 表，一个 family 下有多个 keyword variant
初期建议比例：英文 70%，日文 30%

4.3 YouTube 搜索命中 - 新增能力，但复用现有 gateway

现状：yomiya-service 已有 youtube_api.Client，但只用于获取单个视频 metadata

为什么要加搜索能力：

这是主动发现的核心入口
不能只靠用户导入，要主动按需求词去找

怎么做：

扩展 youtube_api.Client，增加 search.list 调用
每周跑 1 次主扫描，每次跑 8-12 个 query family
搜索结果先进 youtube_candidate_sources 和 youtube_candidate_videos，不直接入库

为什么每周 1 次，不是每天：

省 quota（YouTube Data API 有配额限制）
减少噪音（每天跑会产生大量重复候选）
更适合运营做审核（周报形式，不是实时洪流）
符合当前阶段”把系统摆正”的目标，不追求实时热点

4.4 可处理性信号 - 复用现有 gateway 能力

现状：youtube-gateway 已能判断字幕可用性、获取 transcript

为什么要作为信号：

有字幕的视频才能进入 Yomiya 的处理链
日语主内容才符合产品定位
时长适中（5-30 分钟）更适合学习场景

怎么做：

候选视频入池时，调用 gateway 获取 caption_availability
记录 language_hint、duration_seconds
规则评估：caption_availability='human' AND language_hint='ja' AND duration_seconds BETWEEN 300 AND 1800 → rule_decision='accept'

5. 具体怎么实现：每周调度一次 API 的运转机制

5.1 为什么每周调度一次就能最大化运转

核心思路：不追求实时热点，而是稳定积累高价值候选。

每周运转流程：

周一早上 9:00：scheduler 触发 YouTubeDiscoveryJob
选择本周要跑的 query family：
- 从 youtube_query_families 表中选出 active=true AND suppression_status='active' 的记录
- 按 weekly_weight 排序，取前 8-12 个
- 优先选择 last_run_at 最久的，避免重复扫描
对每个 family 执行搜索：
- 从 youtube_search_seeds 中取该 family 的 keyword variants
- 调用 youtube_api.Client.Search(keyword, maxResults=50)
- 每个 keyword 返回最多 50 个结果
结果写入候选池：
- 新频道 → youtube_candidate_sources (如果不存在)
- 新视频 → youtube_candidate_videos (如果不存在)
- 证据链 → youtube_candidate_evidences (记录”从哪个 query 发现的”)
更新 family 状态：
- 更新 last_run_at = now()
- 如果本周已跑过，设置 cooldown_until = next_monday
生成周报：
- 统计本周新增候选源 Top 10
- 统计本周新增候选视频 Top 20
- 推送到 Admin 后台 + Feishu

为什么这样能最大化运转：

省 quota：每周 8-12 个 query × 50 results = 400-600 个 API 调用，远低于每日配额
高质量：有足够时间让运营审核，不是每天被噪音淹没
可持续：不会因为 quota 耗尽而中断
可调整：如果某个 query family 持续产出低质量候选，可以降低 weekly_weight 或设置 suppression_status='suppressed'

5.2 Query 怎么收：从 ASO 到 Query Family 的转换

输入：ASO 报告中的需求词（如”日语听力”、“JLPT”）

转换规则：

识别需求意图：
- 日语听力 → intent_type = learner_en (学习者意图)
- NHK → intent_type = native_ja (原生内容)
生成 query family：
- family_key = en_japanese_listening_practice
- family_label = “Japanese Listening Practice (English learner intent)”
- intent_type = learner_en
- primary_language = en
- source_type = aso
生成 keyword variants：
- Primary: japanese listening practice
- Secondary: easy japanese listening, japanese listening for beginners
- Fallback: learn japanese listening
写入数据库：

INSERT INTO youtube_query_families (family_key, family_label, intent_type, primary_language, source_type, weekly_weight, active)
VALUES ('en_japanese_listening_practice', 'Japanese Listening Practice', 'learner_en', 'en', 'aso', 10, true);

INSERT INTO youtube_search_seeds (family_id, keyword, language, variant_type, priority, active)
VALUES
  (1, 'japanese listening practice', 'en', 'primary', 1, true),
  (1, 'easy japanese listening', 'en', 'secondary', 2, true),
  (1, 'learn japanese listening', 'en', 'fallback', 3, true);

初期建议的 query family 列表（基于现有 ASO 数据）：

family_key	family_label	intent_type	primary_language	来源理由
`en_japanese_listening_practice`	Japanese Listening Practice	learner_en	en	ASO 高频词”日语听力”
`en_jlpt_listening`	JLPT Listening	learner_en	en	ASO 高频词”JLPT”
`en_japanese_news_learners`	Japanese News for Learners	learner_en	en	ASO 高频词”日语新闻”
`en_japanese_shadowing`	Japanese Shadowing	learner_en	en	学习方法词
`ja_yasashii_nihongo`	やさしい日本語	native_ja	ja	日文补漏
`ja_nihongo_news`	日本語ニュース	native_ja	ja	日文补漏

5.3 用户导入信号怎么聚合

现有流程：

用户导入 YouTube URL → ContentImportApplication → 创建 transcription_tasks

新增流程（不影响现有主链）：

提取频道信息：
- 从 video URL 中提取 channel_id
- 调用 youtube_api.Client.GetChannel(channel_id) 获取频道名称
写入候选源（如果不存在）：

candidateSource := &YouTubeCandidateSource{
    SourceType: "channel",
    ExternalSourceID: channelID,
    SourceName: channelName,
    FirstSeenFrom: "user_import",
    LatestSeenFrom: "user_import",
    ReviewStatus: "new",
}
repo.UpsertCandidateSource(candidateSource)

写入信号：

signal := &YouTubeSourceSignal{
    CandidateSourceID: candidateSource.ID,
    SignalType: "user_import",
    SignalValue: 1.0,
    SignalCount: 1,
    SignalDate: time.Now().Format("2006-01-02"),
}
repo.IncrementSignal(signal)

异步处理：
- 通过 RabbitMQ 发送消息到 youtube_signal_aggregator consumer
- 不阻塞用户导入主链

聚合查询（用于推荐）：

SELECT
    cs.id,
    cs.source_name,
    SUM(CASE WHEN ss.signal_type = 'user_import' AND ss.signal_date >= DATE_SUB(NOW(), INTERVAL 30 DAY) THEN ss.signal_count ELSE 0 END) AS import_count_30d,
    COUNT(DISTINCT CASE WHEN ss.signal_type = 'user_import' AND ss.signal_date >= DATE_SUB(NOW(), INTERVAL 30 DAY) THEN ss.metadata->>'$.user_id' END) AS unique_user_count_30d
FROM youtube_candidate_sources cs
LEFT JOIN youtube_source_signals ss ON cs.id = ss.candidate_source_id
WHERE cs.review_status = 'new'
GROUP BY cs.id
HAVING import_count_30d >= 3 AND unique_user_count_30d >= 2
ORDER BY import_count_30d DESC, unique_user_count_30d DESC
LIMIT 10;

6. 这份文档和 02 / 03 / 04 / 05 的分工

文档	负责什么	不负责什么
`02-当前系统现实.md`	证明当前后端已经有什么对象、字段、关系	不负责讲自动发现方案怎么工作
`03-新资源入库流程.md`	讲”资源确认以后”怎么进入转文字、归类、失败回退	不负责讲系统怎么主动发现候选
`04-内容样本池.md`	记录统一内容资源池、YouTube 当前初判、执行状态、链接附录和简略核验说明	不负责讲自动发现主流程
`05-待决问题.md`	收口暂时没定死的问题	不负责讲当前执行主方案
`09-YouTube主动发现、候选确认与入库衔接.md`	讲自动发现补充层、证据驱动、对象映射、与 03 的衔接	不证明现网已全部实现

一句话区分：

03 管”确认后怎么进系统”
09 管”系统怎么先把值得确认的候选找出来”

7. 最小白先记住这 7 句话

如果第一次看这套方案，先只记这 7 句：

这套方案不是自动入库，而是自动找候选
系统不会替你做最后判断，只会先给你一批值得看的建议
推荐不是靠主管感觉，而是靠用户导入、ASO、搜索命中、字幕可处理性、持续更新这些信号
YouTube 的单条视频，最终更接近 Yomiya 的 Item
YouTube 的频道，最终更接近长期来源层，通常要映射到内部 channel
YouTube 的 playlist / 系列，更接近未来的 Collection / Series candidate
候选确认后，不另起一条新入库链，而是继续接回 03-新资源入库流程.md

8. 我们到底在补什么，不在补什么

这套方案补的是：

主动发现入口（YouTube Data API search.list）
候选池（youtube_candidate_sources + youtube_candidate_videos）
推荐理由（基于 youtube_candidate_evidences 的证据链）
候选确认动作（Admin 后台审核 + 晋升）
确认后回到现有入库主链（接回 transcription_tasks）

这套方案不补的是：

最终内容价值的一票否决权（仍需人工确认）
现有正式内容处理链的重写（复用 content_processing）
首页分发逻辑的直接改造（候选池不直接影响首页）
“所有平台统一自动采集”一次做完（先做 YouTube，再迁移到 Podcast）

所以它的定位非常明确：

它是现有内容系统的一层上游自动化补充，不是把整个内容系统推倒重做。

9. 小白对象映射，先统一语言

这一节非常重要，不然团队很容易把 YouTube 对象、研究对象和系统正式对象混着说。

9.1 先看 YouTube 原生对象

YouTube 这边最常见的 3 个对象是：

Video
Channel
Playlist

9.2 再看 Yomiya 这边大家习惯讨论的对象

Yomiya 这边为了更好理解，经常会用这几个词：

Item
Channel
Collection / Series

但要注意：

当前正式后端现实里，真正承载单条内容的正式主对象还是 news，见 ./02-当前系统现实.md
Collection / Series / Topic 现在还是目标理想，不是现网正式对象，见 ./02-当前系统现实.md

9.3 最推荐的小白映射表

YouTube 对象	在 09 里的理解	确认后更接近 Yomiya 的什么	当前现实提醒
`Video`	候选单条内容	`Item`	当前现实里主要会落到 `news`
`Channel`	候选来源 / 长期监控对象	内部 `channel` 的来源映射层	当前已有 `youtube_channels` 做映射
`Playlist`	候选系列 / 候选合集方向	`Collection / Series candidate`	现在还是目标理想，不是现网正式表

9.4 一张图看懂

flowchart LR
  A[YouTube Video] --> B[候选单条内容]
  B --> C[Yomiya Item 理解层]
  C --> D[当前现实主要落到 news]

  E[YouTube Channel] --> F[候选来源]
  F --> G[内部 channel 映射层]
  G --> H[可晋升为 managed source]

  I[YouTube Playlist] --> J[候选系列 / 候选合集方向]
  J --> K[Collection / Series candidate]
  K --> L[当前仍属目标理想]

这一张图的目的，是让所有人先不要把”未来想怎么承载”和”当前已经怎么落库”混掉。

10. 这套方案的核心思想

我们不直接问：

这个视频我喜不喜欢
这个频道我主观上觉得好不好

我们先问 4 件事：

用户是不是已经在用类似内容（用户导入信号）
市场是不是在要类似内容（ASO 需求信号）
这个内容是不是适合 Yomiya 的学习型处理链（字幕可用性、日语内容、时长适中）
这个来源是不是值得长期监控（频道持续更新、多条视频通过质量代理）

然后系统把这些问题变成”候选 + 证据 + 推荐理由”。

flowchart TD
  A[用户导入信号] --> E[候选发现层]
  B[ASO 需求信号] --> E
  C[YouTube 搜索结果] --> E
  D[第三方补充线索] --> E

  E --> F[候选视频池]
  E --> G[候选频道池]
  F --> H[推荐理由生成]
  G --> H
  H --> I[人工确认]
  I --> J[进入 03 入库主链]
  I --> K[晋升为长期监控来源]
  I --> L[继续观察 / 忽略 / 拉黑]

这就是这套方案和旧人工扩源的本质区别：

旧方式是”先靠感觉挑，再想办法处理”
新方式是”先按信号生成候选，再给出理由，再让人确认”

11. 为什么推荐这些内容，不再靠拍脑袋

11.1 信号来源

第一阶段最重要的信号，建议只保留 4 类：

用户导入信号
- 用户最近导入了什么
- 哪些频道被反复命中
- 实现：youtube_source_signals 表按天聚合
ASO 需求信号
- 用户市场在搜什么
- 哪些主题是最近更强的需求
- 实现：ASO 报告 → youtube_query_families 转换
YouTube 搜索命中
- 哪些 query family 连续出现高相关结果
- 实现：每周调用 search.list，结果写入 youtube_candidate_evidences
可处理性信号
- 有没有字幕
- 是不是日语主内容
- 时长和结构适不适合处理链
- 实现：调用 youtube-gateway 获取 caption_availability，规则评估写入 rule_decision

11.2 为什么这能替代主观偏好

因为它不是在问”主管最近想做什么”，而是在问：

用户最近在消费什么
市场最近在要什么
这个来源是不是能稳定产出
这个内容是不是能真的被处理和复用

11.3 推荐理由应该长什么样

后台不能只给一个分数，必须能直接展示人话理由。

例如：

7 天内被 4 个不同用户导入
命中 Japanese listening practice 这类高需求 query family
最近 10 条内容中 6 条可获取字幕
内容结构稳定，偏新闻讲解 / 慢速讲解 / JLPT 听力
与现有内容库重复度低

也就是说：

推荐必须是”证据驱动”，不是”模型说它高分”。

实现方式：

{
  “candidate_source_id”: 123,
  “source_name”: “Easy Japanese”,
  “reasons”: [
    “30天内被8个不同用户导入”,
    “频道近10个视频中7个可获取日语字幕”,
    “命中英文学习者意图 query：Japanese listening practice”,
    “频道内容结构稳定，偏新闻讲解”
  ],
  “evidence_summary”: {
    “user_import_30d”: 8,
    “unique_users_30d”: 8,
    “caption_available_rate”: 0.7,
    “matched_query_families”: [“en_japanese_listening_practice”]
  }
}

12. 当前推荐策略，为什么这样定

12.1 搜索策略

Yomiya 面向的是”学日语用户”，所以这里要拆开两件事：

目标内容语种
主搜索语种

当前建议是：

目标内容仍然以日语内容为主
搜索时先以英文 learner-intent query 为主
再用日文 query 做补漏

为什么英文为主、日文为辅：

很多面向全球学习者的优质频道用英文标题做 SEO
英文 learner-intent query 更容易命中”适合学习”的内容结构
但如果只搜英文，会漏掉原生日语高质量来源
所以用日文 query 做补漏

例如：

英文主搜索（70%）
- japanese listening practice
- easy japanese listening
- japanese news for learners
- jlpt listening
日文补漏（30%）
- 日本語リスニング
- やさしい日本語
- 日本語ニュース

12.2 执行节奏

第一阶段不建议每天跑，更不建议多语种并跑。

当前建议是：

每周 1 次主扫描
每周 8 到 12 个 query family
先做候选，不直接入库
先做周报，不追求实时热点系统

为什么这样做：

更省 quota
- YouTube Data API 有配额限制
- 每周 8-12 个 query × 50 results = 400-600 个 API 调用
- 远低于每日配额，不会因为 quota 耗尽而中断
更少噪音
- 每天跑会产生大量重复候选
- 周报形式让运营有足够时间审核
更适合运营做审核
- 不是每天被噪音淹没
- 可以集中时间处理一批高质量候选
更符合当前阶段”把系统摆正”的目标
- 先把流程跑通，再考虑提高频率
- 不追求实时热点，而是稳定积累高价值候选

13. 它和 03 怎么衔接：候选确认后的两条路

这套方案最容易让人误会的一点是：

自动发现不是新建一条入库主链。

真正正确的理解应该是：

09 负责”把值得看的候选找出来”
03 负责”确认以后怎么进入转文字、归类、失败回退”

13.1 一张衔接图看懂

flowchart LR
  A[09 自动发现补充层]
  A --> B[候选视频 / 候选频道 / 候选 playlist]
  B --> C[人工确认]
  C -->|确认单条内容| D[03 新资源入库流程]
  C -->|确认频道值得长期跟| E[受管来源监控]
  E --> F[新内容进入 03]
  C -->|确认 playlist 值得继续做| G[Collection / Series 候选研究]
  C -->|忽略 / 继续观察| H[候选池反馈]

13.2 具体衔接方式：两条路

路径 1：单条视频直接入库

适用场景：这个视频质量很高，但不确定频道是否值得长期监控

流程：

Admin 后台看到候选视频推荐
点击”直接入库”
系统调用 ContentImportApplication.ImportYouTubeVideo(videoID)
创建 transcription_tasks
进入 content_processing
产出 news

路径 2：频道晋升为受管源

适用场景：这个频道持续产出高质量内容，值得长期监控

流程：

Admin 后台看到候选频道推荐
点击”晋升为受管源”
关键步骤：绑定内部 channel
- 选择已有内部 channel，或新建内部 channel
- 将候选频道和内部 channel 建立绑定
写入 youtube_channels 表：

managedSource := &YouTubeChannel{
    ChannelID: internalChannelID,  // 必填，绑定内部 channel
    YouTubeChannelID: externalChannelID,
    ChannelName: channelName,
    AutoCrawlEnabled: true,
    CandidateSourceID: candidateSourceID,  // 记录来源
}
repo.CreateYouTubeChannel(managedSource)

后续由 youtube_channel_crawler 自动抓取新视频
新视频自动创建 transcription_tasks
进入 content_processing

13.3 这里最重要的边界

09 之前，系统负责找候选
09 到 03 之间，人工负责确认
进入 03 之后，继续按现有内容处理思路走

所以不要把 09 理解成”自动抓了就自动入库”。

13.4 为什么晋升必须绑定内部 channel

原因：

youtube_channels 不是纯外部频道表，它最终要驱动正式入库流程
入库后的内容需要归属到内部 channel，才能在首页分发时正确展示
如果不绑定，后续无法知道”这个 YouTube 频道的内容应该归到哪个内部频道”

实现：

Admin 后台晋升时，必须显式选择或新建内部 channel
youtube_channels.channel_id 字段保持必填
这样强制”晋升前先绑定内部 channel”

14. 为什么这是一种自动化补充，而不是推翻旧方案

当前已经有两条很重要的现实：

02 证明了系统已有正式内容对象和处理链
03 已经给了新资源确认后的最简处理路径

所以 09 的正确位置，只能是：

对现有方案的自动化采集补充。

它补的是上游，不是重写中游和下游。

这件事带来的好处是：

不需要推翻现有入库主链
不需要把所有新旧文档重写一遍
可以先把 YouTube 扩源做成一个可控补充层
一旦跑顺，再逐步把相同框架迁移到 Podcast 等平台

为什么选择这些能力作为扩展基础：

用户导入信号：已有 ContentImportApplication，只需加统计层
ASO 需求信号：已有 ASO 报告，只需转成 query family
YouTube 搜索：已有 youtube_api.Client，只需扩展 search.list
可处理性信号：已有 youtube-gateway，只需调用现有能力

所有这些都是在已有能力上的扩展，不是从零开始建设。

15. 这套方案是不是也适合播客

答案是：适合，而且大部分思路可以复用。

15.1 能复用的部分

以下这些层，YouTube 和播客都可以共用：

query / seed 生成思路
candidate pool
evidence log
recommendation reasons
review decisions
确认后接回 03

15.2 需要换掉的部分

真正不同的是 adapter：

YouTube 更像 search + channel + playlist
Podcast 更像 节目 + RSS + episode + 榜单 + 推荐来源

也就是说：

框架可以复用
接入器要按平台重写

15.3 为什么现在先写 YouTube

因为 YouTube 现在是最适合先跑通这套机制的平台：

可发现性强（有搜索 API）
频道和 playlist 天然适合做长期监控
视频、频道、playlist 三层结构很适合解释候选与长期来源的关系

所以当前最合理的顺序是：

先把 YouTube 跑通
再把这套候选机制迁移到 Podcast

16. 当前阶段只做什么，不做什么

16.1 当前建议先做

YouTube 候选自动发现补充层
候选视频 / 候选频道池
推荐理由与证据链
人工确认动作
与 03 的主链衔接

16.2 当前先不做

所有平台统一自动采集
全自动入库
完整的 Collection / Series 正式后端建模
首页推荐直接由这套系统全量驱动
复杂的黑盒排序模型

结论很简单：

先把”找源 -> 给理由 -> 人工确认 -> 接回入库主链”这条链跑顺，比一开始追求全自动更重要。

18. 完整实现流程：从 ASO 到周报的全链路

18.1 第一步：初始化 Query Family（一次性）

输入：ASO 报告中的需求词

操作：Admin 后台或脚本初始化

-- 示例：初始化"日语听力"相关的 query family
INSERT INTO youtube_query_families (family_key, family_label, intent_type, primary_language, source_type, weekly_weight, active)
VALUES
  ('en_japanese_listening_practice', 'Japanese Listening Practice', 'learner_en', 'en', 'aso', 10, true),
  ('en_jlpt_listening', 'JLPT Listening', 'learner_en', 'en', 'aso', 8, true),
  ('ja_yasashii_nihongo', 'やさしい日本語', 'native_ja', 'ja', 'aso', 5, true);

-- 为每个 family 添加 keyword variants
INSERT INTO youtube_search_seeds (family_id, keyword, language, variant_type, priority, active)
VALUES
  (1, 'japanese listening practice', 'en', 'primary', 1, true),
  (1, 'easy japanese listening', 'en', 'secondary', 2, true),
  (2, 'jlpt n3 listening', 'en', 'primary', 1, true),
  (2, 'jlpt listening practice', 'en', 'secondary', 2, true),
  (3, 'やさしい日本語', 'ja', 'primary', 1, true);

18.2 第二步：每周一早上 9:00 自动调度

触发器：cmd/scheduler/main.go 中的 cron job

// 每周一早上 9:00 触发
scheduler.AddFunc("0 9 * * 1", func() {
    discoveryService.RunWeeklyDiscovery(ctx)
})

执行逻辑：

选择本周要跑的 query family：

func (s *DiscoveryService) RunWeeklyDiscovery(ctx context.Context) error {
    // 选择 active=true 且不在 cooldown 的 family
    families, err := s.repo.GetActiveQueryFamilies(ctx, &QueryFamilyFilter{
        Active: true,
        NotInCooldown: true,
        Limit: 12,
        OrderBy: "weekly_weight DESC, last_run_at ASC",
    })

    for _, family := range families {
        // 创建 discovery run 记录
        run := &DiscoveryRun{
            TriggerType: "scheduled",
            ScopeType: "family",
            ScopeKey: family.FamilyKey,
            Status: "queued",
        }
        runID, _ := s.repo.CreateDiscoveryRun(ctx, run)

        // 异步执行
        s.queue.Publish("youtube_discovery", &DiscoveryTask{
            RunID: runID,
            FamilyID: family.ID,
        })
    }

    return nil
}

对每个 family 执行搜索：

func (s *DiscoveryService) ExecuteDiscovery(ctx context.Context, task *DiscoveryTask) error {
    // 更新状态为 running
    s.repo.UpdateDiscoveryRun(ctx, task.RunID, &DiscoveryRunUpdate{
        Status: "running",
        StartedAt: time.Now(),
    })

    // 获取该 family 的所有 seeds
    seeds, _ := s.repo.GetSearchSeeds(ctx, task.FamilyID)

    totalQuotaCost := 0
    totalSourceCount := 0
    totalVideoCount := 0

    for _, seed := range seeds {
        // 调用 YouTube Data API
        results, err := s.youtubeClient.Search(ctx, &SearchRequest{
            Query: seed.Keyword,
            MaxResults: 50,
            Type: "video",
            RelevanceLanguage: seed.Language,
        })

        if err != nil {
            // 记录错误但继续
            continue
        }

        totalQuotaCost += 100 // search.list 消耗 100 quota

        // 处理搜索结果
        for _, item := range results.Items {
            // 提取频道信息
            channelID := item.Snippet.ChannelId
            channelTitle := item.Snippet.ChannelTitle

            // Upsert 候选源
            source, _ := s.repo.UpsertCandidateSource(ctx, &CandidateSource{
                SourceType: "channel",
                ExternalSourceID: channelID,
                SourceName: channelTitle,
                FirstSeenFrom: "search",
                LatestSeenFrom: "search",
                ReviewStatus: "new",
            })

            if source.IsNew {
                totalSourceCount++
            }

            // Upsert 候选视频
            video, _ := s.repo.UpsertCandidateVideo(ctx, &CandidateVideo{
                VideoID: item.Id.VideoId,
                SourceID: source.ID,
                Title: item.Snippet.Title,
                PublishedAt: item.Snippet.PublishedAt,
            })

            if video.IsNew {
                totalVideoCount++
            }

            // 写入证据链
            s.repo.CreateEvidence(ctx, &Evidence{
                SubjectType: "source",
                SubjectID: source.ID,
                EvidenceType: "search_hit",
                EvidenceSource: seed.Language + "_query",
                EvidenceKey: task.FamilyKey,
                EvidenceValue: seed.Keyword,
                ObservedAt: time.Now(),
            })

            s.repo.CreateEvidence(ctx, &Evidence{
                SubjectType: "video",
                SubjectID: video.ID,
                EvidenceType: "search_hit",
                EvidenceSource: seed.Language + "_query",
                EvidenceKey: task.FamilyKey,
                EvidenceValue: seed.Keyword,
                ObservedAt: time.Now(),
            })
        }
    }

    // 更新 family 的 last_run_at 和 cooldown
    s.repo.UpdateQueryFamily(ctx, task.FamilyID, &QueryFamilyUpdate{
        LastRunAt: time.Now(),
        CooldownUntil: time.Now().AddDate(0, 0, 7), // 下周一才能再跑
    })

    // 更新 discovery run 状态
    s.repo.UpdateDiscoveryRun(ctx, task.RunID, &DiscoveryRunUpdate{
        Status: "succeeded",
        FinishedAt: time.Now(),
        QuotaCostActual: totalQuotaCost,
        ResultSourceCount: totalSourceCount,
        ResultVideoCount: totalVideoCount,
    })

    return nil
}

18.3 第三步：用户导入时异步写入信号

触发点：ContentImportApplication.ImportYouTubeVideo

func (a *ContentImportApplication) ImportYouTubeVideo(ctx context.Context, url string, userID int64) error {
    // 现有逻辑：创建 transcription_tasks
    // ... 省略 ...

    // 新增逻辑：异步写入信号
    videoID := extractVideoID(url)
    metadata, _ := a.youtubeGateway.GetVideoMetadata(ctx, videoID)

    // 发送到 MQ
    a.mq.Publish("youtube_signal_aggregator", &SignalEvent{
        ChannelID: metadata.ChannelID,
        ChannelName: metadata.ChannelTitle,
        SignalType: "user_import",
        UserID: userID,
        Timestamp: time.Now(),
    })

    return nil
}

消费者逻辑：

func (c *SignalAggregatorConsumer) HandleSignal(ctx context.Context, event *SignalEvent) error {
    // Upsert 候选源
    source, _ := c.repo.UpsertCandidateSource(ctx, &CandidateSource{
        SourceType: "channel",
        ExternalSourceID: event.ChannelID,
        SourceName: event.ChannelName,
        FirstSeenFrom: "user_import",
        LatestSeenFrom: "user_import",
        ReviewStatus: "new",
    })

    // 写入信号（按天聚合）
    today := time.Now().Format("2006-01-02")
    c.repo.IncrementSignal(ctx, &SignalIncrement{
        CandidateSourceID: source.ID,
        YouTubeChannelID: event.ChannelID,
        SignalType: "user_import",
        SignalDate: today,
        SignalValue: 1.0,
        Metadata: map[string]interface{}{
            "user_id": event.UserID,
        },
    })

    // 写入证据链
    c.repo.CreateEvidence(ctx, &Evidence{
        SubjectType: "source",
        SubjectID: source.ID,
        EvidenceType: "user_import_hit",
        EvidenceSource: "import",
        EvidenceValue: fmt.Sprintf("user_%d", event.UserID),
        ObservedAt: event.Timestamp,
    })

    return nil
}

18.4 第四步：生成推荐理由

触发时机：每周扫描完成后，或 Admin 后台查看候选池时

func (s *RecommendationService) GenerateRecommendations(ctx context.Context) ([]*Recommendation, error) {
    // 查询候选源，按信号强度排序
    query := `
        SELECT
            cs.id,
            cs.source_name,
            cs.external_source_id,
            SUM(CASE WHEN ss.signal_type = 'user_import' AND ss.signal_date >= DATE_SUB(NOW(), INTERVAL 30 DAY)
                THEN ss.signal_count ELSE 0 END) AS import_count_30d,
            COUNT(DISTINCT CASE WHEN ss.signal_type = 'user_import' AND ss.signal_date >= DATE_SUB(NOW(), INTERVAL 30 DAY)
                THEN JSON_UNQUOTE(JSON_EXTRACT(ss.metadata, '$.user_id')) END) AS unique_user_count_30d,
            COUNT(DISTINCT CASE WHEN e.evidence_type = 'search_hit' AND e.observed_at >= DATE_SUB(NOW(), INTERVAL 7 DAY)
                THEN e.evidence_key END) AS matched_query_families_7d
        FROM youtube_candidate_sources cs
        LEFT JOIN youtube_source_signals ss ON cs.id = ss.candidate_source_id
        LEFT JOIN youtube_candidate_evidences e ON cs.id = e.subject_id AND e.subject_type = 'source'
        WHERE cs.review_status = 'new' AND cs.suppression_status = 'active'
        GROUP BY cs.id
        HAVING import_count_30d >= 3 OR matched_query_families_7d >= 2
        ORDER BY import_count_30d DESC, unique_user_count_30d DESC, matched_query_families_7d DESC
        LIMIT 20
    `

    rows, _ := s.db.QueryContext(ctx, query)

    recommendations := []*Recommendation{}
    for rows.Next() {
        var rec Recommendation
        rows.Scan(&rec.SourceID, &rec.SourceName, &rec.ExternalSourceID,
                  &rec.ImportCount30d, &rec.UniqueUserCount30d, &rec.MatchedQueryFamilies7d)

        // 生成推荐理由
        reasons := []string{}
        if rec.ImportCount30d > 0 {
            reasons = append(reasons, fmt.Sprintf("30天内被%d个不同用户导入", rec.UniqueUserCount30d))
        }
        if rec.MatchedQueryFamilies7d > 0 {
            // 查询具体命中了哪些 query family
            families := s.getMatchedQueryFamilies(ctx, rec.SourceID)
            for _, f := range families {
                reasons = append(reasons, fmt.Sprintf("命中搜索需求：%s", f.FamilyLabel))
            }
        }

        // 查询该频道的视频字幕可用率
        captionRate := s.getCaptionAvailabilityRate(ctx, rec.SourceID)
        if captionRate > 0.5 {
            reasons = append(reasons, fmt.Sprintf("频道视频中%.0f%%可获取字幕", captionRate*100))
        }

        rec.Reasons = reasons
        recommendations = append(recommendations, &rec)
    }

    return recommendations, nil
}

18.5 第五步：生成周报并推送

触发时机：每周一扫描完成后

func (s *ReportService) GenerateWeeklyReport(ctx context.Context) error {
    // 生成推荐
    recommendations, _ := s.recommendationService.GenerateRecommendations(ctx)

    // 统计本周新增
    stats := s.getWeeklyStats(ctx)

    // 构建 Feishu 卡片
    card := &FeishuCard{
        Header: &CardHeader{
            Title: "YouTube 候选源周报",
        },
        Elements: []CardElement{
            &CardText{
                Content: fmt.Sprintf("本周新增候选源：%d 个\n本周新增候选视频：%d 个",
                                     stats.NewSourceCount, stats.NewVideoCount),
            },
            &CardDivider{},
            &CardText{
                Content: "**推荐晋升的候选源 Top 5**",
            },
        },
    }

    for i, rec := range recommendations[:5] {
        card.Elements = append(card.Elements, &CardText{
            Content: fmt.Sprintf("%d. %s\n   - %s", i+1, rec.SourceName, strings.Join(rec.Reasons, "\n   - ")),
        })
        card.Elements = append(card.Elements, &CardAction{
            Actions: []Action{
                {
                    Tag: "button",
                    Text: "查看详情",
                    URL: fmt.Sprintf("https://admin.yomiya.com/youtube/candidates/%d", rec.SourceID),
                },
                {
                    Tag: "button",
                    Text: "晋升为受管源",
                    URL: fmt.Sprintf("https://admin.yomiya.com/youtube/candidates/%d/promote", rec.SourceID),
                },
            },
        })
    }

    // 发送到 Feishu
    s.feishuClient.SendCard(ctx, card)

    return nil
}

18.6 第六步：人工确认并晋升

Admin 后台操作：

查看候选源列表：
- URL: /admin/youtube/candidates
- 显示推荐理由、信号强度、匹配的 query family
点击”晋升为受管源”：
- 弹出对话框：选择或新建内部 channel
- 确认后调用 API：POST /admin/youtube/candidates/:id/promote

后端逻辑：

func (h *CandidateHandler) PromoteToManagedSource(c *gin.Context) {
    candidateID := c.Param("id")

    var req struct {
        ChannelID int64  `json:"channel_id"`  // 内部 channel ID
        NewChannelName string `json:"new_channel_name"`  // 如果要新建
    }
    c.BindJSON(&req)

    // 获取候选源
    candidate, _ := h.repo.GetCandidateSource(c, candidateID)

    // 如果要新建内部 channel
    var internalChannelID int64
    if req.ChannelID == 0 {
        channel, _ := h.channelRepo.CreateChannel(c, &Channel{
            Name: req.NewChannelName,
            Type: "youtube",
        })
        internalChannelID = channel.ID
    } else {
        internalChannelID = req.ChannelID
    }

    // 创建受管源
    managedSource := &YouTubeChannel{
        ChannelID: internalChannelID,
        YouTubeChannelID: candidate.ExternalSourceID,
        ChannelName: candidate.SourceName,
        AutoCrawlEnabled: true,
        CandidateSourceID: &candidateID,
    }
    h.youtubeChannelRepo.Create(c, managedSource)

    // 更新候选源状态
    h.repo.UpdateCandidateSource(c, candidateID, &CandidateSourceUpdate{
        ReviewStatus: "promoted",
        PromotionReason: "人工确认晋升",
        BoundChannelID: &internalChannelID,
        LastReviewedAt: time.Now(),
    })

    // 记录决策
    h.repo.CreateReviewDecision(c, &ReviewDecision{
        SubjectType: "source",
        SubjectID: candidateID,
        DecisionType: "promote_source",
        OperatorUserID: c.GetInt64("user_id"),
        TargetChannelID: &internalChannelID,
        TargetManagedSourceID: &managedSource.ID,
    })

    c.JSON(200, gin.H{"success": true})
}

18.7 第七步：受管源自动抓取

触发器：现有的 youtube_channel_crawler

// 每小时检查一次受管源
scheduler.AddFunc("0 * * * *", func() {
    crawlerService.CrawlManagedSources(ctx)
})

逻辑：

查询 youtube_channels 中 auto_crawl_enabled=true 的记录
调用 YouTube Data API 获取频道最新视频
对比已有视频，只处理新视频
为新视频创建 transcription_tasks
进入 content_processing 主链

19. 这一节的最终结论

如果只保留一句话来指导后续执行：

09 不是在重写内容系统，而是在内容系统上方补一层 YouTube 主动发现与候选确认机制，用用户需求和数据证据替代纯主观挑选，再把确认后的内容接回现有入库主链。所有实现都基于已有能力扩展：用户导入加统计层、ASO 转 query family、youtube_api 扩展搜索、gateway 复用可处理性判断，每周调度一次 API 就能稳定运转。

17.1 为什么需要这些表

当前 yomiya-service 已有：

youtube_channels：受管频道表
transcription_tasks：转录任务表
news：正式内容表

但缺少：

候选池（搜索结果和用户导入信号没地方沉淀）
证据链（不知道为什么推荐这个候选）
策略层（query family 没有正式建模）

17.2 新增表详细定义

A. `youtube_query_families` - 策略层

作用：定义”为什么搜这一类内容”，是调度、反馈、降权、拉黑的最小单位。

CREATE TABLE youtube_query_families (
  id BIGINT PRIMARY KEY AUTO_INCREMENT,
  family_key VARCHAR(128) UNIQUE NOT NULL COMMENT '稳定唯一标识，如 en_japanese_listening_practice',
  family_label VARCHAR(255) NOT NULL COMMENT '后台展示名',
  intent_type VARCHAR(32) NOT NULL COMMENT 'learner_en / native_ja / topic_expansion',
  primary_language VARCHAR(16) NOT NULL COMMENT '主语种，如 en / ja',
  source_type VARCHAR(32) NOT NULL COMMENT 'aso / import_signal / manual / third_party',
  weekly_weight INT NOT NULL DEFAULT 10 COMMENT '周更扫描时的调度权重',
  active BOOLEAN NOT NULL DEFAULT TRUE COMMENT '是否参与调度',
  last_run_at DATETIME COMMENT '最近运行时间',
  cooldown_until DATETIME COMMENT '降温到何时恢复',
  suppression_status VARCHAR(16) NOT NULL DEFAULT 'active' COMMENT 'active / cooldown / suppressed',
  suppression_until DATETIME COMMENT '临时压制截止时间',
  suppression_reason VARCHAR(255) COMMENT '被压制原因',
  metadata JSON COMMENT 'family 说明、示例 query、来源摘要',
  created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
  updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  INDEX idx_active_weight (active, weekly_weight),
  INDEX idx_suppression_status (suppression_status)
) COMMENT='YouTube 搜索策略 family 表';

为什么需要 family 而不是单个 keyword：

调度、反馈、降权、拉黑都应该作用在 family 层
一个需求可能有多个 keyword variant，但它们是同一个意图
例如 japanese listening practice 和 easy japanese listening 都属于同一个 family

B. `youtube_search_seeds` - 策略执行层

作用：family 下的具体 keyword variant，真正交给 discovery adapter 执行。

CREATE TABLE youtube_search_seeds (
  id BIGINT PRIMARY KEY AUTO_INCREMENT,
  family_id BIGINT NOT NULL COMMENT '关联 youtube_query_families.id',
  keyword VARCHAR(255) NOT NULL COMMENT '具体搜索词',
  language VARCHAR(16) NOT NULL COMMENT '搜索语种',
  variant_type VARCHAR(16) NOT NULL COMMENT 'primary / secondary / fallback',
  priority INT NOT NULL DEFAULT 1 COMMENT 'family 内排序',
  active BOOLEAN NOT NULL DEFAULT TRUE COMMENT '是否启用',
  last_checked_at DATETIME COMMENT '最近执行时间',
  metadata JSON COMMENT '示例命中、历史效果',
  created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
  updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  UNIQUE KEY uk_family_keyword_lang (family_id, keyword, language),
  INDEX idx_family_priority (family_id, priority),
  INDEX idx_active (active),
  FOREIGN KEY (family_id) REFERENCES youtube_query_families(id)
) COMMENT='YouTube 搜索种子表';

C. `youtube_discovery_runs` - 运行历史层

作用：记录每次发现任务的运行历史，服务于周更扫描、手动重跑和 quota 审计。

CREATE TABLE youtube_discovery_runs (
  id BIGINT PRIMARY KEY AUTO_INCREMENT,
  trigger_type VARCHAR(16) NOT NULL COMMENT 'manual / scheduled / event',
  scope_type VARCHAR(16) NOT NULL COMMENT 'family / seed_batch',
  scope_key VARCHAR(255) NOT NULL COMMENT '如 family_key 或批次号',
  status VARCHAR(16) NOT NULL COMMENT 'queued / running / succeeded / failed / partial',
  started_at DATETIME COMMENT '开始时间',
  finished_at DATETIME COMMENT '结束时间',
  quota_cost_estimated INT COMMENT '预估配额消耗',
  quota_cost_actual INT COMMENT '实际配额消耗',
  result_source_count INT COMMENT '新增候选源数量',
  result_video_count INT COMMENT '新增候选视频数量',
  error_message TEXT COMMENT '失败原因',
  metadata JSON COMMENT '执行参数快照',
  created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
  updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  INDEX idx_trigger_type (trigger_type),
  INDEX idx_status (status),
  INDEX idx_started_at (started_at)
) COMMENT='YouTube 发现任务运行历史表';

为什么需要这个表：

审计：知道每周跑了哪些 query，消耗了多少 quota
排查：如果某次扫描失败，可以看到错误信息
复盘：可以统计哪些 query family 产出高质量候选

D. `youtube_candidate_sources` - 候选源池

作用：候选频道 / playlist 池，也是 observed layer 的主表。

CREATE TABLE youtube_candidate_sources (
  id BIGINT PRIMARY KEY AUTO_INCREMENT,
  source_type VARCHAR(16) NOT NULL COMMENT 'channel / playlist',
  external_source_id VARCHAR(128) NOT NULL COMMENT '外部 source ID',
  source_name VARCHAR(500) COMMENT '外部名称',
  canonical_url VARCHAR(1024) COMMENT '标准 URL',
  primary_language VARCHAR(16) COMMENT '主语言提示',
  first_seen_from VARCHAR(32) NOT NULL COMMENT '首次发现来源：search / import_signal',
  latest_seen_from VARCHAR(32) NOT NULL COMMENT '最近一次发现来源',
  review_status VARCHAR(16) NOT NULL DEFAULT 'new' COMMENT 'new / reviewing / watchlist / promoted / rejected',
  promotion_reason VARCHAR(255) COMMENT '晋升原因摘要',
  bound_channel_id BIGINT COMMENT '绑定内部 channels.id',
  suppression_status VARCHAR(16) NOT NULL DEFAULT 'active' COMMENT 'active / cooldown / suppressed',
  suppression_until DATETIME COMMENT '忽略到期时间',
  suppression_reason VARCHAR(255) COMMENT '忽略 / 拉黑原因',
  decision_source VARCHAR(16) COMMENT 'human / rule / system',
  first_seen_at DATETIME NOT NULL COMMENT '首次发现时间',
  last_seen_at DATETIME NOT NULL COMMENT '最近命中时间',
  last_recommended_at DATETIME COMMENT '最近被推送时间',
  last_reviewed_at DATETIME COMMENT '最近被人工处理时间',
  metadata JSON COMMENT '描述、thumbnail、统计摘要',
  created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
  updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  UNIQUE KEY uk_source_type_external_id (source_type, external_source_id),
  INDEX idx_review_status (review_status),
  INDEX idx_first_seen_at (first_seen_at),
  INDEX idx_last_seen_at (last_seen_at),
  INDEX idx_suppression_status (suppression_status),
  INDEX idx_bound_channel_id (bound_channel_id)
) COMMENT='YouTube 候选源池表';

为什么 youtube_channels 不能承担这个职责：

youtube_channels 应该只表示”已晋升的受管源”
搜索发现来的频道，不应该直接写入 youtube_channels
需要先进入候选池，经过人工确认，再晋升

E. `youtube_candidate_videos` - 候选视频池

作用：候选视频池，用于”直接入库”或”继续评估”决策。

CREATE TABLE youtube_candidate_videos (
  id BIGINT PRIMARY KEY AUTO_INCREMENT,
  video_id VARCHAR(32) UNIQUE NOT NULL COMMENT 'YouTube video ID',
  source_id BIGINT COMMENT '关联候选源',
  title VARCHAR(500) COMMENT '标题',
  published_at DATETIME COMMENT '发布时间',
  duration_seconds INT COMMENT '时长',
  transcript_state VARCHAR(16) NOT NULL DEFAULT 'unknown' COMMENT 'unknown / available / unavailable / queued / succeeded / failed',
  caption_availability VARCHAR(16) COMMENT 'unknown / auto / human / none',
  language_hint VARCHAR(16) COMMENT '语言提示',
  rule_decision VARCHAR(16) NOT NULL DEFAULT 'review' COMMENT 'accept / review / defer / reject',
  decision_reason VARCHAR(255) COMMENT '规则判断摘要',
  suppression_status VARCHAR(16) NOT NULL DEFAULT 'active' COMMENT 'active / cooldown / suppressed',
  suppression_until DATETIME COMMENT '忽略到期时间',
  suppression_reason VARCHAR(255) COMMENT '忽略原因',
  decision_source VARCHAR(16) COMMENT 'human / rule / system',
  first_seen_at DATETIME NOT NULL COMMENT '首次发现时间',
  last_seen_at DATETIME NOT NULL COMMENT '最近命中时间',
  last_recommended_at DATETIME COMMENT '最近被推送时间',
  metadata JSON COMMENT 'description、thumbnail、分类摘要',
  created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
  updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  INDEX idx_source_id (source_id),
  INDEX idx_published_at (published_at),
  INDEX idx_transcript_state (transcript_state),
  INDEX idx_language_hint (language_hint),
  INDEX idx_rule_decision (rule_decision),
  INDEX idx_suppression_status (suppression_status),
  FOREIGN KEY (source_id) REFERENCES youtube_candidate_sources(id)
) COMMENT='YouTube 候选视频池表';

为什么需要 rule_decision 字段：

系统可以根据规则自动评估：字幕可用 + 日语内容 + 时长适中 → accept
人工只需要审核 accept 和 review 的候选，不需要看所有候选
defer 表示暂时不确定，reject 表示明确不要

F. `youtube_source_signals` - 信号聚合层

作用：按天沉淀候选源相关信号，建议按日聚合，不直接存海量原始事件。

CREATE TABLE youtube_source_signals (
  id BIGINT PRIMARY KEY AUTO_INCREMENT,
  candidate_source_id BIGINT NOT NULL COMMENT '关联候选源',
  youtube_channel_id VARCHAR(128) NOT NULL COMMENT '外部频道 ID，便于回溯',
  signal_type VARCHAR(32) NOT NULL COMMENT 'user_import / accepted_video / transcript_success / search_hit / manual_keep',
  signal_value DECIMAL(10,4) NOT NULL COMMENT '归一化值',
  signal_count INT NOT NULL DEFAULT 1 COMMENT '当日次数',
  signal_date DATE NOT NULL COMMENT '统计日期',
  metadata JSON COMMENT '明细摘要',
  created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
  updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  UNIQUE KEY uk_source_type_date (candidate_source_id, signal_type, signal_date),
  INDEX idx_youtube_channel_id (youtube_channel_id),
  INDEX idx_signal_type (signal_type),
  INDEX idx_signal_date (signal_date),
  FOREIGN KEY (candidate_source_id) REFERENCES youtube_candidate_sources(id)
) COMMENT='YouTube 候选源信号表';

为什么按天聚合：

如果每次用户导入都写一条记录，会产生海量数据
按天聚合后，查询”最近 30 天被导入多少次”只需要扫描 30 条记录
metadata 字段可以存储明细摘要，如 {“user_ids”: [1, 2, 3]}

G. `youtube_candidate_evidences` - 证据链层

作用：保留推荐证据链，支撑 explainable reasons 和后续审计。

CREATE TABLE youtube_candidate_evidences (
  id BIGINT PRIMARY KEY AUTO_INCREMENT,
  subject_type VARCHAR(16) NOT NULL COMMENT 'source / video / query_family',
  subject_id BIGINT NOT NULL COMMENT '被证明对象的主键',
  evidence_type VARCHAR(32) NOT NULL COMMENT 'search_hit / user_import_hit / aso_match / third_party_seed / manual_review',
  evidence_source VARCHAR(32) NOT NULL COMMENT 'en_query / ja_query / import / aso / third_party / manual',
  evidence_key VARCHAR(255) COMMENT '如 query family key',
  evidence_value VARCHAR(1024) COMMENT '具体值，如命中的 query',
  weight DECIMAL(10,4) COMMENT '证据权重',
  observed_at DATETIME NOT NULL COMMENT '证据发生时间',
  metadata JSON COMMENT '原始返回摘要',
  created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
  INDEX idx_subject (subject_type, subject_id),
  INDEX idx_evidence_type (evidence_type),
  INDEX idx_evidence_source (evidence_source),
  INDEX idx_observed_at (observed_at)
) COMMENT='YouTube 候选证据链表';

为什么需要证据链：

推荐理由必须可追溯：这个候选是从哪个 query 发现的？被多少用户导入过？
审计：如果某个候选被拒绝，可以回溯”为什么当时推荐了它”
反馈：可以统计哪些 query family 产出高质量候选，哪些产出低质量候选

17.3 对现有表的修改

`youtube_channels` - 收敛为受管源表

当前表继续作为 managed source registry，但建议补一个来源追踪字段：

ALTER TABLE youtube_channels
ADD COLUMN candidate_source_id BIGINT COMMENT '记录它是从哪个候选源晋升上来的',
ADD INDEX idx_candidate_source_id (candidate_source_id);

为什么要加这个字段：

可以追溯：这个受管源是从哪个候选源晋升上来的
可以反馈：如果这个受管源后续产出高质量内容，可以提升对应 query family 的权重

注意：

channel_id 继续保持必填，不要把它改回 nullable
这样可以强制”晋升前先绑定内部 channel”

09-YouTube主动发现、候选确认与入库衔接

09-YouTube主动发现、候选确认与入库衔接

1. 这份文档解决什么问题

2. 一句话原则

3. 为什么现在必须补这层

4. 为什么选择这些信号源（基于已有能力扩展）

4.1 用户导入信号 - 已有能力，只需加统计层

4.2 ASO 需求信号 - 已有数据，只需转成 query family

4.3 YouTube 搜索命中 - 新增能力，但复用现有 gateway

4.4 可处理性信号 - 复用现有 gateway 能力

5. 具体怎么实现：每周调度一次 API 的运转机制

5.1 为什么每周调度一次就能最大化运转

5.2 Query 怎么收：从 ASO 到 Query Family 的转换

5.3 用户导入信号怎么聚合

6. 这份文档和 02 / 03 / 04 / 05 的分工

7. 最小白先记住这 7 句话

8. 我们到底在补什么，不在补什么

9. 小白对象映射，先统一语言

9.1 先看 YouTube 原生对象

9.2 再看 Yomiya 这边大家习惯讨论的对象

9.3 最推荐的小白映射表

9.4 一张图看懂

10. 这套方案的核心思想

11. 为什么推荐这些内容，不再靠拍脑袋

11.1 信号来源

11.2 为什么这能替代主观偏好

11.3 推荐理由应该长什么样

12. 当前推荐策略，为什么这样定

12.1 搜索策略

12.2 执行节奏

13. 它和 03 怎么衔接：候选确认后的两条路

13.1 一张衔接图看懂

13.2 具体衔接方式：两条路

13.3 这里最重要的边界

13.4 为什么晋升必须绑定内部 channel

14. 为什么这是一种自动化补充，而不是推翻旧方案

15. 这套方案是不是也适合播客

15.1 能复用的部分

15.2 需要换掉的部分

15.3 为什么现在先写 YouTube

16. 当前阶段只做什么，不做什么

16.1 当前建议先做

16.2 当前先不做

18. 完整实现流程：从 ASO 到周报的全链路

18.1 第一步：初始化 Query Family（一次性）

18.2 第二步：每周一早上 9:00 自动调度

18.3 第三步：用户导入时异步写入信号

18.4 第四步：生成推荐理由

18.5 第五步：生成周报并推送

18.6 第六步：人工确认并晋升

18.7 第七步：受管源自动抓取

19. 这一节的最终结论

17.1 为什么需要这些表

17.2 新增表详细定义

A. youtube_query_families - 策略层

B. youtube_search_seeds - 策略执行层

C. youtube_discovery_runs - 运行历史层

D. youtube_candidate_sources - 候选源池

E. youtube_candidate_videos - 候选视频池

F. youtube_source_signals - 信号聚合层

G. youtube_candidate_evidences - 证据链层

17.3 对现有表的修改

youtube_channels - 收敛为受管源表

A. `youtube_query_families` - 策略层

B. `youtube_search_seeds` - 策略执行层

C. `youtube_discovery_runs` - 运行历史层

D. `youtube_candidate_sources` - 候选源池

E. `youtube_candidate_videos` - 候选视频池

F. `youtube_source_signals` - 信号聚合层

G. `youtube_candidate_evidences` - 证据链层

`youtube_channels` - 收敛为受管源表