829 lines
45 KiB
Python
829 lines
45 KiB
Python
import pandas as pd
|
||
from bs4 import BeautifulSoup
|
||
import os
|
||
import re
|
||
import requests
|
||
from io import BytesIO
|
||
import hashlib
|
||
import concurrent.futures
|
||
import time
|
||
|
||
class Wpdata:
|
||
def __init__(self, data_path: str, input_excel: str, goods_url_path: str, upload_workers: int = 10, upload_image_domain:str = "https://www.fzcaiji.com/upload.php", column_name:str = "商品图片*"):
|
||
self.__data_path__ = data_path
|
||
self.__input_excel__ = input_excel
|
||
self.__goods_url_path__ = goods_url_path
|
||
|
||
self.__upload_workers__ = upload_workers
|
||
self.__upload_image_domain__ = upload_image_domain
|
||
self.__column_name__ = column_name
|
||
|
||
self.__output_csv__ = f"{input_excel}.csv"
|
||
self.__images_txt__ = f"{data_path}_images.txt"
|
||
|
||
# 原始数据转换
|
||
def transform_excel_to_csv(self):
|
||
def match_variant_count(number, group_df, variant_1, title, text, spu, data_rows, variant_2=None,variant_3=None):
|
||
# 提取当前spu分组的价格
|
||
sale_prices = group_df[(group_df['商品属性*'] == 'P') | (group_df['商品属性*'] == 'S')]['商品售价*'].tolist()
|
||
origin_prices = group_df[(group_df['商品属性*'] == 'P') | (group_df['商品属性*'] == 'S')]['商品原价'].tolist()
|
||
|
||
# 提取对应属性的图片，并统计
|
||
M_images = group_df[(group_df['商品属性*'] == 'M') | (group_df['商品属性*'] == 'S')]['商品图片*'].tolist()
|
||
image_list = [url.strip() for url in M_images[0].split(',')]
|
||
|
||
group_df
|
||
# 获取当前分组的URL
|
||
if 'url' in group_df:
|
||
url = group_df['url'].tolist()[0]
|
||
else:
|
||
url = group_df['URL'].tolist()[0]
|
||
|
||
match number:
|
||
case 3:
|
||
# 提取当前spu分组属性为P的值
|
||
all_variants_first = group_df[(group_df['商品属性*'] == 'P')]['款式1'].tolist()
|
||
all_variants_second = group_df[(group_df['商品属性*'] == 'P')]['款式2'].tolist()
|
||
all_variants_three = group_df[(group_df['商品属性*'] == 'P')]['款式3'].tolist()
|
||
|
||
all_variants_count = max(len(all_variants_first), len(all_variants_second), len(all_variants_three))
|
||
|
||
if all_variants_count >= len(image_list):
|
||
|
||
# 计算需要扩大的长度，并进行扩大
|
||
expand_length = all_variants_count - len(image_list)
|
||
expanded_image_list = image_list + [None] * expand_length
|
||
|
||
Handle = title[0].replace(' ', '-').replace('/', '-').lower()
|
||
image_counter = 0
|
||
for k, variant in enumerate(all_variants_first):
|
||
rows = {
|
||
"Handle": Handle,
|
||
"Title": title[0] if k == 0 else "",
|
||
"Body (HTML)": text if k == 0 else "",
|
||
"Vendor": "",
|
||
"Type": "",
|
||
"Tags": "",
|
||
"Published": "TRUE" if k == 0 else "",
|
||
"Option1 Name": variant_1,
|
||
"Option1 Value": all_variants_first[k],
|
||
"Option2 Name": variant_2,
|
||
"Option2 Value": all_variants_second[k],
|
||
"Option3 Name": variant_3,
|
||
"Option3 Value": all_variants_second[k],
|
||
"Variant SKU": f"{spu}{str(k)}",
|
||
"Variant Grams": "0",
|
||
"Variant Inventory Tracker": "shopify",
|
||
"Variant Inventory Qty": "1000",
|
||
"Variant Inventory Policy": "continue",
|
||
"Variant Fulfillment Service": "manual",
|
||
"Variant Price": sale_prices[k],
|
||
"Variant Compare At Price": origin_prices[k],
|
||
"Variant Requires Shipping": "TRUE",
|
||
"Variant Taxable": "TRUE",
|
||
"Variant Barcode": "",
|
||
"Image Src": expanded_image_list[k],
|
||
"Image Position": (image_counter := image_counter + 1) if expanded_image_list[
|
||
k] is not None else "", # 海象运算符
|
||
"Image Alt Text": "",
|
||
"Gift Card": "FALSE" if k == 0 else "",
|
||
"SEO Title": title[0] if k == 0 else "",
|
||
"SEO Description": title[0] if k == 0 else "",
|
||
"Google Shopping / Google Product Category": "",
|
||
"Google Shopping / Gender": "",
|
||
"Google Shopping / Age Group": "",
|
||
"Google Shopping / MPN": "",
|
||
"Google Shopping / AdWords Grouping": "",
|
||
"Google Shopping / AdWords Labels": "",
|
||
"Google Shopping / Condition": "",
|
||
"Google Shopping / Custom Product": "",
|
||
"Google Shopping / Custom Label 0": "",
|
||
"Google Shopping / Custom Label 测试.txt": "",
|
||
"Google Shopping / Custom Label 2": "",
|
||
"Google Shopping / Custom Label 3": "",
|
||
"Google Shopping / Custom Label 4": "",
|
||
"Variant Image": "",
|
||
"IsDraft": "",
|
||
"Variant Weight Unit": "Kg" if all_variants_first[k] else "",
|
||
"Variant Tax Code": "",
|
||
"Cost per item": "",
|
||
"Status": "activate" if k == 0 else "",
|
||
"Collection": "",
|
||
"url": url
|
||
}
|
||
data_rows.append(rows)
|
||
|
||
else: # 变体少于图片数量
|
||
# 计算扩大长度
|
||
expand_length = len(image_list) - all_variants_count
|
||
|
||
expanded_all_variants_first_count = all_variants_first + [None] * expand_length
|
||
expanded_all_variants_two_count = all_variants_second + [None] * expand_length
|
||
expanded_all_variants_three_count = all_variants_three + [None] * expand_length
|
||
# 扩大价格长度（为了统一价格）
|
||
expanded_sale_prices = sale_prices + [None] * expand_length
|
||
expanded_origin_prices = origin_prices + [None] * expand_length
|
||
|
||
Handle = title[0].replace(' ', '-').replace('/', '-').lower()
|
||
image_counter = 0
|
||
|
||
for k, variant in enumerate(image_list):
|
||
rows = {
|
||
"Handle": Handle,
|
||
"Title": title[0] if k == 0 else "",
|
||
"Body (HTML)": text if k == 0 else "",
|
||
"Vendor": "",
|
||
"Type": "",
|
||
"Tags": "",
|
||
"Published": "TRUE" if k == 0 else "",
|
||
"Option1 Name": variant_1 if expanded_all_variants_first_count[k] is not None else "",
|
||
"Option1 Value": expanded_all_variants_first_count[k],
|
||
"Option2 Name": variant_2 if expanded_all_variants_two_count[k] is not None else "",
|
||
"Option2 Value": expanded_all_variants_two_count[k],
|
||
"Option3 Name": variant_3 if expanded_all_variants_three_count[k] is not None else "",
|
||
"Option3 Value": expanded_all_variants_three_count[k],
|
||
"Variant SKU": f"{spu}{str(k)}" if expanded_all_variants_first_count[k] is not None else "",
|
||
"Variant Grams": "0" if expanded_all_variants_first_count[k] is not None else "",
|
||
"Variant Inventory Tracker": "shopify" if expanded_all_variants_first_count[
|
||
k] is not None else "",
|
||
"Variant Inventory Qty": "1000" if expanded_all_variants_first_count[k] is not None else "",
|
||
"Variant Inventory Policy": "continue" if expanded_all_variants_first_count[
|
||
k] is not None else "",
|
||
"Variant Fulfillment Service": "manual" if expanded_all_variants_first_count[
|
||
k] is not None else "",
|
||
"Variant Price": expanded_sale_prices[k],
|
||
"Variant Compare At Price": expanded_origin_prices[k],
|
||
"Variant Requires Shipping": "TRUE" if expanded_all_variants_first_count[k] is not None else "",
|
||
"Variant Taxable": "TRUE" if expanded_all_variants_first_count[k] is not None else "",
|
||
"Variant Barcode": "",
|
||
"Image Src": image_list[k],
|
||
"Image Position": (image_counter := image_counter + 1) if image_list[k] is not None else "",
|
||
"Image Alt Text": "",
|
||
"Gift Card": "FALSE" if k == 0 else "",
|
||
"SEO Title": title[0] if k == 0 else "",
|
||
"SEO Description": title[0] if k == 0 else "",
|
||
"Google Shopping / Google Product Category": "",
|
||
"Google Shopping / Gender": "",
|
||
"Google Shopping / Age Group": "",
|
||
"Google Shopping / MPN": "",
|
||
"Google Shopping / AdWords Grouping": "",
|
||
"Google Shopping / AdWords Labels": "",
|
||
"Google Shopping / Condition": "",
|
||
"Google Shopping / Custom Product": "",
|
||
"Google Shopping / Custom Label 0": "",
|
||
"Google Shopping / Custom Label 测试.txt": "",
|
||
"Google Shopping / Custom Label 2": "",
|
||
"Google Shopping / Custom Label 3": "",
|
||
"Google Shopping / Custom Label 4": "",
|
||
"Variant Image": "",
|
||
"IsDraft": "",
|
||
"Variant Weight Unit": "Kg" if k == 0 else "",
|
||
"Variant Tax Code": "",
|
||
"Cost per item": "",
|
||
"Status": "activate" if k == 0 else "",
|
||
"Collection": "",
|
||
"url": url
|
||
}
|
||
data_rows.append(rows)
|
||
|
||
case 2:
|
||
# 提取当前spu分组属性为P的值
|
||
all_variants_first = group_df[(group_df['商品属性*'] == 'P')]['款式1'].tolist()
|
||
all_variants_second = group_df[(group_df['商品属性*'] == 'P')]['款式2'].tolist()
|
||
|
||
all_variants_count = max(len(all_variants_first), len(all_variants_second))
|
||
|
||
if all_variants_count >= len(image_list):
|
||
|
||
# 计算需要扩大的长度，并进行扩大
|
||
expand_length = all_variants_count - len(image_list)
|
||
expanded_image_list = image_list + [None] * expand_length
|
||
|
||
Handle = title[0].replace(' ', '-').replace('/', '-').lower()
|
||
image_counter = 0
|
||
for k, variant in enumerate(all_variants_first):
|
||
rows = {
|
||
"Handle": Handle,
|
||
"Title": title[0] if k == 0 else "",
|
||
"Body (HTML)": text if k == 0 else "",
|
||
"Vendor": "",
|
||
"Type": "",
|
||
"Tags": "",
|
||
"Published": "TRUE" if k == 0 else "",
|
||
"Option1 Name": variant_1,
|
||
"Option1 Value": all_variants_first[k],
|
||
"Option2 Name": variant_2,
|
||
"Option2 Value": all_variants_second[k],
|
||
"Option3 Name": "",
|
||
"Option3 Value": "",
|
||
"Variant SKU": f"{spu}{str(k)}",
|
||
"Variant Grams": "0",
|
||
"Variant Inventory Tracker": "shopify",
|
||
"Variant Inventory Qty": "1000",
|
||
"Variant Inventory Policy": "continue",
|
||
"Variant Fulfillment Service": "manual",
|
||
"Variant Price": sale_prices[k],
|
||
"Variant Compare At Price": origin_prices[k],
|
||
"Variant Requires Shipping": "TRUE",
|
||
"Variant Taxable": "TRUE",
|
||
"Variant Barcode": "",
|
||
"Image Src": expanded_image_list[k],
|
||
"Image Position": (image_counter := image_counter + 1) if expanded_image_list[
|
||
k] is not None else "", # 海象运算符
|
||
"Image Alt Text": "",
|
||
"Gift Card": "FALSE" if k == 0 else "",
|
||
"SEO Title": title[0] if k == 0 else "",
|
||
"SEO Description": title[0] if k == 0 else "",
|
||
"Google Shopping / Google Product Category": "",
|
||
"Google Shopping / Gender": "",
|
||
"Google Shopping / Age Group": "",
|
||
"Google Shopping / MPN": "",
|
||
"Google Shopping / AdWords Grouping": "",
|
||
"Google Shopping / AdWords Labels": "",
|
||
"Google Shopping / Condition": "",
|
||
"Google Shopping / Custom Product": "",
|
||
"Google Shopping / Custom Label 0": "",
|
||
"Google Shopping / Custom Label 测试.txt": "",
|
||
"Google Shopping / Custom Label 2": "",
|
||
"Google Shopping / Custom Label 3": "",
|
||
"Google Shopping / Custom Label 4": "",
|
||
"Variant Image": "",
|
||
"IsDraft": "",
|
||
"Variant Weight Unit": "Kg" if all_variants_first[k] else "",
|
||
"Variant Tax Code": "",
|
||
"Cost per item": "",
|
||
"Status": "activate" if k == 0 else "",
|
||
"Collection": "",
|
||
"url": url
|
||
}
|
||
data_rows.append(rows)
|
||
|
||
else: # 变体少于图片数量
|
||
# 计算扩大长度
|
||
expand_length = len(image_list) - all_variants_count
|
||
|
||
expanded_all_variants_first_count = all_variants_first + [None] * expand_length
|
||
expanded_all_variants_two_count = all_variants_second + [None] * expand_length
|
||
|
||
# 扩大价格长度（为了统一价格）
|
||
expanded_sale_prices = sale_prices + [None] * expand_length
|
||
expanded_origin_prices = origin_prices + [None] * expand_length
|
||
|
||
Handle = title[0].replace(' ', '-').replace('/', '-').lower()
|
||
image_counter = 0
|
||
|
||
for k, variant in enumerate(image_list):
|
||
rows = {
|
||
"Handle": Handle,
|
||
"Title": title[0] if k == 0 else "",
|
||
"Body (HTML)": text if k == 0 else "",
|
||
"Vendor": "",
|
||
"Type": "",
|
||
"Tags": "",
|
||
"Published": "TRUE" if k == 0 else "",
|
||
"Option1 Name": variant_1 if expanded_all_variants_first_count[k] is not None else "",
|
||
"Option1 Value": expanded_all_variants_first_count[k],
|
||
"Option2 Name": variant_2 if expanded_all_variants_two_count[k] is not None else "",
|
||
"Option2 Value": expanded_all_variants_two_count[k],
|
||
"Option3 Name": "",
|
||
"Option3 Value": "",
|
||
"Variant SKU": f"{spu}{str(k)}" if expanded_all_variants_first_count[k] is not None else "",
|
||
"Variant Grams": "0" if expanded_all_variants_first_count[k] is not None else "",
|
||
"Variant Inventory Tracker": "shopify" if expanded_all_variants_first_count[
|
||
k] is not None else "",
|
||
"Variant Inventory Qty": "1000" if expanded_all_variants_first_count[k] is not None else "",
|
||
"Variant Inventory Policy": "continue" if expanded_all_variants_first_count[
|
||
k] is not None else "",
|
||
"Variant Fulfillment Service": "manual" if expanded_all_variants_first_count[
|
||
k] is not None else "",
|
||
"Variant Price": expanded_sale_prices[k],
|
||
"Variant Compare At Price": expanded_origin_prices[k],
|
||
"Variant Requires Shipping": "TRUE" if expanded_all_variants_first_count[k] is not None else "",
|
||
"Variant Taxable": "TRUE" if expanded_all_variants_first_count[k] is not None else "",
|
||
"Variant Barcode": "",
|
||
"Image Src": image_list[k],
|
||
"Image Position": (image_counter := image_counter + 1) if image_list[k] is not None else "",
|
||
"Image Alt Text": "",
|
||
"Gift Card": "FALSE" if k == 0 else "",
|
||
"SEO Title": title[0] if k == 0 else "",
|
||
"SEO Description": title[0] if k == 0 else "",
|
||
"Google Shopping / Google Product Category": "",
|
||
"Google Shopping / Gender": "",
|
||
"Google Shopping / Age Group": "",
|
||
"Google Shopping / MPN": "",
|
||
"Google Shopping / AdWords Grouping": "",
|
||
"Google Shopping / AdWords Labels": "",
|
||
"Google Shopping / Condition": "",
|
||
"Google Shopping / Custom Product": "",
|
||
"Google Shopping / Custom Label 0": "",
|
||
"Google Shopping / Custom Label 测试.txt": "",
|
||
"Google Shopping / Custom Label 2": "",
|
||
"Google Shopping / Custom Label 3": "",
|
||
"Google Shopping / Custom Label 4": "",
|
||
"Variant Image": "",
|
||
"IsDraft": "",
|
||
"Variant Weight Unit": "Kg" if k == 0 else "",
|
||
"Variant Tax Code": "",
|
||
"Cost per item": "",
|
||
"Status": "activate" if k == 0 else "",
|
||
"Collection": "",
|
||
"url": url
|
||
}
|
||
data_rows.append(rows)
|
||
|
||
case 1:
|
||
all_variants = group_df[(group_df['商品属性*'] == 'P')]['款式1'].tolist()
|
||
|
||
all_variants_count = len(all_variants)
|
||
|
||
if all_variants_count >= len(image_list):
|
||
# 计算扩大长度
|
||
expand_length = all_variants_count - len(image_list)
|
||
expanded_list = image_list + [None] * expand_length
|
||
|
||
# 扩大价格
|
||
expanded_sale_prices = sale_prices + [None] * expand_length
|
||
expanded_origin_prices = origin_prices + [None] * expand_length
|
||
|
||
Handle = title[0].replace(' ', '-').replace('/', '-').lower()
|
||
image_counter = 0
|
||
for k, variant in enumerate(all_variants):
|
||
rows = {
|
||
"Handle": Handle,
|
||
"Title": title[0] if k == 0 else "",
|
||
"Body (HTML)": text if k == 0 else "",
|
||
"Vendor": "",
|
||
"Type": "",
|
||
"Tags": "",
|
||
"Published": "TRUE" if k == 0 else "",
|
||
"Option1 Name": variant_1,
|
||
"Option1 Value": all_variants[k],
|
||
"Option2 Name": "",
|
||
"Option2 Value": "",
|
||
"Option3 Name": "",
|
||
"Option3 Value": "",
|
||
"Variant SKU": f"{spu}{str(k)}",
|
||
"Variant Grams": "0",
|
||
"Variant Inventory Tracker": "shopify",
|
||
"Variant Inventory Qty": "1000",
|
||
"Variant Inventory Policy": "continue",
|
||
"Variant Fulfillment Service": "manual",
|
||
"Variant Price": expanded_sale_prices[k],
|
||
"Variant Compare At Price": expanded_origin_prices[k],
|
||
"Variant Requires Shipping": "TRUE",
|
||
"Variant Taxable": "TRUE",
|
||
"Variant Barcode": "",
|
||
"Image Src": expanded_list[k],
|
||
"Image Position": (image_counter := image_counter + 1) if expanded_list[k] is not None else "",
|
||
"Image Alt Text": "",
|
||
"Gift Card": "FALSE" if k == 0 else "",
|
||
"SEO Title": title[0] if k == 0 else "",
|
||
"SEO Description": title[0] if k == 0 else "",
|
||
"Google Shopping / Google Product Category": "",
|
||
"Google Shopping / Gender": "",
|
||
"Google Shopping / Age Group": "",
|
||
"Google Shopping / MPN": "",
|
||
"Google Shopping / AdWords Grouping": "",
|
||
"Google Shopping / AdWords Labels": "",
|
||
"Google Shopping / Condition": "",
|
||
"Google Shopping / Custom Product": "",
|
||
"Google Shopping / Custom Label 0": "",
|
||
"Google Shopping / Custom Label 测试.txt": "",
|
||
"Google Shopping / Custom Label 2": "",
|
||
"Google Shopping / Custom Label 3": "",
|
||
"Google Shopping / Custom Label 4": "",
|
||
"Variant Image": "",
|
||
"IsDraft": "",
|
||
"Variant Weight Unit": "Kg" if all_variants[k] else "",
|
||
"Variant Tax Code": "",
|
||
"Cost per item": "",
|
||
"Status": "activate" if k == 0 else "",
|
||
"Collection": "",
|
||
"url": url
|
||
}
|
||
data_rows.append(rows)
|
||
|
||
else: # 变体少于图片数量
|
||
# 计算扩大长度
|
||
expand_length = len(image_list) - all_variants_count
|
||
expanded_all_variants_first_count = all_variants + [None] * expand_length
|
||
|
||
# 扩大价格
|
||
expanded_sale_prices = sale_prices + [None] * expand_length
|
||
expanded_origin_prices = origin_prices + [None] * expand_length
|
||
|
||
Handle = title[0].replace(' ', '-').replace('/', '-').lower()
|
||
image_counter = 0
|
||
|
||
for k, variant in enumerate(image_list):
|
||
rows = {
|
||
"Handle": Handle,
|
||
"Title": title[0] if k == 0 else "",
|
||
"Body (HTML)": text if k == 0 else "",
|
||
"Vendor": "",
|
||
"Type": "",
|
||
"Tags": "",
|
||
"Published": "TRUE" if k == 0 else "",
|
||
"Option1 Name": variant_1 if expanded_all_variants_first_count[k] is not None else "",
|
||
"Option1 Value": expanded_all_variants_first_count[k],
|
||
"Option2 Name": "",
|
||
"Option2 Value": "",
|
||
"Option3 Name": "",
|
||
"Option3 Value": "",
|
||
"Variant SKU": f"{spu}{str(k)}" if expanded_all_variants_first_count[k] is not None else "",
|
||
"Variant Grams": "0" if expanded_all_variants_first_count[k] is not None else "",
|
||
"Variant Inventory Tracker": "shopify" if expanded_all_variants_first_count[
|
||
k] is not None else "",
|
||
"Variant Inventory Qty": "1000" if expanded_all_variants_first_count[k] is not None else "",
|
||
"Variant Inventory Policy": "continue" if expanded_all_variants_first_count[
|
||
k] is not None else "",
|
||
"Variant Fulfillment Service": "manual" if expanded_all_variants_first_count[
|
||
k] is not None else "",
|
||
"Variant Price": expanded_sale_prices[k],
|
||
"Variant Compare At Price": expanded_origin_prices[k],
|
||
"Variant Requires Shipping": "TRUE" if expanded_all_variants_first_count[k] is not None else "",
|
||
"Variant Taxable": "TRUE" if expanded_all_variants_first_count[k] is not None else "",
|
||
"Variant Barcode": "",
|
||
"Image Src": image_list[k],
|
||
"Image Position": (image_counter := image_counter + 1) if image_list[k] is not None else "",
|
||
"Image Alt Text": "",
|
||
"Gift Card": "FALSE" if k == 0 else "",
|
||
"SEO Title": title[0] if k == 0 else "",
|
||
"SEO Description": title[0] if k == 0 else "",
|
||
"Google Shopping / Google Product Category": "",
|
||
"Google Shopping / Gender": "",
|
||
"Google Shopping / Age Group": "",
|
||
"Google Shopping / MPN": "",
|
||
"Google Shopping / AdWords Grouping": "",
|
||
"Google Shopping / AdWords Labels": "",
|
||
"Google Shopping / Condition": "",
|
||
"Google Shopping / Custom Product": "",
|
||
"Google Shopping / Custom Label 0": "",
|
||
"Google Shopping / Custom Label 测试.txt": "",
|
||
"Google Shopping / Custom Label 2": "",
|
||
"Google Shopping / Custom Label 3": "",
|
||
"Google Shopping / Custom Label 4": "",
|
||
"Variant Image": "",
|
||
"IsDraft": "",
|
||
"Variant Weight Unit": "Kg" if k == 0 else "",
|
||
"Variant Tax Code": "",
|
||
"Cost per item": "",
|
||
"Status": "activate" if k == 0 else "",
|
||
"Collection": "",
|
||
"url": url
|
||
}
|
||
data_rows.append(rows)
|
||
|
||
case 0:
|
||
Handle = title[0].replace(' ', '-').replace('/', '-').lower()
|
||
image_counter = 0
|
||
for k, variant in enumerate(image_list):
|
||
rows = {
|
||
"Handle": Handle,
|
||
"Title": title[0] if k == 0 else "",
|
||
"Body (HTML)": text if k == 0 else "",
|
||
"Vendor": "",
|
||
"Type": "",
|
||
"Tags": "",
|
||
"Published": "TRUE" if k == 0 else "",
|
||
"Option1 Name": "variant",
|
||
"Option1 Value": "default",
|
||
"Option2 Name": "",
|
||
"Option2 Value": "",
|
||
"Option3 Name": "",
|
||
"Option3 Value": "",
|
||
"Variant SKU": f"{spu}" if k == 0 else "",
|
||
"Variant Grams": "0" if k == 0 else "",
|
||
"Variant Inventory Tracker": "shopify" if k == 0 else "",
|
||
"Variant Inventory Qty": "1000" if k == 0 else "",
|
||
"Variant Inventory Policy": "continue" if k == 0 else "",
|
||
"Variant Fulfillment Service": "manual" if k == 0 else "",
|
||
"Variant Price": sale_prices[0] if k == 0 else "",
|
||
"Variant Compare At Price": origin_prices[0] if k == 0 else "",
|
||
"Variant Requires Shipping": "TRUE" if k == 0 else "",
|
||
"Variant Taxable": "TRUE" if k == 0 else "",
|
||
"Variant Barcode": "",
|
||
"Image Src": image_list[k],
|
||
"Image Position": (image_counter := image_counter + 1) if image_list[k] is not None else "",
|
||
"Image Alt Text": "",
|
||
"Gift Card": "FALSE" if k == 0 else "",
|
||
"SEO Title": title[0] if k == 0 else "",
|
||
"SEO Description": title[0] if k == 0 else "",
|
||
"Google Shopping / Google Product Category": "",
|
||
"Google Shopping / Gender": "",
|
||
"Google Shopping / Age Group": "",
|
||
"Google Shopping / MPN": "",
|
||
"Google Shopping / AdWords Grouping": "",
|
||
"Google Shopping / AdWords Labels": "",
|
||
"Google Shopping / Condition": "",
|
||
"Google Shopping / Custom Product": "",
|
||
"Google Shopping / Custom Label 0": "",
|
||
"Google Shopping / Custom Label 测试.txt": "",
|
||
"Google Shopping / Custom Label 2": "",
|
||
"Google Shopping / Custom Label 3": "",
|
||
"Google Shopping / Custom Label 4": "",
|
||
"Variant Image": "",
|
||
"IsDraft": "",
|
||
"Variant Weight Unit": "Kg" if k == 0 else "",
|
||
"Variant Tax Code": "",
|
||
"Cost per item": "",
|
||
"Status": "activate" if k == 0 else "",
|
||
"Collection": "",
|
||
"url": url
|
||
}
|
||
data_rows.append(rows)
|
||
data_rows = []
|
||
df = pd.read_excel(self.__input_excel__)
|
||
|
||
# 以相同商品spu为一组
|
||
grouped = df.groupby('商品spu')
|
||
|
||
data_rows = []
|
||
|
||
for spu, group_df in grouped:
|
||
print(spu)
|
||
group_df = group_df.fillna('')
|
||
# 获取商品属性为M或S的商品标题
|
||
title = group_df[(group_df['商品属性*'] == 'M') | (group_df['商品属性*'] == 'S')]['商品标题*'].tolist()
|
||
description = group_df[(group_df['商品属性*'] == 'M') | (group_df['商品属性*'] == 'S')]['商品描述'].tolist()
|
||
description.append('')
|
||
text = BeautifulSoup(description[0], 'html.parser').get_text()
|
||
|
||
variant_name_one = group_df[(group_df['商品属性*'] == 'M') | (group_df['商品属性*'] == 'S')]['款式1'].tolist()[
|
||
0]
|
||
variant_name_two = group_df[(group_df['商品属性*'] == 'M') | (group_df['商品属性*'] == 'S')]['款式2'].tolist()[
|
||
0]
|
||
variant_name_three = \
|
||
group_df[(group_df['商品属性*'] == 'M') | (group_df['商品属性*'] == 'S')]['款式3'].tolist()[0]
|
||
variant_count = 0
|
||
|
||
if variant_name_one:
|
||
variant_count += 1
|
||
if variant_name_two:
|
||
variant_count += 1
|
||
if variant_name_three:
|
||
variant_count += 1
|
||
|
||
# 统计个数
|
||
match_variant_count(variant_count, group_df, variant_name_one, title, text, spu, data_rows, variant_name_two,variant_name_three)
|
||
new_df = pd.DataFrame(data_rows)
|
||
new_df.to_csv(self.__output_csv__, index=False, encoding="utf-8")
|
||
|
||
return self
|
||
|
||
# 数据分割
|
||
def split_csv(self, csv_path: str = "", goods_url_path: str = ""):
|
||
if csv_path == "":
|
||
csv_path = self.__output_csv__
|
||
if goods_url_path == "":
|
||
goods_url_path = self.__goods_url_path__
|
||
|
||
with open(goods_url_path, 'r', encoding='utf-8') as f:
|
||
lines = f.readlines()
|
||
|
||
dict_all = {}
|
||
for i in lines:
|
||
url = i.strip()
|
||
if url == '':
|
||
continue
|
||
album = url.rsplit('#', 1)[1]
|
||
third_album = album.split('/')[-1].lstrip() # 最后一个专辑
|
||
# 全部存进字典
|
||
if third_album not in dict_all:
|
||
dict_all[third_album] = []
|
||
|
||
dict_all[third_album].append(url.rsplit('#',1)[0])
|
||
# print(dict_all)
|
||
|
||
# 测试.txt. 读取 CSV 文件
|
||
df = pd.read_csv(csv_path) # 替换为你的 CSV 文件路径
|
||
|
||
output_dir = f"{self.__data_path__}/已分类数据"
|
||
os.makedirs(output_dir, exist_ok=True)
|
||
# 2. 遍历字典的每个键（分类）
|
||
for category, urls in dict_all.items():
|
||
# 3. 筛选出 url 列中包含在当前分类的 URL 的行
|
||
filtered_df = df[df['url'].isin(urls)]
|
||
|
||
# 4. 如果筛选后的 DataFrame 不为空，则删除 'url' 列并保存到 CSV
|
||
if not filtered_df.empty:
|
||
filtered_df = filtered_df.drop(columns=['url']) # 删除 url 列
|
||
output_filename = fr"{output_dir}/{category}.csv"
|
||
filtered_df.to_csv(output_filename, index=False)
|
||
print(f"已保存 {len(filtered_df)} 条数据到 {output_filename}（已删除 url 列）")
|
||
else:
|
||
print(f"分类 {category} 无匹配数据，跳过保存")
|
||
return self
|
||
|
||
# 获取原始图片
|
||
def get_excel_images_to_txt(self):
|
||
def split_image_urls(cell: str):
|
||
"""按 ',http' / ',https' 分割字符串，自动补上 'http'。不校验后缀。"""
|
||
if not isinstance(cell, str) or not cell.strip():
|
||
return []
|
||
|
||
s = re.sub(r',\s*(https?://)', r'||SEP||\1', cell.strip(), flags=re.IGNORECASE)
|
||
parts = s.split('||SEP||')
|
||
|
||
urls = []
|
||
for part in parts:
|
||
part = part.strip().lstrip(',').strip()
|
||
if not part:
|
||
continue
|
||
m = re.search(r'https?://', part, re.IGNORECASE)
|
||
if m:
|
||
part = part[m.start():]
|
||
urls.append(part)
|
||
return urls
|
||
|
||
"""从 Excel 中读取指定列，返回每行的图片 URL 列表。"""
|
||
df = pd.read_excel(self.__input_excel__)
|
||
urls_list = []
|
||
for val in df[self.__column_name__].fillna(''):
|
||
urls_list.append(split_image_urls(val))
|
||
|
||
"""展开 + 去重 + 保存到 txt 文件"""
|
||
all_urls = set() # 用 set 去重
|
||
for row_urls in urls_list:
|
||
for url in row_urls:
|
||
all_urls.add(url.strip())
|
||
|
||
# 写入文件
|
||
with open(self.__images_txt__, "w", encoding="utf-8") as f:
|
||
for url in sorted(all_urls):
|
||
f.write(url + "\n")
|
||
|
||
print(f"✅ 共提取 {len(all_urls)} 张图片，已保存到 {self.__images_txt__}")
|
||
return self
|
||
|
||
# 上传原始图片
|
||
def update_images(self):
|
||
def upload_image(img_url: str):
|
||
"""下载并上传单张图片"""
|
||
HEADERS = {
|
||
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
|
||
"Accept": "image/avif,image/webp,image/apng,image/svg+xml,image/*,*/*;q=0.8",
|
||
"Accept-Language": "en-US,en;q=0.9",
|
||
"Accept-Encoding": "gzip, deflate, br",
|
||
"Referer": "https://www.google.com/"
|
||
}
|
||
try:
|
||
# 通过 requests 下载图片（带代理和请求头）
|
||
resp_img = requests.get(img_url, timeout=15, headers=HEADERS)
|
||
resp_img.raise_for_status()
|
||
img_data = BytesIO(resp_img.content)
|
||
|
||
# 生成哈希文件名
|
||
hash_name = hashlib.sha1(img_url.encode("utf-8")).hexdigest()
|
||
|
||
# 获取扩展名（安全处理）
|
||
ext = img_url.split("?")[0].split(".")[-1].lower()
|
||
if len(ext) > 5 or "/" in ext:
|
||
ext = "jpg"
|
||
|
||
# 上传
|
||
files = {"file": (f"{hash_name}.{ext}", img_data, f"image/{ext}")}
|
||
data = {"source_url": img_url}
|
||
resp = requests.post(self.__upload_image_domain__, files=files, data=data, timeout=30)
|
||
|
||
# 打印结果
|
||
try:
|
||
js = resp.json()
|
||
print(f"✅ {img_url} -> {js.get('url') or js}")
|
||
except Exception:
|
||
print(f"❌ {img_url} -> 上传失败: {resp.text[:100]}")
|
||
|
||
except Exception as e:
|
||
print(f"⚠️ 下载或上传异常: {img_url} -> {e}")
|
||
# 读取本地 txt 文件
|
||
with open(self.__images_txt__, "r", encoding="utf-8") as f:
|
||
urls = [x.strip() for x in f if x.strip()]
|
||
|
||
print(f"共读取到 {len(urls)} 张图片。开始上传...")
|
||
|
||
start = time.time()
|
||
# 使用线程池并发上传
|
||
with concurrent.futures.ThreadPoolExecutor(max_workers=self.__upload_workers__) as executor:
|
||
list(executor.map(upload_image, urls))
|
||
|
||
print(f"\n全部上传完成，用时 {time.time() - start:.2f} 秒。")
|
||
|
||
return self
|
||
|
||
# 替换原始图片
|
||
def replace_excel_image(self):
|
||
"""
|
||
读取 Excel 和 txt，将原图片 URL 替换为新图片 URL，保存到新 Excel
|
||
"""
|
||
def get_uploaded_image_url_php_style(img_url: str) -> str:
|
||
"""
|
||
模拟 PHP 上传脚本的路径生成逻辑：
|
||
- hash = sha1(source_url)
|
||
- 一级目录 = hash 前 1 字符
|
||
- 二级目录 = hash 第 2 字符
|
||
- 文件名 = hash + 扩展名
|
||
"""
|
||
# 计算 sha1
|
||
hash_name = hashlib.sha1(img_url.encode('utf-8')).hexdigest()
|
||
|
||
# 取扩展名（与 PHP 一致：从文件名里提取，默认 jpg）
|
||
ext = os.path.splitext(img_url.split('?')[0])[1].lstrip('.').lower()
|
||
if not ext or len(ext) > 5 or "/" in ext:
|
||
ext = 'jpg'
|
||
|
||
# 两层目录
|
||
dir1 = hash_name[0]
|
||
dir2 = hash_name[1]
|
||
|
||
# 拼接最终 URL
|
||
new_url = f"{self.__upload_image_domain__}/{dir1}/{dir2}/{hash_name}.{ext}"
|
||
|
||
return new_url
|
||
|
||
def split_image_urls(cell: str):
|
||
"""按 ',http' / ',https' 分割字符串，自动补上 'http'。不校验后缀。"""
|
||
if not isinstance(cell, str) or not cell.strip():
|
||
return []
|
||
|
||
s = re.sub(r',\s*(https?://)', r'||SEP||\1', cell.strip(), flags=re.IGNORECASE)
|
||
parts = s.split('||SEP||')
|
||
|
||
urls = []
|
||
for part in parts:
|
||
part = part.strip().lstrip(',').strip()
|
||
if not part:
|
||
continue
|
||
m = re.search(r'https?://', part, re.IGNORECASE)
|
||
if m:
|
||
part = part[m.start():]
|
||
urls.append(part)
|
||
# cell = ',' + cell
|
||
# cell = cell.split(',https://')
|
||
# urls = []
|
||
# for image_url in cell:
|
||
# if '.svg' in image_url:
|
||
# continue
|
||
# if image_url == "":
|
||
# continue
|
||
# urls.append(('https://'+image_url).strip())
|
||
return urls
|
||
|
||
# 读取 txt 中的原图片链接
|
||
with open(self.__images_txt__, "r", encoding="utf-8") as f:
|
||
original_urls = [line.strip() for line in f if line.strip()]
|
||
|
||
# 生成原图片到新图片的映射
|
||
url_mapping = {}
|
||
for old_url in original_urls:
|
||
new_url = get_uploaded_image_url_php_style(old_url)
|
||
url_mapping[old_url] = new_url
|
||
|
||
print(f"📋 共加载 {len(url_mapping)} 个图片映射关系")
|
||
|
||
# 读取 Excel
|
||
df = pd.read_excel(self.__input_excel__)
|
||
|
||
# 替换图片列
|
||
replaced_count = 0
|
||
for idx, cell_value in enumerate(df[self.__column_name__].fillna('')):
|
||
if not cell_value:
|
||
continue
|
||
|
||
# 分割出该单元格的所有图片 URL
|
||
old_urls = split_image_urls(cell_value)
|
||
|
||
# 替换为新 URL
|
||
new_urls = []
|
||
for old_url in old_urls:
|
||
if old_url in url_mapping:
|
||
new_urls.append(url_mapping[old_url])
|
||
replaced_count += 1
|
||
else:
|
||
new_urls.append(old_url) # 如果找不到映射，保持原样
|
||
|
||
# 拼接回去（用逗号分隔）
|
||
df.at[idx, self.__column_name__] = ','.join(new_urls)
|
||
|
||
output_excel = f"{self.__input_excel__}(new_image).xlsx"
|
||
# 保存到新 Excel
|
||
df.to_excel(output_excel, index=False)
|
||
print(f"✅ 替换完成！共替换 {replaced_count} 个图片链接")
|
||
print(f"✅ 已保存到 {output_excel}")
|
||
self.__input_excel__ = output_excel
|
||
|
||
# 一键运行
|
||
def run(self, replace_image: bool = False):
|
||
if replace_image:
|
||
self.get_excel_images_to_txt().update_images().replace_excel_image()
|
||
self.transform_excel_to_csv().split_csv()
|
||
|
||
if __name__ == "__main__":
|
||
wpdata = Wpdata(
|
||
data_path="data", # 已分类数据文件夹保存路径
|
||
input_excel=r"data\jamesavery\info\goods(sort)_2025-12-24.xlsx", # 原始excel路径
|
||
goods_url_path=r"data\jamesavery\urls\goods_2025-12-19.txt_old.txt" # 商品txt路径
|
||
)
|
||
wpdata.run(
|
||
# replace_image=True # 是否上传图片
|
||
)
|