85 lines
3.0 KiB
Python
85 lines
3.0 KiB
Python
import csv
|
||
import traceback
|
||
from openpyxl import load_workbook
|
||
from core import types
|
||
from utils import messages
|
||
|
||
def read_excel_to_dict(excel_path: str, sheet_name: str = 'Sheet1') -> dict[str, types.GoodsInfo]:
|
||
"""
|
||
从 Excel 文件中读取商品数据，返回 {url: GoodsInfo} 的字典。
|
||
|
||
:param excel_path: Excel 文件路径
|
||
:param sheet_name: 工作表名称，默认 'Sheet1'
|
||
:return: dict{GoodsInfo.url: GoodsInfo}
|
||
"""
|
||
wb = load_workbook(excel_path, data_only=True)
|
||
ws = wb[sheet_name]
|
||
|
||
goods_dict = {}
|
||
|
||
# 从第3行开始读取（假设第1、2行为标题/说明）
|
||
for row in ws.iter_rows(min_row=3, values_only=True):
|
||
# 过滤空行（如果整行都为空则跳过）
|
||
if all(cell is None or str(cell).strip() == '' for cell in row):
|
||
continue
|
||
|
||
# 转为字符串列表，None 转为空字符串
|
||
row_data = [str(cell) if cell is not None else '' for cell in row]
|
||
|
||
try:
|
||
goods = types.GoodsInfo.from_row_data(row_data)
|
||
if goods.url and (goods.attribute == "M" or goods.attribute == "S"): # 只有 url 非空才加入字典
|
||
goods_dict[goods.url] = goods
|
||
elif goods.url and goods.attribute == "P":
|
||
if goods.url in goods_dict:
|
||
goods_dict[goods.url].p_lists.append(goods)
|
||
except Exception as e:
|
||
messages.sendError(f"解析行失败（跳过）: {row_data[:5]}... 错误: {e}\n{traceback.format_exc()}")
|
||
continue
|
||
wb.close()
|
||
|
||
return goods_dict
|
||
|
||
def save_lists_to_csv(data: list, save_path: str, template_path: str = ''):
|
||
header = []
|
||
if template_path:
|
||
with open(template_path, 'r', encoding='utf-8-sig') as f:
|
||
template = csv.reader(f)
|
||
for row in template:
|
||
header = list(row)
|
||
data.insert(0, header)
|
||
with open(save_path, 'w', encoding='utf-8', newline="") as f:
|
||
writer = csv.writer(f)
|
||
writer.writerows(data)
|
||
|
||
class WorkBook:
|
||
def __init__(self, template: str, save_path: str, sheet_name: str = 'Sheet1'):
|
||
wb = load_workbook(template)
|
||
ws = wb[sheet_name]
|
||
last_row = ws.max_row
|
||
if ws.cell(row=last_row, column=1).value is None and "此行导入时不可删除" in str(ws.cell(row=2, column=1).value):
|
||
start_row = last_row + 1
|
||
else:
|
||
while ws.cell(row=last_row, column=1).value is not None or any(ws.cell(row=last_row, column=c).value for c in range(2, 30)):
|
||
last_row += 1
|
||
start_row = last_row
|
||
|
||
self.wb = wb
|
||
self.ws = ws
|
||
self.start_row = start_row
|
||
self.save_path = save_path
|
||
|
||
def add_row_of_list(self, data: list):
|
||
for col_idx, value in enumerate(data, start=1):
|
||
try:
|
||
self.ws.cell(row=self.start_row, column=col_idx, value=value)
|
||
except:
|
||
pass
|
||
self.start_row += 1
|
||
|
||
def save(self):
|
||
self.wb.save(self.save_path)
|
||
|
||
def close(self):
|
||
self.wb.close()
|