from database.dao.mysql_dao import MySqlDao from models import Recommend from models.rank.data.config import CustConfig, ImportanceFeaturesMap, ProductConfig, ShopConfig, DeliveryConfig from models.rank.data.utils import sample_data_clear from models.rank import generate_feats_map import os import pandas as pd from utils.reports_process import feats_relation_process, calculate_delivery_by_recommend_data, eval_report_process_pre, eval_report_process 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, recall_count): """获取推荐商户列表""" # 判断product_id是否是新品规 products_in_order = self._dao.get_product_from_order(self._city_uuid)["product_code"].unique().tolist() # recall_count = 100 # 参数调整 if self._product_id in products_in_order: recommend_data = self._recommend_model.get_recommend_list_by_gbdtlr(self._product_id, recall_count=recall_count) else: recommend_data = self._recommend_model.get_recommend_list_by_item2vec(self._product_id, recall_count=recall_count) # # 根据推荐列表获取商户售卖卷烟的月均销量总和 # recommend_list = list(map(lambda x: x["cust_code"], recommend_list)) # order_data = self._dao.get_order_by_cust(self._city_uuid, ) return recommend_data def _generate_feats_map(self, recall_count): """根据召回的推荐列表生成品规-商户features_map""" recommend_data = self._get_recommend_data(recall_count) recommend_list = list(map(lambda x: x["cust_code"], recommend_data)) # 获取卷烟的信息 product_data = self._product_data.copy() # 根据cust_lit获取商户信息和商圈信息 cust_data = self._dao.get_cust_by_ids(self._city_uuid, recommend_list)[CustConfig.FEATURE_COLUMNS] shop_data = self._dao.get_shop_by_ids(self._city_uuid, recommend_list)[ShopConfig.FEATURE_COLUMNS] product_data = sample_data_clear(product_data, ProductConfig) cust_data = sample_data_clear(cust_data, CustConfig) shop_data = sample_data_clear(shop_data, ShopConfig) cust_feats = shop_data.set_index("cust_code") cust_data = cust_data.join(cust_feats, on="BB_RETAIL_CUSTOMER_CODE", how="inner") 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, recall_count): """生成特征相关性分析报告""" feats_map = self._generate_feats_map(recall_count) product_content = self._get_product_content() # 计算SHAP值 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) def generate_product_report(self): """生成推荐品规信息表""" 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") def generate_recommend_report(self, recall_count, delivery_count): """生成推荐报告，包括投放量""" recommend_data = self._get_recommend_data(recall_count) 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 = calculate_delivery_by_recommend_data(recommend_data, recommend_cust_infos, delivery_count) report.to_excel(os.path.join(self._save_dir, "商户售卖推荐表.xlsx"), index=False) def generate_similarity_product_report(self): """生成相似卷烟表""" 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) 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")): print("请先生成'商户售卖推荐表'") 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): """根据推荐列表生成验证报告""" 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=["建议投放量（条）"]) report = eval_report_process(delivery_data, recommend_data) report.to_excel(os.path.join(self._save_dir, "投放验证报告.xlsx"), index=False) def generate_all_data(self, recall_count, delivery_count): self.generate_feats_ralation_report(recall_count) self.generate_product_report() self.generate_recommend_report(recall_count, delivery_count) self.generate_similarity_product_report() # self.generate_eval_data() 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)