large-order-fusion/modules/get_image.py

import os
import time
import requests
import hashlib
import datetime
from dotenv import load_dotenv
from urllib.parse import urlparse
from utils.logger_config import setup_logger

logger = setup_logger(__name__)

load_dotenv()

APP_KEY = os.getenv("IMAGE_GET_APP_KEY")
APP_SECRET = os.getenv("IMAGE_GET_APP_SECRET")


def get_product_image(part_number):
    # 基础参数设置
    base_url = "http://api01pim.gloria.com.cn"
    endpoint = "/pimImgApi/product/img/imgByPartNumber"
    
    # 1. 生成时间戳（格式：YYYY-MM-dd HH:mm:ss）
    timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    
    # 2. 生成签名（appKey + appSecret + timestamp 的 MD5）
    raw_sign = APP_KEY + APP_SECRET + timestamp
    sign = hashlib.md5(raw_sign.encode('utf-8')).hexdigest()
    
    # 3. 设置请求头
    headers = {
        "appKey": APP_KEY,
        "timestamp": timestamp,
        "sign": sign
    }
    
    # 4. 设置查询参数
    params = {"partNumber": part_number}
    
    try:
        # 5. 发送GET请求
        response = requests.get(
            url=f"{base_url}{endpoint}",
            headers=headers,
            params=params,
            timeout=10
        )
        
        # 6. 处理响应
        if response.status_code == 200:
            result = response.json()
            if result.get("code") == 200:
                return result.get("imgPath").replace("nullproductImg", "http://pim.gloria.com.cn/productImg")
            else:
                logger.info(f"接口返回错误: {result.get('msg')} (代码: {result.get('code')})")
                return None
        else:
            logger.info(f"HTTP错误: {response.status_code}")
            return None
            
    except Exception as e:
        logger.info(f"请求异常: {str(e)}")
        return None

def download_image(image_url, filename, save_dir="./data/sku_image/"):
    """
    根据图片 URL 下载图片到本地
    
    :param image_url: 图片的 URL
    :param save_dir: 保存图片的目录，默认为当前目录下的 "images" 文件夹
    :param filename: 自定义文件名（可选），如果不提供则从 URL 中提取
    :return: 下载图片的完整路径，如果失败返回 None
    """
    try:
        # 确保保存目录存在
        os.makedirs(save_dir, exist_ok=True)
        
        # 发送 HTTP 请求获取图片
        response = requests.get(image_url, stream=True, timeout=10)
        response.raise_for_status()  # 检查请求是否成功
        
        # 确定文件名
        filename += '.png'
        
        # 构建完整保存路径
        save_path = os.path.join(save_dir, filename)
        
        # 保存图片
        with open(save_path, 'wb') as f:
            for chunk in response.iter_content(chunk_size=8192):
                f.write(chunk)
        
        logger.info(f"图片已保存至: {save_path}")
        return save_path
    
    except requests.exceptions.RequestException as e:
        logger.info(f"下载图片失败: {e}")
        return None
    except Exception as e:
        logger.info(f"发生错误: {e}")
        return None

def get_sku_image(product_number):
    image_url = get_product_image(product_number)
    logger.info(f"获取到图片公链：{image_url}")
    save_path = download_image(image_url, product_number)

    return {
        "image_url": image_url,
        "image_path": save_path
    }

if __name__ == "__main__":
    # 替换为您的实际款号
    PRODUCT_NUMBER = "195A9M05J06D"
    
    # # 调用接口获取图片路径
    # image_url = get_product_image(PRODUCT_NUMBER)
    
    # if image_url:
    #     logger.info(f"✅ 商品图片获取成功: {image_url}")
    # else:
    #     logger.info("❌ 获取商品图片失败")

    # print(download_image(image_url, "test.png"))

    print(get_sku_image(PRODUCT_NUMBER))