スキャン即時実行 ＆ コマンド安定版

oproxy.py

@@ -1,128 +1,238 @@
 # バージョン情報: Python 3.12+ / FastAPI 0.115.0 / uvicorn 0.30.0 / httpx 0.28.0
-import httpx, asyncio, json, sys, threading, os, argparse, re, unicodedata
+import argparse
+import asyncio
+import json
+import os
+import re
+import sys
+import threading
+import unicodedata
 from datetime import datetime
+
+import httpx
 import uvicorn
 from fastapi import FastAPI, Request
 from starlette.responses import StreamingResponse
 
 app = FastAPI()
 
-C_GRAY, C_CYAN, C_GREEN, C_YELLOW, C_RED, C_RESET = "\033[90m", "\033[96m", "\033[92m", "\033[93m", "\033[91m", "\033[0m"
+# --- カラー・設定 ---
-DEFAULT_REMOTE_PORT, DEFAULT_LOCAL_PORT = 11430, 11434
+C_GRAY, C_CYAN, C_GREEN, C_YELLOW, C_RED, C_RESET = (
+    "\033[90m",
+    "\033[96m",
+    "\033[92m",
+    "\033[93m",
+    "\033[91m",
+    "\033[0m",
+)
+B_GREEN, B_YELLOW, B_RED = "\033[42;30m", "\033[43;30m", "\033[41;37m"
+
 MEM_LIMIT = 16.8
-CONFIG = {"url": f"http://127.0.0.1:{DEFAULT_REMOTE_PORT}", "timeout": httpx.Timeout(None)}
+NAME_MAX_WIDTH = 50
+CONFIG = {"url": "http://127.0.0.1:11430"}
+
 
 def get_ts():
     return f"{C_GRAY}[{datetime.now().strftime('%H:%M:%S.%f')[:-3]}]{C_RESET}"
 
-def clean_text(text):
-    """特殊な空白文字（\xa0等）を通常の空白に置換し、前後のゴミを取る"""
-    return "".join(c for c in text if unicodedata.category(c) != 'Cf').replace('\xa0', ' ').strip()
 
 def get_visual_width(text):
-    """文字の表示幅を計算 (全角2, 半角1)"""
+    return sum(
-    width = 0
+        2 if unicodedata.east_asian_width(c) in ("W", "F", "A") else 1 for c in text
-    for c in text:
+    )
-        if unicodedata.east_asian_width(c) in ('W', 'F', 'A'):
+
-            width += 2
-        else:
-            width += 1
-    return width
 
 def pad_right(text, width):
-    """表示幅に基づいて正確に右パディングする"""
+    plain_text = re.sub(r"\033\[[0-9;]*m", "", text)
-    t = clean_text(text)
+    return text + " " * max(0, width - get_visual_width(plain_text))
-    curr_w = get_visual_width(t)
+
-    return t + ' ' * max(0, width - curr_w)
+
+def draw_progress(current, total, model_name=""):
+    width = 30
+    filled = int(width * current / total)
+    bar = "█" * filled + "░" * (width - filled)
+    percent = (current / total) * 100
+    sys.stdout.write(
+        f"\r{get_ts()} {C_CYAN}[Scanning] |{bar}| {percent:>3.0f}% {C_GRAY}({model_name[:20]}...){C_RESET}"
+    )
+    sys.stdout.flush()
 
-# --- 以下、ロジック部分は維持しつつ表示部のみ微調整 ---
 
 async def check_tool_support(client, model_name):
     try:
         res = await client.post(f"{CONFIG['url']}/api/show", json={"name": model_name})
         if res.status_code == 200:
             info = res.json()
-            # template, system, modelfile 等から tool 関連のキーワードを執念深く探す
+            content = " ".join(
-            look_in = [info.get("template", ""), info.get("system", ""), info.get("modelfile", "")]
+                [
-            content = " ".join(look_in).lower()
+                    info.get("template", ""),
+                    info.get("system", ""),
+                    info.get("modelfile", ""),
+                ]
+            ).lower()
             return any(x in content for x in ["tool", "function", "call", "assistant"])
-    except: pass
+    except:
+        pass
     return False
 
-async def check_connection():
+
+def run_analyze():
+    asyncio.run(analyze_models())
+
+
+async def analyze_models():
     url = CONFIG["url"]
-    print(f"{get_ts()} {C_YELLOW}[Check] {url} 分析中...{C_RESET}")
+    print(f"\n{get_ts()} {C_YELLOW}[Analyze] {url} 接続開始...{C_RESET}")
     try:
         async with httpx.AsyncClient(timeout=10.0) as client:
             res = await client.get(f"{url}/api/tags")
-            if res.status_code == 200:
+            if res.status_code != 200:
-                models = res.json().get('models', [])
+                print(f"{get_ts()} {C_RED}分析エラー: HTTP {res.status_code}{C_RESET}")
-                header_text = f"--- リモートモデル戦力分析 (基準: {MEM_LIMIT}GiB + Tool対応) ---"
+                return
-                print(f"{get_ts()} {C_GREEN}{header_text}{C_RESET}")
 
-                for m in models:
+            models_data = res.json().get("models", [])
-                    name = m['name']
+            total = len(models_data)
-                    size_gb = m['size'] / (1024**3)
+            enriched = []
-                    has_tool = await check_tool_support(client, name)
 
-                    if size_gb > MEM_LIMIT: color, status = C_RED, "❌ MEM "
+            for i, m in enumerate(models_data, 1):
-                    elif not has_tool: color, status = C_YELLOW, "⚠️ TOOL"
+                full_name = m["name"]
-                    else: color, status = C_GREEN, "✅ READY"
+                draw_progress(i, total, full_name.split("/")[-1])
+                size_gb = m["size"] / (1024**3)
+                has_tool = await check_tool_support(client, full_name)
+                score = (
+                    0
+                    if size_gb <= MEM_LIMIT and has_tool
+                    else (1 if size_gb <= MEM_LIMIT else 2)
+                )
+                enriched.append(
+                    {
+                        "full_name": full_name,
+                        "display_name": full_name.split("/")[-1],
+                        "size_gb": size_gb,
+                        "has_tool": has_tool,
+                        "score": score,
+                    }
+                )
 
-                    tool_mark = f"{C_CYAN}[TOOL]{C_RESET}" if has_tool else f"{C_GRAY}[----]{C_RESET}"
+            print("\n")
+            enriched.sort(key=lambda x: (x["score"], x["display_name"], -x["size_gb"]))
 
-                    # カラム幅を微調整 (名前: 70)
+            print(
-                    display_name = pad_right(name, 70)
+                f"{get_ts()} {C_GREEN}--- リモートモデル戦力分析 (Target: {url}) ---{C_RESET}"
-                    print(f"{get_ts()} {color}{status:<8}{C_RESET} {tool_mark} {display_name} {size_gb:>5.1f} GiB")
+            )
+            prefix_width = 32
+            for em in enriched:
+                status = (
+                    f"{B_GREEN} READY{C_RESET}"
+                    if em["score"] == 0
+                    else (
+                        f"{B_YELLOW} TOOL {C_RESET}"
+                        if em["score"] == 1
+                        else f"{B_RED} MEM  {C_RESET}"
+                    )
+                )
+                tool = (
+                    f"{C_CYAN}[TOOL]{C_RESET}"
+                    if em["has_tool"]
+                    else f"{C_GRAY}[----]{C_RESET}"
+                )
+                name, size = em["display_name"], f"{em['size_gb']:>5.1f} GiB"
 
-                print(f"{get_ts()} {C_GREEN}{'-' * 100}{C_RESET}\n")
+                if get_visual_width(name) > NAME_MAX_WIDTH:
-                return True
+                    print(f"{get_ts()} {status} {tool} {name[:NAME_MAX_WIDTH]} {size}")
+                    print(
+                        f"{get_ts()} {' ' * (prefix_width - 15)} {C_GRAY}└ {name[NAME_MAX_WIDTH:]}{C_RESET}"
+                    )
+                else:
+                    print(
+                        f"{get_ts()} {status} {tool} {pad_right(name, NAME_MAX_WIDTH)} {size}"
+                    )
+
+            print(f"{get_ts()} {C_GREEN}{'-' * 80}{C_RESET}")
+            show_help()
     except Exception as e:
-        print(f"{get_ts()} {C_RED}!! 接続失敗 !!: {e}{C_RESET}")
+        print(f"\n{get_ts()} {C_RED}分析失敗: {e}{C_RESET}")
-    return False
+
+
+def show_help():
+    print(f"\n{C_CYAN}[Command Help]{C_RESET}")
+    print(f"  {C_YELLOW}:p [port]{C_RESET}  - 転送先(Ollama)のポートを切り替えて再分析")
+    print(f"  {C_YELLOW}?{C_RESET}          - このヘルプを表示")
+    print(f"  {C_YELLOW}q{C_RESET}          - プロキシを終了")
+    print(f"{C_GRAY}------------------------------------------{C_RESET}\n")
 
-# --- API Route, main, wait_for_quit --- (省略せず統合してください)
 
 @app.api_route("/{path:path}", methods=["GET", "POST", "PUT", "DELETE"])
 async def sticky_proxy(path: str, request: Request):
-    print(f"\n{get_ts()} /{path}: ", end="", flush=True)
+    target_url = f"{CONFIG['url']}/{path}"
-    body = b""; async for chunk in request.stream():
+
-        body += chunk; print(f"{C_CYAN}^{C_RESET}", end="", flush=True)
+    # --- 修正箇所: SyntaxError を防ぐため完全に複数行に展開 ---
-    print(f"{C_YELLOW}|{C_RESET}", end="", flush=True)
+    body = b""
+    async for chunk in request.stream():
+        body += chunk
+    # ------------------------------------------------------
+
+    headers = {
+        k: v
+        for k, v in request.headers.items()
+        if k.lower() not in ["host", "content-length"]
+    }
 
-    headers = {k: v for k, v in request.headers.items() if k.lower() not in ["host", "content-length"]}
     async def stream_response():
         try:
             async with httpx.AsyncClient(timeout=None) as client:
-                async with client.stream(request.method, f"{CONFIG['url']}/{path}", content=body, headers=headers) as response:
+                async with client.stream(
-                    if response.status_code != 200:
+                    request.method, target_url, content=body, headers=headers
-                        err_data = await response.aread(); err_msg = err_data.decode(errors='ignore')
+                ) as response:
-                        print(f" {C_RED}{err_msg}{C_RESET}")
-                        yield json.dumps({"message": {"role": "assistant", "content": f"### Error\n{err_msg}"}, "done": True}).encode()
-                        return
-                    print(f"{C_GREEN}v:{C_RESET}", end="", flush=True)
                     async for chunk in response.aiter_bytes():
-                        print(f"{C_GREEN}v{C_RESET}", end="", flush=True)
                         yield chunk
-                    print(f"{C_YELLOW}*{C_RESET}", end="", flush=True)
+        except Exception as e:
-        except Exception as e: print(f" {C_RED}[Err] {e}{C_RESET}")
+            print(f" {C_RED}[Err] {e}{C_RESET}")
 
     return StreamingResponse(stream_response())
 
-def wait_for_quit():
+
+def interactive_shell():
     while True:
-        line = sys.stdin.readline()
+        try:
-        if line and line.strip().lower() == 'q': os._exit(0)
+            line = sys.stdin.readline().strip().lower()
+            if not line:
+                continue
+
+            if line == "q":
+                os._exit(0)
+            elif line == "?":
+                show_help()
+            elif line.startswith(":p"):
+                parts = line.split()
+                if len(parts) > 1:
+                    new_port = parts[1]
+                    CONFIG["url"] = f"http://127.0.0.1:{new_port}"
+                    threading.Thread(target=run_analyze, daemon=True).start()
+                else:
+                    print(
+                        f"{C_RED}ポート番号を指定してください (例: :p 11435){C_RESET}"
+                    )
+            else:
+                print(
+                    f"{C_GRAY}未知のコマンドです: '{line}' ( ? でヘルプ表示 ){C_RESET}"
+                )
+        except EOFError:
+            break
+
 
 def main():
     parser = argparse.ArgumentParser()
-    parser.add_argument("-r", "--remote", type=int, default=DEFAULT_REMOTE_PORT)
+    parser.add_argument("-r", "--remote", type=int, default=11430)
-    parser.add_argument("-l", "--local", type=int, default=DEFAULT_LOCAL_PORT)
+    parser.add_argument("-l", "--local", type=int, default=11434)
     args = parser.parse_args()
     CONFIG["url"] = f"http://127.0.0.1:{args.remote}"
-    loop = asyncio.new_event_loop(); asyncio.set_event_loop(loop)
+
-    loop.run_until_complete(check_connection())
+    asyncio.run(analyze_models())
-    threading.Thread(target=wait_for_quit, daemon=True).start()
+
+    threading.Thread(target=interactive_shell, daemon=True).start()
     uvicorn.run(app, host="127.0.0.1", port=args.local, log_level="error")
 
-if __name__ == "__main__": main()
+
+if __name__ == "__main__":
+    main()