FastCopyFast/__init__.py at main · hansalemaos/FastCopyFast · GitHub

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
116
117
118
119
120
121
122
# based on this answer: https://stackoverflow.com/a/28129677/15096247
import os
import shutil
import sys
from typing import Union

config = sys.modules[__name__]
config.BUFFER_SIZE = 100 * 1024


class CTError(Exception):
    def __init__(self, errors):
        self.errors = errors


try:
    O_BINARY = os.O_BINARY
except:
    O_BINARY = 0
READ_FLAGS = os.O_RDONLY | O_BINARY
WRITE_FLAGS = os.O_WRONLY | os.O_CREAT | os.O_TRUNC | O_BINARY


def copyfile(src: str, dst: str, copystat: bool = False) -> bool:
    copyok = False
    try:
        fin = os.open(src, READ_FLAGS)
        stat = os.fstat(fin)
        fout = os.open(dst, WRITE_FLAGS, stat.st_mode)
        for x in iter(lambda: os.read(fin, config.BUFFER_SIZE), b""):
            os.write(fout, x)
        copyok = True
    finally:
        try:
            os.close(fin)
        except Exception:
            pass
        try:
            os.close(fout)
        except Exception:
            pass
    if copystat and copyok:
        try:
            shutil.copystat(src, dst)
            return True
        except Exception:
            return False
    if copyok:
        return True
    return False


def movefile(src: str, dst: str, copystat: bool = False) -> bool:
    copyok = False
    try:
        fin = os.open(src, READ_FLAGS)
        stat = os.fstat(fin)
        fout = os.open(dst, WRITE_FLAGS, stat.st_mode)
        for x in iter(lambda: os.read(fin, config.BUFFER_SIZE), b""):
            os.write(fout, x)
        copyok = True
    finally:
        try:
            os.close(fin)
        except Exception:
            pass
        try:
            os.close(fout)
        except Exception:
            pass
    if copystat and copyok:
        try:
            shutil.copystat(src, dst)
        except Exception:
            return False
    if copyok:
        try:
            os.remove(src)
            return True

        except Exception:
            return False
    return False


def copytree(
    src: str,
    dst: str,
    ignore: Union[list, type(None)] = None,
    symlinks: bool = False,
    copystat: bool = False,
    ignore_exceptions=True,
):
    if ignore is None:
        ignore = []
    names = os.listdir(src)

    if not os.path.exists(dst):
        os.makedirs(dst)
    errors = []
    for name in names:
        if name in ignore:
            continue
        srcname = os.path.join(src, name)
        dstname = os.path.join(dst, name)
        try:
            if symlinks and os.path.islink(srcname):
                linkto = os.readlink(srcname)
                os.symlink(linkto, dstname)
            elif os.path.isdir(srcname):
                copytree(srcname, dstname, ignore, symlinks)
            else:
                copyfile(srcname, dstname, copystat)

        except (IOError, os.error) as why:
            errors.append((srcname, dstname, str(why)))
        except CTError as err:
            errors.extend(err.errors)
            print(err)
    if errors:
        if not ignore_exceptions:
            raise CTError(errors)