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
19import calendar20import gzip21import hashlib22import os23import time24
25
26def scantree(path):27"""Recursively yield DirEntry objects for given directory."""28for entry in os.scandir(path):29if entry.is_dir(follow_symlinks=False):30yield from scantree(entry.path)31else:32yield entry33
34# https://www.tutorialspoint.com/how-to-delete-only-empty-folders-in-python
35def delete_empty_folders(root):36for dirpath, dirnames, filenames in os.walk(root, topdown=False):37for dirname in dirnames:38full_path = os.path.join(dirpath, dirname)39if not os.listdir(full_path):40#print(full_path)41os.rmdir(full_path)42
43
44
45def read_or_checksum(path, read : bool, checksum_name : str):46content = None47hexdigest = None48buffer_size : int = 1024*1024 # 1 Mb49with open(path, 'rb') as fd:50if read:51content = bytearray()52file_hash = None53if checksum_name is not None:54file_hash = hashlib.new(checksum_name)55while chunk := fd.read(buffer_size):56if checksum_name is not None:57file_hash.update(chunk)58content += chunk59if checksum_name is not None:60hexdigest = file_hash.hexdigest()61elif checksum_name is not None:62file_hash = hashlib.new(checksum_name)63while chunk := fd.read(buffer_size):64file_hash.update(chunk)65# 3.11+:66#file_hash = hashlib.file_digest(fd, checksum_name)67hexdigest = file_hash.hexdigest()68
69return ( hexdigest, content )70
71def unarchiving(path, data):72if path.endswith(".gz"):73return gzip.decompress(data)74else:75return data76
77
78def time_to_if_modified_since(st_mtime:float):79return time.strftime('%a, %d %b %Y %H:%M:%S GMT', time.gmtime(st_mtime))80
81def if_modified_since_to_time(raw : str):82return calendar.timegm(time.strptime(raw, "%a, %d %b %Y %H:%M:%S %Z"))83
84
85def url_concat(a, b):86sep = "/"87if len(a) > 0 and a[-1] == '/':88sep = ""89return a + sep + b90
91
92