es查询代码
@es_data_router.get("/search", summary="职位搜索") async def search_jobs( # ---------- 关键词 ---------- keyword: Optional[str] = Query(None, description="搜索关键词，如 Java、前端"), # ---------- 筛选 ---------- work_location: Optional[str] = Query(None, description="工作地点（精确匹配）"), edu_require: Optional[str] = Query(None, description="学历要求"), exp_require: Optional[str] = Query(None, description="经验要求"), industry_id: Optional[int] = Query(None, description="行业ID"), company_scale: Optional[int] = Query(None, description="公司规模枚举值"), financing_stage: Optional[int] = Query(None, description="融资阶段枚举值"), status: Optional[int] = Query(1, description="职位状态，默认1=招聘中"), # ---------- 分页 / 排序 ---------- page: int = Query(1, ge=1, description="页码"), page_size: int = Query(10, ge=1, le=50, description="每页条数"), sort: str = Query("time", description="排序：score=相关度，time=发布时间"), es_client: AsyncElasticsearch = Depends(es_client_depend), ): """ 职位搜索思路： 1. bool.must/should：关键词 multi_match（有 keyword 才加） 2. bool.filter：城市、学历、状态等精确条件（不参与算分） 3. from/size：分页 4. sort：有词按相关度，或按 publish_time """ # 索引不存在时友好提示 if not await es_client.indices.exists(index=BOSS_JOB_INDEX_NAME): return { "code": 0, "message": f"索引 {BOSS_JOB_INDEX_NAME} 不存在，请先创建并同步数据", } must_clauses: list[dict[str, Any]] = [] filter_clauses: list[dict[str, Any]] = [] # ----- 1）关键词：打在多个 text 字段，职位名称加权 ----- if keyword and keyword.strip(): must_clauses.append( { "multi_match": { "query": keyword.strip(), "fields": [ "job_name^3", # 名称命中权重更高 "job_desc^2", "duty_require", "enterprise_name", "industry_name", ], "type": "best_fields", "operator": "and", # 可按需改成 or，召回更宽 } } ) # ----- 2）硬筛选：全部放 filter ----- if status is not None: filter_clauses.append({"term": {"status": status}}) if work_location: filter_clauses.append({"term": {"work_location": work_location}}) if edu_require: filter_clauses.append({"term": {"edu_require": edu_require}}) if exp_require: filter_clauses.append({"term": {"exp_require": exp_require}}) if industry_id is not None: filter_clauses.append({"term": {"industry_id": industry_id}}) if company_scale is not None: filter_clauses.append({"term": {"enterpriseInfo_company_scale": company_scale}}) if financing_stage is not None: filter_clauses.append({"term": {"enterpriseInfo_financing_stage": financing_stage}}) # ----- 3）组装 bool ----- bool_query: dict[str, Any] = {} if must_clauses: bool_query["must"] = must_clauses if filter_clauses: bool_query["filter"] = filter_clauses # 既无关键词也无筛选时，匹配全部（仍建议至少有默认 status filter） query: dict[str, Any] = {"bool": bool_query} if bool_query else {"match_all": {}} # ----- 4）排序 ----- # 有关键词且 sort=score → 按相关度；否则按发布时间倒序 if sort == "score" and keyword: sort_clause: Any = ["_score", {"publish_time": {"order": "desc", "missing": "_last"}}] else: sort_clause = [{"publish_time": {"order": "desc", "missing": "_last"}}] # ----- 5）分页：ES 用 from / size ----- from_ = (page - 1) * page_size # 列表页只取卡片字段，减小 _source source_fields = [ "job_id", "job_name", "work_location", "min_salary", "max_salary", "salary_times", "edu_require", "exp_require", "job_tags", "status", "publish_time", "enterprise_id", "enterprise_name", "enterprise_city_name", "enterpriseInfo_company_scale", "enterpriseInfo_financing_stage", "industry_id", "industry_name", ] resp = await es_client.search( index=BOSS_JOB_INDEX_NAME, query=query, sort=sort_clause, from_=from_, size=page_size, source=source_fields, track_total_hits=True, # 拿到准确 total ) hits = resp.get("hits", {}) total = hits.get("total", {}) # ES 7+ total 一般是 {"value": n, "relation": "eq"} total_count = total.get("value", 0) if isinstance(total, dict) else int(total or 0) lists = [] for hit in hits.get("hits", []): item = hit.get("_source") or {} item["_score"] = hit.get("_score") # 可选：方便调相关度 lists.append(item) return { "code": 1, "message": "success", "data": { "lists": lists, "page_info": { "page": page, "page_size": page_size, "total_count": total_count, "total_page": (total_count + page_size - 1) // page_size if page_size else 0, }, }, }
雪花算法
import time import threading class SnowflakeSingleton: """ 单例模式的雪花算法生成唯一ID工具类 """ _instance = None _lock = threading.Lock() # 单例模式的线程锁 # 起始时间戳 (2020-01-01 00:00:00) START_TIMESTAMP = 1577808000000 # 各部分位数 SEQUENCE_BITS = 12 # 序列号位数 WORKER_ID_BITS = 10 # 工作节点ID位数 # 最大取值计算 MAX_WORKER_ID = -1 ^ (-1 << WORKER_ID_BITS) # 1023 MAX_SEQUENCE = -1 ^ (-1 << SEQUENCE_BITS) # 4095 # 移位偏移量计算 WORKER_ID_SHIFT = SEQUENCE_BITS # 12 TIMESTAMP_LEFT_SHIFT = SEQUENCE_BITS + WORKER_ID_BITS # 22 def __new__(cls, worker_id=None): """单例模式实现，确保只创建一个实例""" with cls._lock: # 如果实例不存在，则创建 if not cls._instance: # 首次创建时必须提供worker_id if worker_id is None: raise ValueError("首次初始化必须提供worker_id参数") cls._instance = super().__new__(cls) # 初始化实例变量 cls._instance.worker_id = worker_id cls._instance.last_timestamp = -1 # 上次生成ID的时间戳 cls._instance.sequence = 0 # 序列号 cls._instance.id_lock = threading.Lock() # 生成ID的线程锁 # 如果已经存在实例，再次传入的worker_id必须与初始一致 elif worker_id is not None and worker_id != cls._instance.worker_id: raise ValueError(f"雪花算法实例已初始化，worker_id必须为{cls._instance.worker_id}") return cls._instance @classmethod def get_instance(cls, worker_id=None): """获取单例实例的便捷方法""" return cls(worker_id) def _gen_timestamp(self): """生成当前时间戳（毫秒）""" return int(time.time() * 1000) def get_id(self): """生成下一个唯一ID""" with self.id_lock: # 确保线程安全 timestamp = self._gen_timestamp() # 处理系统时钟回退 if timestamp < self.last_timestamp: raise RuntimeError( f"系统时钟回退，拒绝生成ID。时间差: {self.last_timestamp - timestamp}毫秒" ) # 处理同一毫秒内的序列号 if timestamp == self.last_timestamp: self.sequence = (self.sequence + 1) & self.MAX_SEQUENCE # 序列号达到最大值，等待下一毫秒 if self.sequence == 0: timestamp = self._til_next_millis(self.last_timestamp) else: # 不同时间戳，序列号重置为0 self.sequence = 0 # 更新上次生成ID的时间戳 self.last_timestamp = timestamp # 组合生成ID snowflake_id = ( ((timestamp - self.START_TIMESTAMP) << self.TIMESTAMP_LEFT_SHIFT) | (self.worker_id << self.WORKER_ID_SHIFT) | self.sequence ) return snowflake_id def _til_next_millis(self, last_timestamp): """阻塞到下一个毫秒，直到获得新的时间戳""" timestamp = self._gen_timestamp() while timestamp <= last_timestamp: timestamp = self._gen_timestamp() return timestamp # 调用方法示例 if __name__ == "__main__": # 初始化单例（首次调用必须提供worker_id） snowflake = SnowflakeSingleton.get_instance(worker_id=1) # 生成10个ID并打印 for _ in range(10): print(f"生成的ID: {snowflake.get_id()}") # 在其他地方获取实例（无需再次提供worker_id） another_instance = SnowflakeSingleton.get_instance() print(f"\n验证单例: {snowflake is another_instance}") # 应输出True # 多线程测试 def generate_ids(count): ids = [] for _ in range(count): ids.append(SnowflakeSingleton.get_instance().get_id()) return ids # 创建多个线程同时生成ID threads = [] for i in range(5): t = threading.Thread(target=generate_ids, args=(1000,)) threads.append(t) t.start() for t in threads: t.join() print("\n多线程ID生成测试完成")
