import asyncio import os import re import shlex from datetime import datetime from playwright.async_api import async_playwright # [2026-02-12] Version: 2.1.1-WGP-Final # [2025-07-04] Embed: Version info # [2025-06-21] Language: Japanese class WebGemiAgent: def __init__(self, context, page): self.context = context self.page = page self.input_selector = "div[role='textbox']" self.container_name = "sandbox_container" self.yolo_mode = False async def execute_in_sandbox(self, path, body, cmd): """WGPパケットの展開とコンテナ内実行""" # ヒアドキュメントを利用してファイルを生成し、コマンドを実行 write_script = f"cat << 'EOF' > {path}\n{body}\nEOF" full_script = f"{write_script} && {cmd}" docker_cmd = f"docker exec -e OLLAMA_HOST=host.docker.internal -w /workspace {self.container_name} bash -c {shlex.quote(full_script)}" proc = await asyncio.create_subprocess_shell( docker_cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE ) stdout, stderr = await proc.communicate() return stdout.decode('utf-8', errors='replace'), stderr.decode('utf-8', errors='replace') async def wait_for_wgp_packet(self): """パケットが安定するまで待機""" print("📡 WGP-Packet Receiving", end="", flush=True) last_text = "" stable_count = 0 for _ in range(60): responses = await self.page.query_selector_all(".model-response-text") if responses: current_text = await responses[-1].inner_text() # S式の括弧が閉じているか、前回の確認から変化がないかチェック if current_text == last_text and "(" in current_text and ")" in current_text: stable_count += 1 else: stable_count = 0 last_text = current_text if stable_count >= 3: break print(".", end="", flush=True) await asyncio.sleep(1) print(" [Locked]") return last_text def parse_wgp(self, text): """WGP/1.1 S-Expression Parser""" # 正規表現でS式パケットを抽出 pattern = r'\(WGP1\.1\s+:ID\s+"(?P.*?)"\s+:OP\s+"(?P.*?)"\s+:PATH\s+"(?P.*?)"\s+:CMD\s+"(?P.*?)"\s+:BODY\s+"(?P.*?)"\)' match = re.search(pattern, text, re.DOTALL) if match: return match.groupdict() return None async def run_loop(self): print(f"\n[{datetime.now()}] ⚡ WGP/1.1 インターフェース起動成功。") # ハンドシェイク用プロンプト WGP_HANDSHAKE = ( "これより通信規約 WGP/1.1 を適用する。一切の挨拶、説明を禁止する。\n" "応答は必ず以下のS式パケット1つのみを、Markdownのlispブロックで返せ。\n" "```lisp\n" "(WGP1.1 :ID \"seq-id\" :OP \"EXEC\" :PATH \"/path/to/file\" :CMD \"build && run command\" :BODY \"source code here\")\n" "```" ) await self.page.wait_for_selector(self.input_selector) await self.page.locator(self.input_selector).fill(WGP_HANDSHAKE) await self.page.keyboard.press("Enter") await asyncio.sleep(5) while True: mode_str = "☢️ YOLO" if self.yolo_mode else "🛡️ NORMAL" user_input = input(f"\n🧬 Master [{mode_str}] > ").strip() if not user_input: continue if user_input.lower() == "exit": break if user_input.lower() == "yolo on": self.yolo_mode = True print("☢️ YOLOモード：全自動通信開始。") continue if user_input.lower() == "yolo off": self.yolo_mode = False continue # 送信 target = self.page.locator(self.input_selector) await target.fill(user_input) await self.page.keyboard.press("Enter") # 応答ループ for iteration in range(10): raw_response = await self.wait_for_wgp_packet() packet = self.parse_wgp(raw_response) if packet: print(f"📦 Packet Received. ID: {packet['id']}") if self.yolo_mode or input("▶ Execute? (y/n): ") == "y": out, err = await self.execute_in_sandbox(packet['path'], packet['body'], packet['cmd']) print(f"--- RESULT ---\n{out}{err}\n--------------") # フィードバックパケットの生成 safe_out = out.replace('"', '\\"').replace('\n', '\\n') safe_err = err.replace('"', '\\"').replace('\n', '\\n') feedback = f"(WGP-RESULT :ID \"{packet['id']}\" :STDOUT \"{safe_out}\" :STDERR \"{safe_err}\")" await target.fill(feedback) await self.page.keyboard.press("Enter") if not self.yolo_mode: break else: break else: print(f"❌ Protocol Error or Prose Detected.") break async def main(): async with async_playwright() as p: user_data_dir = os.path.expanduser("~/dev/gchat/gemini_profile") context = await p.chromium.launch_persistent_context( user_data_dir, headless=False, args=["--disable-blink-features=AutomationControlled", "--no-sandbox"] ) page = context.pages[0] # 純粋な文字列としてのURL await page.goto("https://gemini.google.com/app") agent = WebGemiAgent(context, page) await agent.run_loop() if __name__ == "__main__": try: asyncio.run(main()) except KeyboardInterrupt: print("\n☢️ 実験終了。")