Make internal asset policies executable before DB-scale import

Constraint: Internal type enums need a repeatable mapping path into manifest-ready buckets before bulk database exports begin
Rejected: Leave type handling as documentation only | Would force repeated manual filtering and inconsistent ingestion decisions
Confidence: high
Scope-risk: narrow
Directive: Keep internal asset mapping defaults conservative; conditional instrumental variants should stay opt-in until version-aware training is ready
Tested: internal_asset_type_mapper.py on a 6-row sample CSV produced references=2 queries=2 metadata_only=1 excluded=1 with expected type routing
Not-tested: Direct SQL export integration against the live source database

+#!/usr/bin/env python3
+"""Map internal asset type codes into manifest-ready ACR roles.
+
+Input: CSV exported from an internal asset table.
+Output: JSON bundles for references, queries, metadata-only assets, and excluded assets.
+"""
+
+from __future__ import annotations
+
+import argparse
+import csv
+import json
+from pathlib import Path
+from typing import Dict, List, Tuple
+
+REFERENCE = "reference"
+QUERY = "query"
+METADATA = "metadata_only"
+EXCLUDED = "excluded"
+CONDITIONAL = "conditional"
+
+TYPE_POLICY: Dict[int, Dict[str, str]] = {
+    1: {"bucket": REFERENCE, "audio_role": "original_lossy", "train_type": "reference", "priority": "secondary_reference"},
+    2: {"bucket": CONDITIONAL, "audio_role": "inst_with_harmony_lossy", "train_type": "hard_negative", "priority": "conditional"},
+    3: {"bucket": METADATA, "audio_role": "lyrics_txt", "train_type": "none", "priority": "metadata"},
+    4: {"bucket": METADATA, "audio_role": "cover_image", "train_type": "none", "priority": "metadata"},
+    5: {"bucket": METADATA, "audio_role": "license_doc", "train_type": "none", "priority": "metadata"},
+    6: {"bucket": METADATA, "audio_role": "album_info", "train_type": "none", "priority": "metadata"},
+    7: {"bucket": QUERY, "audio_role": "short_video_clip", "train_type": "clean", "priority": "high_value_query"},
+    8: {"bucket": QUERY, "audio_role": "chorus_clip", "train_type": "clean", "priority": "high_value_query"},
+    9: {"bucket": CONDITIONAL, "audio_role": "inst_no_harmony_lossy", "train_type": "hard_negative", "priority": "conditional"},
+    10: {"bucket": CONDITIONAL, "audio_role": "inst_no_harmony_lossless", "train_type": "hard_negative", "priority": "conditional"},
+    11: {"bucket": REFERENCE, "audio_role": "original_lossless", "train_type": "reference", "priority": "primary_reference"},
+    12: {"bucket": CONDITIONAL, "audio_role": "inst_with_harmony_lossless", "train_type": "hard_negative", "priority": "conditional"},
+    13: {"bucket": METADATA, "audio_role": "lyrics_lrc", "train_type": "none", "priority": "metadata"},
+    14: {"bucket": METADATA, "audio_role": "cover_source", "train_type": "none", "priority": "metadata"},
+    16: {"bucket": QUERY, "audio_role": "short_video_clip", "train_type": "clean", "priority": "high_value_query"},
+    17: {"bucket": METADATA, "audio_role": "archive_package", "train_type": "none", "priority": "metadata"},
+    18: {"bucket": QUERY, "audio_role": "demo_audio", "train_type": "clean", "priority": "screen_before_use"},
+    19: {"bucket": METADATA, "audio_role": "sheet_image", "train_type": "none", "priority": "metadata"},
+    20: {"bucket": METADATA, "audio_role": "lyrics_translation", "train_type": "none", "priority": "metadata"},
+}
+
+
+def normalize_row(row: Dict[str, str], args) -> Dict:
+    type_code = int(row[args.type_field])
+    policy = TYPE_POLICY.get(type_code, {"bucket": EXCLUDED, "audio_role": "unknown", "train_type": "none", "priority": "unknown"})
+    canonical_song_id = row.get(args.song_field) or row.get(args.canonical_song_field) or row.get(args.asset_id_field) or "unknown_song"
+    version_id = row.get(args.version_field) or f"{canonical_song_id}_type_{type_code}"
+    record = {
+        "asset_id": row.get(args.asset_id_field),
+        "canonical_song_id": canonical_song_id,
+        "version_id": version_id,
+        "asset_type_code": type_code,
+        "audio_role": policy["audio_role"],
+        "recommended_train_type": policy["train_type"],
+        "priority": policy["priority"],
+        "bucket": policy["bucket"],
+        "audio_path": row.get(args.path_field),
+        "title": row.get(args.title_field),
+        "artist": row.get(args.artist_field),
+        "source_platform": row.get(args.platform_field) or "internal",
+    }
+    return record
+
+
+def to_manifest_record(record: Dict, bucket: str) -> Dict:
+    base = {
+        "song_id": record["canonical_song_id"],
+        "version_id": record["version_id"],
+        "asset_type_code": record["asset_type_code"],
+        "audio_role": record["audio_role"],
+        "audio_path": record["audio_path"],
+        "source_dataset": "internal_assets",
+        "source_platform": record["source_platform"],
+    }
+    if bucket == REFERENCE:
+        return {
+            **base,
+            "type": "reference",
+            "duration": 0.0,
+        }
+    return {
+        **base,
+        "type": record["recommended_train_type"],
+        "duration": 0.0,
+        "offset": None,
+        "segment_type": "external_query",
+    }
+
+
+def route_records(rows: List[Dict], include_conditionals_as: str) -> Tuple[List[Dict], List[Dict], List[Dict], List[Dict]]:
+    references, queries, metadata_only, excluded = [], [], [], []
+    for record in rows:
+        bucket = record["bucket"]
+        if bucket == CONDITIONAL:
+            bucket = include_conditionals_as if include_conditionals_as != "skip" else EXCLUDED
+
+        if bucket == REFERENCE:
+            references.append(to_manifest_record(record, REFERENCE))
+        elif bucket == QUERY:
+            queries.append(to_manifest_record(record, QUERY))
+        elif bucket == METADATA:
+            metadata_only.append(record)
+        else:
+            excluded.append(record)
+    return references, queries, metadata_only, excluded
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("csv_path")
+    parser.add_argument("--output-dir", required=True)
+    parser.add_argument("--asset-id-field", default="id")
+    parser.add_argument("--song-field", default="song_id")
+    parser.add_argument("--canonical-song-field", default="canonical_song_id")
+    parser.add_argument("--version-field", default="version_id")
+    parser.add_argument("--type-field", default="type")
+    parser.add_argument("--path-field", default="audio_path")
+    parser.add_argument("--title-field", default="title")
+    parser.add_argument("--artist-field", default="artist")
+    parser.add_argument("--platform-field", default="source_platform")
+    parser.add_argument("--include-conditionals-as", choices=["skip", "query", "reference"], default="skip")
+    args = parser.parse_args()
+
+    rows = []
+    with open(args.csv_path, newline="") as f:
+        reader = csv.DictReader(f)
+        for row in reader:
+            rows.append(normalize_row(row, args))
+
+    references, queries, metadata_only, excluded = route_records(rows, args.include_conditionals_as)
+
+    out_dir = Path(args.output_dir)
+    out_dir.mkdir(parents=True, exist_ok=True)
+    outputs = {
+        "references.json": references,
+        "queries.json": queries,
+        "metadata_only.json": metadata_only,
+        "excluded.json": excluded,
+        "summary.json": {
+            "input_rows": len(rows),
+            "references": len(references),
+            "queries": len(queries),
+            "metadata_only": len(metadata_only),
+            "excluded": len(excluded),
+            "include_conditionals_as": args.include_conditionals_as,
+        },
+    }
+    for name, payload in outputs.items():
+        (out_dir / name).write_text(json.dumps(payload, indent=2, ensure_ascii=False))
+
+    print(json.dumps(outputs["summary.json"], indent=2, ensure_ascii=False))
+
+
+if __name__ == "__main__":
+    main()

@@ -2,6 +2,34 @@
 
 ## 2026-06-02
 
+### Stage: 将内部素材 type 策略落成可执行映射脚本
+
+完成项：
+- 新增 [acr-engine/scripts/internal_asset_type_mapper.py](../acr-engine/scripts/internal_asset_type_mapper.py)
+- 支持从内部素材 CSV 自动分流到：
+  - `references.json`
+  - `queries.json`
+  - `metadata_only.json`
+  - `excluded.json`
+- 支持 `--include-conditionals-as`，可选把伴奏类临时导出成 `query` 或 `reference`
+- 在 [training-data-and-pgvector-guide.md](./training-data-and-pgvector-guide.md) 增加脚本入口说明
+
+验证结果：
+- 使用 6 行样例 CSV 验证：
+  - `11/1 -> references`
+  - `7/18 -> queries`
+  - `3 -> metadata_only`
+  - `12 -> excluded`
+- 摘要输出：
+  - `references = 2`
+  - `queries = 2`
+  - `metadata_only = 1`
+  - `excluded = 1`
+
+结论：
+- 现在内部素材 type 策略已经不只在文档里，而是可以直接作为批量清洗入口使用
+- 后续如果要从数据库导出 CSV 再转 manifest，已经有第一版可执行桥接脚本
+
 ### Stage: 为内部素材 type 枚举补齐训练参与策略文档
 
 完成项：

@@ -478,5 +478,35 @@ query:
 | 18 | `demo_audio` | `clean` / `augmented` |
 | 2/9/10/12 | `instrumental_variant` | 先不进主训练，或做 hard negative |
 
+## 12.6 现在仓库里已经有可执行映射脚本
+
+脚本：
+- [acr-engine/scripts/internal_asset_type_mapper.py](../acr-engine/scripts/internal_asset_type_mapper.py)
+
+作用：
+- 读取内部素材 CSV
+- 按 `type` 枚举自动分流成：
+  - `references.json`
+  - `queries.json`
+  - `metadata_only.json`
+  - `excluded.json`
+
+最短示例：
+
+```bash
+/usr/local/miniconda3/bin/python acr-engine/scripts/internal_asset_type_mapper.py assets.csv --output-dir out/internal_asset_map
+```
+
+如果你想临时把伴奏类也纳入导出，可用：
+
+```bash
+/usr/local/miniconda3/bin/python acr-engine/scripts/internal_asset_type_mapper.py assets.csv --output-dir out/internal_asset_map --include-conditionals-as query
+```
+
+但默认仍建议：
+- `--include-conditionals-as skip`
+
+这样更符合当前主任务“先把原曲识别打稳，再逐步纳入伴奏版本”的策略。
+
 ## Sources
 - 当前代码事实来自 [acr-engine/src/data/dataset.py](../acr-engine/src/data/dataset.py), [acr-engine/src/data/manifest_tools.py](../acr-engine/src/data/manifest_tools.py), [acr-engine/src/data/external_adapters.py](../acr-engine/src/data/external_adapters.py), [acr-engine/src/utils/audio.py](../acr-engine/src/utils/audio.py), [acr-engine/src/engines/ecapa_embedder.py](../acr-engine/src/engines/ecapa_embedder.py), [acr-engine/train.py](../acr-engine/train.py)