🏷️ types: move to ty

.vscode/extensions.json

@@ -1,7 +1,7 @@
 {
   "recommendations": [
     "ms-python.python",
-    "ms-python.vscode-pylance",
+    "astral-sh.ty",
     "charliermarsh.ruff",
     "esbenp.prettier-vscode",
     "EditorConfig.EditorConfig",

.vscode/settings.json

@@ -1,5 +1,5 @@
 {
-  "python.languageServer": "Pylance",
+  "python.languageServer": "None",
   "python.defaultInterpreterPath": "${workspaceFolder}/.venv/bin/python",
   "python.analysis.inlayHints.functionReturnTypes": true,
   "python.analysis.inlayHints.variableTypes": true,

justfile

@@ -18,7 +18,7 @@ fmt:
   uv run ruff format .
 
 lint:
-  uv run pyright src/yutto packages/biliass/src/biliass tests
+  uv run ty check --error-on-warning src/yutto packages/biliass/src/biliass tests
   uv run ruff check .
   uv run typos
 

packages/biliass/pyproject.toml

@@ -39,11 +39,6 @@ biliass = "biliass.__main__:main"
 features = ["pyo3/extension-module"]
 module-name = "biliass._core"
 
-[tool.pyright]
-include = ["src/biliass"]
-pythonVersion = "3.10"
-typeCheckingMode = "basic"
-
 [build-system]
 requires = ["maturin>=1.4,<2.0"]
 build-backend = "maturin"

pyproject.toml

@@ -45,14 +45,13 @@ yutto = "yutto.__main__:main"
 
 [dependency-groups]
 dev = [
-  "pyright>=1.1.408",
+  "ty>=0.0.31",
   "ruff>=0.15.0",
   "typos>=1.43.3",
   "pytest>=9.0.2",
   "pytest-rerunfailures>=16.1",
   "syrupy>=5.1.0",
   "pytest-codspeed>=4.2.0",
-  "tomli>=2.3.0; python_version >= '3.11'",
 ]
 
 [tool.uv.sources]
@@ -64,11 +63,6 @@ members = ["packages/*"]
 [tool.pytest.ini_options]
 markers = ["api", "e2e", "processor", "biliass", "ignore", "ci_skip", "ci_only"]
 
-[tool.pyright]
-include = ["src/yutto", "packages/biliass/src/biliass", "tests"]
-pythonVersion = "3.10"
-typeCheckingMode = "strict"
-
 [tool.ruff]
 line-length = 120
 target-version = "py310"
@@ -113,8 +107,6 @@ select = [
 ]
 ignore = [
   "E501", # line too long, duplicate with ruff fmt
-  "F401", # imported but unused, duplicate with pyright
-  "F841", # local variable is assigned to but never used, duplicate with pyright
 ]
 future-annotations = true
 

src/yutto/api/ugc_video.py

@@ -358,7 +358,7 @@ def _parse_actor_info(video_info: dict[str, Any]):
     actors: list[Actor] = []
     if video_info.get("staff") and isinstance(video_info["staff"], list):
         _index: int = 0
-        staff_list: list[dict[str, Any]] = video_info["staff"]  # pyright: ignore[reportUnknownVariableType]
+        staff_list: list[dict[str, Any]] = video_info["staff"]
         for staff in staff_list:
             actors.append(
                 Actor(
@@ -371,7 +371,7 @@ def _parse_actor_info(video_info: dict[str, Any]):
             )
             _index += 1
     elif video_info.get("owner") and isinstance(video_info["owner"], dict):
-        staff_info: dict[str, Any] = video_info["owner"]  # pyright: ignore[reportUnknownVariableType]
+        staff_info: dict[str, Any] = video_info["owner"]
         actors.append(
             Actor(
                 name=staff_info["name"],

src/yutto/api/user_info.py

@@ -98,7 +98,7 @@ def _get_mixin_key(string: str) -> str:
     return "".join([string[idx] for idx in char_indices[:32]])
 
 
-def encode_wbi(params: dict[str, Any], wbi_img: WbiImg):
+def encode_wbi(params: dict[str, Any], wbi_img: WbiImg) -> dict[str, Any]:
     img_key = wbi_img["img_key"]
     sub_key = wbi_img["sub_key"]
     illegal_char_remover = re.compile(r"[!'\(\)*]")

src/yutto/cli/cli.py

@@ -56,7 +56,7 @@ def parse_config_path() -> Path | None:
 def cli() -> argparse.ArgumentParser:
     settings_file = parse_config_path()
     if settings_file is None:
-        settings = YuttoSettings()  # pyright: ignore[reportCallIssue]
+        settings = YuttoSettings()  # ty: ignore[missing-argument]
     else:
         Logger.info(f"发现配置文件 {settings_file}，加载中……")
         settings = load_settings_file(settings_file)

src/yutto/cli/settings.py

@@ -101,11 +101,11 @@ class YuttoAuthSettings(BaseModel):
 
 
 class YuttoSettings(BaseModel):
-    basic: Annotated[YuttoBasicSettings, Field(YuttoBasicSettings())]  # pyright: ignore[reportCallIssue]
-    resource: Annotated[YuttoResourceSettings, Field(YuttoResourceSettings())]  # pyright: ignore[reportCallIssue]
-    danmaku: Annotated[YuttoDanmakuSettings, Field(YuttoDanmakuSettings())]  # pyright: ignore[reportCallIssue]
-    batch: Annotated[YuttoBatchSettings, Field(YuttoBatchSettings())]  # pyright: ignore[reportCallIssue]
-    auth: Annotated[YuttoAuthSettings, Field(YuttoAuthSettings())]  # pyright: ignore[reportCallIssue]
+    basic: Annotated[YuttoBasicSettings, Field(YuttoBasicSettings())]  # ty: ignore[missing-argument]
+    resource: Annotated[YuttoResourceSettings, Field(YuttoResourceSettings())]  # ty: ignore[missing-argument]
+    danmaku: Annotated[YuttoDanmakuSettings, Field(YuttoDanmakuSettings())]  # ty: ignore[missing-argument]
+    batch: Annotated[YuttoBatchSettings, Field(YuttoBatchSettings())]  # ty: ignore[missing-argument]
+    auth: Annotated[YuttoAuthSettings, Field(YuttoAuthSettings())]  # ty: ignore[missing-argument]
 
 
 def search_for_settings_file() -> Path | None:
@@ -123,7 +123,7 @@ def search_for_settings_file() -> Path | None:
 
 def load_settings_file(settings_file: Path) -> YuttoSettings:
     with settings_file.open("r", encoding="utf-8") as f:
-        settings_raw: Any = tomllib.loads(f.read())  # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType]
+        settings_raw: Any = tomllib.loads(f.read())
     return YuttoSettings.model_validate(settings_raw)
 
 

src/yutto/input_parser.py

@@ -36,8 +36,8 @@ def alias_parser(file_path: str) -> dict[str, str]:
 
 
 def file_scheme_parser(url: str) -> list[str]:
-    file_url: str = urllib.parse.urlparse(url).path  # pyright: ignore[reportUnknownMemberType,reportAttributeAccessIssue,reportUnknownVariableType]
-    file_path = path_from_cli(urllib.request.url2pathname(file_url))  # pyright: ignore[reportUnknownMemberType,reportUnknownArgumentType]
+    file_url: str = urllib.parse.urlparse(url).path
+    file_path = path_from_cli(urllib.request.url2pathname(file_url))
     Logger.info(f"解析下载列表 {file_path} 中...")
     result: list[str] = []
     with file_path.open("r", encoding="utf-8") as f:

src/yutto/utils/asynclib.py

@@ -23,7 +23,7 @@
 def initial_async_policy():
     if sys.version_info < (3, 11) and platform.system() == "Windows":
         Logger.debug("Windows 平台（Python < 3.11），单独设置 EventLoopPolicy")
-        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())  # pyright: ignore
+        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())  # ty: ignore[unresolved-attribute]
 
 
 class CoroutineWrapper(Generic[RetT]):
@@ -50,9 +50,9 @@ async def sleep_with_status_bar_refresh(seconds: float):
 def async_cache(
     args_to_cache_key: Callable[[inspect.BoundArguments], str],
 ) -> Callable[[Callable[P, Coroutine[Any, Any, RetT]]], Callable[P, Coroutine[Any, Any, RetT]]]:
-    CACHE: dict[str, RetT] = {}
-
     def decorator(fn: Callable[P, Coroutine[Any, Any, RetT]]) -> Callable[P, Coroutine[Any, Any, RetT]]:
+        CACHE: dict[str, RetT] = {}
+
         @wraps(fn)
         async def wrapper(*args: P.args, **kwargs: P.kwargs) -> RetT:
             assert isinstance(fn, types.FunctionType)

src/yutto/utils/fetcher.py

@@ -229,7 +229,7 @@ async def get_size(ctx: FetcherContext, client: AsyncClient, url: str) -> int |
     @MaxRetry(2)
     # 对于相同 session，同样的页面没必要重复 touch
     @async_cache(lambda args: f"client_id={id(args.arguments['client'])}, url={args.arguments['url']}")
-    async def touch_url(ctx: FetcherContext, client: AsyncClient, url: str):
+    async def touch_url(ctx: FetcherContext, client: AsyncClient, url: str) -> None:
         async with ctx.fetch_guard():
             Logger.debug(f"Touch url: {url}")
             await client.get(url)

src/yutto/utils/file_buffer.py

@@ -3,7 +3,7 @@
 import heapq
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, cast
 
 import aiofiles
 
@@ -13,7 +13,12 @@
 if TYPE_CHECKING:
     from types import TracebackType
 
-    from typing_extensions import Self
+    from typing_extensions import Protocol, Self
+
+    class AsyncWritableBinaryFile(Protocol):
+        async def write(self, data: bytes) -> int: ...
+
+        async def close(self) -> None: ...
 
 
 @dataclass(order=True)
@@ -53,7 +58,9 @@ async def __ainit__(self, file_path: str | Path, overwrite: bool = False):
             self.file_path.unlink(missing_ok=True)
         self.buffer = list[BufferChunk]()
         self.written_size = self.file_path.stat().st_size if not overwrite and self.file_path.exists() else 0
-        self.file_obj: aiofiles.threadpool.binary.AsyncBufferedIOBase | None = await aiofiles.open(file_path, "ab")
+        self.file_obj: AsyncWritableBinaryFile | None = cast(
+            "AsyncWritableBinaryFile", await aiofiles.open(file_path, "ab")
+        )
 
     async def write(self, chunk: bytes, offset: int):
         buffer_chunk = BufferChunk(offset, chunk)

src/yutto/utils/functional/data_access.py

@@ -14,4 +14,4 @@ def data_has_chained_keys(data: Any, keys: list[str]) -> bool:
     if not isinstance(data, dict):
         return False
     key, *remaining_keys = keys
-    return data_has_chained_keys(data.get(key, Undefined()), remaining_keys)  # pyright: ignore[reportUnknownMemberType]
+    return data_has_chained_keys(data.get(key, Undefined()), remaining_keys)

tests/test_api/test_bangumi.py

@@ -5,7 +5,7 @@
 from yutto.api.bangumi import (
     get_bangumi_list,
     get_bangumi_playurl,
-    get_bangumi_subtitles,  # pyright: ignore[reportUnusedImport]
+    get_bangumi_subtitles,  # noqa: F401
     get_season_id_by_episode_id,
     get_season_id_by_media_id,
 )

tests/test_utils/test_fetcher.py

@@ -2,13 +2,27 @@
 
 import asyncio
 import ssl
-from typing import Any
+from typing import Any, Protocol, cast
 
 import pytest
 
 from yutto.utils.fetcher import FetcherContext, create_client, create_sync_client, resolve_proxy
 
 
+class _HasSSLContext(Protocol):
+    _ssl_context: ssl.SSLContext
+
+
+class _HasPool(Protocol):
+    _pool: _HasSSLContext
+
+
+def _transport_ssl_context(transport: Any) -> ssl.SSLContext:
+    # Test helper: inspect httpx's private transport internals to assert TLS policy wiring.
+    # If httpx changes `_pool._ssl_context`, this assertion helper will need to be updated too.
+    return cast("_HasPool", transport)._pool._ssl_context
+
+
 def test_resolve_proxy_auto_uses_system_proxy():
     assert resolve_proxy("auto") == (None, True)
 
@@ -34,8 +48,8 @@ def test_resolve_proxy_rejects_invalid_scheme():
 def test_create_client_keeps_download_tls_verification_disabled():
     client = create_client()
     try:
-        transport: Any = client._transport  # pyright: ignore[reportPrivateUsage]
-        ssl_context: ssl.SSLContext = transport._pool._ssl_context
+        transport: Any = client._transport
+        ssl_context = _transport_ssl_context(transport)
         assert ssl_context.verify_mode == ssl.CERT_NONE
         assert not ssl_context.check_hostname
     finally:
@@ -45,8 +59,8 @@ def test_create_client_keeps_download_tls_verification_disabled():
 def test_create_sync_client_follows_default_download_tls_policy():
     client = create_sync_client()
     try:
-        transport: Any = client._transport  # pyright: ignore[reportPrivateUsage]
-        ssl_context: ssl.SSLContext = transport._pool._ssl_context
+        transport: Any = client._transport
+        ssl_context = _transport_ssl_context(transport)
         assert ssl_context.verify_mode == ssl.CERT_NONE
         assert not ssl_context.check_hostname
     finally:
@@ -56,8 +70,8 @@ def test_create_sync_client_follows_default_download_tls_policy():
 def test_create_sync_client_can_enable_tls_verification():
     client = create_sync_client(verify=True)
     try:
-        transport: Any = client._transport  # pyright: ignore[reportPrivateUsage]
-        ssl_context: ssl.SSLContext = transport._pool._ssl_context
+        transport: Any = client._transport
+        ssl_context = _transport_ssl_context(transport)
         assert ssl_context.verify_mode == ssl.CERT_REQUIRED
         assert ssl_context.check_hostname
     finally: