Simon
431 lines
14 KiB
Python
431 lines
14 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
资产推荐服务 (Asset Recommend Service)
|
||
|
||
用于右侧栏智能推荐的资产信息管理：
|
||
- 关键词匹配资产（L1）
|
||
- 画像触发运维提醒（L2）
|
||
- 角色通用资源（L3）
|
||
|
||
支持 YAML 配置文件热重载
|
||
"""
|
||
|
||
import os
|
||
import json
|
||
import logging
|
||
import yaml
|
||
from pathlib import Path
|
||
from typing import List, Dict, Optional, Any
|
||
from datetime import datetime
|
||
from dataclasses import dataclass, asdict
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
@dataclass
|
||
class RecommendItem:
|
||
"""推荐操作项"""
|
||
label: str
|
||
type: str # download, approval, info, doc, guide, link, contact
|
||
url: Optional[str] = None
|
||
value: Optional[str] = None
|
||
copyable: Optional[bool] = False
|
||
description: Optional[str] = None
|
||
approval_type: Optional[str] = None
|
||
|
||
|
||
@dataclass
|
||
class RecommendCard:
|
||
"""推荐卡片"""
|
||
id: str
|
||
layer: str # L1, L2, L3
|
||
layer_label: str
|
||
source: str # dify_intent, keyword_assets, profile_trigger, role_assets
|
||
title: str
|
||
description: Optional[str] = None
|
||
icon: Optional[str] = None
|
||
items: Optional[List[Dict]] = None
|
||
action_url: Optional[str] = None
|
||
action_label: Optional[str] = None
|
||
confidence: Optional[float] = None
|
||
relevance: str = "high" # high, low
|
||
priority: int = 50
|
||
extra: Optional[Dict] = None
|
||
|
||
|
||
class AssetRecommendService:
|
||
"""
|
||
资产推荐服务
|
||
|
||
功能：
|
||
1. 从 YAML 配置文件加载资产数据
|
||
2. 关键词匹配查询
|
||
3. 画像触发规则匹配
|
||
4. 角色资源查询
|
||
5. 配置热重载
|
||
"""
|
||
|
||
_instance = None
|
||
|
||
def __new__(cls, *args, **kwargs):
|
||
"""单例模式"""
|
||
if cls._instance is None:
|
||
cls._instance = super().__new__(cls)
|
||
cls._instance._initialized = False
|
||
return cls._instance
|
||
|
||
def __init__(self, config_path: str = None):
|
||
if self._initialized:
|
||
return
|
||
|
||
self._initialized = True
|
||
|
||
# 配置路径
|
||
self.config_path = config_path or os.environ.get(
|
||
'ASSETS_CONFIG_PATH',
|
||
str(Path(__file__).parent.parent / 'config' / 'assets.yaml')
|
||
)
|
||
|
||
# 数据存储
|
||
self.keyword_assets: Dict[str, Dict] = {}
|
||
self.role_assets: Dict[str, Dict] = {}
|
||
self._alias_map: Dict[str, str] = {}
|
||
|
||
# 加载配置
|
||
self.load()
|
||
|
||
logger.info(f"[AssetRecommend] 服务初始化完成，配置路径: {self.config_path}")
|
||
|
||
def load(self) -> None:
|
||
"""加载配置文件"""
|
||
try:
|
||
if not os.path.exists(self.config_path):
|
||
logger.warning(f"[AssetRecommend] 配置文件不存在: {self.config_path}")
|
||
return
|
||
|
||
with open(self.config_path, 'r', encoding='utf-8') as f:
|
||
config = yaml.safe_load(f)
|
||
|
||
if not config:
|
||
logger.warning("[AssetRecommend] 配置文件为空")
|
||
return
|
||
|
||
# 加载关键词资产
|
||
keyword_assets = config.get('keyword_assets', {})
|
||
self._build_keyword_assets(keyword_assets)
|
||
|
||
# 加载角色资产
|
||
self.role_assets = config.get('role_assets', {})
|
||
|
||
logger.info(
|
||
f"[AssetRecommend] 配置加载完成: "
|
||
f"关键词资产 {len(self.keyword_assets)} 个, "
|
||
f"角色资产 {len(self.role_assets)} 个"
|
||
)
|
||
|
||
except Exception as e:
|
||
logger.error(f"[AssetRecommend] 配置加载失败: {e}")
|
||
|
||
def _build_keyword_assets(self, keyword_assets: Dict) -> None:
|
||
"""构建关键词资产映射，处理别名"""
|
||
self.keyword_assets = {}
|
||
self._alias_map = {}
|
||
|
||
for keyword, asset in keyword_assets.items():
|
||
# 处理别名
|
||
if 'alias_of' in asset:
|
||
self._alias_map[keyword] = asset['alias_of']
|
||
continue
|
||
|
||
# 存储资产
|
||
self.keyword_assets[keyword] = asset
|
||
|
||
def reload(self) -> None:
|
||
"""热重载配置"""
|
||
logger.info("[AssetRecommend] 开始热重载配置")
|
||
self.load()
|
||
logger.info("[AssetRecommend] 配置热重载完成")
|
||
|
||
def get_by_keyword(self, keyword: str) -> Optional[Dict]:
|
||
"""
|
||
根据关键词查询资产
|
||
|
||
Args:
|
||
keyword: 关键词（如 "vpn"、"打印机"）
|
||
|
||
Returns:
|
||
资产信息字典，未找到返回 None
|
||
"""
|
||
# 查别名
|
||
keyword = self._alias_map.get(keyword, keyword)
|
||
|
||
# 查资产
|
||
asset = self.keyword_assets.get(keyword)
|
||
|
||
if asset:
|
||
logger.debug(f"[AssetRecommend] 命中关键词: {keyword}")
|
||
|
||
return asset
|
||
|
||
def match_keywords(self, message: str) -> List[RecommendCard]:
|
||
"""
|
||
从消息中匹配关键词，返回推荐卡片列表
|
||
|
||
Args:
|
||
message: 用户消息
|
||
|
||
Returns:
|
||
推荐卡片列表
|
||
"""
|
||
cards = []
|
||
message_lower = message.lower()
|
||
|
||
for keyword, asset in self.keyword_assets.items():
|
||
if keyword in message_lower:
|
||
card = self._build_card_from_asset(keyword, asset, source='keyword_assets')
|
||
cards.append(card)
|
||
logger.debug(f"[AssetRecommend] 消息匹配关键词: {keyword}")
|
||
|
||
return cards
|
||
|
||
def get_by_role(self, role: str) -> List[RecommendCard]:
|
||
"""
|
||
根据角色获取推荐卡片
|
||
|
||
Args:
|
||
role: 角色标识（如 "new_employee"、"developer"）
|
||
|
||
Returns:
|
||
推荐卡片列表
|
||
"""
|
||
cards = []
|
||
|
||
# 精确匹配
|
||
if role in self.role_assets:
|
||
asset = self.role_assets[role]
|
||
card = self._build_card_from_asset(role, asset, source='role_assets')
|
||
cards.append(card)
|
||
|
||
# 模糊匹配（角色名包含）
|
||
for role_key, asset in self.role_assets.items():
|
||
if role_key in role or role in role_key:
|
||
if role_key != role: # 避免重复
|
||
card = self._build_card_from_asset(role_key, asset, source='role_assets')
|
||
cards.append(card)
|
||
|
||
return cards
|
||
|
||
def _build_card_from_asset(
|
||
self,
|
||
keyword: str,
|
||
asset: Dict,
|
||
source: str,
|
||
layer: str = 'L1',
|
||
layer_label: str = '相关推荐'
|
||
) -> RecommendCard:
|
||
"""从资产数据构建推荐卡片"""
|
||
|
||
import uuid
|
||
|
||
return RecommendCard(
|
||
id=f"rec_{uuid.uuid4().hex[:8]}",
|
||
layer=layer,
|
||
layer_label=layer_label,
|
||
source=source,
|
||
title=asset.get('title', ''),
|
||
description=asset.get('description'),
|
||
icon=asset.get('icon'),
|
||
items=asset.get('items', []),
|
||
relevance='high' if layer == 'L1' else 'low',
|
||
priority=80 if layer == 'L1' else 50
|
||
)
|
||
|
||
def build_ws_message(self, recommends: List[RecommendCard]) -> Dict:
|
||
"""
|
||
构建 WebSocket 推送消息
|
||
|
||
Args:
|
||
recommends: 推荐卡片列表
|
||
|
||
Returns:
|
||
WS 消息字典
|
||
"""
|
||
return {
|
||
'type': 'asset_recommend',
|
||
'data': {
|
||
'recommends': [asdict(rec) for rec in recommends],
|
||
'layered': True
|
||
}
|
||
}
|
||
|
||
def match_profile_triggers(self, profile: Dict) -> List[RecommendCard]:
|
||
"""
|
||
画像触发规则匹配
|
||
|
||
Args:
|
||
profile: 员工画像数据
|
||
|
||
Returns:
|
||
推荐卡片列表（L2）
|
||
"""
|
||
cards = []
|
||
|
||
# 1. 火绒终端版本过旧
|
||
huorong_version = profile.get('huorong_version', '')
|
||
if huorong_version:
|
||
try:
|
||
current = float(huorong_version.replace('5.0.', ''))
|
||
if current < 73: # 假设最新版本是 5.0.73
|
||
cards.append(RecommendCard(
|
||
id=f"rec_huorong_version_{datetime.now().strftime('%Y%m%d')}",
|
||
layer='L2',
|
||
layer_label='运维提醒',
|
||
source='profile_trigger',
|
||
title='火绒终端版本过旧',
|
||
description=f'您的火绒终端版本为 {huorong_version}，最新版本为 5.0.73，建议更新',
|
||
icon='🔶',
|
||
action_url='/itportal/resource/huorong-update',
|
||
relevance='low',
|
||
priority=90
|
||
))
|
||
except (ValueError, AttributeError):
|
||
pass
|
||
|
||
# 2. 火绒病毒库过期
|
||
virusdb_date = profile.get('huorong_virusdb_date')
|
||
if virusdb_date:
|
||
try:
|
||
if isinstance(virusdb_date, str):
|
||
virusdb_date = datetime.fromisoformat(virusdb_date.replace('Z', '+00:00'))
|
||
days_since = (datetime.now() - virusdb_date.replace(tzinfo=None)).days
|
||
if days_since > 7:
|
||
cards.append(RecommendCard(
|
||
id=f"rec_virusdb_{datetime.now().strftime('%Y%m%d')}",
|
||
layer='L2',
|
||
layer_label='运维提醒',
|
||
source='profile_trigger',
|
||
title='火绒病毒库过期',
|
||
description=f'病毒库已 {days_since} 天未更新，建议立即升级',
|
||
icon='🦠',
|
||
action_url='/itportal/resource/huorong-update',
|
||
relevance='low',
|
||
priority=80
|
||
))
|
||
except (ValueError, AttributeError):
|
||
pass
|
||
|
||
# 3. 终端防护离线
|
||
offline_days = profile.get('huorong_offline_days', 0)
|
||
if offline_days > 3:
|
||
cards.append(RecommendCard(
|
||
id=f"rec_offline_{datetime.now().strftime('%Y%m%d')}",
|
||
layer='L2',
|
||
layer_label='运维提醒',
|
||
source='profile_trigger',
|
||
title='终端防护离线',
|
||
description=f'火绒终端已离线 {offline_days} 天，请检查网络连接',
|
||
icon='📡',
|
||
action_url='/itportal/resource/huorong-reconnect',
|
||
relevance='low',
|
||
priority=100 # 最高优先级
|
||
))
|
||
|
||
# 4. 系统补丁缺失
|
||
missing_patches = profile.get('unionsoft_patches_missing', 0)
|
||
if missing_patches > 10:
|
||
cards.append(RecommendCard(
|
||
id=f"rec_patches_{datetime.now().strftime('%Y%m%d')}",
|
||
layer='L2',
|
||
layer_label='运维提醒',
|
||
source='profile_trigger',
|
||
title='系统补丁缺失',
|
||
description=f'您的电脑有 {missing_patches} 个安全补丁未安装，可能存在安全风险',
|
||
icon='🔧',
|
||
action_url='/itportal/resource/windows-update',
|
||
relevance='low',
|
||
priority=90
|
||
))
|
||
|
||
# 5. 联软违规项
|
||
violations = profile.get('unionsoft_violations', [])
|
||
if violations:
|
||
cards.append(RecommendCard(
|
||
id=f"rec_violations_{datetime.now().strftime('%Y%m%d')}",
|
||
layer='L2',
|
||
layer_label='运维提醒',
|
||
source='profile_trigger',
|
||
title='安全策略违规',
|
||
description=f'检测到 {len(violations)} 项违规项',
|
||
icon='⚠️',
|
||
action_url='/itportal/resource/security-compliance',
|
||
relevance='low',
|
||
priority=95,
|
||
extra={'violations': violations}
|
||
))
|
||
|
||
return cards
|
||
|
||
def process_dify_intent(
|
||
self,
|
||
dify_result: Dict,
|
||
employee_id: str,
|
||
profile: Optional[Dict] = None
|
||
) -> List[RecommendCard]:
|
||
"""
|
||
处理 Dify 意图输出，生成资产推荐
|
||
|
||
Args:
|
||
dify_result: Dify 返回的 JSON（包含 intent, need_asset, asset_keywords）
|
||
employee_id: 员工 ID
|
||
profile: 员工画像（可选）
|
||
|
||
Returns:
|
||
推荐卡片列表
|
||
"""
|
||
recommends = []
|
||
|
||
# L1: Dify 意图触发的资产
|
||
if dify_result.get('need_asset') and dify_result.get('asset_keywords'):
|
||
for keyword in dify_result['asset_keywords']:
|
||
asset = self.get_by_keyword(keyword)
|
||
if asset:
|
||
card = self._build_card_from_asset(
|
||
keyword,
|
||
asset,
|
||
source='dify_intent',
|
||
layer='L1',
|
||
layer_label='相关推荐'
|
||
)
|
||
card.confidence = dify_result.get('confidence', 0.85)
|
||
recommends.append(card)
|
||
|
||
# L2: 画像触发（如果提供了 profile）
|
||
if profile:
|
||
profile_recs = self.match_profile_triggers(profile)
|
||
recommends.extend(profile_recs)
|
||
|
||
# L3: 角色通用（如果提供了 profile）
|
||
if profile:
|
||
role = profile.get('position') or profile.get('role', '')
|
||
if role:
|
||
role_recs = self.get_by_role(role)
|
||
for rec in role_recs:
|
||
rec.layer = 'L3'
|
||
rec.layer_label = '常用资源'
|
||
rec.relevance = 'low'
|
||
recommends.extend(role_recs)
|
||
|
||
return recommends
|
||
|
||
|
||
# 全局单例
|
||
_asset_recommend_service: Optional[AssetRecommendService] = None
|
||
|
||
|
||
def get_asset_recommend_service() -> AssetRecommendService:
|
||
"""获取资产推荐服务单例"""
|
||
global _asset_recommend_service
|
||
if _asset_recommend_service is None:
|
||
_asset_recommend_service = AssetRecommendService()
|
||
return _asset_recommend_service
|