SortOrder order = sortOrder.equalsIgnoreCase("asc") ? SortOrder.ASC : SortOrder.DESC;
sourceBuilder.sort(new FieldSortBuilder(sortField).order(order));

sourceBuilder.from((pageNum-1)*pageSize).size(pageSize);

HighlightBuilder highlightBuilder = new HighlightBuilder();
            highlightBuilder.field("name").preTags(HIGHLIGHT_PRE_TAG).postTags(HIGHLIGHT_POST_TAG);
            sourceBuilder.highlighter(highlightBuilder);

if (hit.getHighlightFields()!=null){
                HighlightField nameField = hit.getHighlightFields().get("name");
                if (nameField!=null){
                    Text[] fragments = nameField.fragments();
                    String highlightedName = fragments[0].toString();
                    shop.setName(highlightedName);
                }
            }

SortOrder order = sortOrder.equalsIgnoreCase("asc") ? SortOrder.ASC : SortOrder.DESC;
            sourceBuilder.sort(new FieldSortBuilder(sortField).order(order));

Object[] nextSearchAfter = null;
            if (shops.size() > 0) {
                SearchHit lasthit = searchResponse.getHits().getHits()[shops.size() - 1];
                nextSearchAfter = lasthit.getSortValues();
            }

public SearchResult<Shop> searchShopsAfter(String keywords, int pageSize, Object[] searchAfterSortValues, String sortField, String sortOrder) throws IOException {

        try {
            SearchRequest searchRequest = new SearchRequest(INDEX_NAME);
            SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
            BoolQueryBuilder boolQueryBuilder = new BoolQueryBuilder();

            MatchQueryBuilder matchQueryBuilder = QueryBuilders.matchQuery("name", keywords);
            boolQueryBuilder.must(matchQueryBuilder);

            sourceBuilder.query(boolQueryBuilder);
            sourceBuilder.size(pageSize);

            SortOrder order = sortOrder.equalsIgnoreCase("asc") ? SortOrder.ASC : SortOrder.DESC;
            sourceBuilder.sort(new FieldSortBuilder(sortField).order(order));

            if (searchAfterSortValues != null) {
                sourceBuilder.searchAfter(searchAfterSortValues);
            }

            searchRequest.source(sourceBuilder);
            SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
            List<Shop> shops = convertSearchHitsToShops(searchResponse);

            Object[] nextSearchAfter = null;
            if (shops.size() > 0) {
                SearchHit lasthit = searchResponse.getHits().getHits()[shops.size() - 1];
                nextSearchAfter = lasthit.getSortValues();
            }
            SearchResult<Shop> result = new SearchResult<>();
            result.setItems(shops);
            result.setNextSearchAfter(nextSearchAfter);
            return result;
        } catch (IOException e) {
            log.info("使用 search_after 搜索店铺时发生IO异常", e);
            return null;
        }

    }

//后续请求
            SearchScrollRequest scrollRequest = new SearchScrollRequest(scrollId);
            scrollRequest.scroll(TimeValue.parseTimeValue(scrollTime, null, "scroll"));
            SearchResponse searchResponse = client.scroll(scrollRequest, RequestOptions.DEFAULT);
            return convertSearchHitsToShops(searchResponse);

package com.recommendation.core;

import java.util.List;

/**
 * 推荐引擎核心接口
 * 支持多种推荐策略的统一入口
 */
public interface RecommendationEngine {
  
    /**
     * 获取个性化推荐
     * @param userId 用户ID
     * @param size 推荐数量
     * @param scene 推荐场景(首页/详情页/搜索后等)
     * @return 推荐结果列表
     */
    List<RecommendationResult> recommend(Long userId, Integer size, String scene);
  
    /**
     * 实时更新用户行为
     * @param userBehavior 用户行为数据
     */
    void updateUserBehavior(UserBehavior userBehavior);
}

package com.recommendation.engine;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.cache.annotation.Cacheable;
import lombok.extern.slf4j.Slf4j;

import java.util.*;
import java.util.stream.Collectors;
import java.math.BigDecimal;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * 混合推荐引擎实现
 * 结合协同过滤、内容推荐、热门推荐等多种策略
 */
@Slf4j
@Service
public class HybridRecommendationEngine implements RecommendationEngine {
  
    @Autowired
    private CollaborativeFilteringService cfService;
  
    @Autowired
    private ContentBasedService contentService;
  
    @Autowired
    private HotRecommendationService hotService;
  
    @Autowired
    private UserProfileService userProfileService;
  
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
  
    @Autowired
    private ThreadPoolExecutor recommendationExecutor;
  
    // 推荐策略权重配置
    private static final Map<String, BigDecimal> STRATEGY_WEIGHTS = Map.of(
        "collaborative", new BigDecimal("0.4"),
        "content", new BigDecimal("0.4"),
        "hot", new BigDecimal("0.2")
    );
  
    @Override
    @Cacheable(value = "recommendations", key = "#userId + '_' + #size + '_' + #scene", 
               unless = "#result.size() == 0")
    public List<RecommendationResult> recommend(Long userId, Integer size, String scene) {
        log.info("开始为用户 {} 生成推荐，场景: {}, 数量: {}", userId, scene, size);
      
        try {
            // 获取用户画像
            UserProfile userProfile = userProfileService.getUserProfile(userId);
            if (userProfile == null) {
                return hotService.getHotRecommendations(size, scene);
            }
          
            // 并行执行多种推荐策略
            CompletableFuture<List<RecommendationItem>> cfFuture = 
                CompletableFuture.supplyAsync(() -> 
                    cfService.getCollaborativeRecommendations(userId, size * 2), 
                    recommendationExecutor);
          
            CompletableFuture<List<RecommendationItem>> contentFuture = 
                CompletableFuture.supplyAsync(() -> 
                    contentService.getContentBasedRecommendations(userProfile, size * 2), 
                    recommendationExecutor);
          
            CompletableFuture<List<RecommendationItem>> hotFuture = 
                CompletableFuture.supplyAsync(() -> 
                    hotService.getHotRecommendations(size, scene).stream()
                        .map(r -> new RecommendationItem(r.getShopId(), r.getScore()))
                        .collect(Collectors.toList()), 
                    recommendationExecutor);
          
            // 等待所有任务完成
            CompletableFuture.allOf(cfFuture, contentFuture, hotFuture).join();
          
            // 融合推荐结果
            List<RecommendationItem> hybridResults = mergeRecommendations(
                cfFuture.get(), contentFuture.get(), hotFuture.get()
            );
          
            // 转换为最终结果并应用业务规则
            return buildFinalResults(hybridResults, userProfile, scene, size);
          
        } catch (Exception e) {
            log.error("推荐生成失败，用户: {}, 错误: {}", userId, e.getMessage(), e);
            // 降级到热门推荐
            return hotService.getHotRecommendations(size, scene);
        }
    }
  
    /**
     * 融合多种推荐策略的结果
     */
    private List<RecommendationItem> mergeRecommendations(
            List<RecommendationItem> cfResults,
            List<RecommendationItem> contentResults,
            List<RecommendationItem> hotResults) {
      
        Map<Long, BigDecimal> scoreMap = new HashMap<>();
      
        // 协同过滤结果加权
        cfResults.forEach(item -> 
            scoreMap.merge(item.getShopId(), 
                item.getScore().multiply(STRATEGY_WEIGHTS.get("collaborative")), 
                BigDecimal::add));
      
        // 内容推荐结果加权
        contentResults.forEach(item -> 
            scoreMap.merge(item.getShopId(), 
                item.getScore().multiply(STRATEGY_WEIGHTS.get("content")), 
                BigDecimal::add));
      
        // 热门推荐结果加权
        hotResults.forEach(item -> 
            scoreMap.merge(item.getShopId(), 
                item.getScore().multiply(STRATEGY_WEIGHTS.get("hot")), 
                BigDecimal::add));
      
        return scoreMap.entrySet().stream()
            .map(entry -> new RecommendationItem(entry.getKey(), entry.getValue()))
            .sorted((a, b) -> b.getScore().compareTo(a.getScore()))
            .collect(Collectors.toList());
    }
  
    @Override
    public void updateUserBehavior(UserBehavior userBehavior) {
        // 异步更新用户行为，不阻塞主流程
        CompletableFuture.runAsync(() -> {
            try {
                // 更新实时特征
                userProfileService.updateRealTimeFeatures(userBehavior);
              
                // 发送到Kafka进行离线处理
                kafkaTemplate.send("user_behavior", userBehavior);
              
                // 清除相关缓存
                String cacheKey = "recommendations:" + userBehavior.getUserId() + "*";
                redisTemplate.delete(redisTemplate.keys(cacheKey));
              
            } catch (Exception e) {
                log.error("更新用户行为失败: {}", e.getMessage(), e);
            }
        }, recommendationExecutor);
    }
}

package com.recommendation.algorithm;

import org.springframework.stereotype.Service;
import lombok.extern.slf4j.Slf4j;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.*;
import java.util.stream.Collectors;

/**
 * 基于用户的协同过滤算法
 * 使用改进的余弦相似度计算用户相似性
 */
@Slf4j
@Service
public class CollaborativeFilteringService {
  
    @Autowired
    private UserBehaviorRepository behaviorRepository;
  
    @Autowired
    private ShopRepository shopRepository;
  
    /**
     * 获取协同过滤推荐
     */
    public List<RecommendationItem> getCollaborativeRecommendations(Long userId, Integer size) {
      
        // 1. 获取用户行为数据
        List<UserBehavior> userBehaviors = behaviorRepository.findByUserId(userId);
        if (userBehaviors.isEmpty()) {
            return Collections.emptyList();
        }
      
        // 2. 构建用户-商店评分矩阵
        Map<Long, Map<Long, BigDecimal>> userItemMatrix = buildUserItemMatrix(userId);
      
        // 3. 计算用户相似度
        Map<Long, BigDecimal> userSimilarities = calculateUserSimilarities(userId, userItemMatrix);
      
        // 4. 生成推荐
        return generateRecommendations(userId, userSimilarities, userItemMatrix, size);
    }
  
    /**
     * 构建用户-商店评分矩阵
     */
    private Map<Long, Map<Long, BigDecimal>> buildUserItemMatrix(Long targetUserId) {
      
        // 获取目标用户喜欢的店铺类型
        Set<String> targetUserCategories = getUserPreferredCategories(targetUserId);
      
        // 查找有相似偏好的用户（优化：避免计算所有用户）
        List<Long> similarUsers = behaviorRepository.findUsersByCategories(
            targetUserCategories, 1000); // 限制计算用户数量
      
        Map<Long, Map<Long, BigDecimal>> matrix = new HashMap<>();
      
        for (Long userId : similarUsers) {
            List<UserBehavior> behaviors = behaviorRepository.findByUserId(userId);
            Map<Long, BigDecimal> userRatings = new HashMap<>();
          
            for (UserBehavior behavior : behaviors) {
                // 根据行为类型计算隐式评分
                BigDecimal score = calculateImplicitRating(behavior);
                userRatings.merge(behavior.getShopId(), score, BigDecimal::add);
            }
          
            matrix.put(userId, userRatings);
        }
      
        return matrix;
    }
  
    /**
     * 计算隐式评分
     * 浏览=1分，收藏=3分，评论=5分，高评分评论=8分
     */
    private BigDecimal calculateImplicitRating(UserBehavior behavior) {
        switch (behavior.getBehaviorType()) {
            case "VIEW":
                return BigDecimal.valueOf(1.0);
            case "FAVORITE":
                return BigDecimal.valueOf(3.0);
            case "COMMENT":
                // 根据评论星级调整分数
                Double rating = behavior.getRating();
                if (rating != null && rating >= 4.0) {
                    return BigDecimal.valueOf(8.0);
                } else if (rating != null && rating >= 3.0) {
                    return BigDecimal.valueOf(5.0);
                } else {
                    return BigDecimal.valueOf(2.0);
                }
            case "SHARE":
                return BigDecimal.valueOf(6.0);
            default:
                return BigDecimal.valueOf(1.0);
        }
    }
  
    /**
     * 计算用户相似度（改进的余弦相似度）
     */
    private Map<Long, BigDecimal> calculateUserSimilarities(
            Long targetUserId, 
            Map<Long, Map<Long, BigDecimal>> userItemMatrix) {
      
        Map<Long, BigDecimal> targetUserRatings = userItemMatrix.get(targetUserId);
        if (targetUserRatings == null || targetUserRatings.isEmpty()) {
            return Collections.emptyMap();
        }
      
        Map<Long, BigDecimal> similarities = new HashMap<>();
      
        for (Map.Entry<Long, Map<Long, BigDecimal>> entry : userItemMatrix.entrySet()) {
            Long userId = entry.getKey();
            if (userId.equals(targetUserId)) {
                continue;
            }
          
            Map<Long, BigDecimal> userRatings = entry.getValue();
            BigDecimal similarity = calculateCosineSimilarity(targetUserRatings, userRatings);
          
            if (similarity.compareTo(BigDecimal.valueOf(0.1)) > 0) { // 过滤低相似度用户
                similarities.put(userId, similarity);
            }
        }
      
        return similarities;
    }
  
    /**
     * 余弦相似度计算
     */
    private BigDecimal calculateCosineSimilarity(
            Map<Long, BigDecimal> ratingsA, 
            Map<Long, BigDecimal> ratingsB) {
      
        Set<Long> commonItems = new HashSet<>(ratingsA.keySet());
        commonItems.retainAll(ratingsB.keySet());
      
        if (commonItems.isEmpty()) {
            return BigDecimal.ZERO;
        }
      
        BigDecimal dotProduct = BigDecimal.ZERO;
        BigDecimal normA = BigDecimal.ZERO;
        BigDecimal normB = BigDecimal.ZERO;
      
        for (Long itemId : commonItems) {
            BigDecimal ratingA = ratingsA.get(itemId);
            BigDecimal ratingB = ratingsB.get(itemId);
          
            dotProduct = dotProduct.add(ratingA.multiply(ratingB));
            normA = normA.add(ratingA.multiply(ratingA));
            normB = normB.add(ratingB.multiply(ratingB));
        }
      
        if (normA.equals(BigDecimal.ZERO) || normB.equals(BigDecimal.ZERO)) {
            return BigDecimal.ZERO;
        }
      
        BigDecimal denominator = sqrt(normA).multiply(sqrt(normB));
        return dotProduct.divide(denominator, 4, RoundingMode.HALF_UP);
    }
  
    /**
     * 生成最终推荐列表
     */
    private List<RecommendationItem> generateRecommendations(
            Long targetUserId,
            Map<Long, BigDecimal> userSimilarities,
            Map<Long, Map<Long, BigDecimal>> userItemMatrix,
            Integer size) {
      
        Map<Long, BigDecimal> targetUserRatings = userItemMatrix.get(targetUserId);
        Set<Long> targetUserItems = targetUserRatings.keySet();
      
        Map<Long, BigDecimal> itemScores = new HashMap<>();
      
        for (Map.Entry<Long, BigDecimal> simEntry : userSimilarities.entrySet()) {
            Long similarUserId = simEntry.getKey();
            BigDecimal similarity = simEntry.getValue();
          
            Map<Long, BigDecimal> similarUserRatings = userItemMatrix.get(similarUserId);
          
            for (Map.Entry<Long, BigDecimal> ratingEntry : similarUserRatings.entrySet()) {
                Long shopId = ratingEntry.getKey();
                BigDecimal rating = ratingEntry.getValue();
              
                // 跳过用户已经交互过的商店
                if (targetUserItems.contains(shopId)) {
                    continue;
                }
              
                BigDecimal score = similarity.multiply(rating);
                itemScores.merge(shopId, score, BigDecimal::add);
            }
        }
      
        return itemScores.entrySet().stream()
            .sorted((e1, e2) -> e2.getValue().compareTo(e1.getValue()))
            .limit(size)
            .map(entry -> new RecommendationItem(entry.getKey(), entry.getValue()))
            .collect(Collectors.toList());
    }
  
    // 简化的平方根计算
    private BigDecimal sqrt(BigDecimal value) {
        return BigDecimal.valueOf(Math.sqrt(value.doubleValue()));
    }
}

package com.recommendation.service;

import org.springframework.stereotype.Service;
import org.springframework.data.redis.core.RedisTemplate;
import lombok.extern.slf4j.Slf4j;

import java.time.LocalDateTime;
import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * 用户画像服务
 * 负责用户特征的实时更新和画像构建
 */
@Slf4j
@Service
public class UserProfileService {
  
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
  
    @Autowired
    private UserBehaviorRepository behaviorRepository;
  
    private static final String USER_PROFILE_KEY = "user_profile:";
    private static final String REAL_TIME_FEATURES_KEY = "rt_features:";
  
    /**
     * 获取用户画像
     */
    public UserProfile getUserProfile(Long userId) {
        String key = USER_PROFILE_KEY + userId;
        UserProfile profile = (UserProfile) redisTemplate.opsForValue().get(key);
      
        if (profile == null) {
            profile = buildUserProfile(userId);
            if (profile != null) {
                redisTemplate.opsForValue().set(key, profile, 24, TimeUnit.HOURS);
            }
        }
      
        return profile;
    }
  
    /**
     * 构建用户画像
     */
    private UserProfile buildUserProfile(Long userId) {
        try {
            // 获取用户历史行为
            List<UserBehavior> behaviors = behaviorRepository.findRecentBehaviors(
                userId, LocalDateTime.now().minusDays(30));
          
            if (behaviors.isEmpty()) {
                return null;
            }
          
            UserProfile profile = new UserProfile();
            profile.setUserId(userId);
          
            // 计算用户偏好类别
            Map<String, Long> categoryPreferences = behaviors.stream()
                .collect(Collectors.groupingBy(
                    behavior -> getShopCategory(behavior.getShopId()),
                    Collectors.counting()));
          
            profile.setPreferredCategories(categoryPreferences);
          
            // 计算平均评分
            OptionalDouble avgRating = behaviors.stream()
                .filter(b -> b.getRating() != null)
                .mapToDouble(UserBehavior::getRating)
                .average();
          
            if (avgRating.isPresent()) {
                profile.setAverageRating(BigDecimal.valueOf(avgRating.getAsDouble()));
            }
          
            // 计算活跃度
            long recentBehaviorCount = behaviors.stream()
                .filter(b -> b.getCreateTime().isAfter(LocalDateTime.now().minusDays(7)))
                .count();
          
            profile.setActivityLevel(calculateActivityLevel(recentBehaviorCount));
          
            // 计算偏好价格区间
            profile.setPricePreference(calculatePricePreference(behaviors));
          
            profile.setLastUpdateTime(LocalDateTime.now());
          
            return profile;
          
        } catch (Exception e) {
            log.error("构建用户画像失败，用户ID: {}, 错误: {}", userId, e.getMessage(), e);
            return null;
        }
    }
  
    /**
     * 更新实时特征
     */
    public void updateRealTimeFeatures(UserBehavior userBehavior) {
        String key = REAL_TIME_FEATURES_KEY + userBehavior.getUserId();
      
        try {
            // 更新实时行为计数
            String behaviorCountKey = key + ":behavior_count";
            redisTemplate.opsForValue().increment(behaviorCountKey);
            redisTemplate.expire(behaviorCountKey, 1, TimeUnit.HOURS);
          
            // 更新最近浏览类别
            if ("VIEW".equals(userBehavior.getBehaviorType())) {
                String categoryKey = key + ":recent_categories";
                String category = getShopCategory(userBehavior.getShopId());
                redisTemplate.opsForZSet().add(categoryKey, category, System.currentTimeMillis());
                redisTemplate.expire(categoryKey, 24, TimeUnit.HOURS);
              
                // 只保留最近的10个类别
                redisTemplate.opsForZSet().removeRange(categoryKey, 0, -11);
            }
          
            // 如果是高价值行为，清除推荐缓存
            if (isHighValueBehavior(userBehavior)) {
                String cacheKey = "recommendations:" + userBehavior.getUserId() + "*";
                Set<String> keys = redisTemplate.keys(cacheKey);
                if (!keys.isEmpty()) {
                    redisTemplate.delete(keys);
                }
            }
          
        } catch (Exception e) {
            log.error("更新实时特征失败: {}", e.getMessage(), e);
        }
    }
  
    private boolean isHighValueBehavior(UserBehavior behavior) {
        return "FAVORITE".equals(behavior.getBehaviorType()) 
            || "COMMENT".equals(behavior.getBehaviorType())
            || "SHARE".equals(behavior.getBehaviorType());
    }
}

package com.recommendation.controller;

import org.springframework.web.bind.annotation.*;
import org.springframework.validation.annotation.Validated;
import lombok.extern.slf4j.Slf4j;

import javax.validation.Valid;
import javax.validation.constraints.Max;
import javax.validation.constraints.Min;
import java.util.List;

/**
 * 推荐服务API控制器
 */
@Slf4j
@RestController
@RequestMapping("/api/v1/recommendations")
@Validated
public class RecommendationController {
  
    @Autowired
    private RecommendationEngine recommendationEngine;
  
    @Autowired
    private RecommendationMetricsService metricsService;
  
    /**
     * 获取个性化推荐
     */
    @GetMapping("/users/{userId}")
    public ApiResponse<List<RecommendationResult>> getUserRecommendations(
            @PathVariable Long userId,
            @RequestParam(defaultValue = "10") @Min(1) @Max(50) Integer size,
            @RequestParam(defaultValue = "homepage") String scene) {
      
        long startTime = System.currentTimeMillis();
      
        try {
            List<RecommendationResult> recommendations = 
                recommendationEngine.recommend(userId, size, scene);
          
            // 记录推荐指标
            metricsService.recordRecommendation(userId, scene, recommendations.size());
          
            long duration = System.currentTimeMillis() - startTime;
            log.info("推荐请求完成，用户: {}, 场景: {}, 耗时: {}ms", userId, scene, duration);
          
            return ApiResponse.success(recommendations);
          
        } catch (Exception e) {
            log.error("推荐请求失败，用户: {}, 错误: {}", userId, e.getMessage(), e);
            return ApiResponse.error("推荐服务暂时不可用");
        }
    }
  
    /**
     * 用户行为上报
     */
    @PostMapping("/behaviors")
    public ApiResponse<Void> reportUserBehavior(@Valid @RequestBody UserBehaviorRequest request) {
      
        try {
            UserBehavior behavior = UserBehavior.builder()
                .userId(request.getUserId())
                .shopId(request.getShopId())
                .behaviorType(request.getBehaviorType())
                .rating(request.getRating())
                .createTime(LocalDateTime.now())
                .build();
          
            recommendationEngine.updateUserBehavior(behavior);
          
            return ApiResponse.success();
          
        } catch (Exception e) {
            log.error("用户行为上报失败: {}", e.getMessage(), e);
            return ApiResponse.error("行为上报失败");
        }
    }
  
    /**
     * 获取推荐解释
     */
    @GetMapping("/explanations/{userId}/{shopId}")
    public ApiResponse<RecommendationExplanation> getRecommendationExplanation(
            @PathVariable Long userId,
            @PathVariable Long shopId) {
      
        try {
            RecommendationExplanation explanation = 
                recommendationEngine.explainRecommendation(userId, shopId);
          
            return ApiResponse.success(explanation);
          
        } catch (Exception e) {
            log.error("获取推荐解释失败，用户: {}, 店铺: {}, 错误: {}", 
                     userId, shopId, e.getMessage(), e);
            return ApiResponse.error("获取推荐解释失败");
        }
    }
}