""" HPCI 4軸 (Production Constellation Index) を計算する。 本体プロンプト準拠: - C / Concentration : Σ(share_i)² × 10000 - R / Top-5 concentration (旧称 Repeat／反復参加度): top5 mentions / total mentions × 100 2026-09-10改称。上位5名の占有率であり、曲をまたぐ反復性そのものは測らない。 曲間の重なりは下記 RP_axis 等を補助情報として確認する。 - D / Density : 1曲あたりのユニーク協働者の平均人数 - S / Self-involvement : 本人alias含む曲の比率 × 100 mention の定義: 「1曲につき1人最大1カウント」。 同一曲内で writer / producer 両方に登場しても 1 mention。 サンプリング元権利者: Geniusで "Interpolation" / "Sampled from" タグの付いた writer は、 CSV では `sampling_rights_holders` カラムに移されている前提。 C/R/D 軸の計算から除外する。S 軸は本人 alias のみが対象なので影響なし。 旧 Concentration 版 (按分式 WCI/PCI) とは別物の指標であり、 既存 C_axis との一致チェックは行わない。 """ from __future__ import annotations import argparse import json import os import sys import tempfile from decimal import Decimal, ROUND_HALF_UP from datetime import datetime from pathlib import Path from zoneinfo import ZoneInfo # stateへの書き込みだけ、ワークスペース共通のCASガードを使う。 # 公開コードの読み込み・通常の算出には内部ツールを要求しない。 def is_registry_path(path) -> bool: try: Path(path).absolute().relative_to(Path(__file__).resolve().parents[2] / "state") except ValueError: return False return True def _registry_tools(): sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "tools")) import registry_write return registry_write # このスクリプトを直接実行できるよう sys.path を追加 sys.path.insert(0, str(Path(__file__).resolve().parent)) from aliases import get_aliases # noqa: E402 from load_credits import DEFAULT_ALIASES_CSV, load_album_from_csv, load_person_aliases, Track # noqa: E402 METHOD_VERSION = "hpci_constellation_album_internal_credit_profile" NORMALIZATION_METHOD_VERSION = "hpci_constellation_normalization_empirical" ROUNDING_RULE = "表示は小数第2位までROUND_HALF_UP。中間値は丸めない。" def format_value(value: float | int, places: int = 2) -> str: """本文・図・CLI共通。二進浮動小数点のformat/roundには丸めを任せない。""" quantum = Decimal(1).scaleb(-places) return format(Decimal(str(value)).quantize(quantum, rounding=ROUND_HALF_UP), "f") def write_json_atomic(path: str | Path, value: dict, *, expected_sha256=None) -> None: """同一ディレクトリ内のtmpをfsync後にos.replaceで置換する。""" path = Path(path) if is_registry_path(path): _registry_tools().write_registry(path, value, owner="hpci-pipeline", expected_sha256=expected_sha256) return path.parent.mkdir(parents=True, exist_ok=True) fd, temporary = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".tmp", dir=path.parent) try: with os.fdopen(fd, "w", encoding="utf-8") as stream: json.dump(value, stream, ensure_ascii=False, indent=2, allow_nan=False) stream.write("\n") stream.flush() os.fsync(stream.fileno()) os.replace(temporary, path) finally: if os.path.exists(temporary): os.unlink(temporary) def _build_track_credit_set(t: Track) -> set[str]: """ 1曲の credit set = writers ∪ producers - sampling_rights_holders。 同一曲内の重複は set により自動統合される (1 mention)。 """ s = set(t["writers"]) | set(t["producers"]) s -= set(t["sampling_rights_holders"]) return s # --------------------------------------------------------------------------- # 2026-09-10 追加: 反復参加率 (RP-axis) # # 第三者レビュー (docs/quality-20260910/THIRD-PARTY-REVIEW.md §2) 指摘への対応。 # 既存の R_axis (「上位5名クレジット集中率」に改称予定) は # R = 上位5名の登場曲数合計 / 全mention数 × 100 # であり、これは「クレジットがどれだけ上位5名に集中しているか」という # "集中度" を測るものであって、「曲をまたいで同じ人が繰り返し関わっているか」 # という "反復性" そのものは測っていない。 # 固定チーム5人×5曲 (各人が5曲すべてに登場) と、 # 曲ごとに別の1人×5曲 (誰も2曲以上に登場しない) は、 # どちらも top5 が mention 全体を占めるため R=100 になり区別できない。 # # 反復参加率 (Repeat-Participation rate, RP) はこの穴を埋めるために新設する。 # R_axisの計算式は保持する。人物同定・入力の訂正による数値変更はあり得る。 # # 定義: # RP_axis (反復参加率) = # (2曲以上に登場した協働者の mention 合計) / (全mention数) × 100 # # 固定チーム5人×5曲 → 全員が5曲に登場 → RP=100 # 曲ごとに別の1人×5曲 → 誰も2曲以上に登場しない → RP=0 # → 上の2ケースをRP軸は明確に区別する (Rは両方とも100で区別できない)。 # # 補助指標として mean pairwise Jaccard (曲間クレジット集合の平均重なり率) も # 併せて計算する。RPが「反復した人がどれだけの割合を占めるか」を見るのに対し、 # Jaccardは曲ペアの共通性を測る指標。RP・平均Jaccardは補助情報であり、 # 固定メンバーの確認には曲別クレジット集合を確認する。 # --------------------------------------------------------------------------- def _compute_repeat_participation_rate( collaborator_counts: dict[str, int], total_credit_mentions: int, ) -> dict: """ RP_axis (反復参加率): 2曲以上に登場した協働者の mention 合計 / 全mention数 × 100 R_axis (上位5名クレジット集中率) とは式の異なる補助指標であり、 既存の R_axis 計算には一切影響しない。 """ repeat_contributors = { name: count for name, count in collaborator_counts.items() if count >= 2 } repeat_mentions = sum(repeat_contributors.values()) if total_credit_mentions == 0: rp_axis = 0.0 else: rp_axis = repeat_mentions / total_credit_mentions * 100 return { "RP_axis": rp_axis, "repeat_contributors_count": len(repeat_contributors), "repeat_contributors": sorted( repeat_contributors.items(), key=lambda kv: (-kv[1], kv[0]) ), "repeat_mentions": repeat_mentions, "total_credit_mentions": total_credit_mentions, } def _compute_mean_pairwise_jaccard(track_credit_sets: list[set[str]]) -> dict: """ 補助指標: 全曲ペアの Jaccard 係数 (|A∩B| / |A∪B|) の平均。 曲同士のクレジット集合がどれだけ重なっているかを直接測る。 曲数が1以下の場合は計算不能として None を返す。 """ n = len(track_credit_sets) if n < 2: return {"mean_pairwise_jaccard": None, "pair_count": 0} total = 0.0 pair_count = 0 for i in range(n): for j in range(i + 1, n): a, b = track_credit_sets[i], track_credit_sets[j] union = a | b if not union: continue total += len(a & b) / len(union) pair_count += 1 mean_jaccard = total / pair_count if pair_count else None return {"mean_pairwise_jaccard": mean_jaccard, "pair_count": pair_count} def compute_hpci( album_slug: str, artist: str, title: str, release_year: int, csv_path: str | Path, aliases_csv: str | Path = DEFAULT_ALIASES_CSV, ) -> dict: album = load_album_from_csv(csv_path, album_slug, aliases_csv=aliases_csv) tracks = album["tracks"] total_tracks = len(tracks) if total_tracks == 0: raise ValueError(f"album {album_slug} has 0 tracks") # --- 各曲のクレジットセット (同一曲内重複統合済み, sampling除外済み) --- track_credit_sets: list[set[str]] = [_build_track_credit_set(t) for t in tracks] # --- collaborator_counts: 人物iが何曲に登場したか --- collaborator_counts: dict[str, int] = {} for credit_set in track_credit_sets: for person in credit_set: collaborator_counts[person] = collaborator_counts.get(person, 0) + 1 total_credit_mentions = sum(collaborator_counts.values()) collaborators_total_count = len(collaborator_counts) # --- C-axis --- if total_credit_mentions == 0: C_axis = 0.0 share_by_collaborator: dict[str, float] = {} else: share_by_collaborator = { person: count / total_credit_mentions for person, count in collaborator_counts.items() } # 整数の二乗和から算出し、set反復順序による浮動小数点加算差を避ける。 C_axis = sum(count ** 2 for count in collaborator_counts.values()) * 10000 / total_credit_mentions ** 2 # --- R-axis --- ranked = sorted( collaborator_counts.items(), key=lambda kv: (-kv[1], kv[0]), # count desc, name asc (deterministic) ) top5 = ranked[:5] # tie-breaking なし: 先頭5名のみ top5_collaborators = [name for name, _ in top5] top5_mentions = sum(count for _, count in top5) if total_credit_mentions == 0: R_axis = 0.0 else: R_axis = top5_mentions / total_credit_mentions * 100 # --- D-axis --- collaborator_count_by_track = [len(s) for s in track_credit_sets] if total_tracks == 0: D_axis = 0.0 else: D_axis = sum(collaborator_count_by_track) / total_tracks # 等価形 (検算用): # D_axis_alt = total_credit_mentions / total_tracks # assert abs(D_axis - D_axis_alt) < 1e-9 # --- S-axis --- person_aliases = load_person_aliases(aliases_csv) self_aliases_set = {person_aliases.get(n, n) for n in get_aliases(artist)} self_credit_tracks_count = sum( 1 for s in track_credit_sets if s & self_aliases_set ) tracks_without_self_credit = [ t["title"] for t, s in zip(tracks, track_credit_sets) if not (s & self_aliases_set) ] if total_tracks == 0: S_axis = 0.0 else: S_axis = self_credit_tracks_count / total_tracks * 100 # --- RP-axis (反復参加率) と補助指標 mean pairwise Jaccard --- # 既存 R_axis の値・計算経路には一切影響しない追加ブロック。 repeat_participation = _compute_repeat_participation_rate( collaborator_counts, total_credit_mentions ) pairwise_jaccard = _compute_mean_pairwise_jaccard(track_credit_sets) # --- 正規化 (レーダー描画用) --- C_axis_normalized = min(C_axis / 100, 100.0) R_axis_normalized = min(R_axis, 100.0) D_axis_normalized = min(D_axis * 10, 100.0) S_axis_normalized = min(S_axis, 100.0) # --- 同一曲内同一人物の統合検証 (証跡) --- integration_evidence = [] for t in tracks: w = set(t["writers"]) p = set(t["producers"]) overlap = w & p if overlap: integration_evidence.append({ "track_no": t["track_no"], "title": t["title"], "overlap_persons": sorted(overlap), "raw_mentions_writer_plus_producer": len(t["writers"]) + len(t["producers"]), "after_integration": len(w | p), }) # --- data_quality_flags --- flags = [] if collaborators_total_count <= 5: flags.append("collaborators_under_5") if any(s == set() for s in track_credit_sets): flags.append("credits_incomplete") if share_by_collaborator and max(share_by_collaborator.values()) > 0.8: flags.append("single_dominant_writer") # review_required 発火条件: notes に英語"REVIEW" または 日本語"要確認"/"確認推奨" を含む場合 review_keywords = ("REVIEW", "要確認", "確認推奨") has_review_notes = any( any(kw in (t["notes"] or "") for kw in review_keywords) for t in tracks ) if has_review_notes: flags.append("review_required") # --- source_notes --- notes = [] notes.append( f"mention定義: 1曲につき1人最大1カウント。" f"writer/producer両出は1 mentionに統合。sampling_rights_holdersは除外。" ) if integration_evidence: notes.append( f"{len(integration_evidence)} tracks had writer/producer overlap (integrated to 1 mention each)." ) notes.append("top5: 登場曲数の降順、同数の場合は名前の昇順で先頭5名を採用。同数者の選択はR値に影響しない。") notes.append(f"D_axis 等価検算: D × total_tracks = total_credit_mentions であることをassertチェック済み") # 検算 assert abs(D_axis * total_tracks - total_credit_mentions) < 1e-9, ( f"D_axis equivalence check failed: " f"D({D_axis}) * total_tracks({total_tracks}) != total_credit_mentions({total_credit_mentions})" ) return { "album_slug": album_slug, "artist": artist, "title": title, "release_year": release_year, "total_tracks": total_tracks, "method_version": METHOD_VERSION, "normalization_method_version": NORMALIZATION_METHOD_VERSION, "generated_at": datetime.now(ZoneInfo("Asia/Tokyo")).isoformat(), "axes": { "C_axis": C_axis, "R_axis": R_axis, "D_axis": D_axis, "S_axis": S_axis, }, "axes_normalized": { "C_axis_normalized": C_axis_normalized, "R_axis_normalized": R_axis_normalized, "D_axis_normalized": D_axis_normalized, "S_axis_normalized": S_axis_normalized, }, "intermediate_values": { "collaborator_counts": dict(sorted( collaborator_counts.items(), key=lambda kv: (-kv[1], kv[0]) )), "total_credit_mentions": total_credit_mentions, "collaborators_total_count": collaborators_total_count, "share_by_collaborator": dict(sorted(share_by_collaborator.items())), "top5_collaborators": top5_collaborators, "top5_mentions": top5_mentions, "collaborator_count_by_track": collaborator_count_by_track, "average_collaborators_per_track": D_axis, "self_aliases_used": sorted(self_aliases_set), "self_credit_tracks": self_credit_tracks_count, "tracks_without_self_credit": tracks_without_self_credit, "writer_producer_overlap_integration": integration_evidence, }, "data_quality_flags": flags, "source_notes": notes, "rounding_rule": ROUNDING_RULE, # 2026-09-10 追加 (第三者レビュー対応): R_axisとは式の異なる補助指標。 # 既存キーには一切手を加えていない。 "repeat_participation": { "RP_axis": repeat_participation["RP_axis"], "repeat_contributors_count": repeat_participation["repeat_contributors_count"], "repeat_contributors": repeat_participation["repeat_contributors"], "repeat_mentions": repeat_participation["repeat_mentions"], "mean_pairwise_jaccard": pairwise_jaccard["mean_pairwise_jaccard"], "jaccard_pair_count": pairwise_jaccard["pair_count"], "method_note": ( "RP_axis = (2曲以上に登場した協働者のmention合計) / (全mention数) × 100。" "曲間の重なりの補助指標。固定メンバーの確認には曲別クレジット集合を確認し、" "RP・平均Jaccardは補助情報として扱う。R_axisの式は変更しない。" ), }, } # --- アルバム登録 --- # 入力テンプレートの登録。公開する実作品の範囲はbuild_v5.pyの6作品で固定する。 # take_care_dummyはローカル検証用、icemanは空テンプレートであり公開集計から除外。 # `--album` で受け付ける slug は ALBUM_REGISTRY.keys() から自動生成される。 ALBUM_REGISTRY: dict[str, dict] = { "take_care": { "artist": "Drake", "title": "Take Care", "release_year": 2011, }, "take_care_dummy": { "artist": "Drake", "title": "Take Care (DUMMY)", "release_year": 2011, }, "views": { "artist": "Drake", "title": "Views", "release_year": 2016, }, "scorpion": { "artist": "Drake", "title": "Scorpion", "release_year": 2018, }, "for_all_the_dogs": { "artist": "Drake", "title": "For All the Dogs", "release_year": 2023, }, "gnx": { "artist": "Kendrick Lamar", "title": "GNX", "release_year": 2024, }, "iceman": { "artist": "Drake", "title": "Iceman", "release_year": 2026, }, "ten_til_midnight": { "artist": "Snoop Dogg", "title": "10 Til' Midnight", "release_year": 2026, }, } def _flatten_for_csv(result: dict) -> dict: iv = result["intermediate_values"] return { "album_slug": result["album_slug"], "artist": result["artist"], "title": result["title"], "release_year": result["release_year"], "total_tracks": result["total_tracks"], "C_axis": result["axes"]["C_axis"], "R_axis": result["axes"]["R_axis"], "D_axis": result["axes"]["D_axis"], "S_axis": result["axes"]["S_axis"], "C_axis_normalized": result["axes_normalized"]["C_axis_normalized"], "R_axis_normalized": result["axes_normalized"]["R_axis_normalized"], "D_axis_normalized": result["axes_normalized"]["D_axis_normalized"], "S_axis_normalized": result["axes_normalized"]["S_axis_normalized"], "collaborators_total_count": iv["collaborators_total_count"], "top5_collaborators": "; ".join(iv["top5_collaborators"]), "top5_mentions": iv["top5_mentions"], "total_credit_mentions": iv["total_credit_mentions"], "average_collaborators_per_track": iv["average_collaborators_per_track"], "self_credit_tracks": iv["self_credit_tracks"], "method_version": result["method_version"], "normalization_method_version": result["normalization_method_version"], "data_quality_flags": "; ".join(result["data_quality_flags"]), "source_notes": " | ".join(result["source_notes"]), "generated_at": result["generated_at"], } def _upsert_to_csv(result: dict, csv_path: Path) -> None: """ album_slug をキーとした upsert: - 既存CSVに同じ album_slug があれば、その行を削除 - 最新結果を1行として追加 - 他の album_slug 行は保持 再実行時の行重複を防ぐ。 """ import csv as _csv row = _flatten_for_csv(result) fieldnames = list(row.keys()) other_rows: list[dict] = [] if csv_path.exists(): with csv_path.open("r", encoding="utf-8", newline="") as f: reader = _csv.DictReader(f) for r in reader: if r.get("album_slug") != result["album_slug"]: # 新フィールドの過不足を吸収 (不足列は空文字、余分は無視) normalized = {k: r.get(k, "") for k in fieldnames} other_rows.append(normalized) with csv_path.open("w", encoding="utf-8", newline="") as f: w = _csv.DictWriter(f, fieldnames=fieldnames) w.writeheader() for r in other_rows: w.writerow(r) w.writerow(row) def main() -> int: parser = argparse.ArgumentParser( description="HPCI 4軸 (Production Constellation Index) 算出。" ) parser.add_argument( "--album", required=True, help="ALBUM_REGISTRYに登録されたアルバムslug", ) parser.add_argument( "--csv", default=None, help="入力CSVパス (省略時は data/credits_input_templates/{album}_input.csv)", ) parser.add_argument( "--aliases", default=None, help="人物同定辞書CSVパス", ) parser.add_argument( "--out-json", default=None, help="JSON出力先 (省略時は data/credits/{album}.json)", ) parser.add_argument( "--out-csv", default=None, help="集計CSV出力先 (省略時は data/hpci_results_constellation.csv)", ) # --all 系オプションは試走段階では実装しない (本体プロンプト §9) args = parser.parse_args() # 安全制約: ALBUM_REGISTRY.keys() から自動生成されたホワイトリスト if args.album not in ALBUM_REGISTRY: print( f"ERROR: album '{args.album}' is not in ALBUM_REGISTRY.\n" f"Available albums: {sorted(ALBUM_REGISTRY.keys())}\n" f"新規アルバムを追加する場合は src/compute_hpci.py の ALBUM_REGISTRY に登録してください。\n" f"`--all` や複数アルバム一括処理は試走承認後に追加実装します。", file=sys.stderr, ) return 2 meta = ALBUM_REGISTRY[args.album] project_root = Path(__file__).resolve().parent.parent csv_path = Path(args.csv) if args.csv else project_root / "data" / "credits_input_templates" / f"{args.album}_input.csv" out_json = Path(args.out_json) if args.out_json else project_root / "data" / "credits" / f"{args.album}.json" out_csv = Path(args.out_csv) if args.out_csv else project_root / "data" / "hpci_results_constellation.csv" print(f"=== HPCI 4軸 試走 ({args.album}) ===") print(f" artist: {meta['artist']}") print(f" title: {meta['title']}") print(f" csv: {csv_path}") registry_before = _registry_tools().read_snapshot(out_json, default={}) if is_registry_path(out_json) else None result = compute_hpci( album_slug=args.album, artist=meta["artist"], title=meta["title"], release_year=meta["release_year"], csv_path=csv_path, aliases_csv=Path(args.aliases) if args.aliases else DEFAULT_ALIASES_CSV, ) # 既存ファイルをコピー退避し、読者が途中の欠落を見ないよう最後にatomic置換。 if registry_before is None and out_json.exists(): import shutil backup = out_json.with_suffix(".json.backup_pre_run") shutil.copy2(out_json, backup) print(f" 既存JSON を {backup.name} にコピー退避しました") out_json.parent.mkdir(parents=True, exist_ok=True) write_json_atomic(out_json, result, expected_sha256=registry_before.sha256 if registry_before is not None else None) print(f" → JSON: {out_json}") # CSV出力 (upsert: 同じalbum_slugがあれば置換) out_csv.parent.mkdir(parents=True, exist_ok=True) _upsert_to_csv(result, out_csv) print(f" → CSV : {out_csv}") print("\n--- 4軸結果 ---") for axis in "CRDS": print(f" {axis}_axis: {format_value(result['axes'][axis + '_axis'])} " f"(normalized: {format_value(result['axes_normalized'][axis + '_axis_normalized'])})") print(f" RP_axis: {format_value(result['repeat_participation']['RP_axis'])}") print(f"\n top5: {result['intermediate_values']['top5_collaborators']}") print(f" total_credit_mentions: {result['intermediate_values']['total_credit_mentions']}") print(f" collaborators_total_count: {result['intermediate_values']['collaborators_total_count']}") print(f" data_quality_flags: {result['data_quality_flags']}") return 0 if __name__ == "__main__": sys.exit(main())