一、前言在BOSS直聘這類招聘平臺中企業管理是核心功能之一。企業用戶從注冊到審核通過需要經過一系列完整的流程。本文將詳細介紹如何使用FastAPI Tortoise ORM Redis技術棧實現企業管理系統的后端核心功能包括企業注冊與認證企業列表查詢與分頁企業詳情查看企業審核通過/拒絕企業登錄驗證碼方式數據扁平化處理技巧二、數據庫模型設計2.1 核心數據表結構企業管理涉及多張關聯表主要包括表名說明t_enterprise企業主表存儲企業基本狀態t_enterprise_info企業詳細信息表t_enterprise_qualification企業資質信息表t_enterprise_review企業審核記錄表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_nameform.enterprise_name, enterprise_codestr(uuid.uuid4()), account_statusAccountStatus.PENDING_AUDIT, blacklist_statusBlackListStatus.NOT_BANNED, audit_typeAuditType.NEW_ENTERPRISE_AUTH, submit_timenow(), auth_typeAuthType.ENTERPRISE_LICENSE_AUTH ) # 2. 創建企業詳細信息 await EnterpriseInfo.create( enterprise_nameform.enterprise_name, unified_social_credit_codeform.unified_social_credit_code, legal_representativeform.legal_representative, enterprise_identerprise.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_identerprise.id, business_license_urlsuccess_res[access_url], legal_id_front_urlsuccess_res2[access_url], legal_id_back_urlsuccess_res3[access_url], # ... 其他字段 ) # 5. 自動創建審核記錄 ? 關鍵點 await EnterpriseReview.create( enterprise_identerprise.id, review_resultNone, remark企業注冊提交認證申請等待審核 )3.3 流程說明用戶提交注冊 → 開啟數據庫事務 ├── 創建企業主記錄狀態待審核 ├── 創建企業詳細信息 ├── 上傳資質文件到OSS ├── 創建企業資質記錄 └── 創建審核記錄review_result None → 提交事務四、企業列表查詢與分頁4.1 列表接口實現enterprise_router.get(/list, summary查詢企業列表) async def select_enterprise_list( page: int Query(1, ge1), page_size: int Query(10, ge10, le50), 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__icontainsenterprise_name ) if submit_time_start: query_enterprise query_enterprise.filter( submit_time__gtesubmit_time_start ) if submit_time_end: query_enterprise query_enterprise.filter( submit_time__ltesubmit_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_identerprise_id ).prefetch_related(industry) enterprise_qualification await EnterpriseQualification.get_or_none( enterprise_identerprise_id ) enterprise_review await EnterpriseReview.get_or_none( enterprise_identerprise_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 }扁平化的好處簡化前端模板代碼減少前端渲染計算便于數據導出五、企業詳情查看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(identerprise_id) # 查詢詳細信息預加載行業關聯 enterpriseinfo await EnterpriseInfo.get_or_none( enterprise_identerprise_id ).prefetch_related(industry) # 查詢資質信息 enterprise_qualification await EnterpriseQualification.get_or_none( enterprise_identerprise_id ) # 查詢審核記錄 enterprise_review await EnterpriseReview.get_or_none( enterprise_identerprise_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_identerprise_id ) if enterprise_review is None: # 新建審核記錄 await EnterpriseReview.create( enterprise_identerprise_id, review_resultreview_result, review_reasonreview_reason, review_timenow() ) 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(identerprise_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_phonemobile ) 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_identerprise_id ) # 處理沒有審核記錄的情況兼容老數據 if enterprise_review is None: await EnterpriseReview.create( enterprise_identerprise_id, review_resultNone, remark自動創建審核記錄 ) raise Exception(賬號未審核通過請等待審核) # 審核拒絕 if enterprise_review.review_result 2: raise Exception(f賬號審核未通過: {enterprise_review.review_reason}) # 審核通過驗證驗證碼 if enterprise_review.review_result