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 # 是否上传图片 )