ChingTech OS：NAS 電路圖掛載與三階段非同步搜尋

# config.py
class Settings:
    # NAS 掛載路徑
    nas_mount_path: str = _get_env("NAS_MOUNT_PATH", "/mnt/nas")
    ctos_mount_path: str = _get_env("CTOS_MOUNT_PATH", "/mnt/nas/ctos")
    projects_mount_path: str = _get_env("PROJECTS_MOUNT_PATH", "/mnt/nas/projects")
    circuits_mount_path: str = _get_env("CIRCUITS_MOUNT_PATH", "/mnt/nas/circuits")

# path_manager.py
class PathManager:
    def __init__(self):
        # shared zone 子來源對應（多掛載點）
        self._shared_mounts = {
            "projects": settings.projects_mount_path,   # /mnt/nas/projects
            "circuits": settings.circuits_mount_path,    # /mnt/nas/circuits
        }

def _resolve_shared_path(self, relative_path: str) -> str:
    """解析 shared zone 子來源路徑"""
    first_segment = relative_path.split("/", 1)[0]
    if first_segment in self._shared_mounts:
        mount_path = self._shared_mounts[first_segment]
        rest = relative_path[len(first_segment):].lstrip("/")
        return f"{mount_path}/{rest}" if rest else mount_path
    # 向後相容：fallback 到 projects
    return f"{self._shared_mounts['projects']}/{relative_path}"

# 這些操作在 CIFS 掛載點上都可能阻塞
os.listdir("/mnt/nas/circuits/")      # 卡住
os.walk("/mnt/nas/circuits/")         # 卡住
pathlib.Path("/mnt/nas/circuits/").iterdir()  # 卡住

async def _run_find(args: list[str], timeout: int = 30) -> str:
    """非同步執行 find 指令"""
    proc = None
    try:
        proc = await asyncio.create_subprocess_exec(
            *args,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.DEVNULL,
        )
        stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=timeout)
        return stdout.decode("utf-8", errors="replace").strip()
    except (asyncio.TimeoutError, OSError):
        if proc:
            try:
                proc.kill()
            except ProcessLookupError:
                pass
        return ""

async def _find_matching_dirs(max_depth: int) -> list[str]:
    """在淺層找出名稱匹配任一關鍵字的目錄"""
    tasks = []
    for source in source_paths:
        for kw in keyword_list:
            args = [
                "find", source,
                "-maxdepth", str(max_depth),
                "-type", "d",
                "-iname", f"*{kw}*",
            ]
            tasks.append(_run_find(args, timeout=30))

    results = await asyncio.gather(*tasks)
    dirs = set()
    for output in results:
        for line in output.split("\n"):
            if line:
                dirs.add(line)
    return sorted(dirs)

# 階段 1：淺層 2 層
matched_dirs = await _find_matching_dirs(max_depth=2)
matched_files = await _search_in_dirs(matched_dirs)

# 階段 2：擴展到 3 層
if not matched_files:
    matched_dirs = await _find_matching_dirs(max_depth=3)
    matched_files = await _search_in_dirs(matched_dirs)

# 階段 3：全掃檔名
if not matched_files:
    matched_files = await _search_in_dirs(source_paths)

async def _search_in_dirs(dirs: list[str]) -> list[dict]:
    """在指定目錄中搜尋符合條件的檔案"""
    args = ["find"] + dirs + ["-type", "f"]
    # 所有關鍵字都要匹配路徑（AND 條件）
    for kw in keyword_list:
        args.extend(["-ipath", f"*{kw}*"])
    # 檔案類型過濾
    if type_list:
        args.append("(")
        for i, t in enumerate(type_list):
            if i > 0:
                args.append("-o")
            args.extend(["-iname", f"*.{t}"])
        args.append(")")

    output = await _run_find(args, timeout=120)
    # ... 解析結果 ...

@mcp.tool()
async def download_web_image(
    url: str,
    ctos_user_id: int | None = None,
) -> str:
    """下載網路圖片並準備為回覆訊息"""
    local_path = await download_image_from_url(url)
    if not local_path:
        return f"無法下載圖片：{url}"

    file_info = {
        "type": "image",
        "url": local_path,
        "original_url": url,
        "name": file_name,
    }
    marker = f"[FILE_MESSAGE:{json.dumps(file_info, ensure_ascii=False)}]"
    return f"已下載圖片 {file_name}\n{marker}"

async def download_image_from_url(url: str) -> str | None:
    """下載圖片 URL 到暫存目錄"""
    async with httpx.AsyncClient(follow_redirects=True, timeout=30) as client:
        resp = await client.get(url)
        if resp.status_code != 200:
            return None

        content_type = resp.headers.get("content-type", "")
        if not content_type.startswith("image/"):
            return None

        # 用 URL 的 MD5 作為檔名，避免重複下載
        filename = hashlib.md5(url.encode()).hexdigest()[:12] + ext
        file_path = os.path.join(DOWNLOADED_IMAGE_DIR, filename)

        with open(file_path, "wb") as f:
            f.write(resp.content)

        return file_path

# Line 發送圖片
async def send_nas_file(...):
    if is_image and file_size <= max_image_size:
        # 產生公開下載連結
        download_url = result.full_url.replace("/s/", "/api/public/") + "/download"
        # Line 直接用 URL，不需要先下載
        message_id, error = await push_image(target_id, download_url)

# Telegram 發送檔案
class TelegramBotAdapter:
    async def send_file(self, target, file_url, file_name, **kwargs):
        """先下載檔案到記憶體，再以二進位方式上傳給 Telegram"""
        async with httpx.AsyncClient(follow_redirects=True, timeout=60) as client:
            resp = await client.get(file_url)
            resp.raise_for_status()

        buf = BytesIO(resp.content)
        buf.name = file_name

        msg = await self.bot.send_document(
            chat_id=target,
            document=InputFile(buf, filename=file_name),
        )