CameraMachine/python/service/init_load_source.py

import asyncio
import json
import time
from settings import HLM_HOST
import aiohttp
import os
from utils.utils_func import get_md5, get_modified_time, compare_two_times, check_path

class init_load_source:

    async def load_source(self):
        await self.down_resouce()

    async def down_resouce(self):
        response_data = await self.get_update_file()
        if response_data:
            for relative_file_path, value in response_data.items():
                file_path = f"{os.getcwd()}/{relative_file_path}"
                if os.path.exists(file_path):
                    file_md5 = get_md5(file_path)
                    if file_md5 != value["file_md5"]:
                        print(f"md5 不同 开始下载：{file_path}")
                        await self.async_download_file(value["url"], file_path)
                    else:
                        pass
                else:
                    print(f"文件不存在 开始下载：{file_path}")
                    await self.async_download_file(value["url"], file_path)
        else:
            print("获取更新文件内容失败")

    async def get_update_file(self, type="client_camera", plugins_name="plugins_A"):
        """异步获取指定类型的插件文件更新信息

        Args:
            type (str, optional): 设备类型标识，默认为"client_camera"
            plugins_name (str, optional): 需要查询的插件名称，默认为"plugins_A"

        Returns:
            dict or None: 成功时返回插件文件的JSON数据，失败或未找到时返回None
        """
        url = HLM_HOST + "/api/openai/query_client_addons"
        params = {"type": type}
        try:
            async with aiohttp.ClientSession() as session:
                # 发送初始请求获取插件列表数据
                async with session.get(url, params=params, timeout=10) as response:
                    response.raise_for_status()
                    raw_data = await response.json()

                    if not raw_data or not raw_data.get("data"):
                        return None

                # 从插件列表中查找匹配的插件项
                item = next(
                    (item for item in raw_data["data"]["list"] if item["name"] == plugins_name),
                    None
                )
                if item:
                    # 根据插件项中的URL获取具体文件内容
                    url = item["url"]
                    async with session.get(url, timeout=10) as item_response:
                        item_response.raise_for_status()
                        return await item_response.json()
                return None
        except Exception as e:
            print(f"An error occurred: {e}")
            return None


    async def async_download_file(self, url: str, file_path: str):
        """异步下载文件"""
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as response:
                root_path, file_name = os.path.split(file_path)
                check_path(root_path)
                with open(file_path, 'wb') as f:
                    async for chunk in response.content.iter_chunked(1024):
                        f.write(chunk)