企业官网建设流程全解析

企业端-招聘团队

企业过审之后，得有人来发职位。下面这几块串起来就是：审核通过时自动建招聘团队 → 团队成员凭手机号+验证码登录 → 发职位 → 看自己的职位列表。

@staticmethod async def enterprise_review(enterpriseReviewCreateRequest: EnterpriseReviewCreateRequest): enterprise_id=enterpriseReviewCreateRequest.enterprise_id enterprise_review=await EnterpriseReview.get_or_none(enterprise_id=enterprise_id) if enterprise_review is None: await EnterpriseReview.create( enterprise_id=enterprise_id, review_result=enterpriseReviewCreateRequest.review_result, review_reason=enterpriseReviewCreateRequest.review_reason, remark=enterpriseReviewCreateRequest.remark, audit_time=now() ) else: enterprise_review.review_result=enterpriseReviewCreateRequest.review_result if enterpriseReviewCreateRequest.review_result else enterprise_review.review_result enterprise_review.review_reason=enterpriseReviewCreateRequest.review_reason if enterpriseReviewCreateRequest.review_reason else enterprise_review.review_reason enterprise_review.remark=enterpriseReviewCreateRequest.remark if enterpriseReviewCreateRequest.remark else enterprise_review.remark enterprise_review.audit_time=now() await enterprise_review.save() if enterpriseReviewCreateRequest.review_result==1: enterprise=await Enterprise.get_or_none(id=enterprise_id) enterprise.account_status=AccountStatus.NORMAL await enterprise.save() enterpriseQualification=await EnterpriseQualification.get_or_none(enterprise_id=enterprise_id) await RecruitTeam.create( mobile=enterpriseQualification.contact_phone, name=enterpriseQualification.contact_name, email=enterpriseQualification.contact_email, enterprise_id=enterprise_id, status=TeamMemberStatus.NORMAL ) status = "审核通过" if enterpriseReviewCreateRequest.review_result == 1 else "审核驳回" await EnterpriseReviewHistory.create( enterprise_id=enterprise_id, operation=status, status=status, operator="系统自动审核", remark=enterpriseReviewCreateRequest.remark, created_at=now() )

这是前面管理者端审核的升级版，在"建/改审核记录 + 通过时放开账号"的基础上多做了两件事：

• 审核一通过（review_result == 1），除了把企业account_status置为NORMAL，还会从企业资质里拽出联系人的手机、姓名、邮箱，自动建一条RecruitTeam（招聘团队）记录，状态NORMAL。等于企业过审的同一刻，那个联系人就自动成了招聘团队第一个成员——后面"团队登录"就是对着这张表验的。

• 不管通过还是驳回，末尾都往EnterpriseReviewHistory写一条操作记录（操作人写死"系统自动审核"），相当于留个审计痕，以后查"谁在啥时候改了审核状态"有处可查。

字段名从之前的review_time换成了audit_time，其余逻辑一致。

企业端-登录

async def team_login(loginMobileRequest: LoginMobileRequest): recruit_team=await RecruitTeam.get_or_none(mobile=loginMobileRequest.mobile, status=TeamMemberStatus.NORMAL, is_deleted=DeleteStatus.NOT_DELETED ) if recruit_team is None: raise Exception('手机号不存在或账号状态异常') key = f"boss-api:enterprise-login:sms:{loginMobileRequest.mobile}" redis_code = redis_client.get(key) if redis_code is None: raise Exception("验证码已过期") if redis_code != loginMobileRequest.code: raise Exception("验证码错误") access_token, refresh_token = create_tokens(str(recruit_team.id), loginMobileRequest.mobile) redis_client.delete(key) return { "enterprise_access_token": access_token, "enterprise_refresh_token": refresh_token, } @enterprise_router.post("/login", summary="企业登录", description="企业登录") async def login(loginMobileRequest: LoginMobileRequest): res=await EnterpriseService.team_login(loginMobileRequest) return { "code": 1, "message": "登录成功", "data": res }

团队登录和企业登录（第四节那个）套路一样——都是"手机号 + 短信验证码"，区别在对着哪张表验、验的是谁：

• 用手机号去RecruitTeam查，而且顺手卡了status=NORMAL和is_deleted=NOT_DELETED——被禁用或已删除的成员直接登不进来。

• 查不到抛"手机号不存在或账号状态异常"，这里用的是get_or_none+is None，判断是对的（不像之前企业登录那个filter永远非 None 的坑）。

• Redis 取验证码，过期或不对分别报错；都过了就用recruit_team.id当 subject 发 token，发完把验证码删掉。

企业端-发布职位

@job_router.post("/save", summary="保存职位信息", description="保存职位信息") async def saveJob(jobCreateRequest: JobCreateRequest,team_id=Depends(get_job_info)): await JobService.saveJob(jobCreateRequest,team_id) return { "code": 1, "message": "保存成功" } class JobCreateRequest(BaseModel): job_name: str = Field(..., description="职位名称") # 使用枚举替换原来int类型dept_id department_id: int = Field(..., description="所属部门") work_location: str = Field(..., description="工作地点") min_salary: str = Field(..., description="最低薪资，支持「面议」") max_salary: str | None = Field(None, description="最高薪资") salary_times: str | None = Field(None, description="几薪") exp_require: str = Field(..., description="经验要求") edu_require: str = Field(..., description="学历要求") gender_require: str = Field(..., description="性别要求") recruit_num: str = Field(..., description="招聘人数") job_tags: list[str] | None = Field(None, description='职位标签，示例["五险一金","年终奖"]') job_desc: str = Field(..., description="职位描述") duty_require: str = Field(..., description="任职要求") status: int = Field(..., description="0:草稿,1:招聘中,2:暂停招聘,3:已关闭") @staticmethod async def saveJob(jobCreateRequest: JobCreateRequest,team_id:int): recruitteam =await RecruitTeam.get_or_none(id=team_id) await Job.create(**jobCreateRequest.dict(),publish_time=now(), enterprise_id=recruitteam.enterprise_id,recruit_team_id=team_id)

api层：

POST /save发职位。jobCreateRequest装着职位信息，team_id不从前端点，而是Depends(get_job_info)从登录态里解出来——保证"谁发的职位就挂在谁的团队下"，前端伪造不了归属。

schema层：

职位的数据结构。几个值得说的点：薪资是str而不是数字，因为要兼容"面议"这种非数字值；max_salary、salary_times、job_tags都可空；job_tags是字符串数组，存标签列表；status用 int 表示职位状态——0 草稿、1 招聘中、2 暂停招聘、3 已关闭。注释里那句"用枚举替换原来 int 类型 dept_id"是个优化方向：部门 id 直接写 int 容易传错值，后面用枚举收口会更稳。

service层

先按team_id把招聘团队捞出来——目的是拿到它归属的enterprise_id。然后一条Job.create搞定：用**jobCreateRequest.dict()把 schema 里的字段全展开进职位表，publish_time填当前时间，企业 id 从团队带过来，团队 id 也记上。

这么一来职位天然就和企业、团队绑死了，发布人没法把职位挂到别人的企业底下。

企业端-职位列表

@job_router.get("/list", summary="职位列表") async def selectJobList( page: int = Query(1, description="页码"), page_size: int = Query(10, description="每页数量"), keywords: str | None = Query(None, description="搜索关键词"), status: str | None = Query(None, description="职位状态"), cityname: str | None = Query(None, description="城市名称"), team_id=Depends(get_job_info) ): res=await JobService.selectJobList(page,page_size,keywords,status,cityname,team_id) return { "code": 1, "message": "查询成功", "data": res } @staticmethod async def selectJobList( page: int, page_size: int, keywords: str | None, status: str | None, cityname: str | None, team_id: int ): query=Job.filter(recruit_team_id=team_id) if keywords: query=query.filter(job_name__icontains=keywords) if status: query=query.filter(status=status) if cityname: query=query.filter(cityname=cityname) total_count=await query.count() total_page=math.ceil(total_count/page_size) job_query_set=await query.offset((page-1)*page_size).limit(page_size).all() return { "total_count": total_count, "total_page": total_page, "job_list": job_query_set, "page": page, "page_size": page_size }

先filter(recruit_team_id=team_id)把范围锁死——一个团队只能看自己发的职位，数据隔离在这一层就做了。然后条件叠加：keywords走job_name__icontains（忽略大小写模糊匹配），status、cityname是精确匹配。最后跟前面列表接口一样的套路：算总数、算总页数、offset+limit 分页，把列表和分页信息一起返回。

ES职位搜索

使用ES原因：职位多了之后，光靠数据库 LIKE 查中文又慢又不智能，所以接了 ES 做全文检索。下面三块：客户端怎么建、索引怎么定义、数据怎么灌进去。

ES_HOST = os.getenv("ES_HOST", "http://localhost:9200") es_client: AsyncElasticsearch | None = None async def get_es_client() -> AsyncElasticsearch: """获取 ES 异步客户端单例""" global es_client if es_client is None: es_client = AsyncElasticsearch(ES_HOST) logger.info("ES 客户端初始化成功") return es_client async def close_es_client(): """关闭 ES 客户端连接""" global es_client if es_client is not None: await es_client.close() es_client = None

async def es_client_depend() -> AsyncElasticsearch: return await get_es_client()

ES 客户端做成单例。get_es_client第一次调才真正连，之后全局复用同一个AsyncElasticsearch实例——避免每次请求都重连，省资源。close_es_client在应用关闭时调，关连接、置空。es_client_depend专门给 FastAPI 的Depends用，路由里写Depends(es_client_depend)就能拿到那个单例，不用自己管连接生命周期。

BOSS_JOB_INDEX_NAME = "boss_job_index" @es_data_router.post("/create-index", summary="创建索引") async def create_index(es_client: AsyncElasticsearch = Depends(es_client_depend)): mappings={ "properties":{ "job_id": { "type": "long" }, "job_name": { "type": "text", "analyzer": "ik_max_word" }, "department_id": { "type": "long" }, "work_location": { "type": "keyword" }, "min_salary": { "type": "keyword" }, "max_salary": { "type": "keyword" }, "salary_times": { "type": "keyword" }, "edu_require": { "type": "keyword", }, "exp_require": { "type": "keyword" }, "gender_require": { "type": "keyword" }, "recruit_num": { "type": "long" }, "job_tags": { "type": "keyword" }, "job_desc": { "type": "text", "analyzer": "ik_max_word" }, "duty_require": { "type": "text", "analyzer": "ik_max_word" }, "status": { "type": "integer" }, "publish_time": { "type": "date", # "format": "yyyy-MM-dd HH:mm:ss" }, "enterprise_id": { "type": "long" }, "recruit_team_id": { "type": "long" }, "enterprise_name": { "type": "text", "analyzer": "ik_max_word" }, "enterprise_code": { "type": "keyword" }, "enterprise_city_id": { "type": "long" }, "enterprise_city_name": { "type": "keyword" }, "enterprise_account_status": { "type": "integer" }, "enterprise_create_time": { "type": "date", # "format": "yyyy-MM-dd HH:mm:ss" }, "enterprise_auth_time": { "type": "date", # "format": "yyyy-MM-dd HH:mm:ss" }, "enterprise_auth_type": { "type": "integer" }, "enterprise_risk_level": { "type": "integer" }, "enterprise_blacklist_status": { "type": "integer" }, "enterprise_complaint_count": { "type": "integer" }, "enterprise_company_website": { "type": "keyword" }, "enterprise_email": { "type": "keyword" }, "enterprise_audit_type": { "type": "integer" }, "enterprise_submit_time": { "type": "date", # "format": "yyyy-MM-dd HH:mm:ss" }, "enterpriseInfo_unified_social_credit_code": { "type": "keyword" }, "enterpriseInfo_legal_representative": { "type": "keyword" }, "enterpriseInfo_registered_capital": { "type": "keyword"}, "enterpriseInfo_establish_date": { "type": "date", # "format": "yyyy-MM-dd HH:mm:ss" }, "enterpriseInfo_register_status": { "type": "integer" }, "enterpriseInfo_company_scale": { "type": "keyword" }, "enterpriseInfo_financing_stage": { "type": "keyword" }, "enterpriseInfo_headquarters_address": { "type": "keyword" }, "industry_id": { "type": "long" }, "industry_name": { "type": "text", "analyzer": "ik_max_word" } } } if await es_client.indices.exists(index=BOSS_JOB_INDEX_NAME): return "索引已存在" await es_client.indices.create(index=BOSS_JOB_INDEX_NAME, mappings=mappings) return "索引创建成功" @es_data_router.post("/insert-data", summary="同步数据") async def insert_data(es_client: AsyncElasticsearch = Depends(es_client_depend)): jobs=await Job.all() for job in jobs: enterprise_id = job.enterprise_id enterprise =await Enterprise.get_or_none(id=enterprise_id).prefetch_related("city") enterpriseInfo=await EnterpriseInfo.get_or_none(enterprise_id=enterprise_id).prefetch_related("industry") job_info={ "job_id": job.id, "job_name": job.job_name, "department_id": job.department_id, "work_location": job.work_location, "min_salary": job.min_salary, "max_salary": job.max_salary, "salary_times": job.salary_times, "edu_require": job.edu_require, "exp_require": job.exp_require, "gender_require": job.gender_require, "recruit_num": job.recruit_num, "job_tags": job.job_tags, "job_desc": job.job_desc, "duty_require": job.duty_require, "status": job.status, "publish_time": job.publish_time, "enterprise_id": job.enterprise_id, "recruit_team_id": job.recruit_team_id, "enterprise_name": enterprise.enterprise_name, "enterprise_code": enterprise.enterprise_code, # "enterprise_city_id": enterprise.city.id, # "enterprise_city_name": job.work_location, "enterprise_account_status": enterprise.account_status, "enterprise_create_time": enterprise.create_time, "enterprise_auth_time": enterprise.auth_time, "enterprise_auth_type": enterprise.auth_type, "enterprise_risk_level": enterprise.risk_level, "enterprise_blacklist_status": enterprise.blacklist_status, "enterprise_complaint_count": enterprise.complaint_count, "enterprise_company_website": enterprise.company_website, "enterprise_email": enterprise.email, "enterprise_audit_type": enterprise.audit_type, "enterprise_submit_time": enterprise.submit_time, "enterpriseInfo_unified_social_credit_code": enterpriseInfo.unified_social_credit_code, "enterpriseInfo_legal_representative": enterpriseInfo.legal_representative, "enterpriseInfo_registered_capital": enterpriseInfo.registered_capital, "enterpriseInfo_establish_date": enterpriseInfo.establish_date, "enterpriseInfo_register_status": enterpriseInfo.register_status, "enterpriseInfo_company_scale": enterpriseInfo.company_scale, "enterpriseInfo_financing_stage": enterpriseInfo.financing_stage, "enterpriseInfo_headquarters_address": enterpriseInfo.headquarters_address, "enterpriseInfo_business_scope": enterpriseInfo.business_scope, "industry_id": enterpriseInfo.industry.id, "industry_name": enterpriseInfo.industry.name, } await es_client.index(index=BOSS_JOB_INDEX_NAME, document=job_info)

创建索引：

定义职位索引的字段结构（mapping）。类型分两类，别混：

• 要全文搜的中文文本——job_name、job_desc、duty_require、enterprise_name、industry_name——用text+ik_max_word分词器。ik_max_word是专门切中文的，能把"人力资源"切成"人力/资源/人力资源"那样，搜索才准；ES 默认的英文分词器对中文是一字一切，基本没法用。

• 其余像薪资、状态、各种 code 用keyword/long/integer/date，这些是精确匹配或排序、聚合用的，不分词。

同步数据：

把数据库里的职位全量同步进 ES，相当于"索引建好之后灌数据"。逻辑是：遍历所有Job，每条都去把对应的Enterprise和EnterpriseInfo（含行业）查出来，拼成一个扁平的job_info字典——把职位字段和企业字段揉一块儿存。这种"宽表"做法让搜索时不用再 JOIN 数据库，直接就能按企业名、行业名筛职位。