""" 品牌配置加载器 从YAML配置文件加载品牌识别规则 """ import os import yaml import logging from typing import Dict, List, Any, Optional from pathlib import Path logger = logging.getLogger(__name__) class BrandConfig: """品牌配置管理器""" _instance = None _config_loaded = False def __new__(cls): """单例模式""" if cls._instance is None: cls._instance = super().__new__(cls) return cls._instance def __init__(self): """初始化配置""" if not self._config_loaded: self._priority_brands: List[Dict] = [] self._compound_brands: List[Dict] = [] self._simple_brands: List[Dict] = [] self._brand_variants: Dict[str, str] = {} self._single_word_brands: List[str] = [] self._multi_word_brands: List[str] = [] self._ignore_keywords: set = set() self._prefix_replacements: Dict[str, str] = {} self._brand_description_splits: Dict[str, str] = {} self._load_config() BrandConfig._config_loaded = True def _get_config_path(self) -> Path: """获取配置文件路径""" # 尝试多个可能的路径 possible_paths = [ # 从当前文件位置推断 Path(__file__).parent.parent.parent / "configs" / "brands" / "builtin_brands.yaml", # 从项目根目录 Path("/Users/jinjunqian/PycharmProjects/订单处理新版/ordersys/backend/configs/brands/builtin_brands.yaml"), ] for path in possible_paths: if path.exists(): return path # 如果都找不到，返回第一个路径（会在加载时报错） return possible_paths[0] def _load_config(self): """加载配置文件""" config_path = self._get_config_path() try: if config_path.exists(): with open(config_path, 'r', encoding='utf-8') as f: config = yaml.safe_load(f) if config: self._priority_brands = config.get('priority_brands', []) self._compound_brands = config.get('compound_brands', []) self._simple_brands = config.get('simple_brands', []) self._brand_variants = config.get('brand_variants', {}) self._single_word_brands = config.get('single_word_brands', []) self._multi_word_brands = config.get('multi_word_brands', []) self._ignore_keywords = set(config.get('ignore_keywords', [])) self._prefix_replacements = config.get('prefix_replacements', {}) self._brand_description_splits = config.get('brand_description_splits', {}) logger.info(f"品牌配置加载成功: {config_path}") logger.info(f" - 优先级品牌: {len(self._priority_brands)}") logger.info(f" - 复合品牌: {len(self._compound_brands)}") logger.info(f" - 简单品牌: {len(self._simple_brands)}") logger.info(f" - 品牌变体: {len(self._brand_variants)}") logger.info(f" - 单词品牌: {len(self._single_word_brands)}") logger.info(f" - 多词品牌: {len(self._multi_word_brands)}") else: logger.warning(f"品牌配置文件不存在: {config_path}，使用内置默认配置") self._load_fallback_config() except Exception as e: logger.error(f"加载品牌配置失败: {e}，使用内置默认配置") self._load_fallback_config() def _load_fallback_config(self): """加载内置的后备配置""" # 这里只保留最关键的品牌，完整列表在YAML文件中 self._priority_brands = [ {"keywords": ["lululemon", "露露柠檬"], "canonical": "Lululemon"}, {"keywords": ["marc jacobs"], "canonical": "Marc Jacobs"}, {"keywords": ["arcteryx", "arc'teryx", "始祖鸟"], "canonical": "Arc'teryx"}, ] self._simple_brands = [ {"keyword": "vince", "canonical": "Vince"}, {"keyword": "kenzo", "canonical": "Kenzo"}, {"keyword": "burberry", "canonical": "Burberry"}, ] self._single_word_brands = ["Tods", "SportMax", "MaxMara", "Burberry", "Gucci", "Prada"] self._ignore_keywords = {"小王国", "现货", "女款", "男款", "MC", "LA", "AT"} def reload(self): """重新加载配置""" BrandConfig._config_loaded = False self.__init__() # ======================================== # 属性访问器 # ======================================== @property def priority_brands(self) -> List[Dict]: """获取优先级品牌列表""" return self._priority_brands @property def compound_brands(self) -> List[Dict]: """获取复合品牌列表""" return self._compound_brands @property def simple_brands(self) -> List[Dict]: """获取简单品牌列表""" return self._simple_brands @property def brand_variants(self) -> Dict[str, str]: """获取品牌变体映射""" return self._brand_variants @property def single_word_brands(self) -> List[str]: """获取单词品牌列表""" return self._single_word_brands @property def multi_word_brands(self) -> List[str]: """获取多词品牌列表""" return self._multi_word_brands @property def ignore_keywords(self) -> set: """获取忽略关键词集合""" return self._ignore_keywords @property def prefix_replacements(self) -> Dict[str, str]: """获取前缀替换规则""" return self._prefix_replacements @property def brand_description_splits(self) -> Dict[str, str]: """获取品牌描述拆分规则""" return self._brand_description_splits # ======================================== # 匹配方法 # ======================================== def match_priority_brand(self, product_lower: str) -> Optional[str]: """ 匹配优先级品牌 Args: product_lower: 小写的产品名称 Returns: 匹配到的标准品牌名，未匹配返回None """ for brand_rule in self._priority_brands: keywords = brand_rule.get('keywords', []) canonical = brand_rule.get('canonical') match_type = brand_rule.get('match_type', 'contains') if match_type == 'phrase': # 短语匹配：需要所有关键词都出现 for keyword in keywords: parts = keyword.lower().split() if all(part in product_lower for part in parts): return canonical else: # 默认包含匹配 for keyword in keywords: if keyword.lower() in product_lower: return canonical return None def match_compound_brand(self, product_lower: str) -> Optional[str]: """ 匹配复合品牌 Args: product_lower: 小写的产品名称 Returns: 匹配到的标准品牌名，未匹配返回None """ for brand_rule in self._compound_brands: keywords = brand_rule.get('keywords', []) canonical = brand_rule.get('canonical') for keyword in keywords: if keyword.lower() in product_lower: return canonical return None def match_simple_brand(self, product_lower: str) -> Optional[str]: """ 匹配简单品牌 Args: product_lower: 小写的产品名称 Returns: 匹配到的标准品牌名，未匹配返回None """ for brand_rule in self._simple_brands: keyword = brand_rule.get('keyword', '').lower() canonical = brand_rule.get('canonical') variants = brand_rule.get('variants', []) # 检查主关键词 if keyword and keyword in product_lower: return canonical # 检查变体 for variant in variants: if variant.lower() in product_lower: return canonical return None def preprocess_product_name(self, product_name: str) -> str: """ 预处理产品名称（应用变体替换和前缀处理） Args: product_name: 原始产品名称 Returns: 预处理后的产品名称 """ result = product_name # 应用前缀替换 for pattern, replacement in self._prefix_replacements.items(): if pattern in result: result = result.replace(pattern, replacement) # 应用品牌变体替换（小写比较） result_lower = result.lower() for variant, standard in self._brand_variants.items(): if variant.lower() in result_lower: result = result.replace(variant, standard) result_lower = result.lower() # 应用品牌+描述拆分 for pattern, replacement in self._brand_description_splits.items(): if pattern in result: result = result.replace(pattern, replacement) return result def is_single_word_brand(self, word: str) -> bool: """检查是否为单词品牌""" return word in self._single_word_brands def is_known_multi_word_brand(self, phrase: str) -> Optional[str]: """ 检查是否为已知的多词品牌 Args: phrase: 待检查的短语 Returns: 如果匹配，返回标准品牌名；否则返回None """ phrase_lower = phrase.lower() for brand in self._multi_word_brands: if brand.lower() == phrase_lower or phrase_lower.startswith(brand.lower()): return brand return None def is_ignore_keyword(self, word: str) -> bool: """检查是否为应忽略的关键词""" return word in self._ignore_keywords # 全局单例 brand_config = BrandConfig() def get_brand_config() -> BrandConfig: """获取品牌配置实例""" return brand_config def reload_brand_config(): """重新加载品牌配置""" brand_config.reload()