BrandCultivation/utils/report_utils.py

from database.dao.mysql_dao import MySqlDao
from models import Recommend
from models.rank.data.config import CustConfig, ImportanceFeaturesMap, ProductConfig, DeliveryConfig
from models.rank.data.utils import sample_data_clear
from models.rank import generate_feats_map
from core import get_logger

import os
import pandas as pd
from utils.reports_process import feats_relation_process, build_recommend_report, eval_report_process_pre, eval_report_process

logger = get_logger("utils.report")

class ReportUtils:
    def __init__(self, city_uuid, product_id):
        self._recommend_model = Recommend(city_uuid)
        self._city_uuid = city_uuid
        self._product_id = product_id
        self._dao = MySqlDao()
        self._product_data =  self._dao.get_product_by_id(self._city_uuid, self._product_id)[ProductConfig.FEATURE_COLUMNS]
        self._save_dir = os.path.join("./data/reports", city_uuid, product_id)
        
        if not os.path.exists(self._save_dir):
            os.makedirs(self._save_dir)
        
    def _get_recommend_data(self, cust_code_list):
        """获取推荐商户列表"""
        products_in_order = self._dao.get_product_from_order(self._city_uuid)["product_code"].unique().tolist()
        if self._product_id in products_in_order:
            recommend_data = self._recommend_model.get_recommend_list_by_gbdtlr(
                self._product_id, cust_code_list=cust_code_list
            )
        else:
            recommend_data = self._recommend_model.get_recommend_list_by_item2vec(
                self._product_id, cust_code_list=cust_code_list
            )
        return recommend_data

    def _generate_feats_map(self, cust_code_list):
        """根据召回的推荐列表生成品规-商户features_map"""
        recommend_data = self._get_recommend_data(cust_code_list)
        recommend_list = list(map(lambda x: x["cust_code"], recommend_data))
        product_data = self._product_data.copy()
        cust_data = self._dao.get_cust_by_ids(self._city_uuid, recommend_list)[CustConfig.FEATURE_COLUMNS]
        product_data = sample_data_clear(product_data, ProductConfig)
        cust_data = sample_data_clear(cust_data, CustConfig)
        feats_map = generate_feats_map(product_data, cust_data)
        return feats_map
    
    def _get_product_content(self):
        """获取品规的内容，并以字典的形式返回"""
        product_data = self._product_data.copy()
        filter_dict = product_data.to_dict('records')[0]
        return filter_dict
    
    def generate_feats_ralation_report(self, cust_code_list):
        """生成特征相关性分析报告"""
        logger.info("Generating feature relation report")
        feats_map = self._generate_feats_map(cust_code_list)
        product_content = self._get_product_content()
        shap_result = self._recommend_model._gbdtlr_model.generate_shap_interance(feats_map)
        report = feats_relation_process(shap_result, product_content)
        report.to_excel(os.path.join(self._save_dir, "品规商户特征关系表.xlsx"), index=False)
        logger.info("Feature relation report saved")
        
    def generate_product_report(self):
        """生成推荐品规信息表"""
        logger.info("Generating product report")
        product_data = self._get_product_content()
        with open(os.path.join(self._save_dir, "卷烟信息表.xlsx"), "w", encoding='utf-8-sig') as file:
            for key, value in product_data.items():
                if key != 'product_code':
                    file.write(f"{ImportanceFeaturesMap.PRODUCT_FEATRUES_MAP[key]}, {value}\n")
        logger.info("Product report saved")
                    
    def generate_recommend_report(self, cust_code_list):
        """生成推荐报告"""
        logger.info("Generating recommend report")
        recommend_data = self._get_recommend_data(cust_code_list)
        recommend_list = list(map(lambda x: x["cust_code"], recommend_data))
        recommend_cust_infos = self._dao.get_cust_by_ids(self._city_uuid, recommend_list)
        report = build_recommend_report(recommend_data, recommend_cust_infos)

        report.to_excel(os.path.join(self._save_dir, "商户售卖推荐表.xlsx"), index=False)
        logger.info("Recommend report saved")
        
    def generate_similarity_product_report(self):
        """生成相似卷烟表"""
        logger.info("Generating similarity product report")
        product_similarity_map = self._recommend_model._item2vec_model.generate_product_similarity_map(self._product_id)
        product_similarity_map = product_similarity_map[["product_name", "similarity", "brand_name", "factory_name", "is_low_tar", "is_medium", "is_tiny", "is_coarse", "is_exploding_beads", "is_abnormity", "is_cig", "is_chuangxin", "direct_retail_price", "tbc_total_length", "product_style"]]
        product_similarity_map = product_similarity_map.rename(
            columns={
                "product_name":                 "卷烟名称",
                "similarity":                   "相似度",
                "factory_name":                 "生产厂商",
                "brand_name":                   "品牌名称",
                "is_low_tar":                   "低焦油卷烟",
                "is_medium":                    "中支烟",
                "is_tiny":                      "细支烟",
                "is_coarse":                    "粗支烟",
                "is_exploding_beads":           "爆珠烟",
                "is_abnormity":                 "异形包装",
                "is_cig":                       "雪茄烟",
                "is_chuangxin":                 "创新品类",
                "direct_retail_price":          "卷烟建议零售价",
                "tbc_total_length":             "烟支总长度",
                "product_style":                "包装类型",
            }
        )
        product_similarity_map.to_excel(os.path.join(self._save_dir, "相似卷烟表.xlsx"), index=False)
        logger.info("Similarity product report saved")
        
    def generate_eval_data_pre(self):
        if self._product_id == '350139':
            eval_product_id = "350355"
        else:
            eval_product_id = self._product_id
        eval_order_data = self._dao.get_eval_order_by_product(self._city_uuid, eval_product_id)
        if not os.path.exists(os.path.join(self._save_dir, "商户售卖推荐表.xlsx")):
            logger.error("商户售卖推荐表 not found")
        recommend_data = pd.read_excel(os.path.join(self._save_dir, "商户售卖推荐表.xlsx"))
        report = eval_report_process_pre(eval_order_data, recommend_data)
        
        report.to_excel(os.path.join(self._save_dir, "效果验证表.xlsx"), index=False)
        
    def generate_eval_data(self, start_time, end_time, recommend_data):
        """根据推荐列表生成验证报告"""
        logger.info("Generating eval report")
        if self._product_id == '350139':
            eval_product_id = "350355"
        else:
            eval_product_id = self._product_id
        delivery_data = self._dao.get_delivery_data_by_product(self._city_uuid, eval_product_id, start_time, end_time)
        delivery_data = delivery_data[DeliveryConfig.FEATURE_COLUMNS]
        delivery_data = sample_data_clear(delivery_data, DeliveryConfig)

        recommend_data = recommend_data.drop(columns=["建议投放量（条）"], errors="ignore")

        report = eval_report_process(delivery_data, recommend_data)
        
        report.to_excel(os.path.join(self._save_dir, "投放验证报告.xlsx"), index=False)
        logger.info("Eval report saved")
    
    def generate_all_data(self, cust_code_list):
        logger.info("Generating all reports")
        self.generate_feats_ralation_report(cust_code_list)
        self.generate_product_report()
        self.generate_recommend_report(cust_code_list)
        self.generate_similarity_product_report()
        logger.info("All reports generated")
        
if __name__ == "__main__":
    city_uuid = "00000000000000000000000011445301"
    product_id = '440298'
    start_time = '2025/2/10'
    end_time = '2025/2/16'
    report = ReportUtils(city_uuid, product_id)
    
    report.generate_eval_data(start_time, end_time)