384 lines
14 KiB
Python
384 lines
14 KiB
Python
#############################################################################
|
|
|
|
# Author: Cerys
|
|
# Update: 2026-02-06
|
|
|
|
#############################################################################
|
|
|
|
import os
|
|
import orjson
|
|
from core import excels, types
|
|
from utils import db, messages, files, formats
|
|
|
|
class GoodsModel:
|
|
def __init__(self, db_path: str):
|
|
"""
|
|
商品模型
|
|
|
|
:param db_path: 数据库文件路径(不存在则自动创建并初始化表)
|
|
|
|
"""
|
|
self.__source_folder__ = "sources"
|
|
|
|
create_table = False
|
|
if os.path.exists(db_path) == False:
|
|
create_table = True
|
|
|
|
self.Db = db.Db(db_path)
|
|
if create_table:
|
|
self.__create_table__()
|
|
|
|
def __create_table__(self):
|
|
table_sql = '''
|
|
PRAGMA foreign_keys = false;
|
|
|
|
DROP TABLE IF EXISTS "goods";
|
|
CREATE TABLE "goods" (
|
|
"spu" TEXT NOT NULL,
|
|
"title" TEXT NOT NULL,
|
|
"brand" TEXT,
|
|
"attr" TEXT NOT NULL,
|
|
"desc" TEXT,
|
|
"item1" TEXT,
|
|
"item2" TEXT,
|
|
"item3" TEXT,
|
|
"price" NUMBER NOT NULL,
|
|
"old_price" NUMBER NOT NULL,
|
|
"images" blob NOT NULL,
|
|
"url" TEXT NOT NULL,
|
|
"other_data" blob DEFAULT "{}",
|
|
PRIMARY KEY ("spu")
|
|
);
|
|
|
|
DROP TABLE IF EXISTS "goods_son";
|
|
CREATE TABLE "goods_son" (
|
|
"spu" TEXT NOT NULL,
|
|
"item1" TEXT,
|
|
"item2" TEXT,
|
|
"item3" TEXT,
|
|
"price" NUMBER NOT NULL,
|
|
"old_price" NUMBER NOT NULL,
|
|
"images" blob NOT NULL,
|
|
"url" TEXT,
|
|
"other_data" blob DEFAULT "{}"
|
|
);
|
|
|
|
CREATE INDEX "search_spu"
|
|
ON "goods_son" (
|
|
"spu" ASC
|
|
);
|
|
|
|
PRAGMA foreign_keys = true;
|
|
'''
|
|
self.Db.execute_script(table_sql)
|
|
|
|
def add_goods(self, goods_info: types.GoodsInfo):
|
|
"""
|
|
添加商品
|
|
|
|
:param goods_info: GoodsInfo 类型数据
|
|
|
|
"""
|
|
main_info, son_infos = goods_info.to_db_data()
|
|
if main_info['attr'] == 'M' and len(son_infos) == 0:
|
|
raise Exception(f"变体丢失 {main_info}")
|
|
if len(main_info['images']) == 0:
|
|
raise Exception('无图片')
|
|
|
|
self.Db.table('goods').insert(main_info, True)
|
|
self.Db.table('goods_son').insert_all(son_infos)
|
|
|
|
def is_repeat(self, url):
|
|
"""
|
|
判断是否需要合并
|
|
|
|
:param url: 商品链接
|
|
|
|
"""
|
|
count = self.Db.table('goods_son').where('url', url).count()
|
|
if count > 1:
|
|
return True
|
|
return False
|
|
|
|
def select_goods_to_lists(self) -> list[dict]:
|
|
"""获取商品列表（包含变体）"""
|
|
messages.sendInfo('开始查询')
|
|
db_data = self.Db.table('goods').join('goods_son', 'goods.spu = goods_son.spu').select(
|
|
'''
|
|
goods.spu,
|
|
goods.title,
|
|
goods.brand,
|
|
goods.attr,
|
|
goods.desc,
|
|
goods.item1 as main_item1,
|
|
goods.item2 as main_item2,
|
|
goods.item3 as main_item3,
|
|
goods.price,
|
|
goods.old_price,
|
|
goods.images,
|
|
goods.url,
|
|
goods.other_data,
|
|
goods_son.item1 as son_item1,
|
|
goods_son.item2 as son_item2,
|
|
goods_son.item3 as son_item3,
|
|
goods_son.price as son_price,
|
|
goods_son.old_price as son_old_price,
|
|
goods_son.images as son_images,
|
|
goods_son.url as son_url,
|
|
goods_son.other_data as son_other_data
|
|
'''
|
|
)
|
|
messages.sendInfo('查询完成')
|
|
return db_data
|
|
|
|
def select_goods_to_dict(self, key: str = 'url') -> dict:
|
|
"""
|
|
获取商品列表并转换为字典格式
|
|
|
|
:param key: 字典的键
|
|
|
|
"""
|
|
db_data = self.select_goods_to_lists()
|
|
dict_data = {}
|
|
for item in db_data:
|
|
if item[key] in dict_data:
|
|
dict_data[item[key]]['son_lists'].append({
|
|
'attr_items': [item['son_item1'], item['son_item2'], item['son_item3']],
|
|
'price': item['son_price'],
|
|
'old_price': item['son_old_price'],
|
|
'images': orjson.loads(item['son_images']),
|
|
'url': item['son_url'],
|
|
'son_other_data': orjson.loads(item['son_other_data'])
|
|
})
|
|
else:
|
|
dict_data[item[key]] = {
|
|
'spu': item['spu'],
|
|
'title': item['title'],
|
|
'brand': item['brand'],
|
|
'desc': item['desc'],
|
|
'attr': item['attr'],
|
|
'attr_items': [item['main_item1'], item['main_item2'], item['main_item3']],
|
|
'price': item['price'],
|
|
'old_price': item['old_price'],
|
|
'images': orjson.loads(item['images']),
|
|
'url': item['url'],
|
|
'other_data': orjson.loads(item['other_data'])
|
|
}
|
|
if item['attr'] == 'M':
|
|
if item['son_price'] == None:
|
|
continue
|
|
dict_data[item[key]]['son_lists'] = [
|
|
{
|
|
'attr_items': [item['son_item1'], item['son_item2'], item['son_item3']],
|
|
'price': item['son_price'],
|
|
'old_price': item['son_old_price'],
|
|
'images': orjson.loads(item['son_images']),
|
|
'url': item['son_url'],
|
|
'son_other_data': orjson.loads(item['son_other_data'])
|
|
}
|
|
]
|
|
return dict_data
|
|
|
|
def select_goods_to_goodsinfo(self) -> list[types.GoodsInfo]:
|
|
"""获取商品列表并转换为GoodsInfo对象列表"""
|
|
db_data = self.select_goods_to_lists()
|
|
index_data = {}
|
|
for item in db_data:
|
|
if item['spu'] in index_data:
|
|
index_data[item['spu']].p_lists.append(
|
|
types.GoodsInfo.Attributes(
|
|
attr_items = [item['son_item1'], item['son_item2'], item['son_item3']],
|
|
price = item['son_price'],
|
|
old_price = item['son_old_price'],
|
|
images = orjson.loads(item['son_images']),
|
|
url = item['son_url'],
|
|
other_data = orjson.loads(item['son_other_data'])
|
|
)
|
|
)
|
|
else:
|
|
try:
|
|
index_data[item['spu']] = types.GoodsInfo(
|
|
spu = item['spu'],
|
|
title = item['title'],
|
|
brand = item['brand'],
|
|
attr = item['attr'],
|
|
desc = item['desc'],
|
|
attr_items = [item['main_item1'], item['main_item2'], item['main_item3']],
|
|
price = item['price'],
|
|
old_price = item['old_price'],
|
|
images = orjson.loads(item['images']),
|
|
url = item['url'],
|
|
other_data = orjson.loads(item['other_data'])
|
|
)
|
|
if item['attr'] == 'M':
|
|
index_data[item['spu']].p_lists.append(
|
|
types.GoodsInfo.Attributes(
|
|
attr_items = [item['son_item1'], item['son_item2'], item['son_item3']],
|
|
price = item['son_price'],
|
|
old_price = item['son_old_price'],
|
|
images = orjson.loads(item['son_images']),
|
|
url = item['son_url'],
|
|
other_data = orjson.loads(item['son_other_data'])
|
|
)
|
|
)
|
|
except Exception as e:
|
|
messages.sendError(e)
|
|
return list(index_data.values())
|
|
|
|
def to_amazon_xlsx(self, goods_url_path: str, save_path: str):
|
|
"""
|
|
整数据库导出为亚马逊导入表格
|
|
|
|
:param save_path: 保存路径
|
|
:param goods_url_path: 未去重商品链接路径
|
|
|
|
"""
|
|
goods_data = self.select_goods_to_dict()
|
|
goods_urls = files.load_lines(goods_url_path)
|
|
WookBook = excels.WorkBook(
|
|
f'{self.__source_folder__}/amazon商品模板.xlsx',
|
|
save_path
|
|
)
|
|
|
|
for goods_url in goods_urls:
|
|
key = '#'.join(goods_url.split('#')[:-1])
|
|
if key not in goods_data:
|
|
continue
|
|
|
|
spu = formats.url_to_spu(key)
|
|
category = goods_url.split('#')[-1]
|
|
|
|
WookBook.add_row_of_list([
|
|
goods_data[key]['title'],
|
|
spu,
|
|
goods_data[key]['other_data']['popover'],
|
|
goods_data[key]['other_data']['popover_num'],
|
|
0,
|
|
goods_data[key]['price'],
|
|
goods_data[key]['old_price'],
|
|
orjson.dumps(goods_data[key]['images']),
|
|
goods_data[key]['url'],
|
|
category
|
|
])
|
|
|
|
# for goods_info in goods_info_data:
|
|
# WookBook.add_row_of_list([
|
|
# goods_info.title,
|
|
# goods_info.price,
|
|
# goods_info.images[0],
|
|
# goods_info.other_data['popover'],
|
|
# goods_info.other_data['popover_num'],
|
|
# goods_info.url,
|
|
# orjson.dumps(goods_info.images)
|
|
# ])
|
|
|
|
WookBook.save()
|
|
WookBook.close()
|
|
|
|
def to_shopyy_xlsx(self, save_path: str, save_bitch: int = 150000, shift_urls_path: str = ''):
|
|
"""
|
|
整数据库导出为Shopyy导入表格
|
|
|
|
:param save_path: 保存路径
|
|
:param repeat: 是否合并商品
|
|
:param save_bitch: 超过多少行时自动分割
|
|
|
|
"""
|
|
def get_wookbook(bitch_index: int):
|
|
return excels.WorkBook(
|
|
f'{self.__source_folder__}/shopyy商品模板.xlsx',
|
|
save_path.replace('.xlsx', f'_{bitch_index}.xlsx')
|
|
)
|
|
|
|
def get_shift_url():
|
|
shift_urls = {}
|
|
if shift_urls_path:
|
|
urls = files.load_lines(shift_urls_path)
|
|
for url in urls:
|
|
if url == '':
|
|
continue
|
|
shift_urls[url] = None
|
|
return shift_urls
|
|
|
|
goods_info_data = self.select_goods_to_goodsinfo()
|
|
bitch = 1
|
|
shift_urls = get_shift_url()
|
|
WookBook = get_wookbook(bitch)
|
|
|
|
count = len(goods_info_data)
|
|
for index, goods_info in enumerate(goods_info_data):
|
|
messages.sendInfo(f"{index}/{count}")
|
|
if shift_urls_path and goods_info.url not in shift_urls:
|
|
continue
|
|
|
|
rows_data = goods_info.to_shopyy_row_data()
|
|
for row_data in rows_data:
|
|
WookBook.add_row_of_list(row_data)
|
|
if WookBook.start_row > save_bitch:
|
|
WookBook.save()
|
|
WookBook.close()
|
|
bitch += 1
|
|
WookBook = get_wookbook(bitch)
|
|
|
|
WookBook.save()
|
|
WookBook.close()
|
|
|
|
def to_woo_csv(self, goods_url_path: str, save_path: str):
|
|
"""
|
|
整数据库导出为woo商品CSV表格
|
|
|
|
:param goods_url_path: 未去重链接txt路径
|
|
:param save_path: 保存路径
|
|
|
|
"""
|
|
goods_urls = files.load_lines(goods_url_path)
|
|
category_data = {}
|
|
for goods_url in goods_urls:
|
|
category_name = goods_url.split('#')[-1].replace('/', '>')
|
|
goods_url = '#'.join(goods_url.split('#')[:-1])
|
|
category_data[goods_url] = category_name
|
|
goods_info_data = self.select_goods_to_goodsinfo()
|
|
|
|
bitch = 0
|
|
save_data = []
|
|
for goods_info in goods_info_data:
|
|
rows_data = goods_info.to_woo_row_data(category_data[goods_info.url])
|
|
save_data += rows_data
|
|
if len(save_data) > 10000:
|
|
excels.save_lists_to_csv(save_data, save_path.replace('.csv', f"{bitch}.csv"), r'sources\woo商品模板.csv')
|
|
save_data = []
|
|
bitch += 1
|
|
excels.save_lists_to_csv(save_data, save_path.replace('.csv', f"{bitch}.csv"), r'sources\woo商品模板.csv')
|
|
|
|
def import_shopyy_xlsx(self, excel_path: str):
|
|
"""
|
|
导入shopyy表格为数据库数据
|
|
|
|
:param excel_path: 表格路径
|
|
"""
|
|
shopyy_data = excels.read_excel_to_dict(excel_path)
|
|
for url, goods_info in shopyy_data.items():
|
|
self.add_goods(goods_info)
|
|
|
|
def splice_db(self, db_folder: str):
|
|
"""
|
|
拼接指定文件夹下所有数据库文件
|
|
|
|
:param db_folder: 数据库文件夹路径
|
|
|
|
"""
|
|
db_lists = os.listdir(db_folder)
|
|
db_index = self.select_goods_to_dict()
|
|
for db_name in db_lists:
|
|
splice_db = GoodsModel(f"{db_folder}/{db_name}")
|
|
goods_lists = splice_db.select_goods_to_goodsinfo()
|
|
for goods_info in goods_lists:
|
|
if goods_info.url in db_index:
|
|
continue
|
|
try:
|
|
self.add_goods(goods_info)
|
|
except Exception as e:
|
|
messages.sendError(f"{e} {goods_info.spu}")
|
|
db_index[goods_info.url] = None
|
|
messages.sendInfo(db_name)
|