############################################################################# # 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, main_info: dict, son_infos: list = []): """ 添加商品 :param main_info: 主表信息 :param son_infos: 变体信息列表 """ 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, main_goods_info in shopyy_data.items(): main_info = { 'spu': main_goods_info.spu, 'title': main_goods_info.title, 'brand': main_goods_info.category_name, 'attr': main_goods_info.attribute, 'desc': main_goods_info.desc, 'price': main_goods_info.price, 'old_price': main_goods_info.old_price, 'images': orjson.dumps(main_goods_info.images), 'url': main_goods_info.url, 'other_data': orjson.dumps({}) } for index, item_name in enumerate(main_goods_info.attribute_items): if item_name: main_info[f"item{index+1}"] = item_name son_infos = [] for p_info in main_goods_info.p_lists: item = { 'spu': p_info.spu, 'price': p_info.price, 'old_price': p_info.old_price, 'images': orjson.dumps(p_info.images), 'url': p_info.url, 'other_data': orjson.dumps({}), 'item1': None, 'item2': None, 'item3': None } for index, item_name in enumerate(p_info.attribute_items): if item_name: item[f"item{index+1}"] = item_name son_infos.append(item) self.add_goods(main_info, son_infos) 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.to_db_data()) except Exception as e: messages.sendError(f"{e} {goods_info.spu}") db_index[goods_info.url] = None messages.sendInfo(db_name)