import requestsfrom bs4 import BeautifulSoupfrom django.utils import timezonefrom jobs.models import JobHEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"}def clean_text(value): """清洗文本中的空格、换行等无效字符""" if not value: return "" return " ".join(value.strip().split())def crawl_job_list(url): """爬取岗位列表页并保存岗位信息""" response = requests.get(url, headers=HEADERS, timeout=10) response.encoding = response.apparent_encoding if response.status_code != 200: return {"code": 500, "message": "岗位页面请求失败"} soup = BeautifulSoup(response.text, "html.parser") job_items = soup.select(".job-item") save_count = 0 for item in job_items: title = clean_text(item.select_one(".job-title").get_text()) company = clean_text(item.select_one(".company-name").get_text()) city = clean_text(item.select_one(".job-city").get_text()) salary = clean_text(item.select_one(".job-salary").get_text()) education = clean_text(item.select_one(".job-education").get_text()) experience = clean_text(item.select_one(".job-experience").get_text()) if not title or not company: continue return { "code": 200, "message": "岗位爬取完成", "save_count": save_count, "total_count": len(job_items) }
from collections import defaultdictfrom math import sqrtfrom jobs.models import Jobfrom users.models import UserFavorite, JobApplicationdef cosine_similarity(user_a, user_b, user_job_scores): """计算两个用户之间的余弦相似度""" jobs_a = user_job_scores[user_a] jobs_b = user_job_scores[user_b] common_jobs = set(jobs_a.keys()) & set(jobs_b.keys()) if not common_jobs: return 0 return numerator / (sqrt(sum_a) * sqrt(sum_b))def get_recommend_jobs(current_user_id, limit=10): """为指定用户生成协同过滤岗位推荐结果""" user_job_scores = defaultdict(dict) favorites = UserFavorite.objects.all() for favorite in favorites: user_job_scores[favorite.user_id][favorite.job_id] = 3 applications = JobApplication.objects.all() for application in applications: user_job_scores[application.user_id][application.job_id] = 5 current_scores = user_job_scores.get(current_user_id, {}) if not current_scores: return Job.objects.filter(status=1).order_by("-crawl_time")[:limit] similarity_list = [] for other_user_id in user_job_scores.keys(): if other_user_id == current_user_id: continue similarity = cosine_similarity( current_user_id, other_user_id, user_job_scores ) if similarity > 0: similarity_list.append((other_user_id, similarity)) similarity_list.sort(key=lambda item: item[1], reverse=True) similar_users = similarity_list[:20] recommend_scores = defaultdict(float) for other_user_id, similarity in similar_users: for job_id, score in user_job_scores[other_user_id].items(): if job_id not in current_scores: recommend_scores[job_id] += similarity * score recommend_job_ids = sorted( recommend_scores, key=recommend_scores.get, reverse=True )[:limit] return Job.objects.filter( id__in=recommend_job_ids, status=1 ).order_by("-crawl_time")