mcp-synology-filestation/test_dirsize_md5.py

"""Wegwerfskript: DirSize + MD5 direkt gegen die NAS testen.

Ausfuehren: uv run python test_dirsize_md5.py
"""

import asyncio
import json
import time

import httpx

from mcp_synology_filestation.auth import AuthManager
from mcp_synology_filestation.client import FileStationClient
from mcp_synology_filestation.config import load_config

DIRSIZE_PATHS = ["/test-mcp", "/docker"]
MD5_PATH = "/test-mcp/test.zip"


def pp(label: str, data: object, elapsed_ms: float | None = None) -> None:
    print(f"\n{'='*60}")
    suffix = f"  [{elapsed_ms:.1f} ms]" if elapsed_ms is not None else ""
    print(f"  {label}{suffix}")
    print("=" * 60)
    print(json.dumps(data, indent=2, ensure_ascii=False))


async def raw(http: httpx.AsyncClient, url: str, sid: str, **params) -> dict:
    r = await http.get(url, params={"_sid": sid, **params})
    r.raise_for_status()
    try:
        return r.json()
    except Exception:
        return {"_raw": r.text[:300], "_http_status": r.status_code}


async def probe_dirsize_long(
    http: httpx.AsyncClient, sid: str, api_url: str, path: str
) -> None:
    """Start DirSize and poll v1 every 200ms for up to 15s.

    Goal: find out if 599 means 'task still running' (keep polling)
    or 'task gone' (give up). If the task eventually returns data,
    599 = 'not ready yet'. If it never returns data, 599 = 'task gone'.
    """
    print(f"\n{'#'*60}")
    print(f"  DIRSIZE {path} — long poll (15s, every 200ms)")
    print(f"{'#'*60}")

    t0 = time.perf_counter()
    start_body = await raw(
        http, api_url, sid,
        api="SYNO.FileStation.DirSize", version="2", method="start",
        path=json.dumps([path]),
    )
    elapsed_start = (time.perf_counter() - t0) * 1000
    pp(f"DirSize::start ({path})", start_body, elapsed_start)

    taskid = (start_body.get("data") or {}).get("taskid")
    if not taskid:
        print("[!] No taskid.")
        return

    print(f"\n[*] Polling status v1 every 200ms for up to 15s (taskid={taskid[:12]}...)")
    for attempt in range(75):  # 75 * 200ms = 15s
        if attempt > 0:
            await asyncio.sleep(0.2)
        t = time.perf_counter()
        r = await raw(
            http, api_url, sid,
            api="SYNO.FileStation.DirSize", version="1", method="status",
            taskid=taskid,
        )
        elapsed = (t - t0) * 1000
        success = r.get("success")
        data = (r.get("data") or {})
        finished = data.get("finished")
        error_code = (r.get("error") or {}).get("code")

        if finished:
            print(f"  [{elapsed:.0f}ms] attempt {attempt+1}: FERTIG! "
                  f"num_dir={data.get('num_dir')} "
                  f"num_file={data.get('num_file')} "
                  f"total_size={data.get('total_size')}")
            pp(f"DirSize::status final ({path})", r, elapsed)
            return
        elif success and not finished:
            # Still running — show current progress
            print(f"  [{elapsed:.0f}ms] attempt {attempt+1}: running... "
                  f"num_dir={data.get('num_dir', '?')} "
                  f"num_file={data.get('num_file', '?')} "
                  f"total_size={data.get('total_size', '?')}")
        else:
            print(f"  [{elapsed:.0f}ms] attempt {attempt+1}: error code={error_code}")
            # Continue polling — 599 might mean 'not ready yet'

    print(f"\n[!] No result after 15s — task never returned data.")


async def main() -> None:
    config = load_config()
    auth = AuthManager(config)

    async with FileStationClient(config.base_url, config.connection.verify_ssl) as client:
        client.set_auth_manager(auth)
        await client._ensure_initialized()  # noqa: SLF001
        sid = client.sid
        base = config.base_url

        info = client._api_cache.get("SYNO.FileStation.DirSize", {})  # noqa: SLF001
        api_url = f"{base}/webapi/{info.get('path', 'entry.cgi')}"
        print(f"[*] DirSize API: {api_url}  v{info.get('minVersion')}-v{info.get('maxVersion')}")

        async with httpx.AsyncClient(verify=config.connection.verify_ssl, timeout=30.0) as http:
            for path in DIRSIZE_PATHS:
                await probe_dirsize_long(http, sid, api_url, path)

        await auth.logout(client)
        print("\n[*] Logout OK.")


if __name__ == "__main__":
    asyncio.run(main())