mirror of
https://github.com/esphome/esphome.git
synced 2025-09-07 05:42:20 +01:00
Allow images to be downloaded from URLs (#5214)
Co-authored-by: guillempages <guillempages@users.noreply.github.com> Co-authored-by: Jesse Hills <3060199+jesserockz@users.noreply.github.com>
This commit is contained in:
committed by
GitHub
parent
3ac59180ab
commit
49c09afb87
75
esphome/external_files.py
Normal file
75
esphome/external_files.py
Normal file
@@ -0,0 +1,75 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from pathlib import Path
|
||||
import os
|
||||
from datetime import datetime
|
||||
import requests
|
||||
import esphome.config_validation as cv
|
||||
from esphome.core import CORE, TimePeriodSeconds
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
CODEOWNERS = ["@landonr"]
|
||||
|
||||
NETWORK_TIMEOUT = 30
|
||||
|
||||
IF_MODIFIED_SINCE = "If-Modified-Since"
|
||||
CACHE_CONTROL = "Cache-Control"
|
||||
CACHE_CONTROL_MAX_AGE = "max-age="
|
||||
CONTENT_DISPOSITION = "content-disposition"
|
||||
TEMP_DIR = "temp"
|
||||
|
||||
|
||||
def has_remote_file_changed(url, local_file_path):
|
||||
if os.path.exists(local_file_path):
|
||||
_LOGGER.debug("has_remote_file_changed: File exists at %s", local_file_path)
|
||||
try:
|
||||
local_modification_time = os.path.getmtime(local_file_path)
|
||||
local_modification_time_str = datetime.utcfromtimestamp(
|
||||
local_modification_time
|
||||
).strftime("%a, %d %b %Y %H:%M:%S GMT")
|
||||
|
||||
headers = {
|
||||
IF_MODIFIED_SINCE: local_modification_time_str,
|
||||
CACHE_CONTROL: CACHE_CONTROL_MAX_AGE + "3600",
|
||||
}
|
||||
response = requests.head(url, headers=headers, timeout=NETWORK_TIMEOUT)
|
||||
|
||||
_LOGGER.debug(
|
||||
"has_remote_file_changed: File %s, Local modified %s, response code %d",
|
||||
local_file_path,
|
||||
local_modification_time_str,
|
||||
response.status_code,
|
||||
)
|
||||
|
||||
if response.status_code == 304:
|
||||
_LOGGER.debug(
|
||||
"has_remote_file_changed: File not modified since %s",
|
||||
local_modification_time_str,
|
||||
)
|
||||
return False
|
||||
_LOGGER.debug("has_remote_file_changed: File modified")
|
||||
return True
|
||||
except requests.exceptions.RequestException as e:
|
||||
raise cv.Invalid(
|
||||
f"Could not check if {url} has changed, please check if file exists "
|
||||
f"({e})"
|
||||
)
|
||||
|
||||
_LOGGER.debug("has_remote_file_changed: File doesn't exists at %s", local_file_path)
|
||||
return True
|
||||
|
||||
|
||||
def is_file_recent(file_path: str, refresh: TimePeriodSeconds) -> bool:
|
||||
if os.path.exists(file_path):
|
||||
creation_time = os.path.getctime(file_path)
|
||||
current_time = datetime.now().timestamp()
|
||||
return current_time - creation_time <= refresh.total_seconds
|
||||
return False
|
||||
|
||||
|
||||
def compute_local_file_dir(domain: str) -> Path:
|
||||
base_directory = Path(CORE.data_dir) / domain
|
||||
base_directory.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
return base_directory
|
Reference in New Issue
Block a user