1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
|
"""
What an upload will do, decided before anything is sent — shared by `put` and
the interactive mode, so both refuse the same names and the same collisions.
A name already in the destination folder, compared without case as the node and
the Files tab compare it, or a name the node does not accept (`names.py`), is
found here; what to do about it (refuse the run, skip it, ask) is the caller's.
"""
import os
import stat
from dataclasses import dataclass
from pathlib import Path
from meshbay_common.paths import fold, long_path
from ..errors import USAGE, CliError
from ..output import safe
from .names import same_name, upload_name_problem
from .tree import Tree
@dataclass
class UploadFile:
local: Path
rel: list[str] # remote parts below the destination folder
size: int
def plan_upload(sources: list[str], recursive: bool):
"""Folders to create (parents first) and files to send, as remote parts."""
dirs: list[list[str]] = []
files: list[UploadFile] = []
problems: list[str] = []
seen: dict[str, str] = {}
for src in sources:
path = Path(src).expanduser()
try:
st = os.stat(long_path(path))
except OSError as e:
problems.append(f"{src}: {e.strerror}")
continue
top = path.resolve().name if src.rstrip("/\\") in (".", "..") else path.name
key = fold(top)
if key in seen:
problems.append(f"{src} and {seen[key]} would have the same name there")
continue
seen[key] = src
if stat.S_ISREG(st.st_mode):
files.append(UploadFile(path, [top], st.st_size))
elif stat.S_ISDIR(st.st_mode):
if not recursive:
raise CliError(USAGE, f"{safe(src)} is a folder; use -R.")
for here, subdirs, names in _walk(path):
rel = [top] + list(Path(here).relative_to(path).parts)
dirs.append(rel)
for d in list(subdirs):
if os.path.islink(long_path(Path(here) / d)):
subdirs.remove(d)
problems.append(f"{Path(here) / d} is a link to a folder; not "
"followed")
for n in names:
p = Path(here) / n
try:
st_n = os.stat(long_path(p))
except OSError as e:
problems.append(f"{p}: {e.strerror}")
continue
if stat.S_ISREG(st_n.st_mode):
files.append(UploadFile(p, rel + [n], st_n.st_size))
subdirs.sort()
else:
problems.append(f"{src} is neither a file nor a folder")
return dirs, files, problems
def _walk(top: Path):
"""`os.walk` for a tree that may go past Windows' MAX_PATH: each folder is
listed under its long form (`long_path`, which only prefixes a long path, so
a short top alone would not do) and the paths given back stay plain."""
stack = [top]
while stack:
here = stack.pop()
try:
with os.scandir(long_path(here)) as it:
entries = list(it)
except OSError:
continue
subdirs, names = [], []
for e in entries:
try:
is_dir = e.is_dir()
except OSError:
is_dir = False
(subdirs if is_dir else names).append(e.name)
yield here, subdirs, names # the caller prunes and orders subdirs
stack.extend(here / d for d in reversed(subdirs))
def taken_names(tree: Tree, folder: str, names: list[str]) -> set[str]:
"""The names among `names` already in `folder`, compared without case."""
there = [i.name for i in tree.children(folder, include_unavailable=True)]
return {n for n in names if any(same_name(n, t) for t in there)}
def refused_names(file_names: list[str], dirs: list[list[str]]) -> list[str]:
"""Names the node would refuse — the client's copy of its rule."""
out = []
for name in file_names:
if (why := upload_name_problem(name)):
out.append(f"{name}: {why}")
for d in dirs:
if (why := upload_name_problem(d[-1], is_dir=True)):
out.append(f"{'/'.join(d)}/: {why}")
return out
|