pythonTest/app.py

import requests
import json
import os
from urllib.parse import urlparse
from typing import Optional, Dict, Any
from PIL import Image
import io

token = "Bearer eyJhbGciOiJIUzI1NiJ9.eyJyb2xlIjoiQURNSU4iLCJ1c2VySWQiOjEsImlhdCI6MTc3MjYxOTA2NiwiZXhwIjoxNzczMjIzODY2fQ.IatRhEStT5IUXHsN8_tvxuijNFO0tNkIADYIV79dWMo"

def get_product_detail(mer_id: str, product_id: str) -> Optional[Dict[str, Any]]:
    global token
    """
    获取产品详情
    
    Args:
        mer_id: 商户ID
        product_id: 产品ID
        
    Returns:
        返回JSON格式的产品详情数据，如果请求失败则返回None
    """
    # 构建URL
    base_url = "https://api.tiki-int.site/api/product/detail"
    params = {
        "mer_id": mer_id,
        "product_id": product_id
    }
    
    # 设置请求头
    headers = {
        "accept": "application/json, text/plain, */*",
        "accept-language": "zh-CN,zh;q=0.9,en;q=0.8,zh-TW;q=0.7",
        "cache-control": "no-cache",
        "lang": "en",
        "pragma": "no-cache",
        "priority": "u=1, i",
        "sec-ch-ua": '"Not:A-Brand";v="99", "Google Chrome";v="145", "Chromium";v="145"',
        "sec-ch-ua-mobile": "?0",
        "sec-ch-ua-platform": '"Windows"',
        "sec-fetch-dest": "empty",
        "sec-fetch-mode": "cors",
        "sec-fetch-site": "cross-site",
        "Referer": "https://tiki.fashion11.top/",
        "Referrer-Policy": "strict-origin-when-cross-origin"
    }
    
    try:
        # 发送GET请求
        response = requests.get(
            url=base_url,
            params=params,
            headers=headers,
            timeout=10  # 设置10秒超时
        )
        
        # 检查响应状态
        response.raise_for_status()
        
        # 返回JSON数据
        return response.json()
        
    except requests.exceptions.RequestException as e:
        print(f"请求失败: {e}")
        return None
    except json.JSONDecodeError as e:
        print(f"JSON解析失败: {e}")
        return None


import os
from urllib.parse import urlparse
import requests
from PIL import Image
import io

def download_and_upload_image(image_url):
    """
    下载图片并上传到指定地址，webp格式自动转换为jpg
    
    Args:
        image_url: 图片的URL地址
    
    Returns:
        上传结果的JSON响应
    """
    # 默认上传地址
    upload_url = "https://www.as1688.vip/api/v1/file/upload"
    
    # 从URL中提取文件名和后缀
    parsed_url = urlparse(image_url)
    filename = os.path.basename(parsed_url.path)
    if not filename or '.' not in filename:
        filename = "image.jpg"
    
    # 获取文件后缀
    ext = os.path.splitext(filename)[1].lower()
    
    # 临时文件路径
    temp_file = f"temp_{filename}"
    
    try:
        # 1. 下载图片
        # print(f"正在下载图片: {image_url}")
        img_response = requests.get(image_url, timeout=30)
        img_response.raise_for_status()
        
        # 2. 处理webp格式
        need_convert = False
        if ext == '.webp':
            need_convert = True
            # print("检测到webp格式，正在转换为jpg...")
            
            # 使用PIL转换webp为jpg
            image = Image.open(io.BytesIO(img_response.content))
            
            # 如果是RGBA模式，转换为RGB（去除alpha通道）
            if image.mode == 'RGBA':
                # 创建白色背景
                rgb_image = Image.new('RGB', image.size, (255, 255, 255))
                # 粘贴webp图像，使用alpha通道作为掩码
                rgb_image.paste(image, mask=image.split()[3])
                image = rgb_image
            elif image.mode != 'RGB':
                image = image.convert('RGB')
            
            # 保存为jpg到临时文件
            image.save(temp_file, 'JPEG', quality=85)
            
            # 更新文件名和MIME类型
            base_name = os.path.splitext(filename)[0]
            filename = f"{base_name}.jpg"
            mime_type = 'image/jpeg'
            
            print(f"已转换为: {filename}")
        else:
            # 普通图片，直接保存
            with open(temp_file, 'wb') as f:
                f.write(img_response.content)
            
            # 设置MIME类型
            mime_types = {
                '.jpg': 'image/jpeg',
                '.jpeg': 'image/jpeg',
                '.png': 'image/png',
                '.gif': 'image/gif',
                '.bmp': 'image/bmp',
                '.ico': 'image/x-icon',
                '.svg': 'image/svg+xml'
            }
            mime_type = mime_types.get(ext, 'image/jpeg')
        
        # print(f"图片已保存到: {temp_file}")
        
        # 3. 准备上传
        headers = {
            "accept": "application/json",
            "accept-language": "zh-CN,zh;q=0.9",
            "authorization": token,
            "sec-ch-ua": '"Not:A-Brand";v="99", "Google Chrome";v="145", "Chromium";v="145"',
            "sec-ch-ua-mobile": "?0",
            "sec-ch-ua-platform": '"Windows"'
        }
        
        # 准备文件上传
        with open(temp_file, 'rb') as f:
            files = {
                'file': (filename, f, mime_type)
            }
            
            # print(f"正在上传到: {upload_url}")
            # print(f"文件名: {filename}, 类型: {mime_type}")
            
            upload_response = requests.post(
                upload_url,
                headers=headers,
                files=files,
                timeout=30
            )
            
            upload_response.raise_for_status()
            result = upload_response.json()
            print("上传成功!")
            return result
            
    except requests.exceptions.RequestException as e:
        print(f"网络请求错误: {e}")
        return {"error": str(e)}
    except ImportError:
        print("错误: 请安装PIL库: pip install Pillow")
        return {"error": "缺少PIL库，无法转换webp格式"}
    except Exception as e:
        print(f"其他错误: {e}")
        return {"error": str(e)}
    finally:
        # 清理临时文件
        if os.path.exists(temp_file):
            os.remove(temp_file)
            # print(f"临时文件已删除: {temp_file}")


def batch_create_products(body_data):
    """
    批量创建商品
    
    Args:
        body_data: 可以是字典或JSON字符串
        token: 授权token，如果不提供则使用默认token
    
    Returns:
        接口返回的JSON结果
    """
    # 上传地址
    url = "https://www.as1688.vip/api/v1/product-management/products/batch"
    

    # 请求头
    headers = {
        "accept": "application/json",
        "accept-language": "zh-CN,zh;q=0.9",
        "authorization": token,
        "content-type": "application/json",
        "sec-ch-ua": '"Not:A-Brand";v="99", "Google Chrome";v="145", "Chromium";v="145"',
        "sec-ch-ua-mobile": "?0",
        "sec-ch-ua-platform": '"Windows"',
        "sec-fetch-dest": "empty",
        "sec-fetch-mode": "cors",
        "sec-fetch-site": "same-origin"
    }
    
    # 处理body数据
    if isinstance(body_data, dict):
        json_body = body_data
    elif isinstance(body_data, str):
        json_body = json.loads(body_data)
    else:
        raise ValueError("body_data必须是字典或JSON字符串")
    
    try:
        # 发送POST请求
        response = requests.post(
            url,
            headers=headers,
            json=json_body,
            timeout=30
        )
        
        # 检查响应
        response.raise_for_status()
        
        # 返回JSON结果
        return response.json()
        
    except requests.exceptions.RequestException as e:
        print(f"请求错误: {e}")
        if hasattr(e, 'response') and e.response:
            print(f"响应内容: {e.response.text}")
        return {"error": str(e)}
    except json.JSONDecodeError as e:
        print(f"JSON解析错误: {e}")
        return {"error": str(e)}


def get_product_list(category_id, page=1):
    """最简单的版本，只返回商品列表"""
    url = f"https://api.tiki-int.site/api/product/search?category_id={category_id}&page={page}&limit=15&type=1"
    
    try:
        return requests.get(url).json()
    except:
        return []
# 使用示例
if __name__ == "__main__":
    for index in range(1, 214):
        listData = get_product_list(24, index)
        for item in listData['data']['list']:
            print(index,item['product_id'], item['mer_id'])
            # 测试函数
            result = get_product_detail(item['mer_id'],item['product_id'])

            print(result['data']['goods'])
            
            content = result['data']['goods']['content']
            title = result['data']['goods']['title']
            stock = result['data']['goods']['stock']
            images = result['data']['goods']['images'].split(",")
            market_price = result['data']['goods']['sales_price']
            # 上传图片获取URL
            uploaded_image_urls = []
            for image_url in images:
                print(f"处理图片: {image_url}")
                resultFile = download_and_upload_image(image_url)

                if ('data' in resultFile):
                    uploadFileURL = resultFile['data']
                    uploaded_image_urls.append(uploadFileURL)
            print(title, stock, market_price)
            if (len(uploaded_image_urls) > 0 and market_price > 0):
                #      
                uploadInfo = {"products":[{"name":title,"category":"全球购","price":str(market_price),"stock":stock,"description":content,"images":uploaded_image_urls}]}
                print(uploadInfo["products"][0]["category"])
                uploadInfoReturn = batch_create_products(uploadInfo)
                print(uploadInfoReturn)