205 lines
7.9 KiB
Python
205 lines
7.9 KiB
Python
from datetime import timedelta
|
|
from typing import Any, Dict
|
|
from backup.file import File
|
|
from aiohttp.client_exceptions import ClientResponseError
|
|
from injector import inject, singleton
|
|
|
|
from ..config import Config, Setting
|
|
from ..exceptions import (BackupFolderInaccessible, BackupFolderMissingError,
|
|
GoogleDrivePermissionDenied, LogInToGoogleDriveError)
|
|
from ..time import Time
|
|
from .driverequests import DriveRequests
|
|
from ..logger import getLogger
|
|
|
|
logger = getLogger(__name__)
|
|
|
|
FOLDER_MIME_TYPE = 'application/vnd.google-apps.folder'
|
|
FOLDER_NAME = 'Home Assistant Backups'
|
|
FOLDER_CACHE_SECONDS = 60 * 31 # 31 minutes
|
|
|
|
|
|
@singleton
|
|
class FolderFinder():
|
|
@inject
|
|
def __init__(self, config: Config, time: Time, drive_requests: DriveRequests):
|
|
self.config = config
|
|
self.drivebackend: DriveRequests = drive_requests
|
|
self.time = time
|
|
|
|
# The cached folder id
|
|
self._folderId = None
|
|
|
|
# When the fodler id was last cached
|
|
self._folder_queryied_last = None
|
|
|
|
# These get set when an existing folder is found and should cause the UI to
|
|
# prompt for what to do about it.
|
|
self._existing_folder = None
|
|
self._use_existing = None
|
|
self._folder_details = None
|
|
|
|
def resolveExisting(self, val):
|
|
if self._existing_folder:
|
|
self._use_existing = val
|
|
else:
|
|
self._use_existing = None
|
|
|
|
def _isSharedDrive(self, folder):
|
|
driveId = folder.get("driveId", None)
|
|
return driveId and len(driveId) > 0
|
|
|
|
def currentIsSharedDrive(self):
|
|
return self._folder_details and self._isSharedDrive(self._folder_details)
|
|
|
|
async def get(self):
|
|
if self._existing_folder and self._use_existing is not None:
|
|
if self._use_existing:
|
|
await self.save(self._existing_folder)
|
|
else:
|
|
await self.create()
|
|
self._use_existing = None
|
|
if not self._folder_queryied_last or self._folder_queryied_last + timedelta(seconds=FOLDER_CACHE_SECONDS) < self.time.now():
|
|
try:
|
|
self._folderId = await self._readFolderId()
|
|
except (BackupFolderMissingError, BackupFolderInaccessible):
|
|
if not self.config.get(Setting.SPECIFY_BACKUP_FOLDER):
|
|
# Search for a folder, they may have created one before
|
|
self._existing_folder = await self._search()
|
|
if self._existing_folder:
|
|
self._folderId = self._existing_folder.get('id')
|
|
else:
|
|
# Create folder, since no other folder is available
|
|
await self.create()
|
|
else:
|
|
raise
|
|
self._folder_queryied_last = self.time.now()
|
|
return self._folderId
|
|
|
|
def getExisting(self):
|
|
return self._existing_folder
|
|
|
|
async def save(self, folder: Any) -> str:
|
|
if not isinstance(folder, str):
|
|
self._folder_details = folder
|
|
folder = folder.get('id')
|
|
else:
|
|
self._folder_details = None
|
|
logger.info("Saving backup folder: " + folder)
|
|
File.write(self.config.get(Setting.FOLDER_FILE_PATH), folder)
|
|
self._folderId = folder
|
|
self._folder_queryied_last = self.time.now()
|
|
self._existing_folder = None
|
|
|
|
def reset(self):
|
|
if File.exists(self.config.get(Setting.FOLDER_FILE_PATH)):
|
|
File.delete(self.config.get(Setting.FOLDER_FILE_PATH))
|
|
self._folderId = None
|
|
self._folder_queryied_last = None
|
|
self._existing_folder = None
|
|
|
|
def getCachedFolder(self):
|
|
return self._folderId
|
|
|
|
def deCache(self):
|
|
self._folderId = None
|
|
self._folder_queryied_last = None
|
|
|
|
async def _readFolderId(self) -> str:
|
|
# First, check if we cached the drive folder
|
|
if not File.exists(self.config.get(Setting.FOLDER_FILE_PATH)):
|
|
raise BackupFolderMissingError()
|
|
else:
|
|
folder_id: str = File.read(self.config.get(Setting.FOLDER_FILE_PATH)).strip()
|
|
if await self._verify(folder_id):
|
|
return folder_id
|
|
else:
|
|
raise BackupFolderInaccessible(folder_id)
|
|
|
|
async def _search(self) -> str:
|
|
folders = []
|
|
|
|
try:
|
|
async for child in self.drivebackend.query("mimeType='" + FOLDER_MIME_TYPE + "'"):
|
|
if self._isValidFolder(child):
|
|
folders.append(child)
|
|
except ClientResponseError as e:
|
|
# 404 means the folder doesn't exist (maybe it got moved?)
|
|
if e.status == 404:
|
|
"Make Error"
|
|
raise LogInToGoogleDriveError()
|
|
else:
|
|
raise e
|
|
|
|
if len(folders) == 0:
|
|
return None
|
|
|
|
folders.sort(key=lambda c: Time.parse(c.get("modifiedTime")))
|
|
# Found a folder, which means we're probably using the add-on from a
|
|
# previous (or duplicate) installation. Record and return the id but don't
|
|
# persist it until the user chooses to do so.
|
|
folder = folders[len(folders) - 1]
|
|
logger.info("Found " + folder.get('name'))
|
|
return folder
|
|
|
|
async def _verify(self, id):
|
|
if self.drivebackend.isCustomCreds():
|
|
# If the user is using custom creds and specifying the backup folder, then chances are the
|
|
# app doesn't have permission to access the parent folder directly. Ironically, we can still
|
|
# query for children and add/remove backups. Not a huge deal, just
|
|
# means we can't verify the folder still exists, isn't trashed, etc. Just let it be valid
|
|
# and handle potential errors elsewhere.
|
|
return True
|
|
# Query drive for the folder to make sure it still exists and we have the right permission on it.
|
|
try:
|
|
folder = await self.drivebackend.get(id)
|
|
if not self._isValidFolder(folder):
|
|
logger.info("Provided backup folder {0} is invalid".format(id))
|
|
return False
|
|
self._folder_details = folder
|
|
return True
|
|
except ClientResponseError as e:
|
|
if e.status == 404:
|
|
# 404 means the folder doesn't exist (maybe it got moved?) but can also mean that we
|
|
# just don't have permission to see the folder. Often we can still upload into it, so just
|
|
# let it pass without further verification and let other error handling (on upload) identify problems.
|
|
return True
|
|
else:
|
|
raise e
|
|
except GoogleDrivePermissionDenied:
|
|
# Lost permission on the backup folder
|
|
return False
|
|
|
|
def _isValidFolder(self, folder) -> bool:
|
|
try:
|
|
caps = folder.get('capabilities')
|
|
if folder.get('trashed'):
|
|
return False
|
|
elif not caps['canAddChildren']:
|
|
return False
|
|
elif not caps['canListChildren']:
|
|
return False
|
|
elif not caps.get('canDeleteChildren', False) and not caps.get('canRemoveChildren', False):
|
|
if self._isSharedDrive(folder) and caps.get("canTrashChildren", False):
|
|
# Allow folders in shared drives if you can still trash items inside it.
|
|
return True
|
|
return False
|
|
elif folder.get("mimeType") != FOLDER_MIME_TYPE:
|
|
return False
|
|
except Exception:
|
|
return False
|
|
return True
|
|
|
|
async def create(self) -> str:
|
|
logger.info('Creating folder "{}" in "My Drive"'.format(FOLDER_NAME))
|
|
file_metadata: Dict[str, str] = {
|
|
'name': FOLDER_NAME,
|
|
'mimeType': FOLDER_MIME_TYPE,
|
|
'appProperties': {
|
|
"backup_folder": "true",
|
|
},
|
|
}
|
|
folder = await self.drivebackend.createFolder(file_metadata)
|
|
self._folder_details = folder
|
|
await self.save(folder)
|
|
return folder.get('id')
|