reposync

1
#
2
# RepoSync.Provider.Base
3
#
4
# Copyright (c) 2023-2024 Владислав Щапов aka Vladislav Shchapov <vladislav@shchapov.ru>
5
#
6
# Licensed under the Apache License, Version 2.0 (the "License");
7
# you may not use this file except in compliance with the License.
8
# You may obtain a copy of the License at
9
#
10
#     http://www.apache.org/licenses/LICENSE-2.0
11
#
12
# Unless required by applicable law or agreed to in writing, software
13
# distributed under the License is distributed on an "AS IS" BASIS,
14
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15
# See the License for the specific language governing permissions and
16
# limitations under the License.
17
#
18

19
import calendar
20
import gzip
21
import hashlib
22
import os
23
import time
24

25

26
def scantree(path):
27
    """Recursively yield DirEntry objects for given directory."""
28
    for entry in os.scandir(path):
29
        if entry.is_dir(follow_symlinks=False):
30
            yield from scantree(entry.path)
31
        else:
32
            yield entry
33

34
# https://www.tutorialspoint.com/how-to-delete-only-empty-folders-in-python
35
def delete_empty_folders(root):
36
   for dirpath, dirnames, filenames in os.walk(root, topdown=False):
37
      for dirname in dirnames:
38
         full_path = os.path.join(dirpath, dirname)
39
         if not os.listdir(full_path): 
40
            #print(full_path)
41
            os.rmdir(full_path)
42

43

44

45
def read_or_checksum(path, read : bool, checksum_name : str):
46
    content   = None
47
    hexdigest = None
48
    buffer_size : int = 1024*1024 # 1 Mb
49
    with open(path, 'rb') as fd:
50
        if read:
51
            content = bytearray()
52
            file_hash = None
53
            if checksum_name is not None:
54
                file_hash = hashlib.new(checksum_name)
55
            while chunk := fd.read(buffer_size):
56
                if checksum_name is not None:
57
                    file_hash.update(chunk)
58
                content += chunk
59
            if checksum_name is not None:
60
                hexdigest = file_hash.hexdigest()
61
        elif checksum_name is not None:
62
            file_hash = hashlib.new(checksum_name)
63
            while chunk := fd.read(buffer_size):
64
                file_hash.update(chunk)
65
            # 3.11+:
66
            #file_hash = hashlib.file_digest(fd, checksum_name)
67
            hexdigest = file_hash.hexdigest()  
68

69
    return ( hexdigest, content )
70

71
def unarchiving(path, data):
72
    if path.endswith(".gz"):
73
        return gzip.decompress(data)
74
    else:
75
        return data
76

77

78
def time_to_if_modified_since(st_mtime:float):
79
    return time.strftime('%a, %d %b %Y %H:%M:%S GMT', time.gmtime(st_mtime))
80

81
def if_modified_since_to_time(raw : str):
82
    return calendar.timegm(time.strptime(raw, "%a, %d %b %Y %H:%M:%S %Z"))
83

84

85
def url_concat(a, b):
86
    sep = "/"
87
    if len(a) > 0 and a[-1] == '/':
88
        sep = ""
89
    return a + sep + b
90

91

92