一、前言

在BOSS直聘这类招聘平台中，企业管理是核心功能之一。企业用户从注册到审核通过，需要经过一系列完整的流程。本文将详细介绍如何使用FastAPI + Tortoise ORM + Redis技术栈，实现企业管理系统的后端核心功能，包括：

• 企业注册与认证
• 企业列表查询与分页
• 企业详情查看
• 企业审核（通过/拒绝）
• 企业登录（验证码方式）
• 数据扁平化处理技巧

二、数据库模型设计

2.1 核心数据表结构

企业管理涉及多张关联表，主要包括：

2.2 关键枚举类型

class AccountStatus(str, Enum): NORMAL = "normal" # 正常状态 PENDING_AUDIT = "pending" # 待审核 FROZEN = "frozen" # 已冻结 class BlackListStatus(int, Enum): NOT_BANNED = 0 # 未被拉黑 BANNED = 1 # 已被拉黑 class AuditType(str, Enum): NEW_ENTERPRISE_AUTH = "new" # 新企业认证 ANNUAL_REVIEW = "annual" # 年度审核 class AuthType(str, Enum): ENTERPRISE_LICENSE_AUTH = "license" # 营业执照认证

三、企业注册与认证流程

3.1 企业注册接口

企业注册是整个流程的起点，用户提交企业信息和相关资质文件后，系统会自动创建审核记录。

接口定义

@enterprise_router.post("/save", summary="保存企业信息") async def saveEnterpriseInfo( form: EnterpriseCreateRequest = Depends(get_enterprise_form), business_license_file: UploadFile = File(None), legal_id_front_file: UploadFile = File(None), legal_id_back_file: UploadFile = File(None), ): await EnterpriseService.saveEnterpriseInfo( form, business_license_file, legal_id_front_file, legal_id_back_file ) return {"code": 1, "message": "保存成功"}

3.2 核心注册逻辑

@staticmethod async def saveEnterpriseInfo( form: EnterpriseCreateRequest, business_license_file: UploadFile, legal_id_front_file: UploadFile, legal_id_back_file: UploadFile, ): async with in_transaction() as conn: # 1. 创建企业主记录 enterprise = await Enterprise.create( enterprise_name=form.enterprise_name, enterprise_code=str(uuid.uuid4()), account_status=AccountStatus.PENDING_AUDIT, blacklist_status=BlackListStatus.NOT_BANNED, audit_type=AuditType.NEW_ENTERPRISE_AUTH, submit_time=now(), auth_type=AuthType.ENTERPRISE_LICENSE_AUTH ) # 2. 创建企业详细信息 await EnterpriseInfo.create( enterprise_name=form.enterprise_name, unified_social_credit_code=form.unified_social_credit_code, legal_representative=form.legal_representative, enterprise_id=enterprise.id, # ... 其他字段 ) # 3. 上传并保存资质文件 success_res = await AliyunOSSTool.upload_file(business_license_file) success_res2 = await AliyunOSSTool.upload_file(legal_id_front_file) success_res3 = await AliyunOSSTool.upload_file(legal_id_back_file) # 4. 创建企业资质记录 await EnterpriseQualification.create( enterprise_id=enterprise.id, business_license_url=success_res["access_url"], legal_id_front_url=success_res2["access_url"], legal_id_back_url=success_res3["access_url"], # ... 其他字段 ) # 5. 自动创建审核记录 ⭐ 关键点 await EnterpriseReview.create( enterprise_id=enterprise.id, review_result=None, remark="企业注册提交认证申请，等待审核" )

3.3 流程说明

用户提交注册 → 开启数据库事务 ├── 创建企业主记录（状态：待审核） ├── 创建企业详细信息 ├── 上传资质文件到OSS ├── 创建企业资质记录 └── 创建审核记录（review_result = None） → 提交事务

四、企业列表查询与分页

4.1 列表接口实现

@enterprise_router.get("/list", summary="查询企业列表") async def select_enterprise_list( page: int = Query(1, ge=1), page_size: int = Query(10, ge=10, le=50), enterprise_name: str = Query(None), submit_time_start: str = Query(None), submit_time_end: str = Query(None), ): res = await EnterpriseService.select_enterprise_list( page, page_size, enterprise_name, submit_time_start, submit_time_end ) return {"code": 1, "message": "查询成功", "data": res}

4.2 列表查询核心逻辑

关键点：数据扁平化处理

@staticmethod async def select_enterprise_list(page, page_size, enterprise_name, submit_time_start, submit_time_end): # 1. 构建查询条件 query_enterprise = Enterprise.all() if enterprise_name: query_enterprise = query_enterprise.filter( enterprise_name__icontains=enterprise_name ) if submit_time_start: query_enterprise = query_enterprise.filter( submit_time__gte=submit_time_start ) if submit_time_end: query_enterprise = query_enterprise.filter( submit_time__lte=submit_time_end ) # 2. 分页计算 total_count = await query_enterprise.count() total_page = math.ceil(total_count / page_size) offset = (page - 1) * page_size # 3. 执行分页查询 query_enterprise = await query_enterprise.offset(offset).limit(page_size) # 4. 数据扁平化处理 ⭐ 关键技巧 enterprise_list = [] for enterprise in query_enterprise: enterprise_id = enterprise.id # 查询关联数据 enterpriseinfo = await EnterpriseInfo.get_or_none( enterprise_id=enterprise_id ).prefetch_related("industry") enterprise_qualification = await EnterpriseQualification.get_or_none( enterprise_id=enterprise_id ) enterprise_review = await EnterpriseReview.get_or_none( enterprise_id=enterprise_id ) # 处理审核状态 audit_status = 0 reject_reason = None if enterprise_review: audit_status = enterprise_review.review_result if enterprise_review.review_result else 0 reject_reason = enterprise_review.review_reason # 获取行业名称 industry_name = "" if enterpriseinfo and enterpriseinfo.industry: industry_name = enterpriseinfo.industry.name # 构建扁平数据结构 item = { "id": enterprise.id, "enterprise_name": enterprise.enterprise_name, "legal_representative": enterpriseinfo.legal_representative if enterpriseinfo else "", "contact_name": enterprise_qualification.contact_name if enterprise_qualification else "", "contact_phone": enterprise_qualification.contact_phone if enterprise_qualification else "", "industry": industry_name, "submit_time": enterprise.submit_time.isoformat() if enterprise.submit_time else "", "audit_status": audit_status, "reject_reason": reject_reason, "business_license_url": enterprise_qualification.business_license_url if enterprise_qualification else "", } enterprise_list.append(item) return { "list": enterprise_list, "total": total_count, "total_page": total_page, "page": page, "page_size": page_size }

4.3 数据扁平化技巧

为什么需要扁平化？

原始数据结构是嵌套对象：

{ "enterprise": {...}, "enterprise_info": {...}, "enterprise_qualification": {...}, "enterprise_review": {...} }

前端表格组件el-table需要扁平结构：

{ "id": 1, "enterprise_name": "示例科技", "legal_representative": "张三", "audit_status": 0 }

扁平化的好处：

1. 简化前端模板代码
2. 减少前端渲染计算
3. 便于数据导出

五、企业详情查看

5.1 详情接口

@enterprise_router.get("/detail/{enterprise_id}", summary="查询企业详情") async def select_enterprise_by_id(enterprise_id: int): res = await EnterpriseService.select_enterprise_by_id(enterprise_id) return {"code": 1, "message": "查询成功", "data": res}

5.2 详情查询实现

@staticmethod async def select_enterprise_by_id(enterprise_id: int): # 查询企业主表 enterprise = await Enterprise.get_or_none(id=enterprise_id) # 查询详细信息（预加载行业关联） enterpriseinfo = await EnterpriseInfo.get_or_none( enterprise_id=enterprise_id ).prefetch_related("industry") # 查询资质信息 enterprise_qualification = await EnterpriseQualification.get_or_none( enterprise_id=enterprise_id ) # 查询审核记录 enterprise_review = await EnterpriseReview.get_or_none( enterprise_id=enterprise_id ) # 组装扁平数据 return { "id": enterprise.id if enterprise else None, "enterprise_name": enterprise.enterprise_name if enterprise else "", "unified_social_credit_code": enterpriseinfo.unified_social_credit_code if enterpriseinfo else "", "legal_representative": enterpriseinfo.legal_representative if enterpriseinfo else "", "registered_capital": enterpriseinfo.registered_capital if enterpriseinfo else "", "establish_date": str(enterpriseinfo.establish_date) if enterpriseinfo else "", "industry": enterpriseinfo.industry.name if enterpriseinfo and enterpriseinfo.industry else "", "contact_name": enterprise_qualification.contact_name if enterprise_qualification else "", "contact_phone": enterprise_qualification.contact_phone if enterprise_qualification else "", "audit_status": enterprise_review.review_result if enterprise_review and enterprise_review.review_result else 0, "reject_reason": enterprise_review.review_reason if enterprise_review else None, "business_license_url": enterprise_qualification.business_license_url if enterprise_qualification else "", }

六、企业审核流程

6.1 审核接口

@enterprise_router.post("/review", summary="企业审核") async def enterprise_review(enterpriseReviewCreateRequest: EnterpriseReviewCreateRequest): await EnterpriseService.enterprise_review(enterpriseReviewCreateRequest) return {"code": 1, "message": "审核完成"}

6.2 审核逻辑实现

@staticmethod async def enterprise_review(enterpriseReviewCreateRequest: EnterpriseReviewCreateRequest): enterprise_id = enterpriseReviewCreateRequest.enterprise_id review_result = enterpriseReviewCreateRequest.review_result review_reason = enterpriseReviewCreateRequest.review_reason # 1. 查询或创建审核记录 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=review_result, review_reason=review_reason, review_time=now() ) else: # 更新已有审核记录 enterprise_review.review_result = review_result enterprise_review.review_reason = review_reason enterprise_review.review_time = now() await enterprise_review.save() # 2. 审核通过时，更新企业状态 if review_result == 1: # 通过 enterprise = await Enterprise.get_or_none(id=enterprise_id) enterprise.account_status = AccountStatus.NORMAL await enterprise.save() elif review_result == 2: # 拒绝 # 可选：添加拒绝处理逻辑 pass # TODO: 发送短信/邮件通知企业用户

6.3 审核状态流转

企业注册 → 待审核 (review_result = None) │ ├── 审核通过 → review_result = 1 → account_status = NORMAL │ └── 审核拒绝 → review_result = 2 → account_status 保持 PENDING 记录拒绝原因 review_reason

七、企业登录（验证码方式）

7.1 验证码发送

@enterprise_router.post("/send_sms_code/{mobile}", summary="发送验证码") async def send_sms_code(mobile: str): res = await EnterpriseService.send_sms_code(mobile) return {"code": 1, "message": "发送成功", "data": res}

7.2 验证码登录

@enterprise_router.post("/login", summary="企业登录") async def login(loginMobileRequest: LoginMobileRequest): res = await EnterpriseService.login(loginMobileRequest) return {"code": 1, "message": "登录成功", "data": res}

7.3 登录逻辑实现

@staticmethod async def login(loginMobileRequest: LoginMobileRequest): mobile = loginMobileRequest.mobile code = loginMobileRequest.code # 1. 通过手机号查找企业 enterprise_qualifications = await EnterpriseQualification.filter( contact_phone=mobile ) if not enterprise_qualifications: raise Exception("手机号不存在") for qualification in enterprise_qualifications: enterprise_id = qualification.enterprise_id # 2. 检查审核记录 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=None, remark="自动创建审核记录" ) raise Exception("账号未审核通过，请等待审核") # 审核拒绝 if enterprise_review.review_result == 2: raise Exception(f"账号审核未通过: {enterprise_review.review_reason}") # 审核通过，验证验证码 if enterprise_review.review_result