app/device.py - master - drresist/geekmagic-api

app/device.py
195 строк4 KB
MFesenko
feat: rewrite to modules
14 май 2026, 21:27
14 май 2026, 21:2745efc4f
import time
from pathlib import Path
from typing import Optional

import requests

from .config import (
    DELETE_URL,
    ENABLE_REMOTE_CLEANUP,
    FILELIST_URL,
    GEEKMAGIC_BASE,
    HTTP_HEADERS,
    REQUEST_TIMEOUT,
    SET_URL,
    THEME_PARAM,
    UPLOAD_URL,
)


def request_get(
    url: str,
    params: Optional[dict] = None,
    retries: int = 3,
    response_limit: int = 5000,
) -> str:
    last_error = None

    for attempt in range(1, retries + 1):
        try:
            response = requests.get(
                url,
                params=params,
                headers=HTTP_HEADERS,
                timeout=REQUEST_TIMEOUT,
            )
            response.raise_for_status()
            return response.text[:response_limit]

        except requests.RequestException as exc:
            last_error = exc
            time.sleep(0.4 * attempt)

    raise RuntimeError(
        f"GET failed after {retries} retries: "
        f"url={url}, params={params}, error={last_error}"
    )


def request_post_file(
    url: str,
    file_path: Path,
    filename: str,
    retries: int = 3,
) -> tuple[int, str]:
    last_error = None

    for attempt in range(1, retries + 1):
        try:
            with file_path.open("rb") as f:
                files = {"file": (filename, f, "image/jpeg")}

                response = requests.post(
                    url,
                    files=files,
                    headers=HTTP_HEADERS,
                    timeout=REQUEST_TIMEOUT,
                )

                response.raise_for_status()
                return response.status_code, response.text[:1000]

        except requests.RequestException as exc:
            last_error = exc

            try:
                if wait_until_file_exists(filename, timeout_seconds=2.0):
                    return 0, f"Upload response failed, but file exists: {exc}"
            except Exception:
                pass

            time.sleep(0.7 * attempt)

    return 0, f"Upload failed after {retries} retries: {last_error}"


def set_theme(theme_id: int) -> str:
    return request_get(
        SET_URL,
        params={THEME_PARAM: theme_id},
        retries=3,
    )


def set_image(filename: str) -> str:
    return request_get(
        SET_URL,
        params={"img": f"/image//{filename}"},
        retries=3,
    )


def upload_image(file_path: Path, filename: str) -> tuple[int, str]:
    return request_post_file(
        UPLOAD_URL,
        file_path=file_path,
        filename=filename,
        retries=3,
    )


def get_filelist_raw() -> str:
    return request_get(
        FILELIST_URL,
        params={"_": int(time.time() * 1000)},
        retries=3,
        response_limit=50000,
    )


def remote_image_exists(filename: str) -> bool:
    try:
        response = requests.get(
            f"{GEEKMAGIC_BASE}/image//{filename}",
            headers=HTTP_HEADERS,
            timeout=REQUEST_TIMEOUT,
        )

        return response.status_code == 200 and len(response.content) > 0

    except requests.RequestException:
        return False


def wait_until_file_exists(filename: str, timeout_seconds: float = 12.0) -> bool:
    deadline = time.time() + timeout_seconds

    while time.time() < deadline:
        if remote_image_exists(filename):
            return True

        try:
            filelist = get_filelist_raw()
            if filename in filelist:
                return True
        except Exception:
            pass

        time.sleep(0.5)

    return False


def delete_remote_image(filename: str) -> dict:
    """
    Firmware endpoint:

    GET /delete?file=/image//filename.jpg
    """

    if not ENABLE_REMOTE_CLEANUP:
        return {
            "ok": False,
            "skipped": True,
            "reason": "remote cleanup disabled",
        }

    if not filename.startswith("notify_") or not filename.endswith(".jpg"):
        return {
            "ok": False,
            "skipped": True,
            "reason": f"refusing to delete non-notification file: {filename}",
        }

    try:
        response = requests.get(
            DELETE_URL,
            params={"file": f"/image//{filename}"},
            headers=HTTP_HEADERS,
            timeout=REQUEST_TIMEOUT,
        )

        response.raise_for_status()

        return {
            "ok": True,
            "url": response.url,
            "response": response.text[:300],
        }

    except requests.RequestException as exc:
        return {
            "ok": False,
            "filename": filename,
            "error": str(exc),
        }