import io from collections import Counter from django.db.models import Count, Q from django.http import HttpResponse from rest_framework import viewsets, permissions, generics, status from rest_framework.decorators import action from rest_framework.response import Response from rest_framework.views import APIView from rest_framework.filters import SearchFilter from rest_framework.pagination import PageNumberPagination from django_filters.rest_framework import DjangoFilterBackend import openpyxl from openpyxl.styles import Font, PatternFill, Alignment, Border, Side from openpyxl.worksheet.datavalidation import DataValidation from .models import Job, JobFavorite from .serializers import JobListSerializer, JobDetailSerializer, JobFavoriteSerializer from .filters import JobFilter from apps.accounts.permissions import IsAdminOrSuperAdmin, IsSeeker from apps.organizations.models import Organization from apps.applications.models import Application from .school_data import classify_school class JobPublicPagination(PageNumberPagination): page_size = 20 page_size_query_param = 'page_size' max_page_size = 500 class JobPublicViewSet(viewsets.ReadOnlyModelViewSet): """公开只读，仅返回已发布职位""" queryset = Job.objects.filter(status='published').select_related('organization') filterset_class = JobFilter filter_backends = [DjangoFilterBackend, SearchFilter] search_fields = ['title', 'description', 'location'] permission_classes = [permissions.AllowAny] pagination_class = JobPublicPagination def get_serializer_class(self): if self.action == 'retrieve': return JobDetailSerializer return JobListSerializer @action(detail=True, methods=['post'], permission_classes=[IsSeeker]) def favorite(self, request, pk=None): job = self.get_object() fav, created = JobFavorite.objects.get_or_create(user=request.user, job=job) if not created: fav.delete() return Response({'collected': False}) return Response({'collected': True}) class MyFavoritesView(generics.ListAPIView): """求职者的收藏列表""" serializer_class = JobFavoriteSerializer permission_classes = [IsSeeker] def get_queryset(self): return JobFavorite.objects.filter(user=self.request.user).select_related( 'job', 'job__organization' ) class JobManageViewSet(viewsets.ModelViewSet): """管理端：公司管理员管理本公司职位""" permission_classes = [IsAdminOrSuperAdmin] def get_serializer_class(self): if self.action in ['retrieve', 'create', 'update', 'partial_update']: return JobDetailSerializer return JobListSerializer def get_queryset(self): user = self.request.user if user.is_superadmin: return Job.objects.all().select_related('organization') # 防御 organization 为空的情况 if not user.organization_id: return Job.objects.none() return Job.objects.filter(organization=user.organization).select_related('organization') def perform_create(self, serializer): if self.request.user.is_admin: # Admin 强制使用自己公司，忽略请求体中的 organization_id serializer.save(organization=self.request.user.organization) else: # 超管需要在请求体中提供 organization_id serializer.save() @action(detail=False, methods=['get'], url_path='template') def download_template(self, request): """下载职位导入Excel模板""" wb = openpyxl.Workbook() ws = wb.active ws.title = '职位导入模板' # 表头定义: (列名, 是否必填, 列宽, 示例值) columns = [ ('职位名称', True, 20, 'Python开发工程师'), ('职位类别', True, 15, '技术'), ('工作地点', True, 15, '北京'), ('薪资范围', True, 15, '15k-25k'), ('学历要求', True, 15, '本科及以下'), ('招聘人数', False, 12, '3'), ('职位描述', False, 40, '负责后端系统开发与维护'), ('状态', False, 12, '已发布'), ] if request.user.is_superadmin: columns.insert(0, ('所属公司名称', True, 25, '华远科技有限公司')) col_map_by_name = {name: idx for idx, (name, _, _, _) in enumerate(columns, 1)} header_font = Font(bold=True, color='FFFFFF', size=11) required_fill = PatternFill(start_color='2196F3', end_color='2196F3', fill_type='solid') optional_fill = PatternFill(start_color='90CAF9', end_color='90CAF9', fill_type='solid') thin_border = Border( left=Side(style='thin'), right=Side(style='thin'), top=Side(style='thin'), bottom=Side(style='thin'), ) for col_idx, (name, required, width, _) in enumerate(columns, 1): cell = ws.cell(row=1, column=col_idx, value=f'{name}（必填）' if required else name) cell.font = header_font cell.fill = required_fill if required else optional_fill cell.alignment = Alignment(horizontal='center') cell.border = thin_border ws.column_dimensions[openpyxl.utils.get_column_letter(col_idx)].width = width # 写入示例数据行 for col_idx, (_, _, _, example) in enumerate(columns, 1): cell = ws.cell(row=2, column=col_idx, value=example) cell.border = thin_border # 数据验证：学历要求下拉 edu_col = openpyxl.utils.get_column_letter(col_map_by_name['学历要求']) edu_dv = DataValidation(type='list', formula1='"博士,硕士,本科及以下"', allow_blank=False) edu_dv.error = '请选择：博士、硕士、本科及以下' edu_dv.errorTitle = '学历要求错误' ws.add_data_validation(edu_dv) edu_dv.add(f'{edu_col}2:{edu_col}500') # 数据验证：状态下拉 status_col = openpyxl.utils.get_column_letter(col_map_by_name['状态']) status_dv = DataValidation(type='list', formula1='"草稿,已发布"', allow_blank=True) status_dv.error = '请选择：草稿、已发布' status_dv.errorTitle = '状态错误' ws.add_data_validation(status_dv) status_dv.add(f'{status_col}2:{status_col}500') # 数据验证：所属公司下拉（超管） if request.user.is_superadmin: org_names = list(Organization.objects.values_list('name', flat=True).order_by('name')) if org_names: # 将公司列表写入隐藏sheet，用名称引用实现下拉 org_ws = wb.create_sheet('_org_data') for i, name in enumerate(org_names, 1): org_ws.cell(row=i, column=1, value=name) org_ws.sheet_state = 'hidden' org_col = openpyxl.utils.get_column_letter(col_map_by_name['所属公司名称']) org_dv = DataValidation( type='list', formula1=f"'_org_data'!$A$1:$A${len(org_names)}", allow_blank=False ) org_dv.error = '请从下拉列表中选择公司，不可填写系统外的公司' org_dv.errorTitle = '公司名称错误' ws.add_data_validation(org_dv) org_dv.add(f'{org_col}2:{org_col}500') # 添加说明 sheet info_ws = wb.create_sheet('填写说明') info_ws.column_dimensions['A'].width = 20 info_ws.column_dimensions['B'].width = 60 notes = [ ('字段', '说明'), ('职位名称', '必填，最多100个字符'), ('职位类别', '必填，如：技术、产品、设计、运营、市场等'), ('工作地点', '必填，如：北京、上海、深圳'), ('薪资范围', '必填，如：15k-25k、面议'), ('学历要求', '必填，可选值：博士、硕士、本科及以下'), ('招聘人数', '选填，默认为1，填写正整数'), ('职位描述', '选填，职位的详细描述'), ('状态', '选填，可选值：草稿、已发布，默认为草稿'), ] if request.user.is_superadmin: notes.insert(1, ('所属公司名称', '超管必填，填写系统中已有的公司名称')) for row_idx, (field, desc) in enumerate(notes, 1): a = info_ws.cell(row=row_idx, column=1, value=field) b = info_ws.cell(row=row_idx, column=2, value=desc) if row_idx == 1: a.font = Font(bold=True) b.font = Font(bold=True) buf = io.BytesIO() wb.save(buf) buf.seek(0) response = HttpResponse( buf.getvalue(), content_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet' ) response['Content-Disposition'] = 'attachment; filename="job_import_template.xlsx"' return response @action(detail=False, methods=['post'], url_path='import-excel') def import_excel(self, request): """通过Excel批量导入职位""" file = request.FILES.get('file') if not file: return Response({'detail': '请上传Excel文件'}, status=status.HTTP_400_BAD_REQUEST) if not file.name.endswith(('.xlsx', '.xls')): return Response({'detail': '仅支持 .xlsx 或 .xls 格式'}, status=status.HTTP_400_BAD_REQUEST) try: wb = openpyxl.load_workbook(file, read_only=True) except Exception: return Response({'detail': 'Excel文件解析失败，请检查文件格式'}, status=status.HTTP_400_BAD_REQUEST) ws = wb.active rows = list(ws.iter_rows(min_row=1, values_only=True)) if len(rows) < 2: return Response({'detail': 'Excel中没有数据行'}, status=status.HTTP_400_BAD_REQUEST) # 解析表头，建立列名到索引的映射 headers = [str(h).replace('（必填）', '').strip() if h else '' for h in rows[0]] col_map = {name: idx for idx, name in enumerate(headers)} user = request.user is_super = user.is_superadmin education_choices = {'博士', '硕士', '本科及以下'} status_choices = {'draft', 'published', '草稿', '已发布'} status_map = {'草稿': 'draft', '已发布': 'published', 'draft': 'draft', 'published': 'published'} # 超管需要提前加载公司名映射 org_map = {} if is_super: org_map = {o.name: o for o in Organization.objects.all()} created = 0 errors = [] for row_idx, row in enumerate(rows[1:], start=2): def get_val(col_name): idx = col_map.get(col_name) if idx is not None and idx < len(row): v = row[idx] return str(v).strip() if v is not None else '' return '' title = get_val('职位名称') category = get_val('职位类别') location = get_val('工作地点') salary = get_val('薪资范围') education = get_val('学历要求') headcount_str = get_val('招聘人数') description = get_val('职位描述') job_status = get_val('状态') # 跳过全空行 if not any([title, category, location, salary]): continue # 校验必填字段 missing = [] if not title: missing.append('职位名称') if not category: missing.append('职位类别') if not location: missing.append('工作地点') if not salary: missing.append('薪资范围') if not education: education = '本科及以下' if education not in education_choices: errors.append(f'第{row_idx}行：学历要求必须为 博士/硕士/本科及以下') continue if missing: errors.append(f'第{row_idx}行：缺少必填字段 {", ".join(missing)}') continue # 招聘人数 headcount = 1 if headcount_str: try: headcount = int(float(headcount_str)) if headcount < 1: raise ValueError except (ValueError, TypeError): errors.append(f'第{row_idx}行：招聘人数必须为正整数') continue # 状态：支持中文和英文 job_status = status_map.get(job_status, 'draft') # 确定所属公司 org = None if is_super: org_name = get_val('所属公司名称') if not org_name: errors.append(f'第{row_idx}行：超级管理员必须填写所属公司名称') continue org = org_map.get(org_name) if not org: errors.append(f'第{row_idx}行：公司"{org_name}"不存在') continue else: org = user.organization if not org: return Response({'detail': '当前账号未关联公司，无法导入'}, status=status.HTTP_400_BAD_REQUEST) Job.objects.create( organization=org, title=title, category=category, location=location, salary=salary, education=education, headcount=headcount, description=description, status=job_status, ) created += 1 result = {'created': created, 'errors': errors} if created == 0 and errors: return Response(result, status=status.HTTP_400_BAD_REQUEST) return Response(result) class DashboardView(APIView): """管理后台 Dashboard 统计数据""" permission_classes = [IsAdminOrSuperAdmin] def get(self, request): user = request.user is_super = user.is_superadmin # 根据角色过滤数据范围 if is_super: jobs_qs = Job.objects.all() apps_qs = Application.objects.all() else: jobs_qs = Job.objects.filter(organization=user.organization) apps_qs = Application.objects.filter(job__organization=user.organization) # 职位统计 job_stats = jobs_qs.aggregate( published=Count('id', filter=Q(status='published')), closed=Count('id', filter=Q(status='closed')), draft=Count('id', filter=Q(status='draft')), total=Count('id'), ) # 投递统计 app_stats = apps_qs.aggregate( total=Count('id'), pending=Count('id', filter=Q(status='pending')), viewed=Count('id', filter=Q(status='viewed')), interviewing=Count('id', filter=Q(status='interviewing')), hired=Count('id', filter=Q(status='hired')), rejected=Count('id', filter=Q(status='rejected')), ) # 按投递人去重统计学历和院校分布（每人只算一次，取最高学历） DEGREE_RANK = {'博士': 4, 'MBA': 3, '硕士': 2, '本科': 1, '其他': 0} school_type_counter = Counter() degree_counter = Counter() def pick_top_edu(edu_list): """从教育经历中取最高学历的那条""" best = None best_rank = -1 for edu in edu_list: d = edu.get('degree', '').strip() rank = DEGREE_RANK.get(d, 0) if rank > best_rank: best_rank = rank best = edu return best or (edu_list[0] if edu_list else None) # 按 applicant_id 去重，每个人只取一条投递快照 seen_applicants = set() for applicant_id, snapshot in apps_qs.values_list('applicant_id', 'resume_snapshot'): if applicant_id in seen_applicants: continue seen_applicants.add(applicant_id) edu_list = snapshot.get('education', []) if snapshot else [] edu_list = [e for e in edu_list if e.get('school', '').strip() or e.get('degree', '').strip()] top = pick_top_edu(edu_list) if top: degree = top.get('degree', '').strip() if degree: degree_counter[degree] += 1 school_name = top.get('school', '').strip() school_type_counter[classify_school(school_name)] += 1 # 如果没有投递数据，从所有简历统计 if not seen_applicants: from apps.resumes.models import Resume for resume in Resume.objects.all(): edu_list = [e for e in (resume.education or []) if e.get('school', '').strip() or e.get('degree', '').strip()] top = pick_top_edu(edu_list) if top: degree = top.get('degree', '').strip() if degree: degree_counter[degree] += 1 school_name = top.get('school', '').strip() school_type_counter[classify_school(school_name)] += 1 # 保证 985/211/其他 顺序固定 school_type_dist = [ {'name': '985', 'value': school_type_counter.get('985', 0)}, {'name': '211', 'value': school_type_counter.get('211', 0)}, {'name': '其他', 'value': school_type_counter.get('其他', 0)}, ] degree_dist = [{'name': k, 'value': v} for k, v in degree_counter.items()] # 职位类别分布（top 10） category_dist = list( jobs_qs.values('category') .annotate(count=Count('id')) .order_by('-count')[:10] ) return Response({ 'job_stats': job_stats, 'app_stats': app_stats, 'school_type_dist': school_type_dist, 'degree_dist': degree_dist, 'category_dist': [{'name': c['category'], 'value': c['count']} for c in category_dist], })