#!/usr/bin/env python3 """seisaku.py — 別々の AI が話し合って、価値のある動画の企画を作る「制作チーム」. 合議（みんなで1つの結論に寄せる）はしない。批判は案を消すためではなく、強くするために使う。 流れ（動画/_制作/{名前}/）: 1. 3社が別々に企画書を1本ずつ作る（互いに見ない） Claude … 案_Claude_1.md（チャットの中で Claude が書く） Gemini … 案_Gemini_1.md（このスクリプト。Google 検索つき） codex … 案_codex_1.md （このスクリプト。ウェブ検索つき） 2. 攻めと直し：各社が、ほかの2本を「誰が書いたか伏せて」読み、直し方つきで攻める 批評_{回}_{批評した人}.md。「弱い」だけの指摘は禁止。言い過ぎは、事実の範囲で一番強い言い方に直す。案を捨てる権限は誰にもない 3. 書き直し：作者が、自分の案への批評を読み、取り込むかを自分で決めて書き直す 案_{作者}_{回+1}.md 2と3を 回数.txt の回数（無ければ2回）くり返す 4. 並べる.md … 最後の3本を、作者を伏せて並べる（YouTube で調べた数字を横に添える） 比べる.md … 作者ごとに「最初の版」と「最後の版」を、どちらがどちらか伏せて並べる（話し合いで良くなったかをエイイチが判定する） 答え_見ないで.json … 伏せた名前の対応 Claude の番になると、ここで止まって 進み具合.md に「Claude の番」と書き、Mac に通知を出す。 Claude がファイルを置けば、次の見回りで続きから進む。 """ import json, os, random, re, time from datetime import datetime import debate_runner as dr cr = dr.cr ROOT = "_制作" MARK, DONE, STOP, RUNNING = "かける.txt", "かけた.txt", "止める.txt", "実行中.txt" AUTHORS = ["Claude", "Gemini", "codex"] AUTO = {"Gemini", "codex"} FORMAT = """\ # 企画書の形（この見出しで、この順に書く） ## 題名 ## サムネイルの文字（大きく2〜3行） ## 冒頭15秒のセリフ（ハコブくんが話す） ## 見せ場（見た人が新しく知ること・誰かに話したくなること） ## ロボットに詳しい人の目（この動画にしかない見方） ## 構成（5〜7節。1節1〜2行） ## 事実の主張と出典（動画で言う主張を1行ずつ。出典の URL・発表日・発表元。確かめられないものは「未確認」と書く） ## 撮るもの・作るもの（撮影は2〜3時間以内） ## YouTubeで調べる言葉（入口になる言葉を2つ。1行に1つ） """ GROUND = """\ # 制作チームの決まり（全員に共通） あなたは YouTube チャンネル「フィジカルAI解説｜Pixels to Motors」の動画の企画を作る制作チームの一員です。 チームの目的は、全員が反対しない案を作ることではなく、見た人にとって価値があり、たくさんの人が押して最後まで見る企画を作ることです。 - 尖った所・話題性のある所は、チームの財産です。消さずに、事実の範囲で一番強く伝わる形にする - 事実と違うこと・確かめられないことを、確かなこととして言わない。言い過ぎは、事実に合わせて言い方を直す（題材は捨てない） - 時事のニュースは、出典の URL と発表日を必ず添える。検索で見つけた要約ではなく、元の記事・発表で確かめる - 釣り（中身と合わない題名やサムネイル）・煽り・人や会社を悪く言う演出はしない - 日本語で書く。英語の文を混ぜない。自分の名前や役の名前を名乗らない """ DRAFT = """\ # あなたの仕事：企画書を1本作る 依頼を読んで、あなたが一番価値があると思う動画の企画を1本だけ作ってください。題材は自由です。 ほかのメンバーも別々に企画を作っています。似た案にしないために、あなたにしか出せない切り口を選んでください。 無難な案は求めていません。話題性があって、事実で裏付けられる、尖った案を作ってください。 """ + FORMAT CRITIQUE = """\ # あなたの仕事：ほかのメンバーの企画書を、強くするために攻める 下に、ほかのメンバーが作った企画書が2本あります（誰が書いたかは伏せてあります）。1本ずつ、次の形で書いてください。 ## 案A ### この案のいちばん尖っている所（守る・もっと強めるべき所） ### 弱い所と、その直し方（弱い所1つにつき、必ず具体的な直し方を書く。「弱い」だけの指摘は書かない） ### 事実の直し（間違い・言い過ぎ・出典がない所を、事実の範囲で一番強い言い方に書き直す。出典の URL と日付を添える。確かめられなければ「未確認」） ### もっと見られるための一手（題名・サムネイル・冒頭・見せ場のどれか1つを、具体的な文で） ## 案B （同じ形） 決まり：案を「やめるべき」「別の題材にすべき」と言わない。題材を活かしたまま強くする方法だけを書く。 """ REWRITE = """\ # あなたの仕事：自分の企画書を書き直す 下に、あなたが書いた企画書と、それへのほかのメンバーからの批評があります。 この企画の作者はあなたです。批評を取り込むかどうかは、あなたが決めてください。 - 取り込むと良くなる直しは取り込む。事実の直しは必ず取り込む - 案が平凡になる直し・尖りを消す直しは、理由を書いて断ってよい - 題材は変えない（磨く） はじめに「## 受け取った直し」として、直しごとに「取り込んだ／断った」と理由を1行ずつ書き、 そのあとに書き直した企画書の全文を、下の形で書いてください。 """ + FORMAT def log(*a): dr.log(*a) def path(d, name): return os.path.join(d, name) def exists(d, name): return os.path.isfile(path(d, name)) def read(d, name): return dr.read(path(d, name)) def strip_head(text): """書き出しのヘッダ（日時など）を除いた本文""" return text.split("\n---\n", 1)[1].strip() if "\n---\n" in text[:600] else text.strip() def write_doc(d, name, title, meta, body): dr.write(path(d, name), f"# {title}\n\n- 日時: {datetime.now(dr.JST).strftime('%Y-%m-%d %H:%M')}（JST）\n- {meta}\n\n---\n\n{body}") def load(d): p = path(d, "状態.json") if os.path.isfile(p): return json.load(open(p, encoding="utf-8")) rnd = random.Random() labels = {} for c in AUTHORS: # 批評する人ごとに、ほかの2本を 案A・案B に（順は毎回ばらばら） others = [a for a in AUTHORS if a != c] rnd.shuffle(others) labels[c] = {"案A": others[0], "案B": others[1]} return {"labels": labels, "cost_in": 0, "cost_out": 0, "codex": 0, "start": datetime.now(dr.JST).isoformat(timespec="minutes")} def save(d, st): tmp = path(d, "状態.json.tmp") with open(tmp, "w", encoding="utf-8") as f: json.dump(st, f, ensure_ascii=False, indent=1) os.replace(tmp, path(d, "状態.json")) def progress(d, st, rounds, now, note=""): usd = st["cost_in"] / 1e6 * 2.0 + st["cost_out"] / 1e6 * 12.0 rows = [] for a in AUTHORS: vs = [v for v in range(1, rounds + 2) if exists(d, f"案_{a}_{v}.md")] cs = [k for k in range(1, rounds + 1) if exists(d, f"批評_{k}_{a}.md")] rows.append(f"| {a} | {('第' + str(max(vs)) + '版') if vs else 'まだ'} | {('・'.join(f'{k}回目' for k in cs)) or 'まだ'} |") L = [f"# 進み具合（制作チーム：{os.path.basename(d)}）", "", f"- いま：**{now}**", f"- 往復の回数：{rounds} 回（攻めと書き直しを1往復と数える）", f"- 始めた時刻 {st['start'].replace('T', ' ')}・最後に書いた時刻 {datetime.now(dr.JST).strftime('%Y-%m-%d %H:%M')}", f"- Gemini の費用の目安 約{usd * 155:,.0f}円・codex {st['codex']} 回（ChatGPT Plus の枠）"] if note: L.append(f"- {note}") L += ["", "| 作者 | 企画書 | 攻めた回 |", "|---|---|---|"] + rows + [ "", "ファイル：案_{作者}_{版}.md（企画書）・批評_{回}_{批評した人}.md（攻め）。最後に 並べる.md と 比べる.md ができる"] dr.write(path(d, "進み具合.md"), "\n".join(L)) def run_gemini(st, key, model, task, body): ans, meta = dr.gemini(key, model, GROUND + "\n" + task, body, search=True) st["cost_in"] += dr.LAST_USAGE["inp"] st["cost_out"] += dr.LAST_USAGE["out"] return ans, meta def run_codex(st, d, task, body): ans, meta = dr.codex(d, GROUND + "\n" + task + "\n\n" + body, web=True) st["codex"] += 1 return ans, meta def critique_input(d, st, critic, version): lab = st["labels"][critic] parts = [] for L in ("案A", "案B"): parts.append(f"==================== {L} ====================\n\n" + strip_head(read(d, f"案_{lab[L]}_{version}.md"))) return "# 依頼\n\n" + read(d, "依頼.md") + "\n\n# ほかのメンバーの企画書\n\n" + "\n\n".join(parts) def section_for(text, label): """批評の中から、案A・案B のどちらか1本ぶんを抜き出す""" text = strip_head(text) m = re.search(rf"^#+\s*{label}\s*$(.*?)(?=^#+\s*案[AB]\s*$|\Z)", text, re.M | re.S) return m.group(1).strip() if m else "" def rewrite_input(d, st, author, k): mine = strip_head(read(d, f"案_{author}_{k}.md")) notes = [] for critic in AUTHORS: if critic == author or not exists(d, f"批評_{k}_{critic}.md"): continue lab = {v: L for L, v in st["labels"][critic].items()}[author] sec = section_for(read(d, f"批評_{k}_{critic}.md"), lab) if sec: notes.append(f"==================== メンバーからの批評 ====================\n\n{sec}") return ("# 依頼\n\n" + read(d, "依頼.md") + "\n\n# あなたの企画書（いまの版）\n\n" + mine + "\n\n# あなたの企画書への批評\n\n" + ("\n\n".join(notes) or "（批評が見つかりませんでした）")) def claude_turn(d, st, rounds, what, name): progress(d, st, rounds, f"Claude の番：{what}（{name} を待っています）") flag = path(d, "Claudeの番.txt") if not os.path.exists(flag): dr.write(flag, f"{datetime.now(dr.JST).strftime('%Y-%m-%d %H:%M')}\n{what}\n{name} を置けば続きます\n") dr.notify(f"制作チーム {os.path.basename(d)}：Claude の番（{what}）") log(f" Claude の番：{what}（{name}）") def finish(d, st, rounds): rnd = random.Random() key = {"並べる": {}, "比べる": {}} order = AUTHORS[:] rnd.shuffle(order) L = [f"# 並べる：最後の3本（{os.path.basename(d)}・作者は伏せてあります）", ""] words = [] for i, a in enumerate(order, 1): key["並べる"][f"案{i}"] = a body = strip_head(read(d, f"案_{a}_{rounds + 1}.md")) body = re.sub(r"^##\s*受け取った直し.*?(?=^##\s*題名)", "", body, flags=re.M | re.S).strip() L += [f"## 案{i}", "", body, ""] m = re.search(r"^##\s*YouTubeで調べる言葉[^\n]*$(.*?)(?=^##|\Z)", body, re.M | re.S) if m: words += [w.strip(" -・*") for w in m.group(1).strip().splitlines() if w.strip(" -・*")][:2] # YouTube で入口の言葉を調べる（小さなチャンネルの実績を横に添える） if words: sub = path(d, "調べ") os.makedirs(sub, exist_ok=True) with open(path(sub, "キーワード.txt"), "w", encoding="utf-8") as f: f.write("\n".join(dict.fromkeys(words)) + "\n") try: import youtube_market youtube_market.run(sub) L += ["---", "", "## YouTube で入口の言葉を調べた数字", "", "（調べ/結果.md の要点。小さなチャンネル＝登録者1万人未満の本編を見る）", ""] res = open(path(sub, "結果.md"), encoding="utf-8").read() L += [x for x in res.splitlines() if x.startswith("## ") or x.startswith("- 本編") or x.startswith("- ")][:80] except Exception as e: L += ["", f"（YouTube の調べはできませんでした: {e}）"] dr.write(path(d, "並べる.md"), "\n".join(L)) C = [f"# 比べる：話し合いで良くなったか（{os.path.basename(d)}）", "", "作者ごとに、最初の版と最後の版を、どちらがどちらか伏せて並べました。", "それぞれ「X と Y のどちらが良いか」と一言の理由を書いてください。答えは 答え_見ないで.json にあります。", ""] for i, a in enumerate(order, 1): pair = [("最初", strip_head(read(d, f"案_{a}_1.md"))), ("最後", strip_head(read(d, f"案_{a}_{rounds + 1}.md")))] rnd.shuffle(pair) key["比べる"][f"組{i}"] = {"X": pair[0][0], "Y": pair[1][0], "作者": a} for tag, (_, body) in zip(("X", "Y"), pair): body = re.sub(r"^##\s*受け取った直し.*?(?=^##\s*題名)", "", body, flags=re.M | re.S).strip() C += [f"## 組{i}・{tag}", "", body, ""] C += [f"**組{i}の判定**：（X か Y）　理由：", "", "---", ""] dr.write(path(d, "比べる.md"), "\n".join(C)) with open(path(d, "答え_見ないで.json"), "w", encoding="utf-8") as f: json.dump(key, f, ensure_ascii=False, indent=1) def run(name, d, dry): files = cr.listdir_nfc(d) if "依頼.md" not in files: log(f"× {name}: 依頼.md がありません") return rounds = 2 if "回数.txt" in files: try: rounds = max(1, int(read(d, "回数.txt").strip())) except ValueError: pass st = load(d) save(d, st) if dry: log(f"［試し］制作チーム {name}") return key = dr.gemini_key() model, _ = dr.pick_model(key) dr.write(path(d, RUNNING), f"開始 {datetime.now(dr.JST).strftime('%Y-%m-%d %H:%M:%S')}\nPID {os.getpid()}\n") flag = path(d, "Claudeの番.txt") if os.path.exists(flag): # 待っていた Claude のファイルが置かれたら、印を外して進む want = open(flag, encoding="utf-8").read().splitlines()[-1].split(" ")[0] if exists(d, want): os.remove(flag) else: os.remove(path(d, RUNNING)) return log(f" まだ Claude の番です（{want}）") try: req = "# 依頼\n\n" + read(d, "依頼.md") posts = {} # 持ち場.json … 作者ごとに、企画のもとにする材料を分ける（同じ材料だと題材が似るため） if exists(d, "持ち場.json"): posts = json.load(open(path(d, "持ち場.json"), encoding="utf-8")) # 1. 別々に企画書 for a in ("Gemini", "codex"): if not exists(d, f"案_{a}_1.md"): progress(d, st, rounds, f"{a} が企画書の第1版を作っています") log(f" {a} 第1版") mine = req + (f"\n\n# あなたの持ち場（この材料から企画を作る。ほかのメンバーは別の持ち場から作っている）\n\n{posts[a]}" if a in posts else "") ans, meta = run_gemini(st, key, model, DRAFT, mine) if a == "Gemini" else run_codex(st, d, DRAFT, mine) write_doc(d, f"案_{a}_1.md", f"企画書 第1版（{a}）", meta, ans) save(d, st) if not exists(d, "案_Claude_1.md"): return claude_turn(d, st, rounds, "企画書の第1版", "案_Claude_1.md") # 2・3. 攻めと書き直しを往復 for k in range(1, rounds + 1): for c in ("Gemini", "codex"): if not exists(d, f"批評_{k}_{c}.md"): progress(d, st, rounds, f"{k}往復目：{c} がほかの2本を攻めています") log(f" {k}往復目 {c} の批評") body = critique_input(d, st, c, k) ans, meta = run_gemini(st, key, model, CRITIQUE, body) if c == "Gemini" else run_codex(st, d, CRITIQUE, body) write_doc(d, f"批評_{k}_{c}.md", f"{k}往復目の批評（{c}）", meta, ans) save(d, st) if not exists(d, f"批評_{k}_Claude.md"): dr.write(path(d, f"Claudeへ_批評_{k}.md"), CRITIQUE + "\n\n" + critique_input(d, st, "Claude", k)) return claude_turn(d, st, rounds, f"{k}往復目の批評（Claudeへ_批評_{k}.md を読む）", f"批評_{k}_Claude.md") for a in ("Gemini", "codex"): if not exists(d, f"案_{a}_{k + 1}.md"): progress(d, st, rounds, f"{k}往復目：{a} が自分の企画書を書き直しています") log(f" {k}往復目 {a} の書き直し") body = rewrite_input(d, st, a, k) ans, meta = run_gemini(st, key, model, REWRITE, body) if a == "Gemini" else run_codex(st, d, REWRITE, body) write_doc(d, f"案_{a}_{k + 1}.md", f"企画書 第{k + 1}版（{a}）", meta, ans) save(d, st) if not exists(d, f"案_Claude_{k + 1}.md"): dr.write(path(d, f"Claudeへ_書き直し_{k}.md"), REWRITE + "\n\n" + rewrite_input(d, st, "Claude", k)) return claude_turn(d, st, rounds, f"{k}往復目の書き直し（Claudeへ_書き直し_{k}.md を読む）", f"案_Claude_{k + 1}.md") # 4. 並べる・比べる progress(d, st, rounds, "最後の3本を並べて、YouTube の数字を調べています") finish(d, st, rounds) if os.path.exists(path(d, MARK)): os.rename(path(d, MARK), path(d, DONE)) progress(d, st, rounds, "終わり。並べる.md（最後の3本）と 比べる.md（良くなったかの判定）を見てください") dr.notify(f"制作チームが終わりました：{os.path.basename(d)}") log(f"✓ 制作チーム {name} が終わりました") except Exception as e: dr.write(path(d, "エラー.txt"), f"{datetime.now(dr.JST).strftime('%Y-%m-%d %H:%M:%S')}\n\n{e}\n") progress(d, st, rounds, "エラーで止まりました（エラー.txt）。次の見回りで続きから試します") log(f"× 制作チーム {name}: {e}") finally: save(d, st) if os.path.exists(path(d, RUNNING)): os.remove(path(d, RUNNING)) def sweep(dry): real = cr.listdir_nfc(dr.VIDEOS).get(dr.nfc(ROOT)) if not real: return False root = os.path.join(dr.VIDEOS, real) for name, r in sorted(cr.listdir_nfc(root).items()): d = os.path.join(root, r) if not os.path.isdir(d): continue f = cr.listdir_nfc(d) if RUNNING in f: if time.time() - os.path.getmtime(path(d, f[RUNNING])) < 3 * 3600: log(f"制作チーム {name} が動いています") return True os.remove(path(d, f[RUNNING])) if MARK in f and STOP not in f: # Claude の番で待っているときは、Claude のファイルが置かれるまで何もしない（Gemini を無駄に呼ばない） run(f"{ROOT}/{name}", d, dry) return True return False