Redis搜索应用

|
import re
 
# 定义停用词集合 - 这些词出现频率太高，对搜索意义不大
STOP_WORDS = {
    '的', '了', '在', '是', '我', '有', '和', '就', '不', '人', '都', '一', '一个', '上', '也', '很', '到', '说', '要', '去', '你', '会', '着', '没有', '看', 
}
 
# 中文分词的正则表达式：匹配中文字符
CHINESE_WORDS_RE = re.compile(r'[\u4e00-\u9fff]+')
 
def tokenize_chinese(content):
    """
    将中文文档内容分词，去除停用词
    参数: content - 原始文档内容
    返回: 去重后的词汇集合
    """
    words = set()
    
    # 使用正则表达式提取中文词汇
    for match in CHINESE_WORDS_RE.finditer(content):
        word = match.group()
        
        # 只保留长度至少为2的词
        if len(word) >= 2:
            words.add(word)
    
    # 返回去除停用词后的词汇集合
    return words - STOP_WORDS
 
# 测试中文分词功能
tutorial_content = "Redis缓存优化策略详解，提升系统性能的关键技术"
tokens = tokenize_chinese(tutorial_content)
print(f"分词结果: {tokens}")
# 输出: {'redis', '缓存', '优化', '策略', '详解', '提升', '系统', '性能', '关键', '技术'}

|
def index_tutorial(conn, tutorial_id, content):
    """
    为教程建立倒排索引
    参数: 
        conn - Redis连接对象
        tutorial_id - 教程唯一标识
        content - 教程内容
    返回: 添加到索引中的词汇数量
    """
    # 对教程内容进行分词
    words = tokenize_chinese(content)
    
    # 使用管道批量执行Redis命令，提高性能
    pipeline = conn.pipeline(True)
    
    # 将教程ID添加到每个词对应的SET中
    for word in words:
        pipeline.sadd(f'idx:{word

|
import uuid
 
def _set_common(conn, method, names, ttl=30, execute=True):
    """
    执行SET操作的通用辅助函数
    参数:
        conn - Redis连接对象
        method - SET操作方法名（sinterstore, sunionstore, sdiffstore）
        names - 要操作的SET名称列表
        ttl - 结果SET的过期时间（秒）
        execute - 是否立即执行命令
    返回: 结果SET的ID
    """
    # 生成唯一的临时SET标识符
    result_id = str(uuid.uuid4())
    
    # 设置管道，确保操作的原子性
    pipeline = conn.pipeline(True) 

|
import re
 
# 查询解析的正则表达式：匹配带前缀的词
QUERY_RE = re.compile(r"[+-]?[\u4e00-\u9fff]+")
 
def parse_query(query):
    """
    解析搜索查询，支持同义词和排除词
    参数: query - 用户输入的搜索查询
    返回: (同义词组列表, 排除词列表)
    
    查询语法说明:
    - 普通词: 必须包含的词
    - +词: 同义词（与前面的词是同义词）
    - -词: 排除词（不包含的词）
    
    示例: "缓存 +缓存策略 +缓存优化 数据库 -nosql"
    """
    unwanted = set()  # 排除词集合

|
def search_and_sort(conn, query, result_id=None, ttl=300, sort="-updated", start=0, num=20):
    """
    搜索并排序结果
    参数:
        conn - Redis连接对象
        query - 搜索查询
        result_id - 之前的搜索结果ID（用于分页）
        ttl - 结果缓存时间
        sort - 排序字段（-updated表示按更新时间降序）
        start - 起始位置（分页）
        num - 返回数量
    返回: (总数量, 结果列表, 结果ID)
    """
    # 判断是否降序排序
    desc = sort.startswith('-'

|
def search_and_zsort(conn, query, result_id=None, ttl=300, 
                    like_weight=1, time_weight=0, author_weight=0,
                    start=0, num=20, desc=True):
    """
    使用ZSET进行复合排序搜索
    参数:
        conn - Redis连接对象
        query - 搜索查询
        result_id - 之前的搜索结果ID
        ttl - 结果缓存时间
        like_weight - 点赞数权重

|
def string_to_score(string, ignore_case=False):
    """
    将字符串转换为数值分数，用于ZSET排序
    参数:
        string - 要转换的字符串
        ignore_case - 是否忽略大小写
    返回: 转换后的数值分数
    """
    if ignore_case:
        string = string.lower()
    
    # 只取前6个字符，避免数值过大
    pieces = list(map(ord, string[:6]))
    
    # 如果字符串不足6个字符，用-1填充
    while len(pieces) 

|
def cpc_to_ecpm(views, clicks, cpc):
    """
    将CPC广告转换为eCPM
    参数:
        views - 广告展示次数
        clicks - 广告点击次数
        cpc - 每次点击的费用
    返回: eCPM值
    """
    if views == 0:
        return 0
    return 1000.0 * cpc * clicks / views
 
def cpa_to_ecpm(views, actions, cpa):
    """
    将CPA广告转换为eCPM
    参数:
        views - 广告展示次数
        actions - 用户行动次数

|
def index_ad(conn, ad_id, locations, content, ad_type, value):
    """
    为广告建立索引
    参数:
        conn - Redis连接对象
        ad_id - 广告唯一标识
        locations - 目标地理位置列表
        content - 广告内容描述
        ad_type - 广告类型（cpm, cpc, cpa）
        value - 广告价值（费用或出价）
    """
    pipeline = conn.pipeline(True)
    
    # 为每个地理位置建立索引
    for location in locations:
        pipeline.sadd(f'req:{location}', ad_id)
    

|
def target_ads(conn, locations, content):
    """
    广告定向主函数
    参数:
        conn - Redis连接对象
        locations - 用户地理位置
        content - 页面内容
    返回: (目标ID, 广告ID)
    """
    pipeline = conn.pipeline(True)
    
    # 第一步：位置匹配
    matched_ads, base_ecpm = match_location(pipeline, locations)
    
    # 第二步：内容匹配和eCPM计算
    words, targeted_ads = finish_scoring(
        pipeline, matched_ads, base_ecpm, content
    )
    
    # 记录广告展示
    pipeline.incr(

|
def record_targeting_result(conn, target_id, ad_id, words):
    """
    记录广告定向结果，用于学习优化
    参数:
        conn - Redis连接对象
        target_id - 定向ID
        ad_id - 广告ID
        words - 匹配的词汇
    """
    pipeline = conn.pipeline(True)
    
    # 获取广告的所有词汇
    ad_terms = conn.smembers(f'terms:{ad_id}')
    
    # 找出实际匹配的词汇
    matched_words = list(words &

|
def index_job(conn, job_id, required_skills):
    """
    为职位建立技能索引
    参数:
        conn - Redis连接对象
        job_id - 职位唯一标识
        required_skills - 必需技能列表
    """
    pipeline = conn.pipeline(True)
    
    # 为每个技能建立索引
    for skill in required_skills:
        pipeline.sadd(f'idx:skill:{skill}', job_id)
    
    # 记录职位需要的技能总数
    pipeline.zadd('idx:jobs:req', {job_id: len

|
def find_partial_matches(conn, candidate_skills, min_match_ratio=0.5):
    """
    找到部分匹配的职位
    参数:
        conn - Redis连接对象
        candidate_skills - 求职者技能列表
        min_match_ratio - 最小匹配比例
    返回: [(职位ID, 匹配比例), ...]
    """
    # 构建技能权重字典
    skills = {}
    for skill in set(candidate_skills):
        skills[f'skill:{skill}'] = 1
    
    # 计算求职者具备的技能数量