#!/usr/bin/env python3 """kikaku.py — 動画の企画を、AI に100回以上議論させる「企画会議」. debate_runner.py の見回り（launchd・5分ごと）から呼ばれる。手で動かすときも debate_runner.py --once でよい。 置き場所（動画/_企画/{会議の名前}/）: 依頼.md Claude が書く。何の企画を決めたいか・前提・Claude の最初の案 回数.txt 何回議論するか（無ければ100）。数字だけを書く かける.txt 印。Claude が置く。終わると かけた.txt に名前を変える 止める.txt エイイチが置くと、その回が終わったところで止まる（あとで消せば続きから） 進み具合.md 毎回書き換える。いま何回目か・どの段階か・いまの企画ボード・かかった費用の目安 ログ.md 1回ごとの発言を、上に足していく（新しいものが上） 状態.json 続きから再開するための記録 最終案.md 100回終わったときの企画ボード 段階（回数を100としたとき。回数を変えると比で伸び縮みする）: 1〜20 出す 企画の案を広く出す 21〜60 揉む 案の穴を突き、直し、合わせ、捨てる 61〜90 絞る 4つの軸と見込みの再生数で点を付け、上位に絞る 91〜100 仕上げる 最終の3本を、題名・サムネイルの文字・冒頭5秒・構成・見込みまで書く 話す役: 10回に1回は codex（数字と釣りの番）。残りは Gemini が、次の役を順に回す 経営者の視聴者・PM の視聴者・開発者の視聴者・YouTube の企画の編集者・穴を突く役・ロボットに詳しくない一般の視聴者・エイイチの目的を守る役 Claude は最初の案（依頼.md の中）と、終わったあとのまとめで加わる """ import json, os, re, signal, subprocess, sys, time from datetime import datetime import debate_runner as dr cr = dr.cr ROOT = "_企画" MARK, DONE, STOP, RUNNING = "かける.txt", "かけた.txt", "止める.txt", "実行中.txt" PRICE = {"pro": (2.0, 12.0), "flash": (0.75, 3.75)} # 1百万トークンあたりのドル（入力・出力の目安。Pro は 3.1 Pro preview、Flash は 3.x Flash） YEN = 155 # 円の換算の目安 MAX_PARALLEL = 3 # 同時に回す会議の数の上限（会議は API を待つだけで Mac は軽い） PRO_ROLES = ("穴を突く",) # この役と「仕上げる」段階だけ Pro。ほかは安い Flash PERSONAS = [ ("経営者の視聴者", "飲食・小売・病院などでロボット導入を判断する立場。忙しく、数字とお金で考える。押したくなる題名か、最後まで見たいかを正直に言う"), ("PM の視聴者", "ロボット導入のプロジェクトを任された人。現場とメーカーの間に立つ。自分の仕事に使えるかで見る"), ("開発者の視聴者", "ロボットや AI のソフトを書く人。前提の甘さ・ありきたりな話にすぐ気づく。新しく分かることがあるかで見る"), ("YouTube の企画の編集者", "日本の YouTube でどんな題材・題名・サムネイルが表示され、押され、見続けられるかに詳しい。釣りはしないが、入口の作りには厳しい"), ("穴を突く役", "今の案のいちばん弱い所を探して突く。みんなが賛成している案ほど疑う"), ("ロボットに詳しくない一般の視聴者", "ロボットにも AI にも詳しくない、普段は身近なニュースや雑学の動画を見る人。おすすめに出てきたとき、題名とサムネイルだけで押すかどうかを決める。知らない言葉が1つでもあれば押さない"), ("エイイチの目的を守る役", "チャンネルの目的（仕事につながる人に見てもらう）と、守ること（釣りをしない・顔は出さない・声は AI）からずれていないかを見る"), ] PHASES = [(0.20, "出す", "企画の案を広く出す段階。いまのボードにない新しい案を1〜3本足す。似た案を足さない"), (0.60, "揉む", "案を揉む段階。ボードの案の穴を突き、直し、合わせ、見込みのない案は理由を書いて捨てる"), (0.90, "絞る", "絞る段階。4つの軸（ターゲットの役に立つ・最後まで分かる・エイイチの見立てがある・事実が正しい）と、公開から3か月の見込みの再生数で点を付け、上位5本に絞る"), (1.01, "仕上げる", "仕上げる段階。上位3本について、題名・サムネイルの文字・冒頭5秒のセリフ・構成（4〜6つの節）・見込みの再生数と根拠・撮影や試算で要るものを書く。前の回の書き方をよくする")] WAY_HEAD = """ # この会議のやり方 - これは、動画の企画を決めるための長い会議です。あなたは今回、1回分だけ発言します - 自分の名前（Gemini・codex・Claude など）や役の名前を名乗らない。発言の中身だけを書く - 前の人の発言に、必ず1つは具体的に反応する（賛成でも反論でもよい）。同じことを言い直さない - 「企画ボード」は、会議の今の結論をまとめた板です。あなたの発言を受けて、ボードを書き直してください """ WAY_BOARD_VIDEO = """- ボードは2,500字以内。案ごとに、題名の案・中身・どの柱か・見込みの再生数（根拠つき）・いまの評価・残っている疑問 を書く - 見込みの再生数は、依頼に書いたチャンネルの実際の数字を土台にする。根拠のない楽観はしない """ WAY_TAIL = """ # 出す形（この形を必ず守る） 【発言】 （あなたの発言。400字くらいまで） 【ボード】 （書き直した企画ボードの全文） """ GROUND = dr.THEME_GROUND + WAY_HEAD + WAY_BOARD_VIDEO + WAY_TAIL # 動画の会議（前と同じ） CODEX_ROLE_VIDEO = ("# あなたの役：数字と釣りの番（codex）\nボードの見込みの再生数の根拠と計算を確かめ、甘い所は直す。" "題名やサムネイルの案が中身より言い過ぎていないか（釣りになっていないか）も止める。必要ならウェブで確かめる。\n") # 会議ごとの差し替え（フォルダに置けば、その会議だけ動画用の決まりの代わりに使う） # 前提.md … 議論の前提とボードの書き方（THEME_GROUND と、ボードの2行の代わり） # 段階.json … [[比, "名前", "やること"], ...]（PHASES の代わり） # 番.md … 10回に1回の codex の役（CODEX_ROLE_VIDEO の代わり） # 役.json … Gemini が回す役（PERSONAS の代わり。前からある） def pick_flash(key): """使える Flash の中で、版の数字がいちばん大きい正式版（無ければ preview）""" d = dr.http(f"{dr.GEMINI_BASE}/models?pageSize=1000", key) c = [] for m in d.get("models", []): mm = dr.MODEL_RE.match(m.get("name", "")) if mm and mm.group(3) == "flash" and "generateContent" in m.get("supportedGenerationMethods", []): c.append(((float(mm.group(2)), not mm.group(4)), mm.group(1))) return max(c)[1] if c else None def phase_of(t, total, phases=None): phases = phases or PHASES for edge, name, how in phases: if t <= max(1, round(total * edge)): return name, how return phases[-1][1], phases[-1][2] def split(ans): m = re.search(r"【発言】\s*(.*?)\s*【ボード】\s*(.*)", ans, re.S) if not m: return ans.strip(), None board = m.group(2).strip() return m.group(1).strip(), (board if len(board) > 50 else None) def load(d): p = os.path.join(d, "状態.json") if os.path.isfile(p): return json.load(open(p, encoding="utf-8")) return {"turn": 0, "board": "（まだ案はありません）", "recent": [], "inp": 0, "out": 0, "codex": 0, "start": datetime.now(dr.JST).isoformat(timespec="minutes")} def save(d, st): tmp = os.path.join(d, "状態.json.tmp") with open(tmp, "w", encoding="utf-8") as f: json.dump(st, f, ensure_ascii=False, indent=1) os.replace(tmp, os.path.join(d, "状態.json")) def progress(d, st, total, now_phase, note=""): usd = sum(st.get(f"inp_{k}", 0) / 1e6 * PRICE[k][0] + st.get(f"out_{k}", 0) / 1e6 * PRICE[k][1] for k in PRICE) \ + st["inp"] / 1e6 * PRICE["pro"][0] + st["out"] / 1e6 * PRICE["pro"][1] # inp/out は Pro だけだった前の版の分 bar = "■" * (st["turn"] * 20 // total) + "□" * (20 - st["turn"] * 20 // total) L = [f"# 進み具合（{os.path.basename(d)}）", "", f"- {bar} **{st['turn']} / {total} 回**（段階：{now_phase}）", f"- 始めた時刻 {st['start'].replace('T', ' ')}・最後に書いた時刻 {datetime.now(dr.JST).strftime('%Y-%m-%d %H:%M')}", f"- Gemini の費用の目安 約{usd * YEN:,.0f}円（Pro {st.get('n_pro', 0)} 回・Flash {st.get('n_flash', 0)} 回）・codex {st['codex']} 回（ChatGPT Plus の枠）", "- 止めたいときは、このフォルダに 止める.txt を置く（いまの回が終わったら止まる。消せば続きから）"] if note: L.append(f"- {note}") L += ["", "## 直近の発言", ""] for r in st["recent"][-3:][::-1]: L += [f"**{r['n']}回目｜{r['who']}**", "", r["say"][:400], ""] L += ["## いまの企画ボード", "", st["board"], ""] dr.write(os.path.join(d, "進み具合.md"), "\n".join(L)) def append_log(d, n, total, phase, who, say): p = os.path.join(d, "ログ.md") old = open(p, encoding="utf-8").read() if os.path.isfile(p) else f"# 会議のログ（{os.path.basename(d)}・新しい発言が上）\n\n" head, _, rest = old.partition("\n\n") entry = f"## {n} / {total}｜{phase}｜{who}（{datetime.now(dr.JST).strftime('%H:%M')}）\n\n{say}\n\n" dr.write(p, head + "\n\n" + entry + rest) def run(name, d, dry): files = cr.listdir_nfc(d) if "依頼.md" not in files: dr.log(f"× {name}: 依頼.md がありません") return total = 100 if "回数.txt" in files: try: total = max(10, int(open(os.path.join(d, files["回数.txt"]), encoding="utf-8").read().strip())) except ValueError: pass personas = PERSONAS if "役.json" in files: # 会議ごとに役を差し替える（[["名前","説明"], ...]） personas = [tuple(x) for x in json.load(open(os.path.join(d, files["役.json"]), encoding="utf-8"))] ground, phases, codex_role = GROUND, PHASES, CODEX_ROLE_VIDEO if "前提.md" in files: ground = dr.read(os.path.join(d, files["前提.md"])) + WAY_HEAD + WAY_TAIL if "段階.json" in files: phases = [tuple(x) for x in json.load(open(os.path.join(d, files["段階.json"]), encoding="utf-8"))] if "番.md" in files: codex_role = dr.read(os.path.join(d, files["番.md"])) + "\n" st = load(d) if dry: dr.log(f"［試し］{name}: {st['turn']} / {total} 回まで済み") return req = dr.read(os.path.join(d, files["依頼.md"])) key = dr.gemini_key() model, _ = dr.pick_model(key) flash = pick_flash(key) or model dr.write(os.path.join(d, RUNNING), f"開始 {datetime.now(dr.JST).strftime('%Y-%m-%d %H:%M:%S')}\nPID {os.getpid()}\n") dr.log(f"▶ 企画会議 {name}: {st['turn'] + 1} 回目から {total} 回目まで（Pro {model}・Flash {flash}）") try: while st["turn"] < total: if os.path.exists(os.path.join(d, STOP)): progress(d, st, total, phase_of(st["turn"] + 1, total, phases)[0], "止める.txt があるので止まっています") dr.log(f" 止める.txt があるので {st['turn']} 回目で止めました") return n = st["turn"] + 1 phase, how = phase_of(n, total, phases) recent = "\n\n".join(f"［{r['n']}回目｜{r['who']}］\n{r['say']}" for r in st["recent"][-6:]) or "（まだ発言はありません）" body = (f"# 依頼\n\n{req}\n\n# いまの企画ボード\n\n{st['board']}\n\n# 直近の発言\n\n{recent}\n\n" f"# 今回\n\n{n} / {total} 回目。段階は「{phase}」。{how}") for attempt in range(3): try: if n % 10 == 0: who = "codex（数字と釣りの番）" first = codex_role.split("\n")[0] who = "codex（" + (first.split("：", 1)[-1].split("（")[0].strip() if "：" in first else "番") + "）" ans, _ = dr.codex(d, ground + "\n" + codex_role + "\n" + body, web=True) st["codex"] += 1 else: pname, pdesc = personas[(n - 1) % len(personas)] who = f"Gemini（{pname}）" search = phase in ("出す", "揉む") and ("編集者" in pname or "一般" in pname or "ニュース" in pname) tier = "pro" if (phase == "仕上げる" or any(r in pname for r in PRO_ROLES) or flash == model) else "flash" ans, _ = dr.gemini(key, model if tier == "pro" else flash, ground + f"\n# あなたの役：{pname}\n{pdesc}\n", body, search=search) st[f"inp_{tier}"] = st.get(f"inp_{tier}", 0) + dr.LAST_USAGE["inp"] st[f"out_{tier}"] = st.get(f"out_{tier}", 0) + dr.LAST_USAGE["out"] st[f"n_{tier}"] = st.get(f"n_{tier}", 0) + 1 who += "・Pro" if tier == "pro" else "" break except Exception as e: dr.log(f" {n}回目 失敗（{attempt + 1}回目）: {str(e)[:200]}") if attempt == 2: raise time.sleep(30) say, board = split(ans) if board: st["board"] = board st["turn"] = n st["recent"] = (st["recent"] + [{"n": n, "who": who, "say": say}])[-12:] save(d, st) try: os.utime(os.path.join(d, RUNNING)) except OSError: pass append_log(d, n, total, phase, who, say) progress(d, st, total, phase) dr.log(f" {n}/{total} {phase} {who}") if n in (total // 4, total // 2, total * 3 // 4): dr.notify(f"企画会議 {os.path.basename(d)}：{n} / {total} 回") dr.write(os.path.join(d, "最終案.md"), f"# 最終案（{os.path.basename(d)}・{total}回の会議のあと）\n\n" + st["board"]) mk = os.path.join(d, MARK) if os.path.exists(mk): os.rename(mk, os.path.join(d, DONE)) progress(d, st, total, "終わり", "会議は終わりました。最終案.md を見てください") dr.log(f"✓ 企画会議 {name}: {total} 回終わりました") dr.notify(f"企画会議が終わりました：{os.path.basename(d)}") except Exception as e: dr.write(os.path.join(d, "エラー.txt"), f"{datetime.now(dr.JST).strftime('%Y-%m-%d %H:%M:%S')}\n\n{e}\n") progress(d, st, total, phase_of(st["turn"] + 1, total, phases)[0], f"エラーで止まりました（エラー.txt）。次の見回りで続きから試します") dr.log(f"× 企画会議 {name}: {e}") dr.notify(f"企画会議が止まりました：{os.path.basename(d)}") finally: p = os.path.join(d, RUNNING) if os.path.exists(p): os.remove(p) SITE_ROOT = os.path.join(os.path.dirname(os.path.dirname(dr.VIDEOS)), "サイト改善", "_企画") # PixelstoMotors/サイト改善/_企画/ def roots(): out = [] real = cr.listdir_nfc(dr.VIDEOS).get(dr.nfc(ROOT)) if real: out.append((ROOT, os.path.join(dr.VIDEOS, real))) if os.path.isdir(SITE_ROOT): out.append(("サイト改善/_企画", SITE_ROOT)) return out def meetings(): for label, root in roots(): for name, r in sorted(cr.listdir_nfc(root).items()): d = os.path.join(root, r) if os.path.isdir(d): yield f"{label}/{name}", d, cr.listdir_nfc(d) def is_running(d, f): """実行中.txt があり、その PID が生きているか、起こしてから20分以内なら「動いている」""" if RUNNING not in f: return False p = os.path.join(d, f[RUNNING]) try: txt = open(p, encoding="utf-8").read() age = time.time() - os.path.getmtime(p) except OSError: return False m = re.search(r"PID (\d+)", txt) if m: try: os.kill(int(m.group(1)), 0) return True except ProcessLookupError: return False except PermissionError: return True return age < 20 * 60 def spawn(label, d): """会議を別の作業として起こし、見張り係はすぐ見回りに戻る（会議が終わるまで待たない）""" dr.write(os.path.join(d, RUNNING), f"起動 {datetime.now(dr.JST).strftime('%Y-%m-%d %H:%M:%S')}（見張り係が起こしたところ）\n") os.makedirs(cr.CONF, exist_ok=True) logf = open(os.path.join(cr.CONF, "kikaku.log"), "a") subprocess.Popen([sys.executable, os.path.abspath(__file__), "--run", d, label], cwd=dr.HERE, stdin=subprocess.DEVNULL, stdout=logf, stderr=subprocess.STDOUT, start_new_session=True, env=os.environ.copy()) dr.log(f"▶ 企画会議 {label} を別の作業として起こしました") def sweep(dry): """印のついた会議を、同時に MAX_PARALLEL 本まで起こす。会議の終わりは待たない。 ほかの議論（台本の議論・テーマの議論）の見回りも続けられるよう、いつも False を返す""" ms = list(meetings()) running = [] for label, d, f in ms: if is_running(d, f): running.append(label) elif RUNNING in f and not dry: os.remove(os.path.join(d, f[RUNNING])) # PID が死んでいる・古い印は、止まった跡 f.pop(RUNNING, None) if running: dr.log(f"企画会議 動いている: {'・'.join(running)}") for label, d, f in ms: if len(running) >= MAX_PARALLEL: break if label in running or MARK not in f or STOP in f: continue if dry: dr.log(f"［試し］企画会議 {label} を起こします") else: spawn(label, d) running.append(label) return False if __name__ == "__main__": if len(sys.argv) >= 4 and sys.argv[1] == "--run": d, label = sys.argv[2], sys.argv[3] try: run(label, d, False) except Exception as e: dr.log(f"× 企画会議 {label}: {e}") p = os.path.join(d, RUNNING) if os.path.exists(p) and "PID" not in open(p, encoding="utf-8").read(): os.remove(p) sys.exit(1)