优化项目结构,添加注释
This commit is contained in:
383
core/goods.py
Normal file
383
core/goods.py
Normal file
@@ -0,0 +1,383 @@
|
||||
#############################################################################
|
||||
|
||||
# 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.GoodsInfoP(
|
||||
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.GoodsInfoP(
|
||||
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)
|
||||
Reference in New Issue
Block a user