108 lines
2.7 KiB
Python
108 lines
2.7 KiB
Python
import os
|
|
import uuid
|
|
import logging
|
|
import mimetypes
|
|
|
|
import requests
|
|
|
|
import config
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
IMAGE_MIME_TYPES = {
|
|
"image/jpeg",
|
|
"image/png",
|
|
"image/bmp",
|
|
"image/gif",
|
|
"image/webp",
|
|
"image/tiff",
|
|
}
|
|
|
|
EXT_MAP = {
|
|
"image/jpeg": ".jpg",
|
|
"image/png": ".png",
|
|
"image/bmp": ".bmp",
|
|
"image/gif": ".gif",
|
|
"image/webp": ".webp",
|
|
"image/tiff": ".tiff",
|
|
}
|
|
|
|
|
|
def _detect_ext(response) -> str:
|
|
content_type = response.headers.get("Content-Type", "").split(";")[0].strip().lower()
|
|
if content_type in EXT_MAP:
|
|
return EXT_MAP[content_type]
|
|
|
|
url_path = response.url.split("?")[0]
|
|
guessed, _ = mimetypes.guess_type(url_path)
|
|
if guessed and guessed in EXT_MAP:
|
|
return EXT_MAP[guessed]
|
|
|
|
return ".jpg"
|
|
|
|
|
|
def fetch_wallpaper() -> str:
|
|
cfg = config.load()
|
|
api = cfg["api"]
|
|
wallpaper_cfg = cfg["wallpaper"]
|
|
|
|
url = api["url"]
|
|
mode = api.get("mode", "direct")
|
|
timeout = api.get("timeout", 30)
|
|
|
|
cache_dir = wallpaper_cfg.get("cache_dir", "./cache")
|
|
cache_dir = _resolve_cache(cache_dir)
|
|
os.makedirs(cache_dir, exist_ok=True)
|
|
|
|
if mode == "json":
|
|
image_url = _fetch_json(url, api.get("json_image_field", "url"), timeout)
|
|
response = _download(image_url, timeout)
|
|
else:
|
|
response = _download(url, timeout)
|
|
|
|
ext = _detect_ext(response)
|
|
filename = f"{uuid.uuid4().hex}{ext}"
|
|
filepath = os.path.join(cache_dir, filename)
|
|
|
|
with open(filepath, "wb") as f:
|
|
f.write(response.content)
|
|
|
|
logger.info("Wallpaper saved to: %s", filepath)
|
|
max_cache = cfg["wallpaper"].get("max_cache_files", 10)
|
|
_cleanup_cache(cache_dir, keep=max_cache)
|
|
return filepath
|
|
|
|
|
|
def _fetch_json(url: str, field: str, timeout: int) -> str:
|
|
r = requests.get(url, timeout=timeout)
|
|
r.raise_for_status()
|
|
data = r.json()
|
|
image_url = data.get(field)
|
|
if not image_url:
|
|
raise ValueError(f"Field '{field}' not found in JSON response")
|
|
return image_url
|
|
|
|
|
|
def _download(url: str, timeout: int) -> requests.Response:
|
|
r = requests.get(url, timeout=timeout, stream=True)
|
|
r.raise_for_status()
|
|
return r
|
|
|
|
|
|
def _resolve_cache(cache_dir: str) -> str:
|
|
if os.path.isabs(cache_dir):
|
|
return cache_dir
|
|
base = os.path.dirname(os.path.abspath(__file__))
|
|
return os.path.normpath(os.path.join(base, cache_dir))
|
|
|
|
|
|
def _cleanup_cache(cache_dir: str, keep: int) -> None:
|
|
try:
|
|
files = [os.path.join(cache_dir, f) for f in os.listdir(cache_dir)]
|
|
files = [f for f in files if os.path.isfile(f)]
|
|
files.sort(key=os.path.getmtime, reverse=True)
|
|
for old in files[keep:]:
|
|
os.remove(old)
|
|
except Exception:
|
|
pass
|