large-order-fusion/modules/query_db.py

import os
import time
import numpy as np
from typing import Literal
from datetime import datetime, date, timedelta
from utils.logger_config import setup_logger
from utils.tools import load_sql_file, query_clickhouse

logger = setup_logger(__name__)


def get_sku_info(goods_id: str, color_id: str) -> list:
    """根据商品ID和颜色ID获取商品信息"""
    sql_query = load_sql_file("get_sku_info.sql")
    sql_query = sql_query.replace("goods_id_holder", goods_id).replace("color_id_holder", color_id)
    
    result = query_clickhouse(sql_query)

    logger.info(f"查询商品信息成功！")
    return result

def get_sku_id(goods_code: str, color_code: str) -> list:
    """根据商品CODE和颜色CODE获取商品ID和颜色ID"""
    sql_query = load_sql_file("get_sku_id.sql")
    sql_query = sql_query.replace("g_code_holder", goods_code).replace("cl_code_holder", color_code)
    
    result = query_clickhouse(sql_query)

    logger.info(f"查询商品ID，颜色ID成功！")
    return result

def get_core_sku_freq(
    goods_code: str,
    color_code: str,
    goods_id: str,
    color_id: str,
    query_type: Literal["22", "33"],
    start_date: str = None,
    end_date: str = None
) -> list:
    """获取主推款近一个月的销售频次"""
    sql_query = load_sql_file(f"core_sku_freq{query_type}.sql")

    if not start_date:
        start_date = str(date.today() - timedelta(days=30))
    if not end_date:
        end_date = str(date.today())

    sql_query = sql_query.replace("g_code_holder", goods_code).replace("cl_code_holder", color_code)
    sql_query = sql_query.replace("goods_id_holder", goods_id).replace("color_id_holder", color_id)
    sql_query = sql_query.replace("date_start_holder", start_date).replace("date_end_holder", end_date)

    result = query_clickhouse(sql_query)

    logger.info(f"查询主推款销售频次成功！")
    return result

def get_fusion_order(
    goods_code: str,
    color_code: str,
    goods_id: str,
    color_id: str,
    query_type: Literal["22", "33"],
    start_date: str = None,
    end_date: str = None
) -> list:
    """获取组合订单"""
    sql_query = load_sql_file(f"get_fusion{query_type}.sql")

    if not start_date:
        start_date = str(date.today() - timedelta(days=30))
    if not end_date:
        end_date = str(date.today())

    sql_query = sql_query.replace("g_code_holder", goods_code).replace("cl_code_holder", color_code)
    sql_query = sql_query.replace("goods_id_holder", goods_id).replace("color_id_holder", color_id)
    sql_query = sql_query.replace("date_start_holder", start_date).replace("date_end_holder", end_date)

    result = query_clickhouse(sql_query)

    if query_type == "22":
        flat_list = '\n'.join(result).split('\n')
        result_list = [flat_list[i:i+2] for i in range(0, len(flat_list), 2)]
    elif query_type == "33":
        flat_list = '\n'.join(result).split('\n')
        result_list = [flat_list[i:i+3] for i in range(0, len(flat_list), 3)]
    else:
        logger.warning(f"请输入正确的查询类型（22或33）！")

    logger.info(f"查询组合订单成功！")
    return result_list

def get_sales(
    goods_code: str,
    color_code: str,
    goods_id: str,
    color_id: str,
    query_type: Literal['22', '33'],
    sub_goods_id: str = None,
    sub_color_id: str = None,
    start_date: str = None,
    end_date: str = None,
) -> list:
    sql_query = load_sql_file(f"count_sale{query_type}.sql")

    if not start_date:
        start_date = str(date.today() - timedelta(days=30))
    if not end_date:
        end_date = str(date.today())

    sql_query = sql_query.replace('g_code_holder', goods_code).replace('cl_code_holder', color_code)
    sql_query = sql_query.replace('goods_id_holder', goods_id).replace('color_id_holder', color_id)
    sql_query = sql_query.replace('date_start_holder', start_date).replace('date_end_holder', end_date)

    if query_type == '22':
        result = query_clickhouse(sql_query)
        logger.info(f"查询连带销售金额成功！")
        return result

    if query_type == '33':
        sql_query = sql_query.replace('goods_third_id_holder', sub_goods_id).replace('color_third_id_holder', sub_color_id)
        result = query_clickhouse(sql_query)
        logger.info(f"查询连带销售金额成功！")
        return result
    


if __name__ == '__main__':
    
    good_code = "1E5C6M200"
    color_code = "91Y"
    
    # 商品ID & 颜色ID
    result_id = get_sku_id(good_code, color_code)
    
    good_id = result_id[0]
    color_id = result_id[1]

    # 主商品频次
    result_freq = get_core_sku_freq(
        good_code,
        color_code,
        good_id,
        color_id,
        "22"
    )

    # 主商品信息
    result_primary_good = get_sku_info(good_id, color_id)
    print(result_primary_good)

    # # 查询连带款
    # result_combine = get_fusion_order(
    #     good_code,
    #     color_code,
    #     good_id,
    #     color_id,
    #     "33"
    # )

    # print(result_combine)
    # print(int("11")*2)

    # # combine_dict = {}

    # # for combine in result_combine:
    # #     good_color = combine[0]
    # #     count = combine[1]

    # #     good_id = good_color.split("-")[0]
    # #     color_id = good_color.split("-")[1]

    # #     result_code = get_sku_info(good_id, color_id)

    # #     good_code = result_code[0]
    # #     good_name = result_code[1]
    # #     color_code = result_code[2]
    # #     color_name = result_code[3]
    # #     good_season = result_code[4]
    # #     good_type = result_code[5]

    # #     logger.info(f"result_code: {result_code}")

    # #     logger.info(f"good_id: {good_id}, color_id: {color_id}")


    # # print(result_combine)
    # # print(len(result_combine))
    

    
    # result = get_sales(
    #     goods_code="1E5C6M200",
    #     color_code="91Y",
    #     goods_id="2586599925281280",
    #     color_id="2055025368795724",
    #     query_type="22",
    #     start_date="2025-09-06",
    #     end_date="2025-09-09"
    # )

    # print(result)