#!/usr/bin/env python3
"""debate_runner.py — 動画の台本を、ほかの AI に議論させる「議論係」.
Claude だけで決めると、同じ書き手の盲点が残る。そこで役を分けた3社に台本を読ませ、
1回目は別々に指摘、2回目はお互いの指摘に賛成か反論かを1回だけ言わせる。
採るかどうかは、最後にエイイチが決める（AI の多数決では決めない）。
役:
Claude … 台本の作り手。構成と流れの指摘、最後の論点表（まとめ.md）。チャットの中で書く
Gemini … 視聴者役。経営者・PM・開発者になりきって、飽きる所・止まる言葉を言う（API）
codex … 事実の番。1回目は事実確認。2回目は「直す案が事実をゆがめないか」だけを見る
（ChatGPT のログインで動く。codex_runner.py と同じ本体を使う）
置き場所（動画/{動画名}/議論/）:
依頼.md Claude が書く。台本と、チャンネルの前提
1_Claude.md Claude が書く（1回目）
1_codex.md 無ければ codex にかける。前の codex_検証結果 を写して置けば、それを使う
1_Gemini.md このスクリプトが書く
2_Gemini.md このスクリプトが書く（2回目）
2_codex.md このスクリプトが書く（2回目）
かける.txt 印。Claude が置く。終わると かけた.txt に名前を変える
キー:
Gemini の API キーは Mac のキーチェーン（名前 gemini_api_key）から読む。ファイルには書かない。
入れ方: security add-generic-password -a "$USER" -s gemini_api_key -w
使い方（Mac のターミナルで）:
python3 debate_runner.py --check キーとモデルを確かめるだけ（議論はしない）
python3 debate_runner.py --once 印のついた動画を1本、議論にかける
python3 debate_runner.py --dry-run 何をするかを出すだけ
python3 debate_runner.py --install 5分ごとに見回るように登録する（一度だけ。以後は印を置けば自動で回る）
python3 debate_runner.py --uninstall 自動の見回りを外す
Python の標準の道具だけで動く。台本（narration.py など）には触らない。ファイルは消さない。
"""
import argparse, json, os, re, subprocess, sys, time, unicodedata, urllib.error, urllib.request
from datetime import datetime, timezone, timedelta
HERE = os.path.dirname(os.path.abspath(__file__))
VIDEOS = os.path.dirname(HERE) # 動画/
sys.path.insert(0, HERE)
import codex_runner as cr # codex の本体の場所と環境を使い回す
JST = timezone(timedelta(hours=9))
SUB = "議論"
MARK = "かける.txt"
DONE = "かけた.txt"
RUNNING = "実行中.txt"
REQ = "依頼.md"
THEMES = "_議論" # 動画/_議論/{テーマ}/ … 台本ではなく、企画や方針を議論するとき（印と結果はフォルダの中に直接置く）
ROLE_GEMINI = "役_Gemini.md" # テーマの議論で、Gemini の1回目の役（あれば視聴者役の代わりに使い、Google 検索も使わせる）
ROLE_CODEX = "役_codex.md" # テーマの議論で、codex の1回目の役
KEYCHAIN_SERVICE = "gemini_api_key"
GEMINI_BASE = "https://generativelanguage.googleapis.com/v1beta"
GEMINI_MODEL = os.environ.get("GEMINI_MODEL", "") # 空なら、使える中でいちばん上の Pro を選ぶ
GEMINI_TIMEOUT = 10 * 60
CODEX_EFFORT = "medium" # 2回目は検索なしで、直す案の事実だけを見る
CODEX_TIMEOUT = 30 * 60
# ───────── 共通の前提（3社に同じものを渡す） ─────────
GROUND = """\
# 議論の前提（全員に共通）
あなたは、YouTube の解説動画の台本を良くするための議論に参加しています。
参加者は Claude（台本の作り手）・Gemini（視聴者役）・codex（事実の番）の3つの AI です。
採るかどうかは、最後に人間（エイイチ）が決めます。あなたは材料を出す役です。
## 判定の軸（この4つだけで評価する）
1. ターゲットの役に立つか（ロボティクスに興味がある経営者・PM・マネージャー・開発者）
2. 最後まで分かるか（耳で聞いて分かる平易な言葉か・冗長でないか・飽きないか）
3. エイイチ（チャンネルの人）の見立て・実感が入っているか
4. 事実は正しいか（言い過ぎていないか）
## 出してはいけない提案
- 再生数を稼ぐための誇張・釣り・煽り。中身と合わないタイトルやサムネイル
- 事実を曲げる・不安をあおる・他社や人を悪く言う演出
- 「ユーモアや実体験を AI が作って足す」案（実体験は本人に聞いて入れる決まり）
再生数が伸びそうでも、上の4つの軸に反する案は出さないこと。
## 書き方
- 日本語で書く。英語の文を混ぜない
- 台本の行は「文番号（S001 など）」で指す
- 指摘には重さを付ける：【高】直さないと伝わらない・誤り ／【中】直すと良くなる ／【低】好み
- 直す案は、元の文を置き換える短い案文で出す（長い書き直しはしない）
"""
GEMINI_R1 = """\
# あなたの役：視聴者（Gemini）
次の3人になりきって、この台本の動画を最初から最後まで見たつもりで読んでください。
- 経営者：飲食チェーンの役員。ロボット導入の判断をする立場。数字とお金に敏感で、忙しい
- PM：サービスロボットの導入プロジェクトを任された人。現場とメーカーの間に立つ
- 開発者：ロボットのソフトを書く人。試算の前提の甘さに気づく
## 出してほしいもの（この順で）
1. 離脱しそうな所：3人それぞれ、どの文番号のあたりで「もういいか」と思うか。理由を1行で
2. 止まる言葉：耳で聞いて意味が取れない言葉・数字の詰め込み。文番号と言い換え案
3. 物足りない所：その人が知りたいのに出てこないこと
4. 良かった所：残すべき所（2〜3個）
5. いちばん直すべき3つ：重さ付きで
ほかの AI の意見はまだ見せていません。自分の目で見て書いてください。
"""
ROUND2 = """\
# 2回目：ほかの AI の指摘への返事（{who}）
下に、3つの AI の1回目の指摘があります（あなた自身のものも含みます）。
ほかの AI の指摘1つずつに、次のどれかで答えてください。新しい指摘は足さないこと。
- 賛成：理由を1行
- 反論：理由を1行（直すと何が悪くなるか）
- 保留：判断に人間の情報が要るもの（何を聞けばよいか）
{extra}
最後に「人間が先に決めるべき3つ」を、重さ付きで挙げてください。
話を丸くまとめようとせず、意見が割れる所は割れたまま書いてください。
"""
CODEX_EXTRA = """\
あなたは事実の番です。賛成・反論の判断は「その直す案を採ったとき、事実が正しいままか・
言い過ぎにならないか」だけで決めてください。構成・面白さ・言葉の好みには意見しないこと
（その指摘は「事実の面では問題なし」とだけ書く）。話の筋や構成を組み替える提案もしないこと。
作業フォルダの narration.py の冒頭のコメントに、数字の出どころが書いてあります。
"""
GEMINI_EXTRA = """\
あなたは視聴者役のままです。経営者・PM・開発者の目で、その直し方で見やすくなるかを判断してください。
codex（事実の番）の1回目の指摘には返事をしないこと（事実の判断は codex に任せる）。返事をするのは Claude の指摘だけ。
賛成ばかりにしないこと。視聴者として本当に見やすくなるかを考え、反論すべき所は反論してください。
"""
THEME_EXTRA = """\
賛成ばかりにしないこと。ほかの AI の数字や見立てに、根拠が弱い所・甘い所があれば反論してください。
"""
THEME_CODEX_EXTRA = """\
あなたは数字の番です。ほかの AI が出した数字（再生数の見込み・ほかのチャンネルの数字など）について、
根拠が確かか・言い過ぎていないかを中心に返事をしてください。
"""
def log(*a):
print(datetime.now(JST).strftime("%Y-%m-%d %H:%M:%S"), *a, flush=True)
def nfc(s):
return unicodedata.normalize("NFC", s)
def notify(msg):
try:
t = msg.replace("\\", "\\\\").replace('"', '\\"')
subprocess.run(["osascript", "-e", f'display notification "{t}" with title "議論係"'],
capture_output=True, timeout=20)
except Exception:
pass
def read(path):
b, _ = cr.read_stable(path)
return b.decode("utf-8", errors="replace")
def write(path, text):
with open(path, "w", encoding="utf-8") as f:
f.write(text.rstrip() + "\n")
# ───────── Gemini ─────────
def gemini_key():
r = subprocess.run(["/usr/bin/security", "find-generic-password", "-s", KEYCHAIN_SERVICE, "-w"],
capture_output=True, text=True)
key = r.stdout.strip()
if r.returncode != 0 or not key:
raise RuntimeError("キーチェーンに gemini_api_key が見つかりません。"
" security add-generic-password -a \"$USER\" -s gemini_api_key -w で入れてください")
return key
def http(url, key, body=None, timeout=60):
data = json.dumps(body).encode("utf-8") if body is not None else None
req = urllib.request.Request(url, data=data, method="POST" if data else "GET",
headers={"x-goog-api-key": key, "Content-Type": "application/json"})
try:
with urllib.request.urlopen(req, timeout=timeout) as r:
return json.loads(r.read().decode("utf-8"))
except urllib.error.HTTPError as e:
detail = e.read().decode("utf-8", "replace")[:800]
raise RuntimeError(f"Gemini の API がエラーを返しました（{e.code}）: {detail}")
MODEL_RE = re.compile(r"^models/(gemini-(\d+(?:\.\d+)?)-(pro|flash)(-preview(?:-\d{2}-\d{4})?)?)$") # customtools・tts などの変わり種は外す
def pick_model(key):
"""使えるモデルの一覧から選ぶ。Pro を優先し、版の数字が大きいもの、正式版を優先"""
if GEMINI_MODEL:
return GEMINI_MODEL, []
out, url = [], f"{GEMINI_BASE}/models?pageSize=1000"
d = http(url, key)
cands = []
for m in d.get("models", []):
name = m.get("name", "")
mm = MODEL_RE.match(name)
if not mm or "generateContent" not in m.get("supportedGenerationMethods", []):
continue
model, ver, kind, pre = mm.groups()
cands.append(((kind == "pro", float(ver), not pre), model))
cands.sort(reverse=True)
if not cands:
raise RuntimeError("使える Gemini のモデルが見つかりません")
return cands[0][1], [c[1] for c in cands[:6]]
THEME_GROUND = """\
# 議論の前提（全員に共通）
あなたは、YouTube チャンネル「フィジカルAI解説｜Pixels to Motors」の企画や方針を決めるための議論に参加しています。
参加者は Claude・Gemini・codex の3つの AI です。採るかどうかは、最後に人間（エイイチ）が決めます。あなたは材料を出す役です。
- 再生数を稼ぐための誇張・釣り・煽り、中身と合わないタイトルやサムネイルは、伸びそうでも提案しない
- 日本語で書く。英語の文を混ぜない
- 数字を言うときは、根拠（実際のデータ・調べた出典・推測）のどれかを必ず添える。推測は推測と書く
- 話を丸くまとめず、ほかの AI と意見が割れる所は割れたまま書く
"""
LAST_USAGE = {"inp": 0, "out": 0}
def gemini(key, model, system, user, search=False):
body = {
"systemInstruction": {"parts": [{"text": system}]},
"contents": [{"role": "user", "parts": [{"text": user}]}],
"generationConfig": {"temperature": 0.7},
}
if search:
body["tools"] = [{"google_search": {}}]
t0 = time.time()
d = http(f"{GEMINI_BASE}/models/{model}:generateContent", key, body, timeout=GEMINI_TIMEOUT)
parts = (d.get("candidates") or [{}])[0].get("content", {}).get("parts", [])
text = "\n".join(p.get("text", "") for p in parts if not p.get("thought")).strip()
if not text:
raise RuntimeError(f"Gemini の答えが空でした: {json.dumps(d, ensure_ascii=False)[:600]}")
gm = (d.get("candidates") or [{}])[0].get("groundingMetadata", {})
srcs = [c.get("web", {}) for c in gm.get("groundingChunks", []) if c.get("web")]
if srcs:
text += "\n\n## Gemini が検索で見た出典\n\n" + "\n".join(f"- {w.get('title', '')}: {w.get('uri', '')}" for w in srcs)
u = d.get("usageMetadata", {})
LAST_USAGE.update(inp=u.get("promptTokenCount", 0) or 0, out=(u.get("candidatesTokenCount", 0) or 0) + (u.get("thoughtsTokenCount", 0) or 0))
meta = f"{'Google 検索あり・' if search else ''}モデル {model}・{int(time.time() - t0)}秒・入力 {u.get('promptTokenCount', '?')}・出力 {u.get('candidatesTokenCount', '?')} トークン"
return text, meta
# ───────── codex ─────────
def codex(wdir, prompt, web=False):
tmp = os.path.join(os.environ.get("TMPDIR", "/tmp"), f"debate_codex_{os.getpid()}.md")
cmd = [cr.CODEX, "exec"] + (["-c", "tools.web_search=true"] if web else []) + ["-c", f'model_reasoning_effort="{CODEX_EFFORT}"', "-m", cr.MODEL,
"-s", "read-only", "--skip-git-repo-check", "-C", wdir, "-o", tmp, "-"]
t0 = time.time()
r = subprocess.run(cmd, input=prompt, text=True, encoding="utf-8", capture_output=True,
timeout=CODEX_TIMEOUT, env=cr.env_for_codex(), cwd=wdir)
ans = ""
if os.path.isfile(tmp):
with open(tmp, encoding="utf-8", errors="replace") as f:
ans = f.read().strip()
os.remove(tmp)
if r.returncode != 0 or not ans:
raise RuntimeError(f"codex が失敗しました（終了コード {r.returncode}）:\n" + "\n".join((r.stderr or "").splitlines()[-30:]))
tk = re.findall(r"tokens used\s*\n?\s*([\d,]+)", r.stderr or "")
return ans, f"モデル {cr.MODEL}・考える深さ {CODEX_EFFORT}・{int(time.time() - t0)}秒・{tk[-1] if tk else '?'} トークン"
# ───────── 1本分 ─────────
def head(title, name, meta):
return f"# {title}（{name}）\n\n- 日時: {datetime.now(JST).strftime('%Y-%m-%d %H:%M')}（JST）\n- {meta}\n\n---\n\n"
def run_one(name, vdir, dry, ddir=None):
theme = ddir is not None
ddir = ddir or os.path.join(vdir, SUB)
vdir = vdir or ddir
files = cr.listdir_nfc(ddir)
p = lambda n: os.path.join(ddir, files.get(n, n))
if REQ not in files:
log(f"× {name}: {SUB}/{REQ} がありません")
return
req = read(p(REQ))
todo = [f for f in ("1_Gemini.md", "1_codex.md", "2_Gemini.md", "2_codex.md") if f not in files]
log(f"▶ {name}: これから作るもの → {'・'.join(todo) or 'なし'}")
if "1_Claude.md" not in files:
log(" 注意: 1_Claude.md がありません（Claude 抜きで進めます）")
if dry:
return
write(p(RUNNING), f"開始 {datetime.now(JST).strftime('%Y-%m-%d %H:%M:%S')}\nPID {os.getpid()}\n")
try:
key = gemini_key()
model, _ = pick_model(key)
# 1回目
ground = THEME_GROUND if theme else GROUND
if "1_Gemini.md" not in files:
log(f" 1回目 Gemini（{model}）…")
if theme and ROLE_GEMINI in files:
ans, meta = gemini(key, model, ground + "\n" + read(p(ROLE_GEMINI)), req, search=True)
else:
ans, meta = gemini(key, model, GROUND + "\n" + GEMINI_R1, req)
write(p("1_Gemini.md"), head("1回目 Gemini（視聴者役）", name, meta) + ans)
files["1_Gemini.md"] = "1_Gemini.md"
if "1_codex.md" not in files:
log(" 1回目 codex（事実確認）…")
if theme and ROLE_CODEX in files:
ans, meta = codex(vdir, ground + "\n" + read(p(ROLE_CODEX)) + "\n\n" + req, web=True)
else:
ans, meta = codex(vdir, GROUND + "\n# あなたの役：事実の番（codex）\n台本の事実確認をしてください。"
"観点は事実と言い過ぎだけ。筋・構成・見立ては変えない。\n\n" + req, web=True)
write(p("1_codex.md"), head("1回目 codex（事実の番）", name, meta) + ans)
files["1_codex.md"] = "1_codex.md"
# 2回目（3つの1回目をまとめて見せる）
r1 = ""
labels = (("Claude", "1_Claude.md"), ("Gemini", "1_Gemini.md"), ("codex", "1_codex.md")) if theme else \
(("Claude（台本の作り手）", "1_Claude.md"), ("Gemini（視聴者役）", "1_Gemini.md"), ("codex（事実の番）", "1_codex.md"))
for who, f in labels:
if f in files:
r1 += f"\n\n==================== {who} の1回目 ====================\n\n" + read(p(f))
body2 = ("# 依頼\n\n" if theme else "# 台本と前提\n\n") + req + "\n\n# 1回目の意見" + r1
if "2_Gemini.md" not in files:
log(" 2回目 Gemini…")
if theme:
ans, meta = gemini(key, model, ground + "\n" + ROUND2.format(who="Gemini", extra=THEME_EXTRA), body2)
else:
ans, meta = gemini(key, model, GROUND + "\n" + ROUND2.format(who="Gemini・視聴者役", extra=GEMINI_EXTRA), body2)
write(p("2_Gemini.md"), head("2回目 Gemini", name, meta) + ans)
if "2_codex.md" not in files:
log(" 2回目 codex…")
if theme:
ans, meta = codex(vdir, ground + "\n" + ROUND2.format(who="codex", extra=THEME_CODEX_EXTRA) + "\n\n" + body2)
else:
ans, meta = codex(vdir, GROUND + "\n" + ROUND2.format(who="codex・事実の番", extra=CODEX_EXTRA) + "\n\n" + body2)
write(p("2_codex.md"), head("2回目 codex", name, meta) + ans)
except Exception as e:
log(f"× {name}: {e}")
write(p("エラー.txt"), f"{datetime.now(JST).strftime('%Y-%m-%d %H:%M:%S')}\n\n{e}\n")
notify(f"議論が止まりました：{name}")
return
finally:
if os.path.exists(p(RUNNING)):
os.remove(p(RUNNING))
if os.path.exists(p(MARK)):
os.rename(p(MARK), p(DONE))
log(f"✓ {name}: 議論が終わりました → 1_・2_ のファイル。次は Claude がまとめる番")
notify(f"議論が終わりました：{name}")
def research(dry):
"""動画/_調べ/{名前}/ に かける.txt があれば、YouTube を検索して数字を取る（youtube_market.py）"""
real = cr.listdir_nfc(VIDEOS).get(nfc("_調べ"))
if not real:
return False
root = os.path.join(VIDEOS, real)
for name, r in sorted(cr.listdir_nfc(root).items()):
d = os.path.join(root, r)
f = cr.listdir_nfc(d) if os.path.isdir(d) else {}
if MARK in f and "キーワード.txt" in f:
log(f"▶ YouTube の調べ _調べ/{name}")
if dry:
return True
try:
import youtube_market
n = youtube_market.run(d)
os.rename(os.path.join(d, f[MARK]), os.path.join(d, DONE))
log(f"✓ _調べ/{name}: {n} 語を調べました → 結果.md")
notify(f"YouTube の調べが終わりました：{name}")
except Exception as e:
write(os.path.join(d, "エラー.txt"), f"{datetime.now(JST).strftime('%Y-%m-%d %H:%M:%S')}\n\n{e}\n")
os.rename(os.path.join(d, f[MARK]), os.path.join(d, "失敗.txt"))
log(f"× _調べ/{name}: {e}")
return True
return False
def sweep(dry):
if research(dry):
return
import neta # 動画/_ネタ/ … ネタ出し（neta.py）
if neta.sweep(dry):
return
import seisaku # 動画/_制作/ … 制作チーム（seisaku.py）
if seisaku.sweep(dry):
return
import kikaku # 動画/_企画/ … 100回の企画会議（kikaku.py）
if kikaku.sweep(dry):
return
troot = cr.listdir_nfc(VIDEOS).get(nfc(THEMES))
if troot:
tdir = os.path.join(VIDEOS, troot)
for name, real in sorted(cr.listdir_nfc(tdir).items()):
d = os.path.join(tdir, real)
if not os.path.isdir(d):
continue
f = cr.listdir_nfc(d)
if RUNNING in f:
log(f"いま動いています: {THEMES}/{name}（今回は何もしません）")
return
if MARK in f:
run_one(f"{THEMES}/{name}", None, dry, ddir=d)
return
for name, real in sorted(cr.listdir_nfc(VIDEOS).items()):
if name.startswith("_"):
continue
vdir = os.path.join(VIDEOS, real)
sub = cr.listdir_nfc(vdir).get(nfc(SUB))
if not sub:
continue
f = cr.listdir_nfc(os.path.join(vdir, sub))
if RUNNING in f:
log(f"いま動いています: {name}（今回は何もしません）")
return
if MARK in f:
run_one(name, vdir, dry)
return # 一度に1本だけ
log("印（議論/かける.txt・_議論/{テーマ}/かける.txt）のついたものはありません")
LABEL = "jp.adaptdesign.pixelstomotors.debate-runner"
PLIST = os.path.expanduser(f"~/Library/LaunchAgents/{LABEL}.plist")
LOGFILE = os.path.join(cr.CONF, "debate_runner.log")
def install():
"""5分ごとに見回るように登録する。印（かける.txt）を置けば、ターミナルを開かなくても議論が回る"""
os.makedirs(os.path.dirname(PLIST), exist_ok=True)
os.makedirs(cr.CONF, exist_ok=True)
cal = "".join(f"Minute{m}" for m in range(2, 60, 5))
plist = f"""
Label{LABEL}
ProgramArguments{sys.executable}{os.path.abspath(__file__)}--once
StartCalendarInterval{cal}
RunAtLoad
WorkingDirectory{HERE}
EnvironmentVariables
PATH/usr/bin:/bin:/usr/sbin:/sbin
CODEX_HOME{cr.CODEX_HOME}
LANGja_JP.UTF-8
StandardOutPath{LOGFILE}
StandardErrorPath{LOGFILE}
"""
with open(PLIST, "w") as f:
f.write(plist)
subprocess.run(["launchctl", "unload", PLIST], capture_output=True)
r = subprocess.run(["launchctl", "load", PLIST], capture_output=True, text=True)
if r.returncode != 0:
raise RuntimeError(f"登録できませんでした: {r.stderr.strip()}")
print(f"登録しました。5分ごとに見回ります（いまも1回動きます）。記録: {LOGFILE}")
def uninstall():
subprocess.run(["launchctl", "unload", PLIST], capture_output=True)
if os.path.exists(PLIST):
os.remove(PLIST)
print("自動の見回りを外しました")
def check():
key = gemini_key()
log(f"キー: 見つかりました（頭 {key[:4]}…・{len(key)} 文字）")
model, top = pick_model(key)
log(f"選んだモデル: {model}")
if top:
log("候補の上位:", "・".join(top))
ans, meta = gemini(key, model, "日本語で、ひとことで答えてください。", "動作確認です。「準備できました」とだけ返してください。")
log(f"Gemini の返事: {ans}（{meta}）")
log(f"codex の本体: {'あります' if os.path.isfile(cr.CODEX) else '見つかりません'}（{cr.CODEX}）")
if __name__ == "__main__":
ap = argparse.ArgumentParser(description="動画の台本を、Gemini と codex に議論させる")
ap.add_argument("--check", action="store_true", help="キーとモデルを確かめるだけ")
ap.add_argument("--once", action="store_true", help="印のついた動画を1本、議論にかける")
ap.add_argument("--dry-run", action="store_true", help="何をするかを出すだけ")
ap.add_argument("--install", action="store_true", help="5分ごとに見回るように登録する")
ap.add_argument("--uninstall", action="store_true", help="自動の見回りを外す")
a = ap.parse_args()
try:
if a.install:
install()
elif a.uninstall:
uninstall()
elif a.check:
check()
else:
sweep(a.dry_run)
except Exception as e:
log("×", e)
sys.exit(1)