fusselkater/sync_kiwix
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

run black

Browse source
This commit is contained in:
Jan-Marten Brüggemann 2024-09-14 02:48:29 +02:00
parent 12be2f9c45
commit 27675b60d1
1 changed files with 84 additions and 37 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

103
sync_kiwix.py
View file

@ -14,41 +14,63 @@ from tqdm import tqdm
KIWIX_BASE_URL = "https://download.kiwix.org/zim/" KIWIX_BASE_URL = "https://download.kiwix.org/zim/"
class CacheException(Exception): class CacheException(Exception):
pass pass
class NoMatchException(Exception): class NoMatchException(Exception):
def __init__(self, message: str, wiki: str): def __init__(self, message: str, wiki: str):
super().__init__(message) super().__init__(message)
self.wiki = wiki self.wiki = wiki
class DownloadException(Exception): class DownloadException(Exception):
pass pass
def parse_args() -> argparse.Namespace: def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser( parser = argparse.ArgumentParser(
prog="sync_kiwix", prog="sync_kiwix",
description="Synchronize zim files for kiwix", description="Synchronize zim files for kiwix",
) )
parser.add_argument('wiki', type=str, nargs="+", help="Wikis to synchronize") parser.add_argument("wiki", type=str, nargs="+", help="Wikis to synchronize")
parser.add_argument('destination', type=str, help="Destination directory") parser.add_argument("destination", type=str, help="Destination directory")
parser.add_argument('--cache', action=argparse.BooleanOptionalAction, default=True, help="Use filelist cache (default: yes)") parser.add_argument(
parser.add_argument('--cache-file', type=str, default=os.path.join(os.getcwd(), ".cache"), help="Path to cache file (default: ./.cache)") "--cache",
parser.add_argument('--cache-max-age', type=int, default=1, help="Maximum age of the cached file list in days") action=argparse.BooleanOptionalAction,
default=True,
help="Use filelist cache (default: yes)",
)
parser.add_argument(
"--cache-file",
type=str,
default=os.path.join(os.getcwd(), ".cache"),
help="Path to cache file (default: ./.cache)",
)
parser.add_argument(
"--cache-max-age",
type=int,
default=1,
help="Maximum age of the cached file list in days",
)
return parser.parse_args() return parser.parse_args()
def load_cache(cache_file: str) -> list: def load_cache(cache_file: str) -> list:
try: try:
with open(file=cache_file, mode='r') as file: with open(file=cache_file, mode="r") as file:
return json.load(file) return json.load(file)
except json.decoder.JSONDecodeError: except json.decoder.JSONDecodeError:
raise CacheException('Could not decode JSON') raise CacheException("Could not decode JSON")
def save_cache(cache_file: str, zim_list: list): def save_cache(cache_file: str, zim_list: list):
with open(file=cache_file, mode='w') as file: with open(file=cache_file, mode="w") as file:
json.dump(zim_list, file) json.dump(zim_list, file)
def retreive_zim_list(session: requests.Session, url: str) -> list: def retreive_zim_list(session: requests.Session, url: str) -> list:
try: try:
response = session.get(url) response = session.get(url)
@ -66,20 +88,23 @@ def retreive_zim_list(session: requests.Session, url: str) -> list:
for file in files: for file in files:
file_base, file_extension = os.path.splitext(file) file_base, file_extension = os.path.splitext(file)
file_base_without_date, file_date = file_base.rsplit('_', 1) file_base_without_date, file_date = file_base.rsplit("_", 1)
result.append({ result.append(
"url": urljoin(url, file), {
"url_base": url, "url": urljoin(url, file),
"file": file, "url_base": url,
"file_base": file_base, "file": file,
"file_date": file_date, "file_base": file_base,
"file_extension": file_extension, "file_date": file_date,
"file_base_without_date": file_base_without_date "file_extension": file_extension,
}) "file_base_without_date": file_base_without_date,
}
)
return result return result
def print_fuzzy_matches(wiki: str, zim_list: list): def print_fuzzy_matches(wiki: str, zim_list: list):
print(f"Here is a list of zim files similar to your request:", file=sys.stderr) print(f"Here is a list of zim files similar to your request:", file=sys.stderr)
@ -87,6 +112,7 @@ def print_fuzzy_matches(wiki: str, zim_list: list):
if wiki in zim["url"]: if wiki in zim["url"]:
print(f" - {zim['file_base_without_date']} ({zim['url']})", file=sys.stderr) print(f" - {zim['file_base_without_date']} ({zim['url']})", file=sys.stderr)
def generate_download_list(wikis: list, zim_list: list) -> list: def generate_download_list(wikis: list, zim_list: list) -> list:
result = list() result = list()
@ -99,7 +125,9 @@ def generate_download_list(wikis: list, zim_list: list) -> list:
current_zim = zim current_zim = zim
continue continue
current_zim_date = datetime.datetime.strptime(current_zim["file_date"], "%Y-%m") current_zim_date = datetime.datetime.strptime(
current_zim["file_date"], "%Y-%m"
)
zim_date = datetime.datetime.strptime(zim["file_date"], "%Y-%m") zim_date = datetime.datetime.strptime(zim["file_date"], "%Y-%m")
if zim_date > current_zim_date: if zim_date > current_zim_date:
current_zim = zim current_zim = zim
@ -111,6 +139,7 @@ def generate_download_list(wikis: list, zim_list: list) -> list:
return result return result
def check_is_newer(path: str, last_modified: str) -> bool: def check_is_newer(path: str, last_modified: str) -> bool:
try: try:
mtime_local = datetime.datetime.fromtimestamp(os.path.getmtime(path)) mtime_local = datetime.datetime.fromtimestamp(os.path.getmtime(path))
@ -120,6 +149,7 @@ def check_is_newer(path: str, last_modified: str) -> bool:
mtime_remote = datetime.datetime.strptime(last_modified, "%a, %d %b %Y %H:%M:%S %Z") mtime_remote = datetime.datetime.strptime(last_modified, "%a, %d %b %Y %H:%M:%S %Z")
return mtime_remote.date() > mtime_local.date() return mtime_remote.date() > mtime_local.date()
def download_zim(session: requests.Session, zim: dict, dst: str): def download_zim(session: requests.Session, zim: dict, dst: str):
# Get sha256 digest from server # Get sha256 digest from server
@ -137,7 +167,9 @@ def download_zim(session: requests.Session, zim: dict, dst: str):
digest_value = base64.b64decode(data) digest_value = base64.b64decode(data)
break break
else: else:
raise DownloadException(f"Could not get SHA-256 digest for {zim['file']}. Aborting.") raise DownloadException(
f"Could not get SHA-256 digest for {zim['file']}. Aborting."
)
sha256 = hashlib.sha256() sha256 = hashlib.sha256()
try: try:
@ -145,13 +177,20 @@ def download_zim(session: requests.Session, zim: dict, dst: str):
response.raise_for_status() response.raise_for_status()
if not check_is_newer(dst, response.headers.get("Last-Modified")): if not check_is_newer(dst, response.headers.get("Last-Modified")):
print(f"{zim['file_base_without_date']} was not updated on server. Skipping...") print(
f"{zim['file_base_without_date']} was not updated on server. Skipping..."
)
return return
total_size = int(response.headers.get("content-length", 0)) total_size = int(response.headers.get("content-length", 0))
block_size = 1024 block_size = 1024
with tqdm(total=total_size, unit="B", unit_scale=True, desc=zim['file_base_without_date']) as progress_bar: with tqdm(
with open(dst + '.part', "wb") as file: total=total_size,
unit="B",
unit_scale=True,
desc=zim["file_base_without_date"],
) as progress_bar:
with open(dst + ".part", "wb") as file:
for block in response.iter_content(block_size): for block in response.iter_content(block_size):
progress_bar.update(len(block)) progress_bar.update(len(block))
file.write(block) file.write(block)
@ -161,13 +200,16 @@ def download_zim(session: requests.Session, zim: dict, dst: str):
raise DownloadException(str(err)) raise DownloadException(str(err))
if digest_value != sha256.digest(): if digest_value != sha256.digest():
raise DownloadException(f"Checksum Error for {zim['file_base_without_date']}. Aborting.") raise DownloadException(
f"Checksum Error for {zim['file_base_without_date']}. Aborting."
)
sha256_hex = sha256.hexdigest() sha256_hex = sha256.hexdigest()
with open(dst + '.sha256sum', 'w') as file: with open(dst + ".sha256sum", "w") as file:
file.write(f"{sha256_hex} {dst}\n") file.write(f"{sha256_hex} {dst}\n")
os.rename(dst + '.part', dst) os.rename(dst + ".part", dst)
def main(): def main():
args = parse_args() args = parse_args()
@ -180,7 +222,9 @@ def main():
if args.cache == True: if args.cache == True:
try: try:
mtime = datetime.datetime.fromtimestamp(os.path.getmtime(args.cache_file)) mtime = datetime.datetime.fromtimestamp(os.path.getmtime(args.cache_file))
if datetime.datetime.now() - mtime < datetime.timedelta(days=args.cache_max_age): if datetime.datetime.now() - mtime < datetime.timedelta(
days=args.cache_max_age
):
zim_list = load_cache(cache_file=args.cache_file) zim_list = load_cache(cache_file=args.cache_file)
except (FileNotFoundError, CacheException): except (FileNotFoundError, CacheException):
pass pass
@ -205,10 +249,13 @@ def main():
# Download zim files # Download zim files
try: try:
for zim in download_list: for zim in download_list:
dst_path = os.path.join(args.destination, zim["file_base_without_date"] + zim["file_extension"]) dst_path = os.path.join(
args.destination, zim["file_base_without_date"] + zim["file_extension"]
)
download_zim(session=session, zim=zim, dst=dst_path) download_zim(session=session, zim=zim, dst=dst_path)
except DownloadException as err: except DownloadException as err:
raise SystemExit(str(err)) raise SystemExit(str(err))
if __name__ == "__main__": if __name__ == "__main__":
main() main()