2023-06-21 17:39:33 +03:00
|
|
|
import logging
|
|
|
|
from os import makedirs, path
|
|
|
|
from pathlib import Path
|
|
|
|
from typing import List, Union
|
2023-03-20 13:03:03 +02:00
|
|
|
from zipfile import ZipFile
|
2022-08-11 12:27:17 +03:00
|
|
|
|
2023-03-20 13:03:03 +02:00
|
|
|
import aiofiles
|
2023-03-12 15:54:06 +02:00
|
|
|
|
2023-06-21 17:39:33 +03:00
|
|
|
logger = logging.getLogger(__name__)
|
2022-08-08 15:53:26 +03:00
|
|
|
|
2023-06-21 17:39:33 +03:00
|
|
|
USERS_WITH_CONTEXT: List[int] = []
|
2023-03-17 14:52:16 +02:00
|
|
|
|
2023-03-09 12:33:02 +02:00
|
|
|
|
2023-06-21 17:39:33 +03:00
|
|
|
async def extract_and_save(handle: ZipFile, filename: str, destpath: Union[str, Path]):
|
2023-03-20 13:03:03 +02:00
|
|
|
"""Extract and save file from archive
|
|
|
|
|
2023-06-21 17:39:33 +03:00
|
|
|
### Args:
|
|
|
|
* handle (`ZipFile`): ZipFile handler
|
|
|
|
* filename (`str`): File base name
|
|
|
|
* path (`Union[str, Path]`): Path where to store
|
2023-03-20 13:03:03 +02:00
|
|
|
"""
|
|
|
|
data = handle.read(filename)
|
2023-06-21 17:39:33 +03:00
|
|
|
filepath = path.join(str(destpath), filename)
|
2023-03-20 13:03:03 +02:00
|
|
|
try:
|
|
|
|
makedirs(path.dirname(filepath), exist_ok=True)
|
|
|
|
async with aiofiles.open(filepath, "wb") as fd:
|
|
|
|
await fd.write(data)
|
2023-06-21 17:39:33 +03:00
|
|
|
logger.debug("Unzipped %s", filename)
|
2023-03-20 13:03:03 +02:00
|
|
|
except IsADirectoryError:
|
|
|
|
makedirs(filepath, exist_ok=True)
|
|
|
|
except FileNotFoundError:
|
|
|
|
pass
|
|
|
|
return
|