#!/usr/bin/env python3 """site_team.py — サイトの入口のページを、3社の AI の「制作チーム」で作る（合議はしない）. 合議（全員の批判を通る案を残す）にすると、残るのは価値のある案ではなく、誰も反論できない無難な案になる。 そこで、ひとつにまとめず、作者ごとの案を磨いて並べる。 流れ（1つの依頼につき）: 1. 3社（Claude・Gemini・codex）が、互いに見せずに企画書を1本ずつ書く（v1） 2. ほかの2社が評者として攻める。指摘は必ず「弱い所」と「こう直せばもっと強くなる」の組。 案を捨てる・題材を変えさせる権限は誰にもない。事実の誤りは、ログにある範囲でいちばん強い言い方に直して示す 3. 作者が書き直す。指摘を取り込むかは作者が決め、尖りは守る。2〜3を 回数.txt の回数（既定2）くり返す 4. 3本を並べたまま終える（並べた.md）。選ぶのはエイイチ 5. 話し合いで良くなったかを確かめる（比べる.md）。各作者の最初の版と最後の版を、 どちらがどちらか伏せて並べる。答えは _答え（見ないで）.json 評者にも作者の名前は伏せる（案A・案B・案C と呼ぶ）。どの社がどの案かは _対応（見ないで）.json 置き場所: PixelstoMotors/サイト改善/制作/{依頼の名前}/ 依頼.md Claude が書く。何を作るか・守ること・材料（数字・ログ一覧） 回数.txt 書き直しの回数（無ければ2） かける.txt 印。終わると かけた.txt に名前を変える 止める.txt 置くと、いまの段が終わったところで止まる（消せば続きから。できたファイルは作り直さない） 使い方（Mac のターミナルで）: python3 site_team.py --check 3社が答えるかを確かめるだけ python3 site_team.py --once 印のついた依頼を1つ回す python3 site_team.py --install 5分ごとに見回るように登録する（YouTube の会議とは別に回る） """ import argparse, json, os, random, re, subprocess, sys, tempfile, time from concurrent.futures import ThreadPoolExecutor from datetime import datetime HERE = os.path.dirname(os.path.abspath(__file__)) # PixelstoMotors/サイト改善/_仕組み P2M = os.path.dirname(os.path.dirname(HERE)) # PixelstoMotors BASE = os.path.join(P2M, "デジタルツイン学習R2R2R", "動画", "_制作ベース") sys.path.insert(0, BASE) import debate_runner as dr # Gemini・codex の呼び方とキーを使い回す cr = dr.cr JOBS = os.path.join(os.path.dirname(HERE), "制作") MARK, DONE, STOP, RUNNING = "かける.txt", "かけた.txt", "止める.txt", "実行中.txt" AUTHORS = ["Claude", "Gemini", "codex"] CLAUDE = os.path.expanduser("~/.npm-global/bin/claude") CLAUDE_TIMEOUT = 25 * 60 PATH = ":".join([os.path.expanduser("~/.npm-global/bin"), "/opt/homebrew/bin", "/usr/local/bin", "/usr/bin", "/bin", "/usr/sbin", "/sbin"]) LABEL = "jp.adaptdesign.pixelstomotors.site-team" PLIST = os.path.expanduser(f"~/Library/LaunchAgents/{LABEL}.plist") LOGFILE = os.path.join(cr.CONF, "site_team.log") GROUND = """\ # 前提（全員に共通） あなたは、Webサイト「Pixels to Motors」（pixelstomotors.com）の、人の入口になるページを作る制作チームの一員です。 チームは3つの AI で、最後に人間（エイイチ）が選びます。 目指すのは、届けたい相手が「これは保存しておきたい」と思う、価値のあるページです。誰にも怒られない無難なページではありません。 ## 守ること（エイイチが決めたこと） - 個々のログ（/record/…）の形式は変えない。入口のページから送るだけ - 題材を広げない。入口のページは、依頼の「ログ一覧」にある中身だけで作る。ログにない事実・数字を足さない - トップや SNS に相談の窓口を出さない。新しいキャッチコピー・肩書きを足さない - 中身より言い過ぎた見出し・煽り・他社を悪く言う演出はしない。ただし、ログにある事実の範囲でいちばん強く言うのはよい - 日本語で書く。英語の文を混ぜない。数字には根拠（依頼の実際の数字・ログ・推測）を添え、推測は推測と書く """ PLAN = """\ ## 企画書の形（3,000字以内。この見出しで書く） ### ページの名前と URL の案 ### 誰に（1人に絞る。その人がどこから来て、何に困っているか） ### この人がこのページで得るもの（1文） ### 最初の画面（見出しと、その下の3行。実際の文面で書く） ### 見せ場（このページにしかないもの。ログのどの事実か） ### 構成（4〜6節。節ごとに、何を書くかと、使うログ /record/… とその中の事実） ### 束ねるログ（/record/… の一覧） ### 今あるページとのつなぎ方（トップ・シリーズ・今ある LP） ### 公開後に見る数字（30件たまるまでは判断しない） ### ゴールにどう効くか（依頼にゴールの表があれば、その行ごとに。効く理由と、効かないかもしれない理由の両方） """ AUTHOR_V1 = """\ # あなたの役：作者 入口のページを1本、作者として企画してください。ほかの作者の案は見ていません。見る必要もありません。 - 依頼に「達成したいこと」と「ゴール」があれば、それに効くことを最優先にする - 自分がいちばん価値があると思う題材と切り口を、ログ一覧から選ぶ。ほかの人も選びそうな無難な切り口より、あなたにしか出せない切り口を選ぶ - 届けたい相手を1人に絞り、その人に深く刺さるように尖らせる - 迷ったら、平均的な案ではなく、強い案を選ぶ。言い過ぎは後で直せるが、弱さは直せない """ + PLAN CRITIC = """\ # あなたの役：評者（{me}） 作者が書いた「{target}」の企画書を、もっと強くするために攻めてください。ほかの案とは比べません。 - 「強い」とは、依頼の「達成したいこと」と「ゴール」に効くこと。ゴールに効かない所をいちばん重く見る - 指摘は必ず「弱い所」と「こう直せばもっと強くなる（具体の書き換え案・文面）」の組で書く。弱い所だけの指摘は書かない - 案を捨てる・題材や相手を変えるよう求めない。作者が選んだ題材と相手のまま、強くする - 事実の誤り（ログ一覧にないログ・ログにない中身・言い過ぎ）は、ログにある範囲でいちばん強い言い方に書き直して示す - 指摘には重さを付ける：【高】直さないと伝わらない・誤り ／【中】直すと強くなる ／【低】好み。全部で5〜8個まで - 最後に「守るべき尖り」として、この案のいちばん良い所を1つ書く（作者が書き直しで消さないように） """ REWRITE = """\ # あなたの役：作者（書き直し） あなたが書いた企画書（{target}）に、2人の評者から指摘が届きました。 - 取り込むかどうかは、作者のあなたが決める。評者の言いなりにならない。案の尖り（いちばん良い所）は守る - 事実の誤りの指摘だけは、必ず直す（ログにある範囲でいちばん強い言い方に） - 出す形：まず【採否】として、指摘ごとに「採る／一部採る／採らない」と理由を1行ずつ。次に【企画書】として、書き直した企画書の全文 """ + PLAN VALUE_CRITIC = """\ # あなたの役：評者（{me}） 作者が書いた「{target}」のページを、もっと価値のあるものにするために攻めてください。ほかの案とは比べません。 問うのは1つだけ：**読む人（ロボット導入を判断する経営者・任された PM）が、このページで初めて得るものは何か。** - ほかのサイトやカタログでも手に入ること、読む人の判断に効かないことは「削る」か「この現場の数字で強くする」かを、書き換えの文面つきで示す - いちばん刺さる項目と、いちばん弱い項目を1つずつ挙げ、弱い方の直し方を書く - 案を捨てさせない。作者の切り口のまま強くする。指摘は5〜8個まで - 事実の確かめは「事実の照合」の材料と照らして行う。合わない所は、ログにある範囲でいちばん強い言い方に直して示す """ REWRITE_PAGE = """\ # あなたの役：作者（書き直し） あなたが書いたページ（{target}）に、2人の評者から指摘が届きました。 - 取り込むかどうかは、作者のあなたが決める。いちばん刺さる所は守る - 事実の誤りの指摘だけは、必ず直す - 出す形：まず【採否】として、指摘ごとに「採る／一部採る／採らない」と理由を1行ずつ。次に【企画書】として、書き直したページの全文 """ AUTHOR_PAGE = """\ # あなたの役：作者 依頼の「ページの形」に沿って、ページそのものを書いてください。ほかの作者のページは見ていません。 - 依頼の後ろにある「持ち寄った材料」（ログからまとめた項目・外の文脈・事実の照合）を全部使う - 読む人が「入れる前に知りたかった」と思うことを、いちばん前に出す。無難にまとめない """ def now(): return datetime.now(dr.JST).strftime("%Y-%m-%d %H:%M") def claude(prompt, cwd=None, tools=None): env = dict(os.environ) for k in ("ANTHROPIC_API_KEY", "ANTHROPIC_AUTH_TOKEN"): env.pop(k, None) env.update({"ENABLE_CLAUDEAI_MCP_SERVERS": "false", "PATH": PATH, "LANG": "ja_JP.UTF-8"}) with tempfile.TemporaryDirectory() as tmp, tempfile.TemporaryFile("w+", encoding="utf-8") as fo, \ tempfile.TemporaryFile("w+", encoding="utf-8") as fe: cmd = [CLAUDE, "-p", prompt, "--output-format", "text", "--permission-mode", "default"] if tools: cmd += ["--allowedTools", ",".join(tools)] p = subprocess.Popen(cmd, cwd=cwd or tmp, stdout=fo, stderr=fe, stdin=subprocess.DEVNULL, env=env, start_new_session=True) try: p.wait(timeout=CLAUDE_TIMEOUT) except subprocess.TimeoutExpired: p.kill() raise RuntimeError("Claude が時間内に答えませんでした") fo.seek(0); fe.seek(0) out, err = fo.read().strip(), fe.read().strip() if p.returncode != 0 or not out: raise RuntimeError(f"Claude が失敗しました（終了コード {p.returncode}）: {err[-600:]}") return out class Caller: def __init__(self): self.key = dr.gemini_key() self.model, _ = dr.pick_model(self.key) def ask(self, who, system, user, wdir, search=False, files=False): for attempt in range(3): try: if who == "Gemini": ans, _ = dr.gemini(self.key, self.model, system, user, search=search) elif who == "codex": ans, _ = dr.codex(wdir, system + "\n\n" + user, web=False) else: tools = (["Read", "Glob", "Grep"] if files else []) + (["WebSearch", "WebFetch"] if search else []) ans = claude(system + "\n\n" + user, cwd=wdir if files else None, tools=tools or None) return ans.strip() except Exception as e: dr.log(f" {who} 失敗（{attempt + 1}回目）: {str(e)[:200]}") if attempt == 2: raise time.sleep(30) def rd(p): return open(p, encoding="utf-8").read() def wr(p, text): os.makedirs(os.path.dirname(p), exist_ok=True) dr.write(p, text) def plan_part(ans): m = re.search(r"【企画書】\s*(.*)", ans, re.S) return (m.group(1) if m else ans).strip() def progress(d, lines): wr(os.path.join(d, "進み具合.md"), f"# 進み具合（{os.path.basename(d)}）\n\n- 最後に書いた時刻 {now()}\n" + "".join(f"- {x}\n" for x in lines)) def run(d, dry=False): name = os.path.basename(d) req = rd(os.path.join(d, "依頼.md")) rounds = 2 if os.path.exists(os.path.join(d, "回数.txt")): try: rounds = max(1, min(4, int(rd(os.path.join(d, "回数.txt")).strip()))) except ValueError: pass mp = os.path.join(d, "_対応（見ないで）.json") if os.path.exists(mp): who_of = json.load(open(mp, encoding="utf-8")) else: labels = ["案A", "案B", "案C"] sh = AUTHORS[:] random.shuffle(sh) who_of = dict(zip(labels, sh)) if not dry: wr(mp, json.dumps(who_of, ensure_ascii=False, indent=1)) labels = sorted(who_of) pack_tasks = None if os.path.exists(os.path.join(d, "持ち寄り.json")): pack_tasks = json.load(open(os.path.join(d, "持ち寄り.json"), encoding="utf-8")) if dry: dr.log(f"［試し］{name}: 書き直し {rounds} 回・{who_of}") return call = Caller() wr(os.path.join(d, RUNNING), f"開始 {now()}\nPID {os.getpid()}\n") done = [] pool = ThreadPoolExecutor(max_workers=3) def stopped(): if os.path.exists(os.path.join(d, STOP)): progress(d, done + ["止める.txt があるので止まっています（消せば続きから）"]) return True return False def step(jobs): """jobs: [(出力パス, 社, system, user[, {search, files}])]。まだ無いものだけを同時に""" todo = [j for j in jobs if not os.path.exists(j[0])] futs = [(j[0], pool.submit(call.ask, j[1], GROUND + "\n" + j[2], j[3], d, **(j[4] if len(j) > 4 else {}))) for j in todo] err = None for path, f in futs: # できた分は残す（続きから再開したときに作り直さない） try: wr(path, f.result() + "\n") except Exception as e: err = err or e if err: raise err try: dr.log(f"▶ 制作チーム {name}: 書き直し {rounds} 回") author_v1, critic, rewrite = AUTHOR_V1, CRITIC, REWRITE if pack_tasks: # 0. 持ち寄り：3社がそれぞれにしか取れない材料を作る（Claude と codex は 材料/ のログ全文を読む。Gemini は Web 検索） opt = {"Claude": {"files": True}, "codex": {}, "Gemini": {"search": True}} step([(os.path.join(d, "持ち寄り", f"{w}.md"), w, pack_tasks[w]["役"], req, opt[w]) for w in AUTHORS]) packs = "\n\n".join(f"## {pack_tasks[w]['名前']}\n\n" + rd(os.path.join(d, "持ち寄り", f"{w}.md")) for w in AUTHORS) req_raw, req = req, req + "\n\n# 持ち寄った材料\n\n" + packs author_v1, critic, rewrite = AUTHOR_PAGE, VALUE_CRITIC, REWRITE_PAGE done.append("0. 持ち寄りの材料（3社）ができました") # 比べる相手：Claude 1本だけで、同じ生の材料から同じページを書く（持ち寄りも話し合いもなし） step([(os.path.join(d, "1本", "v1.md"), "Claude", AUTHOR_PAGE + "\n- 材料/ のログ全文を自分で読み、必要なら Web 検索もしてよい。持ち寄った材料は無い", req_raw, {"files": True, "search": True})]) done.append("0. 比べる相手（Claude 1本の版）ができました") progress(d, done) # 1. 互いに見せずに書く step([(os.path.join(d, L, "v1.md"), who_of[L], author_v1, req) for L in labels]) done.append("1. 3本の企画書（v1）ができました") progress(d, done) for r in range(1, rounds + 1): if stopped(): return # 2. ほかの2社が攻める（名前は伏せる） jobs = [] for L in labels: cur = rd(os.path.join(d, L, f"v{r}.md")) for C in labels: if C == L: continue sysm = critic.format(me=f"評者（{C}の作者）", target=L) jobs.append((os.path.join(d, L, f"批評{r}_{C}の作者から.md"), who_of[C], sysm, f"{req}\n\n# 作者の企画書（{L}・第{r}版）\n\n{cur}")) step(jobs) done.append(f"2. {r}回目の批評ができました") progress(d, done) if stopped(): return # 3. 作者が書き直す jobs = [] for L in labels: cur = rd(os.path.join(d, L, f"v{r}.md")) crit = "\n\n".join(f"## 評者（{C}の作者）から\n\n" + rd(os.path.join(d, L, f"批評{r}_{C}の作者から.md")) for C in labels if C != L) jobs.append((os.path.join(d, L, f"書き直し{r}.md"), who_of[L], rewrite.format(target=L), f"{req}\n\n# あなたの企画書（{L}・第{r}版）\n\n{cur}\n\n# 届いた指摘\n\n{crit}")) step(jobs) for L in labels: v = os.path.join(d, L, f"v{r + 1}.md") if not os.path.exists(v): wr(v, plan_part(rd(os.path.join(d, L, f"書き直し{r}.md"))) + "\n") done.append(f"3. {r}回目の書き直し（v{r + 1}）ができました") progress(d, done) last = rounds + 1 # 4. 3本を並べる wr(os.path.join(d, "並べた.md"), f"# 並べた（{name}・最後の版）\n\nどの社がどの案かは伏せています（_対応（見ないで）.json）。選ぶのはエイイチ。\n\n" + "\n\n---\n\n".join(f"# {L}\n\n" + rd(os.path.join(d, L, f"v{last}.md")) for L in labels)) # 5. 最初と最後を伏せて比べる if pack_tasks: cands = [("1本（Claude だけ）", rd(os.path.join(d, "1本", "v1.md")))] + [(f"{L}（話し合いのあと）", rd(os.path.join(d, L, f"v{last}.md"))) for L in labels] random.shuffle(cands) marks = ["①", "②", "③", "④"] wr(os.path.join(d, "_答え（見ないで）.json"), json.dumps({m: c[0] for m, c in zip(marks, cands)}, ensure_ascii=False, indent=1)) wr(os.path.join(d, "比べる.md"), "# 比べる（1本のAIより、話し合いの方が良いか）\n\n4本のうち1本は Claude だけで書いた版、3本は3社の持ち寄りと話し合いのあとの版です。どれがどれかは伏せています。" "いちばん良いものと、いちばん悪いものを選んでから、_答え（見ないで）.json を開いてください。\n\n" + "\n\n---\n\n".join(f"# {m}\n\n" + c[1] for m, c in zip(marks, cands))) ans, body = None, None else: ans, body = {}, [] for L in (labels if ans is not None else []): pair = [("v1", rd(os.path.join(d, L, "v1.md"))), (f"v{last}", rd(os.path.join(d, L, f"v{last}.md")))] random.shuffle(pair) ans[L] = {"甲": pair[0][0], "乙": pair[1][0]} body.append(f"# {L}\n\n## 甲\n\n{pair[0][1]}\n\n## 乙\n\n{pair[1][1]}") if ans is not None: wr(os.path.join(d, "_答え（見ないで）.json"), json.dumps(ans, ensure_ascii=False, indent=1)) wr(os.path.join(d, "比べる.md"), "# 比べる（話し合いで良くなったか）\n\n各案の最初の版と最後の版を、どちらがどちらか伏せて並べています。" "案ごとに「甲と乙のどちらが良いか」を決めてから、_答え（見ないで）.json を開いてください。\n\n" + "\n\n---\n\n".join(body)) os.rename(os.path.join(d, MARK), os.path.join(d, DONE)) done.append("4. 並べた.md と 比べる.md ができました。終わりです") progress(d, done) dr.log(f"✓ 制作チーム {name}: 終わりました") dr.notify(f"サイトの制作チームが終わりました：{name}") except Exception as e: wr(os.path.join(d, "エラー.txt"), f"{now()}\n\n{e}\n") progress(d, done + [f"エラーで止まりました（エラー.txt）。次の見回りで続きから試します"]) dr.log(f"× 制作チーム {name}: {e}") dr.notify(f"サイトの制作チームが止まりました：{name}") finally: pool.shutdown(wait=False) p = os.path.join(d, RUNNING) if os.path.exists(p): os.remove(p) def alive(d): p = os.path.join(d, RUNNING) if not os.path.exists(p): return False m = re.search(r"PID (\d+)", rd(p)) if m: try: os.kill(int(m.group(1)), 0) return True except ProcessLookupError: os.remove(p) return False except PermissionError: return True return False def once(dry): if not os.path.isdir(JOBS): dr.log("制作/ がありません") return for name in sorted(os.listdir(JOBS)): d = os.path.join(JOBS, name) if not os.path.isdir(d): continue f = set(os.listdir(d)) if alive(d): dr.log(f"制作チーム {name} が動いています") return if MARK in f and STOP not in f and "依頼.md" in f: run(d, dry) return dr.log("印（制作/{名前}/かける.txt）のついたものはありません") def check(): call = Caller() dr.log(f"Gemini のモデル: {call.model}") for who in AUTHORS: t0 = time.time() a = call.ask(who, "短く答えてください。", "「準備できました」とだけ答えてください。", HERE) dr.log(f"{who}: {a[:40]}（{int(time.time() - t0)}秒）") def install(): 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(4, 60, 5)) plist = f""" Label{LABEL} ProgramArguments{sys.executable}{os.path.abspath(__file__)}--once StartCalendarInterval{cal} RunAtLoad AbandonProcessGroup WorkingDirectory{HERE} EnvironmentVariables PATH{PATH} CODEX_HOME{cr.CODEX_HOME} LANGja_JP.UTF-8 StandardOutPath{LOGFILE} StandardErrorPath{LOGFILE} """ open(PLIST, "w").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}") if __name__ == "__main__": ap = argparse.ArgumentParser(description="サイトの入口のページを、3社の AI の制作チームで作る") ap.add_argument("--check", action="store_true") ap.add_argument("--once", action="store_true") ap.add_argument("--dry-run", action="store_true") ap.add_argument("--install", action="store_true") a = ap.parse_args() try: if a.install: install() elif a.check: check() else: once(a.dry_run) except Exception as e: dr.log("×", e) sys.exit(1)