BrandCultivation/models/recommend.py

from database.dao.mysql_dao import MySqlDao
from database.db.redis_db import RedisDatabaseHelper
import os
from models.item2vec.inference import Item2VecModel
from models.rank.data.config import CustConfig, ProductConfig
from models.rank.data.utils import sample_data_clear
from models.rank import GbdtLrModel, generate_feats_map
import pandas as pd
from core import get_logger

logger = get_logger("models.recommend")


CORE_RERANK_CONFIG = {
    "existing": {
        "core_model_weight": 0.75,
        "core_quality_weight": 0.15,
        "core_boost": 35,
        "low_model_threshold": 35,
        "low_model_weight": 0.85,
        "low_quality_weight": 0.10,
        "low_core_boost": 65,
        "normal_model_weight": 0.90,
    },
    "new": {
        "core_model_weight": 0.55,
        "core_quality_weight": 0.25,
        "core_boost": 50,
        "normal_model_weight": 0.90,
    },
}


class Recommend:
    def __init__(self, city_uuid):
        self._redis = RedisDatabaseHelper().redis
        self._dao = MySqlDao()
        
        self._load_molde(city_uuid)
        
    def _load_molde(self, city_uuid):
        """加载推演模型"""
        self._city_uuid = city_uuid
        gbdtlr_model_path = os.path.join("./models/rank/weights", city_uuid, "gbdtlr_model.pkl")
        self._gbdtlr_model = GbdtLrModel(gbdtlr_model_path)
        self._item2vec_model = Item2VecModel(city_uuid)
        logger.info(f"Models loaded for city_uuid={city_uuid}")
        
    def _get_itemcf_recall(self, product_id):
        """协同召回"""
        key = f"fc:{self._city_uuid}:{product_id}"
        recall_list = self._redis.zrevrange(key, 0, -1, withscores=False)
        return recall_list
    
    def get_recal_cust(self, product_id, cust_code_list):
        """通过协同过滤召回与核心零售户列表取并集，得到待推荐商户列表"""
        itemcf_recall_list = self._get_itemcf_recall(product_id)
        seen = set(itemcf_recall_list)
        extra = [c for c in cust_code_list if c not in seen]
        result = list(itemcf_recall_list) + extra
        logger.info(f"Recall completed: {len(result)} customers (itemcf={len(itemcf_recall_list)}, core_extra={len(extra)}) for product {product_id}")
        return result

    def get_recommend_list_by_gbdtlr(self, product_id, cust_code_list=None):
        """根据gbdt_lr获取商户推荐列表"""
        logger.info(f"GBDT-LR recommend started for product {product_id}")
        # 获取召回的商户列表
        if cust_code_list is None:
            cust_code_list = []
        recall_cust_list = self.get_recal_cust(product_id, cust_code_list)
        # 获取卷烟数据
        product_data = self._dao.get_product_by_id(self._city_uuid, product_id)[ProductConfig.FEATURE_COLUMNS]
        product_data = sample_data_clear(product_data, ProductConfig)
        
        # 获取整合商户数据
        cust_data = self._dao.get_cust_by_ids(self._city_uuid, recall_cust_list)[CustConfig.FEATURE_COLUMNS]
        # shop_data = self._dao.get_shop_by_ids(self._city_uuid, recall_cust_list)[ShopConfig.FEATURE_COLUMNS]
        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")

        # 按 recall_cust_list 顺序对齐 cust_data，确保 feats_map 行顺序与 recall_list 一致
        # 否则 get_recommend_list 中 zip(recall_list, scores) 会错配商户ID和分数
        cust_codes_in_data = set(cust_data["cust_code"].tolist())
        ordered_recall_list = [c for c in recall_cust_list if c in cust_codes_in_data]
        cust_order = {code: i for i, code in enumerate(ordered_recall_list)}
        cust_data = cust_data.sort_values("cust_code", key=lambda x: x.map(cust_order)).reset_index(drop=True)

        # 获取推理用的feats_map
        feats_map = generate_feats_map(product_data, cust_data)
        recommend_list = self._gbdtlr_model.get_recommend_list(feats_map, ordered_recall_list)
        recommend_list = self._rerank_existing_product(recommend_list, cust_code_list)
        # recommend_list = self.filter_recommend_list(recommend_list)
        logger.info(f"GBDT-LR recommend completed: {len(recommend_list)} results")
        return recommend_list
    
    def get_recommend_list_by_item2vec(self, product_id, cust_code_list=None):
        """根据item2vec获取商户推荐列表，核心商户并入候选集统一评分"""
        if cust_code_list is None:
            cust_code_list = []
        logger.info(f"Item2Vec recommend started for product {product_id}")
        recommend_list = self._item2vec_model.get_recommend_cust_list(product_id, cust_code_list=cust_code_list)
        recommend_list = recommend_list.drop(columns=["sale_qty"])
        recommend_list = recommend_list.to_dict(orient='records')
        recommend_list = self._rerank_new_product(recommend_list, cust_code_list)
        # recommend_list = self.filter_recommend_list(recommend_list)
        logger.info(f"Item2Vec recommend completed: {len(recommend_list)} results")
        return recommend_list

    def _rerank_existing_product(self, recommend_list, core_cust_list):
        """Rerank existing-product results with core-customer boosts, then sort by score."""
        core_set = {str(cust_code) for cust_code in (core_cust_list or [])}
        if not core_set or not recommend_list:
            return recommend_list

        quality_score_map = self._build_quality_score_map(core_set)
        cfg = CORE_RERANK_CONFIG["existing"]

        for item in recommend_list:
            cust_code = str(item["cust_code"])
            model_score = float(item.get("recommend_score", 0) or 0)
            is_core = cust_code in core_set
            quality_score = quality_score_map.get(cust_code, 60.0)

            if is_core:
                if model_score >= cfg["low_model_threshold"]:
                    final_score = (
                        model_score * cfg["core_model_weight"]
                        + quality_score * cfg["core_quality_weight"]
                        + cfg["core_boost"]
                    )
                else:
                    final_score = (
                        model_score * cfg["low_model_weight"]
                        + quality_score * cfg["low_quality_weight"]
                        + cfg["low_core_boost"]
                    )
            else:
                final_score = model_score * cfg["normal_model_weight"]

            item["model_score"] = model_score
            item["is_core_cust"] = is_core
            item["core_quality_score"] = quality_score if is_core else None
            item["recommend_score"] = min(float(final_score), 100.0)

        recommend_list.sort(key=lambda x: x["recommend_score"], reverse=True)
        logger.info(f"Core boost rerank completed for existing product: core_count={len(core_set)}")
        return recommend_list

    def _rerank_new_product(self, recommend_list, core_cust_list):
        """Rerank Item2Vec cold-start results with core-customer boosts, then sort by score."""
        core_set = {str(cust_code) for cust_code in (core_cust_list or [])}
        if not core_set or not recommend_list:
            return recommend_list

        quality_score_map = self._build_quality_score_map(core_set)
        cfg = CORE_RERANK_CONFIG["new"]

        for item in recommend_list:
            cust_code = str(item["cust_code"])
            model_score = float(item.get("recommend_score", 0) or 0)
            is_core = cust_code in core_set
            quality_score = quality_score_map.get(cust_code, 60.0)

            if is_core:
                final_score = (
                    model_score * cfg["core_model_weight"]
                    + quality_score * cfg["core_quality_weight"]
                    + cfg["core_boost"]
                )
            else:
                final_score = model_score * cfg["normal_model_weight"]

            item["item2vec_score"] = model_score
            item["is_core_cust"] = is_core
            item["core_quality_score"] = quality_score if is_core else None
            item["recommend_score"] = min(float(final_score), 100.0)

        recommend_list.sort(key=lambda x: x["recommend_score"], reverse=True)
        logger.info(f"Core boost rerank completed for new product: core_count={len(core_set)}")
        return recommend_list

    def _build_quality_score_map(self, cust_list):
        """Build a 0-100 business-quality score for candidate customers."""
        if not cust_list:
            return {}

        unique_cust_list = list(dict.fromkeys(str(cust_code) for cust_code in cust_list))
        cust_data = self._dao.get_cust_by_ids(self._city_uuid, unique_cust_list)
        if cust_data.empty:
            return {cust_code: 60.0 for cust_code in unique_cust_list}

        score_map = {}
        for _, row in cust_data.iterrows():
            cust_code = row.get("cust_code")
            if pd.isna(cust_code):
                continue

            score_map[str(cust_code)] = self._calculate_core_quality_score(row)

        for cust_code in unique_cust_list:
            score_map.setdefault(cust_code, 60.0)
        return score_map

    def _calculate_core_quality_score(self, row):
        """Calculate a 0-100 quality score using only fields defined in CustConfig."""
        field_scores = [
            ("terminal_star_name", {
                "五星终端": 100,
                "四星终端": 90,
                "三星终端": 80,
                "二星终端": 70,
                "一星终端": 60,
                "其他": 50,
                "无": 40,
            }, 0.18),
            ("cooperate_codename", {
                "好": 90,
                "较好": 75,
                "一般": 60,
            }, 0.14),
            ("store_appearance_name", {
                "好": 90,
                "较好": 75,
                "一般": 60,
                "差": 40,
            }, 0.12),
            ("is_modern_terminalname", {
                "是": 85,
                "否": 55,
            }, 0.10),
            ("modern_terminal_name", {
                "直营终端": 95,
                "合作终端": 90,
                "加盟终端": 85,
                "一般现代终端": 75,
                "普通终端": 60,
                "无法识别": 50,
            }, 0.08),
            ("cooperate_type_name", {
                "品牌加盟": 90,
                "冠名加盟": 85,
                "无": 55,
            }, 0.08),
            ("creditclass_name", {
                "AAA": 95,
                "AA": 90,
                "A": 85,
                "C": 60,
                "D": 45,
            }, 0.10),
            ("counter_status_name", {
                "有": 80,
                "计划中": 65,
                "无": 50,
            }, 0.05),
            ("counter_put_type_name", {
                "独立陈列": 85,
                "混杂陈列": 70,
                "无陈列": 50,
            }, 0.05),
            ("back_counter_status_name", {
                "有": 80,
                "计划中": 65,
                "无": 50,
            }, 0.04),
            ("back_counter_put_type_name", {
                "独立陈列": 85,
                "混杂陈列": 70,
                "无陈列": 50,
            }, 0.03),
            ("back_counter_has_show_name", {
                "有": 80,
                "无": 50,
            }, 0.03),
        ]

        weighted_score = 0.0
        total_weight = 0.0

        for field, score_map, weight in field_scores:
            score = self._score_by_config_value(row, field, score_map)
            if score is None:
                continue
            weighted_score += score * weight
            total_weight += weight

        for field, weight in [("counter_number", 0.05), ("back_counter_number", 0.05)]:
            score = self._counter_score(self._get_row_value(row, field))
            if score is None:
                continue
            weighted_score += score * weight
            total_weight += weight

        if total_weight == 0:
            return 60.0

        return round(weighted_score / total_weight, 4)

    def _score_by_config_value(self, row, field, score_map):
        if field not in CustConfig.FEATURE_COLUMNS:
            return None

        value = self._get_row_value(row, field)
        if pd.isna(value):
            return None

        text = str(value)
        if text not in CustConfig.ONEHOT_CAT.get(field, []):
            return None

        return float(score_map.get(text, 60.0))

    @staticmethod
    def _get_row_value(row, field):
        if field not in row.index:
            return None
        return row.get(field)

    @staticmethod
    def _counter_score(value):
        if pd.isna(value):
            return None

        try:
            number = float(value)
        except (TypeError, ValueError):
            return None

        if number <= 0:
            return 50.0
        if number >= 4:
            return 90.0
        return 50.0 + number * 10.0
        
    def filter_recommend_list(self, recommend_list):
        """过滤掉已经歇业的商铺"""
        cust_set = set(self._dao.get_cust_list(self._city_uuid))
        filter_recommend_list = [
            item for item in recommend_list
            if item["cust_code"] in cust_set
        ]
        return filter_recommend_list


if __name__ == "__main__":
    city_uuid = "00000000000000000000000011445301"
    product_id = '350139'
    recommend = Recommend(city_uuid)
    recommend_list = recommend.get_recommend_list_by_gbdtlr(product_id)
    # for i in recommend_list:
    #     print(i)