1093 lines
35 KiB
Python
1093 lines
35 KiB
Python
#!/usr/bin/env python3
|
||
# encoding: utf-8
|
||
|
||
from __future__ import annotations
|
||
|
||
__author__ = "ChenyangGao <https://chenyanggao.github.io>"
|
||
__all__ = ["P115SharePath", "P115ShareFileSystem"]
|
||
|
||
import errno
|
||
|
||
from collections import deque
|
||
from collections.abc import (
|
||
AsyncIterator, Callable, Coroutine, Iterable, Iterator, Mapping,
|
||
MutableMapping, Sequence,
|
||
)
|
||
from copy import deepcopy
|
||
from datetime import datetime
|
||
from functools import cached_property, partial
|
||
from os import fspath, stat_result, PathLike
|
||
from posixpath import join as joinpath
|
||
from re import compile as re_compile
|
||
from stat import S_IFDIR, S_IFREG
|
||
from time import time
|
||
from typing import cast, overload, Any, Literal, Never, Self
|
||
|
||
from dictattr import AttrDict
|
||
from iterutils import run_gen_step, run_gen_step_iter, Yield, YieldFrom
|
||
from p115client import check_response, normalize_attr, P115URL
|
||
from posixpatht import escape, joins, splits, path_is_dir_form
|
||
|
||
from .client import P115Client
|
||
from .fs_base import IDOrPathType, P115PathBase, P115FileSystemBase
|
||
|
||
|
||
CRE_SHARE_LINK_search1 = re_compile(r"(?:/s/|share\.115\.com/)(?P<share_code>[a-z0-9]+)\?password=(?P<receive_code>[a-z0-9]{4})").search
|
||
CRE_SHARE_LINK_search2 = re_compile(r"(?P<share_code>[a-z0-9]+)-(?P<receive_code>[a-z0-9]{4})").search
|
||
|
||
|
||
class P115SharePath(P115PathBase):
|
||
fs: P115ShareFileSystem
|
||
|
||
@property
|
||
def ancestors(self, /) -> list[dict]:
|
||
try:
|
||
return self["ancestors"]
|
||
except KeyError:
|
||
ancestors = self.fs.get_ancestors(self.id)
|
||
self.attr["path"] = joins([a["name"] for a in ancestors])
|
||
return ancestors
|
||
|
||
@property
|
||
def path(self, /) -> str:
|
||
try:
|
||
return self["path"]
|
||
except KeyError:
|
||
self.ancestors
|
||
return self.attr["path"]
|
||
|
||
@overload
|
||
def search(
|
||
self,
|
||
/,
|
||
async_: Literal[False] = False,
|
||
**payload,
|
||
) -> Iterator[P115SharePath]:
|
||
...
|
||
@overload
|
||
def search(
|
||
self,
|
||
/,
|
||
async_: Literal[True],
|
||
**payload,
|
||
) -> AsyncIterator[P115SharePath]:
|
||
...
|
||
def search(
|
||
self,
|
||
/,
|
||
async_: Literal[False, True] = False,
|
||
**payload,
|
||
) -> Iterator[P115SharePath] | AsyncIterator[P115SharePath]:
|
||
return self.fs.search(self, async_=async_, **payload)
|
||
|
||
|
||
class P115ShareFileSystem(P115FileSystemBase[P115SharePath]):
|
||
share_code: str
|
||
receive_code: str
|
||
path_to_id: MutableMapping[str, int]
|
||
id_to_attr: MutableMapping[int, AttrDict]
|
||
pid_to_children: MutableMapping[int, tuple[AttrDict, ...]]
|
||
full_loaded: bool
|
||
path_class = P115SharePath
|
||
|
||
def __init__(
|
||
self,
|
||
/,
|
||
client: str | P115Client,
|
||
share_code: str,
|
||
receive_code: str = "",
|
||
request: None | Callable = None,
|
||
async_request: None | Callable = None,
|
||
):
|
||
"""115 分享链接的文件系统封装
|
||
|
||
支持以下几种格式的链接（括号内的字符表示可有可无）：
|
||
- http(s)://115.com/s/{share_code}?password={receive_code}(#)
|
||
- http(s)://share.115.com/{share_code}?password={receive_code}(#)
|
||
- (/){share_code}-{receive_code}(/)
|
||
"""
|
||
super().__init__(client, request, async_request)
|
||
self.__dict__.update(
|
||
id=0,
|
||
path="/",
|
||
share_code=share_code,
|
||
receive_code=receive_code,
|
||
path_to_id={"/": 0},
|
||
id_to_attr={},
|
||
pid_to_children={},
|
||
full_loaded=False,
|
||
)
|
||
|
||
@classmethod
|
||
def from_url(
|
||
cls,
|
||
/,
|
||
client: str | P115Client,
|
||
url: str,
|
||
request: None | Callable = None,
|
||
async_request: None | Callable = None,
|
||
) -> Self:
|
||
m = CRE_SHARE_LINK_search1(url)
|
||
if m is None:
|
||
m = CRE_SHARE_LINK_search2(url)
|
||
if m is None:
|
||
raise ValueError("not a valid 115 share link")
|
||
return cls(
|
||
client,
|
||
share_code=m["share_code"],
|
||
receive_code=m["receive_code"] or "",
|
||
request=request,
|
||
async_request=async_request,
|
||
)
|
||
|
||
def __repr__(self, /) -> str:
|
||
cls = type(self)
|
||
module = cls.__module__
|
||
name = cls.__qualname__
|
||
if module != "__main__":
|
||
name = module + "." + name
|
||
return f"<{name}(client={self.client!r}, share_code={self.share_code!r}, receive_code={self.receive_code!r}, id={self.id!r}, path={self.path!r}) at {hex(id(self))}>"
|
||
|
||
def __setattr__(self, attr, val, /) -> Never:
|
||
raise TypeError("can't set attributes")
|
||
|
||
@overload
|
||
def downlist(
|
||
self,
|
||
/,
|
||
id: int,
|
||
*,
|
||
async_: Literal[False] = False,
|
||
) -> dict:
|
||
...
|
||
@overload
|
||
def downlist(
|
||
self,
|
||
/,
|
||
id: int,
|
||
*,
|
||
async_: Literal[True],
|
||
) -> Coroutine[Any, Any, dict]:
|
||
...
|
||
def downlist(
|
||
self,
|
||
/,
|
||
id: int,
|
||
*,
|
||
async_: Literal[False, True] = False,
|
||
) -> dict | Coroutine[Any, Any, dict]:
|
||
"""获取分享链接的某个文件夹中可下载的文件的列表（只含文件，不含文件夹，任意深度，简略信息）
|
||
"""
|
||
return check_response(self.client.share_downlist( # type: ignore
|
||
{
|
||
"share_code": self.share_code,
|
||
"receive_code": self.receive_code,
|
||
"cid": id,
|
||
},
|
||
request=self.async_request if async_ else self.request,
|
||
async_=async_,
|
||
))
|
||
|
||
@overload
|
||
def fs_files(
|
||
self,
|
||
/,
|
||
payload: dict,
|
||
async_: Literal[False] = False,
|
||
) -> dict:
|
||
...
|
||
@overload
|
||
def fs_files(
|
||
self,
|
||
/,
|
||
payload: dict,
|
||
async_: Literal[True],
|
||
) -> Coroutine[Any, Any, dict]:
|
||
...
|
||
def fs_files(
|
||
self,
|
||
/,
|
||
payload: dict,
|
||
async_: Literal[False, True] = False,
|
||
) -> dict | Coroutine[Any, Any, dict]:
|
||
"""获取分享链接的某个文件夹中的文件和子文件夹的列表（包含详细信息）
|
||
:param payload:
|
||
- id: int | str = 0
|
||
- limit: int = 32
|
||
- offset: int = 0
|
||
- asc: 0 | 1 = <default> # 是否升序排列
|
||
- o: str = <default>
|
||
# 用某字段排序：
|
||
# - "file_name": 文件名
|
||
# - "file_size": 文件大小
|
||
# - "user_ptime": 创建时间/修改时间
|
||
"""
|
||
return check_response(self.client.share_snap( # type: ignore
|
||
{
|
||
**payload,
|
||
"share_code": self.share_code,
|
||
"receive_code": self.receive_code,
|
||
},
|
||
base_url=True,
|
||
request=self.async_request if async_ else self.request,
|
||
async_=async_,
|
||
))
|
||
|
||
@overload
|
||
def fs_search(
|
||
self,
|
||
payload: str | dict,
|
||
/,
|
||
async_: Literal[False] = False,
|
||
) -> dict:
|
||
...
|
||
@overload
|
||
def fs_search(
|
||
self,
|
||
payload: str | dict,
|
||
/,
|
||
async_: Literal[True],
|
||
) -> Coroutine[Any, Any, dict]:
|
||
...
|
||
def fs_search(
|
||
self,
|
||
payload: str | dict,
|
||
/,
|
||
async_: Literal[False, True] = False,
|
||
) -> dict | Coroutine[Any, Any, dict]:
|
||
if isinstance(payload, str):
|
||
payload = {"share_code": self.share_code, "receive_code": self.receive_code, "cid": self.id, "search_value": payload}
|
||
else:
|
||
payload = {"share_code": self.share_code, "receive_code": self.receive_code, "cid": self.id, **payload}
|
||
return check_response(self.client.share_search( # type: ignore
|
||
payload,
|
||
request=self.async_request if async_ else self.request,
|
||
async_=async_,
|
||
))
|
||
|
||
@cached_property
|
||
def create_time(self, /) -> datetime:
|
||
"分享的创建时间"
|
||
return datetime.fromtimestamp(self.create_timestamp)
|
||
|
||
@cached_property
|
||
def create_timestamp(self, /) -> int:
|
||
"分享的创建时间"
|
||
return int(self.shareinfo["create_time"])
|
||
|
||
@cached_property
|
||
def snap_id(self, /) -> int:
|
||
"获取这个分享的 id"
|
||
return int(self.shareinfo["snap_id"])
|
||
|
||
@cached_property
|
||
def user_id(self, /) -> int:
|
||
"获取分享者的用户 id"
|
||
return int(self.sharedata["userinfo"]["user_id"])
|
||
|
||
@property
|
||
def sharedata(self, /) -> dict:
|
||
"获取分享的首页数据"
|
||
return self.fs_files({"limit": 1})["data"]
|
||
|
||
@property
|
||
def shareinfo(self, /) -> dict:
|
||
"获取分享信息"
|
||
return self.sharedata["shareinfo"]
|
||
|
||
@overload
|
||
def _search_item(
|
||
self,
|
||
id: int,
|
||
/,
|
||
async_: Literal[False] = False,
|
||
) -> AttrDict:
|
||
...
|
||
@overload
|
||
def _search_item(
|
||
self,
|
||
id: int,
|
||
/,
|
||
async_: Literal[True],
|
||
) -> Coroutine[Any, Any, AttrDict]:
|
||
...
|
||
def _search_item(
|
||
self,
|
||
id: int,
|
||
/,
|
||
async_: Literal[False, True] = False,
|
||
) -> AttrDict | Coroutine[Any, Any, AttrDict]:
|
||
dq = deque((self.attr(0),))
|
||
get, put = dq.popleft, dq.append
|
||
if async_:
|
||
async def request():
|
||
while dq:
|
||
async for attr in self.iterdir(get(), async_=True):
|
||
if attr["id"] == id:
|
||
return attr
|
||
if attr["is_directory"]:
|
||
put(attr)
|
||
self.__dict__["full_loaded"] = True
|
||
raise FileNotFoundError(errno.ENOENT, f"no such id: {id!r}")
|
||
return request()
|
||
else:
|
||
while dq:
|
||
for attr in self.iterdir(get()):
|
||
if attr["id"] == id:
|
||
return attr
|
||
if attr["is_directory"]:
|
||
put(attr)
|
||
self.__dict__["full_loaded"] = True
|
||
raise FileNotFoundError(errno.ENOENT, f"no such id: {id!r}")
|
||
|
||
@overload
|
||
def _attr(
|
||
self,
|
||
id: int,
|
||
/,
|
||
async_: Literal[False] = False,
|
||
) -> AttrDict:
|
||
...
|
||
@overload
|
||
def _attr(
|
||
self,
|
||
id: int,
|
||
/,
|
||
async_: Literal[True],
|
||
) -> Coroutine[Any, Any, AttrDict]:
|
||
...
|
||
def _attr(
|
||
self,
|
||
id: int,
|
||
/,
|
||
async_: Literal[False, True] = False,
|
||
) -> AttrDict | Coroutine[Any, Any, AttrDict]:
|
||
share_code = self.share_code
|
||
receive_code = self.receive_code
|
||
def gen_step():
|
||
try:
|
||
return self.id_to_attr[id]
|
||
except KeyError:
|
||
pass
|
||
if self.full_loaded:
|
||
raise FileNotFoundError(errno.ENOENT, f"no such id: {id!r}")
|
||
if id == 0:
|
||
attr = self.id_to_attr[0] = AttrDict({
|
||
"id": 0,
|
||
"parent_id": 0,
|
||
"name": "",
|
||
"path": "/",
|
||
"is_directory": True,
|
||
"size": None,
|
||
"time": self.create_timestamp,
|
||
"ico": "folder",
|
||
"ancestors": [{"id": 0, "name": ""}],
|
||
"share_code": share_code,
|
||
"receive_code": receive_code,
|
||
})
|
||
return attr
|
||
# NOTE: quick detection of id existence
|
||
yield partial(
|
||
self.client.share_download_url,
|
||
{
|
||
"share_code": share_code,
|
||
"receive_code": receive_code,
|
||
"file_id": id,
|
||
},
|
||
strict=False,
|
||
request=self.async_request if async_ else self.request,
|
||
async_=async_,
|
||
)
|
||
return (yield partial(self._search_item, id, async_=async_))
|
||
return run_gen_step(gen_step, async_=async_)
|
||
|
||
@overload
|
||
def _attr_path(
|
||
self,
|
||
path: str | PathLike[str] | Sequence[str],
|
||
/,
|
||
pid: None | int = None,
|
||
ensure_dir: bool = False,
|
||
*,
|
||
async_: Literal[False] = False,
|
||
) -> AttrDict:
|
||
...
|
||
@overload
|
||
def _attr_path(
|
||
self,
|
||
path: str | PathLike[str] | Sequence[str],
|
||
/,
|
||
pid: None | int = None,
|
||
ensure_dir: bool = False,
|
||
*,
|
||
async_: Literal[True],
|
||
) -> Coroutine[Any, Any, AttrDict]:
|
||
...
|
||
def _attr_path(
|
||
self,
|
||
path: str | PathLike[str] | Sequence[str],
|
||
/,
|
||
pid: None | int = None,
|
||
ensure_dir: bool = False,
|
||
*,
|
||
async_: Literal[False, True] = False,
|
||
) -> AttrDict | Coroutine[Any, Any, AttrDict]:
|
||
def gen_step():
|
||
nonlocal path, pid, ensure_dir
|
||
|
||
if isinstance(path, PathLike):
|
||
path = fspath(path)
|
||
if pid is None:
|
||
pid = self.id
|
||
if not path or path == ".":
|
||
return (yield partial(self._attr, pid, async_=async_))
|
||
|
||
parents = 0
|
||
if isinstance(path, str):
|
||
if not ensure_dir:
|
||
ensure_dir = path_is_dir_form(path)
|
||
patht, parents = splits(path)
|
||
if not (patht or parents):
|
||
return (yield partial(self._attr, pid, async_=async_))
|
||
else:
|
||
if not ensure_dir:
|
||
ensure_dir = path[-1] == ""
|
||
patht = [path[0], *(p for p in path[1:] if p)]
|
||
if patht == [""]:
|
||
return self._attr(0)
|
||
elif patht and patht[0] == "":
|
||
pid = 0
|
||
|
||
ancestor_patht: list[str] = []
|
||
if pid == 0:
|
||
if patht[0] != "":
|
||
patht.insert(0, "")
|
||
else:
|
||
ancestors = yield partial(self.get_ancestors, pid, async_=async_)
|
||
if parents:
|
||
if parents >= len(ancestors):
|
||
pid = 0
|
||
else:
|
||
pid = cast(int, ancestors[-parents-1]["id"])
|
||
ancestor_patht = ["", *(a["name"] for a in ancestors[1:-parents])]
|
||
else:
|
||
ancestor_patht = ["", *(a["name"] for a in ancestors[1:])]
|
||
if not patht:
|
||
return (yield partial(self._attr, pid, async_=async_))
|
||
|
||
if pid == 0:
|
||
dirname = ""
|
||
ancestors_paths: list[str] = [(dirname := f"{dirname}/{escape(name)}") for name in patht[1:]]
|
||
else:
|
||
dirname = joins(ancestor_patht)
|
||
ancestors_paths = [(dirname := f"{dirname}/{escape(name)}") for name in patht]
|
||
|
||
fullpath = ancestors_paths[-1]
|
||
path_to_id = self.path_to_id
|
||
if path_to_id:
|
||
if not ensure_dir and (id := path_to_id.get(fullpath)):
|
||
return (yield partial(self._attr, id, async_=async_))
|
||
if (id := path_to_id.get(fullpath + "/")):
|
||
return (yield partial(self._attr, id, async_=async_))
|
||
if self.full_loaded:
|
||
raise FileNotFoundError(
|
||
errno.ENOENT,
|
||
f"no such path {fullpath!r} (in {pid!r})",
|
||
)
|
||
|
||
parent: int | AttrDict
|
||
for i in reversed(range(len(ancestors_paths)-1)):
|
||
if path_to_id and (id := path_to_id.get((dirname := ancestors_paths[i]) + "/")):
|
||
parent = yield partial(self._attr, id, async_=async_)
|
||
i += 1
|
||
break
|
||
else:
|
||
i = 0
|
||
parent = pid
|
||
|
||
if pid == 0:
|
||
i += 1
|
||
|
||
attr: AttrDict
|
||
last_idx = len(patht) - 1
|
||
if async_:
|
||
for i, name in enumerate(patht[i:], i):
|
||
async def step():
|
||
nonlocal attr, parent
|
||
async for attr in self.iterdir(parent, async_=True):
|
||
if attr["name"] == name:
|
||
if ensure_dir or i < last_idx:
|
||
if attr["is_directory"]:
|
||
parent = attr
|
||
break
|
||
else:
|
||
break
|
||
else:
|
||
if isinstance(parent, AttrDict):
|
||
parent = parent["id"]
|
||
raise FileNotFoundError(
|
||
errno.ENOENT,
|
||
f"no such file {name!r} (in {parent} @ {joins(patht[:i])!r})",
|
||
)
|
||
yield step
|
||
else:
|
||
for i, name in enumerate(patht[i:], i):
|
||
for attr in self.iterdir(parent):
|
||
if attr["name"] == name:
|
||
if ensure_dir or i < last_idx:
|
||
if attr["is_directory"]:
|
||
parent = attr
|
||
break
|
||
else:
|
||
break
|
||
else:
|
||
if isinstance(parent, AttrDict):
|
||
parent = parent["id"]
|
||
raise FileNotFoundError(
|
||
errno.ENOENT,
|
||
f"no such file {name!r} (in {parent} @ {joins(patht[:i])!r})",
|
||
)
|
||
return attr
|
||
return run_gen_step(gen_step, async_=async_)
|
||
|
||
@overload
|
||
def attr(
|
||
self,
|
||
id_or_path: IDOrPathType = "",
|
||
/,
|
||
pid: None | int = None,
|
||
ensure_dir: bool = False,
|
||
*,
|
||
async_: Literal[False] = False,
|
||
) -> AttrDict:
|
||
...
|
||
@overload
|
||
def attr(
|
||
self,
|
||
id_or_path: IDOrPathType = "",
|
||
/,
|
||
pid: None | int = None,
|
||
ensure_dir: bool = False,
|
||
*,
|
||
async_: Literal[True],
|
||
) -> Coroutine[Any, Any, AttrDict]:
|
||
...
|
||
def attr(
|
||
self,
|
||
id_or_path: IDOrPathType = "",
|
||
/,
|
||
pid: None | int = None,
|
||
ensure_dir: bool = False,
|
||
*,
|
||
async_: Literal[False, True] = False,
|
||
) -> AttrDict | Coroutine[Any, Any, AttrDict]:
|
||
"获取属性"
|
||
def gen_step():
|
||
path_class = type(self).path_class
|
||
if isinstance(id_or_path, path_class):
|
||
attr = id_or_path.attr
|
||
elif isinstance(id_or_path, AttrDict):
|
||
attr = id_or_path
|
||
elif isinstance(id_or_path, int):
|
||
attr = yield partial(self._attr, id_or_path, async_=async_)
|
||
else:
|
||
return (yield partial(
|
||
self._attr_path,
|
||
id_or_path,
|
||
pid=pid,
|
||
ensure_dir=ensure_dir,
|
||
async_=async_,
|
||
))
|
||
if ensure_dir and not attr["is_directory"]:
|
||
raise NotADirectoryError(
|
||
errno.ENOTDIR,
|
||
f"{attr['id']} (id={attr['id']}) is not directory"
|
||
)
|
||
return attr
|
||
return run_gen_step(gen_step, async_=async_)
|
||
|
||
@overload
|
||
def dirlen(
|
||
self,
|
||
id_or_path: IDOrPathType = "",
|
||
/,
|
||
pid: None | int = None,
|
||
*,
|
||
async_: Literal[False] = False,
|
||
) -> int:
|
||
...
|
||
@overload
|
||
def dirlen(
|
||
self,
|
||
id_or_path: IDOrPathType = "",
|
||
/,
|
||
pid: None | int = None,
|
||
*,
|
||
async_: Literal[True],
|
||
) -> Coroutine[Any, Any, int]:
|
||
...
|
||
def dirlen(
|
||
self,
|
||
id_or_path: IDOrPathType = "",
|
||
/,
|
||
pid: None | int = None,
|
||
*,
|
||
async_: Literal[False, True] = False,
|
||
) -> int | Coroutine[Any, Any, int]:
|
||
"文件夹中的项目数（直属的文件和目录计数）"
|
||
def gen_step():
|
||
id = yield partial(self.get_id, id_or_path, pid=pid, async_=async_)
|
||
if (children := self.pid_to_children.get(id)) is not None:
|
||
return len(children)
|
||
resp = yield partial(
|
||
self.fs_files,
|
||
{"cid": id, "limit": 1},
|
||
async_=async_,
|
||
)
|
||
return resp["data"]["count"]
|
||
return run_gen_step(gen_step, async_=async_)
|
||
|
||
@overload
|
||
def get_ancestors(
|
||
self,
|
||
id_or_path: IDOrPathType = "",
|
||
/,
|
||
pid: None | int = None,
|
||
*,
|
||
async_: Literal[False] = False,
|
||
) -> list[dict]:
|
||
...
|
||
@overload
|
||
def get_ancestors(
|
||
self,
|
||
id_or_path: IDOrPathType = "",
|
||
/,
|
||
pid: None | int = None,
|
||
*,
|
||
async_: Literal[True],
|
||
) -> Coroutine[Any, Any, list[dict]]:
|
||
...
|
||
def get_ancestors(
|
||
self,
|
||
id_or_path: IDOrPathType = "",
|
||
/,
|
||
pid: None | int = None,
|
||
*,
|
||
async_: Literal[False, True] = False,
|
||
) -> list[dict] | Coroutine[Any, Any, list[dict]]:
|
||
"获取各个上级目录的少量信息（从根目录到当前目录）"
|
||
def gen_step():
|
||
attr = yield partial(self.attr, id_or_path, pid=pid, async_=async_)
|
||
return deepcopy(attr["ancestors"])
|
||
return run_gen_step(gen_step, async_=async_)
|
||
|
||
@overload
|
||
def get_url(
|
||
self,
|
||
id_or_path: IDOrPathType,
|
||
/,
|
||
pid: None | int = None,
|
||
headers: None | Mapping = None,
|
||
*,
|
||
async_: Literal[False] = False,
|
||
) -> P115URL:
|
||
...
|
||
@overload
|
||
def get_url(
|
||
self,
|
||
id_or_path: IDOrPathType,
|
||
/,
|
||
pid: None | int = None,
|
||
headers: None | Mapping = None,
|
||
*,
|
||
async_: Literal[True],
|
||
) -> Coroutine[Any, Any, P115URL]:
|
||
...
|
||
def get_url(
|
||
self,
|
||
id_or_path: IDOrPathType,
|
||
/,
|
||
pid: None | int = None,
|
||
headers: None | Mapping = None,
|
||
*,
|
||
async_: Literal[False, True] = False,
|
||
) -> P115URL | Coroutine[Any, Any, P115URL]:
|
||
"获取下载链接"
|
||
def gen_step():
|
||
if isinstance(id_or_path, int):
|
||
id = id_or_path
|
||
else:
|
||
attr = yield partial(self.attr, id_or_path, pid=pid, async_=async_)
|
||
if attr["is_directory"]:
|
||
raise IsADirectoryError(errno.EISDIR, f"{attr['path']!r} (id={attr['id']!r}) is a directory")
|
||
id = attr["id"]
|
||
return (yield partial(
|
||
self.client.share_download_url,
|
||
{
|
||
"share_code": self.share_code,
|
||
"receive_code": self.receive_code,
|
||
"file_id": id,
|
||
},
|
||
headers=headers,
|
||
use_web_api=attr.get("violated", False) and attr["size"] < 1024 * 1024 * 115,
|
||
request=self.async_request if async_ else self.request,
|
||
async_=async_,
|
||
))
|
||
return run_gen_step(gen_step, async_=async_)
|
||
|
||
@overload
|
||
def iterdir(
|
||
self,
|
||
id_or_path: IDOrPathType = "",
|
||
/,
|
||
pid: None | int = None,
|
||
start: int = 0,
|
||
stop: None | int = None,
|
||
page_size: int = 1_000,
|
||
refresh: bool = False,
|
||
*,
|
||
async_: Literal[False] = False,
|
||
**payload,
|
||
) -> Iterator[AttrDict]:
|
||
...
|
||
@overload
|
||
def iterdir(
|
||
self,
|
||
id_or_path: IDOrPathType = "",
|
||
/,
|
||
pid: None | int = None,
|
||
start: int = 0,
|
||
stop: None | int = None,
|
||
page_size: int = 1_000,
|
||
refresh: bool = False,
|
||
*,
|
||
async_: Literal[True],
|
||
**payload,
|
||
) -> AsyncIterator[AttrDict]:
|
||
...
|
||
def iterdir(
|
||
self,
|
||
id_or_path: IDOrPathType = "",
|
||
/,
|
||
pid: None | int = None,
|
||
start: int = 0,
|
||
stop: None | int = None,
|
||
page_size: int = 1_000,
|
||
refresh: bool = False,
|
||
*,
|
||
async_: Literal[False, True] = False,
|
||
**payload,
|
||
) -> Iterator[AttrDict] | AsyncIterator[AttrDict]:
|
||
"""迭代获取目录内直属的文件或目录的信息
|
||
:param payload:
|
||
- limit: int = 32
|
||
- offset: int = 0
|
||
- asc: 0 | 1 = <default> # 是否升序排列
|
||
- o: str = <default>
|
||
# 用某字段排序：
|
||
# - 文件名："file_name"
|
||
# - 文件大小："file_size"
|
||
# - 文件种类："file_type"
|
||
# - 修改时间："user_utime"
|
||
# - 创建时间："user_ptime"
|
||
# - 上次打开时间："user_otime"
|
||
"""
|
||
path_class = type(self).path_class
|
||
if page_size <= 0:
|
||
page_size = 1_000
|
||
share_code = self.share_code
|
||
receive_code = self.receive_code
|
||
def gen_step():
|
||
nonlocal start, stop
|
||
if stop is not None and (start >= 0 and stop >= 0 or start < 0 and stop < 0) and start >= stop:
|
||
return ()
|
||
if isinstance(id_or_path, int):
|
||
attr = yield partial(self._attr, id_or_path, async_=async_)
|
||
elif isinstance(id_or_path, AttrDict):
|
||
attr = id_or_path
|
||
elif isinstance(id_or_path, path_class):
|
||
attr = id_or_path.attr
|
||
else:
|
||
attr = yield partial(
|
||
self._attr_path,
|
||
id_or_path,
|
||
pid=pid,
|
||
ensure_dir=True,
|
||
async_=async_,
|
||
)
|
||
#if not attr["is_directory"]:
|
||
try:
|
||
isDir = attr["is_directory"]
|
||
except:
|
||
attr=attr()
|
||
isDir = attr["is_directory"]
|
||
if not isDir:
|
||
raise NotADirectoryError(
|
||
errno.ENOTDIR,
|
||
f"{attr['path']!r} (id={attr['id']!r}) is not a directory",
|
||
)
|
||
id = attr["id"]
|
||
ancestors = attr["ancestors"]
|
||
children: Sequence[AttrDict]
|
||
try:
|
||
if refresh:
|
||
raise KeyError
|
||
children = self.pid_to_children[id]
|
||
except KeyError:
|
||
payload["cid"] = id
|
||
payload["limit"] = page_size
|
||
offset = int(payload.setdefault("offset", 0))
|
||
if offset < 0:
|
||
offset = payload["offset"] = 0
|
||
else:
|
||
payload["offset"] = 0
|
||
dirname = attr["path"]
|
||
get_files = self.fs_files
|
||
path_to_id = self.path_to_id
|
||
ls: list[AttrDict] = []
|
||
add = ls.append
|
||
resp = yield partial(get_files, payload, async_=async_)
|
||
|
||
try:
|
||
data = resp["data"]
|
||
except:
|
||
resp=resp()
|
||
data = resp["data"]
|
||
for attr in map(normalize_attr, data["list"]):
|
||
attr["ancestors"] = [*ancestors, {"id": attr["id"], "name": attr["name"]}]
|
||
path = attr["path"] = joinpath(dirname, escape(attr["name"]))
|
||
attr["share_code"] = share_code
|
||
attr["receive_code"] = receive_code
|
||
path_to_id[path + "/"[:attr["is_directory"]]] = attr["id"]
|
||
add(attr)
|
||
for _ in range((data["count"] - 1) // page_size):
|
||
payload["offset"] += page_size
|
||
resp = yield partial(get_files, payload, async_=async_)
|
||
try:
|
||
data = resp["data"]
|
||
except:
|
||
resp=resp()
|
||
data = resp["data"]
|
||
for attr in map(normalize_attr, data["list"]):
|
||
attr["ancestors"] = [*ancestors, {"id": attr["id"], "name": attr["name"]}]
|
||
path = attr["path"] = joinpath(dirname, escape(attr["name"]))
|
||
attr["share_code"] = share_code
|
||
attr["receive_code"] = receive_code
|
||
path_to_id[path + "/"[:attr["is_directory"]]] = attr["id"]
|
||
add(attr)
|
||
children = self.pid_to_children[id] = tuple(ls)
|
||
self.id_to_attr.update((attr["id"], attr) for attr in children)
|
||
else:
|
||
count = len(children)
|
||
if start < 0:
|
||
start += count
|
||
if start < 0:
|
||
start = 0
|
||
if stop is None:
|
||
stop = count
|
||
elif stop < 0:
|
||
stop += count
|
||
if start >= stop or stop <= 0 or start >= count:
|
||
return ()
|
||
key: None | Callable
|
||
match payload.get("o"):
|
||
case "file_name":
|
||
key = lambda attr: attr["name"]
|
||
case "file_size":
|
||
key = lambda attr: attr.get("size") or 0
|
||
case "file_type":
|
||
key = lambda attr: attr.get("ico", "folder" if attr["is_directory"] else "")
|
||
case "user_utime" | "user_ptime" | "user_otime":
|
||
key = lambda attr: attr["time"]
|
||
case _:
|
||
key = None
|
||
if key:
|
||
children = sorted(children, key=key, reverse=payload.get("asc", True))
|
||
return YieldFrom(children[start:stop])
|
||
return run_gen_step_iter(gen_step, may_call=False, async_=async_)
|
||
|
||
@overload
|
||
def receive(
|
||
self,
|
||
ids: int | str | Iterable[int | str],
|
||
/,
|
||
to_pid: int = 0,
|
||
*,
|
||
async_: Literal[False] = False,
|
||
) -> dict:
|
||
...
|
||
@overload
|
||
def receive(
|
||
self,
|
||
ids: int | str | Iterable[int | str],
|
||
/,
|
||
to_pid: int = 0,
|
||
*,
|
||
async_: Literal[True],
|
||
) -> Coroutine[Any, Any, dict]:
|
||
...
|
||
def receive(
|
||
self,
|
||
ids: int | str | Iterable[int | str],
|
||
/,
|
||
to_pid: int = 0,
|
||
*,
|
||
async_: Literal[False, True] = False,
|
||
) -> dict | Coroutine[Any, Any, dict]:
|
||
"""接收分享文件到网盘
|
||
:param ids: 要转存到文件 id（这些 id 归属分享链接）
|
||
:param to_pid: 你的网盘的一个目录 id（这个 id 归属你的网盘）
|
||
"""
|
||
def gen_step():
|
||
nonlocal ids
|
||
if isinstance(ids, int):
|
||
ids = str(ids)
|
||
elif isinstance(ids, Iterable):
|
||
ids = ",".join(map(str, ids))
|
||
if not ids:
|
||
raise ValueError("no id (to file) to receive")
|
||
payload = {
|
||
"share_code": self.share_code,
|
||
"receive_code": self.receive_code,
|
||
"file_id": ids,
|
||
"cid": to_pid,
|
||
}
|
||
return (yield partial(
|
||
self.client.share_receive,
|
||
payload,
|
||
request=self.async_request if async_ else self.request,
|
||
async_=async_,
|
||
))
|
||
return run_gen_step(gen_step, async_=async_)
|
||
|
||
@overload
|
||
def search(
|
||
self,
|
||
id_or_path: IDOrPathType = "",
|
||
/,
|
||
pid: None | int = None,
|
||
page_size: int = 1_000,
|
||
*,
|
||
async_: Literal[False] = False,
|
||
**payload,
|
||
) -> Iterator[P115SharePath]:
|
||
...
|
||
@overload
|
||
def search(
|
||
self,
|
||
id_or_path: IDOrPathType = "",
|
||
/,
|
||
pid: None | int = None,
|
||
page_size: int = 1_000,
|
||
*,
|
||
async_: Literal[True],
|
||
**payload,
|
||
) -> AsyncIterator[P115SharePath]:
|
||
...
|
||
def search(
|
||
self,
|
||
id_or_path: IDOrPathType = "",
|
||
/,
|
||
pid: None | int = None,
|
||
page_size: int = 1_000,
|
||
*,
|
||
async_: Literal[False, True] = False,
|
||
**payload,
|
||
) -> Iterator[P115SharePath] | AsyncIterator[P115SharePath]:
|
||
"""搜索目录
|
||
|
||
:param payload:
|
||
- share_code: str = <default> 💡 分享码
|
||
- receive_code: str = <default> 💡 接收码（即密码）
|
||
- limit: int = 32 💡 一页大小，意思就是 page_size
|
||
- offset: int = 0 💡 索引偏移，索引从 0 开始计算
|
||
- search_value: str = "." 💡 搜索文本，仅支持搜索文件名
|
||
- suffix: str = <default> 💡 文件后缀（扩展名），优先级高于 `type`
|
||
- type: int = <default> 💡 文件类型
|
||
|
||
- 0: 全部
|
||
- 1: 文档
|
||
- 2: 图片
|
||
- 3: 音频
|
||
- 4: 视频
|
||
- 5: 压缩包
|
||
- 6: 应用
|
||
- 7: 书籍
|
||
- 99: 仅文件
|
||
"""
|
||
if page_size <= 0:
|
||
page_size = 1_000
|
||
def gen_step():
|
||
attr = yield self.attr(id_or_path, pid=pid, async_=async_)
|
||
if attr["is_directory"]:
|
||
payload["cid"] = attr["id"]
|
||
else:
|
||
payload["cid"] = attr["parent_id"]
|
||
payload["limit"] = page_size
|
||
offset = int(payload.setdefault("offset", 0))
|
||
if offset < 0:
|
||
payload["offset"] = 0
|
||
search = self.fs_search
|
||
while True:
|
||
resp = yield search(payload, async_=async_)
|
||
ls = resp["data"]["list"]
|
||
if not ls:
|
||
return
|
||
for attr in ls:
|
||
attr = normalize_attr(attr)
|
||
yield Yield(P115SharePath(self, attr))
|
||
offset = payload["offset"] = offset + resp["page_size"]
|
||
if offset >= resp["count"] or offset >= 10_000:
|
||
break
|
||
if offset + page_size > 10_000:
|
||
payload["page_size"] = 10_000 - offset
|
||
return run_gen_step_iter(gen_step, may_call=False, async_=async_)
|
||
|
||
@overload
|
||
def stat(
|
||
self,
|
||
id_or_path: IDOrPathType = "",
|
||
/,
|
||
pid: None | int = None,
|
||
*,
|
||
async_: Literal[False] = False,
|
||
) -> stat_result:
|
||
...
|
||
@overload
|
||
def stat(
|
||
self,
|
||
id_or_path: IDOrPathType = "",
|
||
/,
|
||
pid: None | int = None,
|
||
*,
|
||
async_: Literal[True],
|
||
) -> Coroutine[Any, Any, stat_result]:
|
||
...
|
||
def stat(
|
||
self,
|
||
id_or_path: IDOrPathType = "",
|
||
/,
|
||
pid: None | int = None,
|
||
*,
|
||
async_: Literal[False, True] = False,
|
||
) -> stat_result | Coroutine[Any, Any, stat_result]:
|
||
"检查路径的属性，就像 `os.stat`"
|
||
def gen_step():
|
||
attr = yield partial(self.attr, id_or_path, pid=pid, async_=async_)
|
||
is_dir = attr["is_directory"]
|
||
timestamp: float = attr["timestamp"]
|
||
return stat_result((
|
||
(S_IFDIR if is_dir else S_IFREG) | 0o444, # mode
|
||
cast(int, attr["id"]), # ino
|
||
cast(int, attr["parent_id"]), # dev
|
||
1, # nlink
|
||
self.user_id, # uid
|
||
1, # gid
|
||
cast(int, 0 if is_dir else attr["size"]), # size
|
||
timestamp, # atime
|
||
timestamp, # mtime
|
||
timestamp, # ctime
|
||
))
|
||
return run_gen_step(gen_step, async_=async_)
|
||
|