Skip to content

Commit

Permalink
0.30.6:
Browse files Browse the repository at this point in the history
重构path_lib以支持3.12
将AlistPath* 添加到 __init__.py 中。
  • Loading branch information
lee-cq committed Feb 25, 2024
1 parent 6fe5d66 commit b45fb6e
Show file tree
Hide file tree
Showing 7 changed files with 1,930 additions and 192 deletions.
1 change: 1 addition & 0 deletions alist_sdk/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,3 +3,4 @@
from .models import *
from .err import *
from .version import __version__
from .path_lib import AlistPath, PureAlistPath, AlistServer, login_server
51 changes: 51 additions & 0 deletions alist_sdk/alistpath.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,51 @@
#!/usr/bin/python3
# -*- coding: utf-8 -*-
"""
@File Name : alistpath.py
@Author : LeeCQ
@Date-Time : 2024/2/25 12:40
"""
import os
import posixpath

from posixpath import *

__all__ = [*posixpath.__all__, "splitroot"]


def splitroot(p):
"""Split a pathname into drive, root and tail. On Posix, drive is always
empty; the root may be empty, a single slash, or two slashes. The tail
contains anything after the root. For example:
splitroot('foo/bar') == ('', '', 'foo/bar')
splitroot('/foo/bar') == ('', '/', 'foo/bar')
splitroot('//foo/bar') == ('', '//', 'foo/bar')
splitroot('///foo/bar') == ('', '/', '//foo/bar')
splitroot('http://server/path/to/file') == ('http://server', '/', 'path/to/file')
"""
p = os.fspath(p)
if isinstance(p, bytes):
sep = b"/"
empty = b""
else:
sep = "/"
empty = ""

if p.startswith("http://") or p.startswith("https://"):
# Absolute path, e.g.: 'http://server/path/to/file'
return "/".join(p.split("/", 3)[:3]), "/", p.split("/", 3)[-1]

elif p[:1] != sep:
# Relative path, e.g.: 'foo'
return empty, empty, p

elif p[1:2] != sep or p[2:3] == sep:
# Absolute path, e.g.: '/foo', '///foo', '////foo', etc.
return empty, sep, p[1:]

else:
# Precisely two leading slashes, e.g.: '//foo'. Implementation defined per POSIX, see
# https://pubs.opengroup.org/onlinepubs/9699919799/basedefs/V1_chap04.html#tag_04_13
return empty, p[:2], p[2:]
200 changes: 20 additions & 180 deletions alist_sdk/path_lib.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,19 +2,14 @@
像使用Path一样的易于使用Alist中的文件
"""
import fnmatch
import os
import posixpath
import re
import sys
from pathlib import PurePosixPath, PurePath

from functools import lru_cache, cached_property
from urllib.parse import quote_from_bytes as urlquote_from_bytes
from typing import Iterator

from pydantic import BaseModel
from httpx import URL

from alist_sdk import alistpath
from alist_sdk.py312_pathlib import PurePosixPath
from alist_sdk import AlistError, RawItem
from alist_sdk.client import Client

Expand All @@ -29,12 +24,12 @@ class AlistServer(BaseModel):


def login_server(
server: str,
token=None,
username=None,
password=None,
has_opt=False,
**kwargs,
server: str,
token=None,
username=None,
password=None,
has_opt=False,
**kwargs,
):
""""""
if token is None:
Expand All @@ -53,173 +48,18 @@ def login_server(
)


# noinspection PyMethodMayBeStatic
class _AlistFlavour:
sep = "/"
altsep = ""
has_drv = True
pathmod = posixpath

is_supported = True

def __init__(self):
self.join = self.sep.join

def parse_parts(self, parts):
parsed = []
sep = self.sep
altsep = self.altsep
drv = root = ""

it = reversed(parts)
for part in it:
if not part:
continue
if altsep:
part = part.replace(altsep, sep)
drv, root, rel = self.splitroot(part)
if sep in rel:
for x in reversed(rel.split(sep)):
if x and x != ".":
parsed.append(sys.intern(x))
else:
if rel and rel != ".":
parsed.append(sys.intern(rel))
if drv or root:
if not drv:
# If no drive is present, try to find one in the previous
# parts. This makes the result of parsing e.g.
# ("C:", "/", "a") reasonably intuitive.
# noinspection PyAssignmentToLoopOrWithParameter
for part in it:
if not part:
continue
if altsep:
part = part.replace(altsep, sep)
drv = self.splitroot(part)[0]
if drv:
break
break
if drv or root:
parsed.append(drv + root)
parsed.reverse()
return drv, root, parsed

def join_parsed_parts(self, drv, root, parts, drv2, root2, parts2):
"""
Join the two paths represented by the respective
(drive, root, parts) tuples. Return a new (drive, root, parts) tuple.
"""
if root2:
if not drv2 and drv:
return drv, root2, [drv + root2] + parts2[1:]
elif drv2:
if drv2 == drv or self.casefold(drv2) == self.casefold(drv):
# Same drive => second path is relative to the first
return drv, root, parts + parts2[1:]
else:
# Second path is non-anchored (common case)
return drv, root, parts + parts2
return drv2, root2, parts2

def splitroot(self, part, sep=sep):
if part and part[0] == sep:
stripped_part = part.lstrip(sep)
# According to POSIX path resolution:
# http://pubs.opengroup.org/onlinepubs/009695399/basedefs/xbd_chap04.html#tag_04_11
# "A pathname that begins with two successive slashes may be
# interpreted in an implementation-defined manner, although more
# than two leading slashes shall be treated as a single slash".
if len(part) - len(stripped_part) == 2:
return "", sep * 2, stripped_part
else:
return "", sep, stripped_part
else:
return "", "", part

def casefold(self, s):
return s

def casefold_parts(self, parts):
return parts

def compile_pattern(self, pattern):
return re.compile(fnmatch.translate(pattern)).fullmatch

def make_uri(self, path):
# We represent the path using the local filesystem encoding,
# for portability to other applications.
bpath = bytes(path)
return "file://" + urlquote_from_bytes(bpath)
class PureAlistPath(PurePosixPath):
_flavour = alistpath

def is_absolute(self):
"""True if the path is absolute (has both a root and, if applicable,
a drive)."""
if self._flavour is alistpath:
# ntpath.isabs() is defective - see GH-44626.
return bool(self.drive and self.root)

# noinspection PyProtectedMember,PyUnresolvedReferences
class PureAlistPath(PurePosixPath):
_flavour = _AlistFlavour()

@classmethod
def _parse_args(cls, args):
# This is useful when you don't want to create an instance, just
# canonicalize some constructor arguments.
parts = []
for a in args:
if isinstance(a, PurePath):
parts += a._parts
else:
a = os.fspath(a)
if isinstance(a, str):
# Force-cast str subclasses to str (issue #21127)
parts.append(str(a))
else:
raise TypeError(
"argument should be a str object or an os.PathLike "
"object returning str, not %r" % type(a)
)
return cls._flavour.parse_parts(parts)

@classmethod
def _from_parts(cls, args):
# We need to call _parse_args on the instance, to get the
# right flavour.
args = list(args)
self = object.__new__(cls)
if isinstance(args[0], str) and args[0].startswith("http"):
_u = URL(args[0])
server = (
f"{_u.scheme}://{_u.host}:{_u.port}".replace(":80", "")
.replace(":443", "")
.replace(":None", "")
)
args[0] = _u.path
elif isinstance(args[0], cls):
server = args[0].server
else:
server = ""

drv, root, parts = self._parse_args(args)
self._drv = drv or server
self._root = root
self._parts = parts
self.server = server
return self

@classmethod
def _from_parsed_parts(cls, drv, root, parts):
self = object.__new__(cls)
self._drv = drv
self._root = root
self._parts = parts
return self

def _make_child(self, args):
drv, root, parts = self._parse_args(args)
drv, root, parts = self._flavour.join_parsed_parts(
self._drv, self._root, self._parts, drv, root, parts
)
return self._from_parsed_parts(drv, root, parts)

def __new__(cls, *args):
return cls._from_parts(args)
return self._flavour.isabs(str(self))

def as_posix(self) -> str:
return str(self).replace(self.drive, "")
Expand Down Expand Up @@ -266,7 +106,7 @@ def stat(self) -> RawItem:
data = _raw.data
return data
if _raw.code == 500 and (
"object not found" in _raw.message or "storage not found" in _raw.message
"object not found" in _raw.message or "storage not found" in _raw.message
):
raise FileNotFoundError(_raw.message)
raise AlistError(_raw.message)
Expand Down Expand Up @@ -295,7 +135,7 @@ def iterdir(self) -> Iterator["AlistPath"]:
raise

for item in (
self._client.list_files(self.as_posix(), refresh=True).data.content or []
self._client.list_files(self.as_posix(), refresh=True).data.content or []
):
yield self.joinpath(item.name)

Expand Down
Loading

0 comments on commit b45fb6e

Please sign in to comment.