招聘信息抓取爬虫
思路
- 数据抓取：定时从招聘网站获取最新职位条目。
- 数据比对：将抓取结果与历史保存的条目（从阿里云 tablestore 获取）进行比对。
- 更新处理：对新增或更新条目：
- 写入阿里云 tablestore
- 抓取条目详情页 → qwen-flash 总结 → 企业微信通知
flowchart TD
A[定时触发: 阿里云函数计算 6h一次] --> B[抓取招聘网站最新条目]
B --> C[从tablestore中读取历史保存的条目]
C --> D{是否有更新?}
D -- 否 --> E[结束流程]
D -- 是 --> F[保存更新条目到tablestore]
F --> H[进入条目详情页爬取详细信息]
H --> I[调用 qwen-flash 总结]
I --> J[推送总结结果到企业微信]
J --> E
tablestore
步骤：新建表格实例，创建表格，设置主键，开启公网访问。给子用户设置 OTS 的权限并保存访问 id 和 key。
self.ots_endpoint = 'https://******.cn-beijing.ots.aliyuncs.com' # 请替换为您的实际endpoint
self.access_key_id = '******' # 请替换为您的AccessKey ID
self.access_key_secret = '******' # 请替换为您的AccessKey Secret
self.instance_name = '******' # 请替换为您的实例名
self.table_name = '******' # 表格名称
设置后尝试在本地读写一下看看是否正常工作。
就业信息网爬虫
tsinghua_scraper.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
import json
import os
import time
from datetime import datetime
import re
from bs4 import BeautifulSoup, Tag
from typing import List
from tablestore import *
class TsinghuaCareerScraper:
def __init__(self):
self.base_url = "https://career.cic.tsinghua.edu.cn/xsglxt/f/jyxt/anony/xxfb"
self.webhook_url = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=****************"
self.headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
}
self.session = requests.Session()
retry = Retry(total=3, backoff_factor=0.2, status_forcelist=[429, 500, 502, 503, 504])
adapter = HTTPAdapter(max_retries=retry)
self.session.mount('https://', adapter)
self.session.headers.update(self.headers)
self.date_pattern = re.compile(r'\d{4}-\d{2}-\d{2}|\d{4}/\d{2}/\d{2}|\d{2}-\d{2}')
def fetch_page(self):
try:
response = self.session.get(self.base_url, timeout=60)
response.raise_for_status()
response.encoding = 'utf-8'
return response.text
except requests.RequestException:
return None
def parse_content(self, html):
if not html:
return []
soup = BeautifulSoup(html, 'lxml')
job_items = []
try:
content_div = soup.find('div', class_='list') or soup.find('ul', class_='list') or soup
possible_selectors = [
'tr[onclick]',
'a[href^="/xsglxt/f/jyxt/anony/xxfb/info/"]',
'div.news-item',
'li.news-item',
'tr',
'a[href*="info"]'
]
items = []
for selector in possible_selectors:
found = content_div.select(selector)
if found and len(found) > 2:
items = found
break
if not items:
items = [link for link in content_div.find_all('a', href=True) if link.get_text(strip=True) and '查询' not in link.get_text() and '更多' not in link.get_text()]
valid_items = []
for item in items:
job_info = self.extract_job_info(item)
if job_info and job_info['title'] and len(job_info['title']) > 5:
invalid_keywords = ['查询', '更多', 'javascript:', '返回', '首页']
if not any(keyword in job_info['title'] for keyword in invalid_keywords):
valid_items.append(job_info)
job_items = valid_items[:20]
except Exception:
pass
return job_items
def extract_job_info(self, item):
try:
job_info = {'title': '', 'company': '', 'date': '', 'url': '', 'description': ''}
title_text = item.get_text(strip=True) if item.name == 'a' else (item.find('a').get_text(strip=True) if item.find('a') else item.get_text(strip=True))
job_info['title'] = ' '.join(title_text.split())
base_url = "https://career.cic.tsinghua.edu.cn"
link_elem = item if item.name == 'a' else item.find('a')
if link_elem and link_elem.get('href'):
href = link_elem['href']
if href.startswith('/'):
job_info['url'] = base_url + href
else:
job_info['url'] = href
if 'showZwxx?zpxxid=' in str(item):
zpxxid_match = re.search(r'zpxxid=(\d+)', str(item))
if zpxxid_match:
job_info['url'] = f"{base_url}/xsglxt/f/jyxt/anony/showZwxx?zpxxid={zpxxid_match.group(1)}&type="
title_parts = job_info['title'].split('————')
if len(title_parts) >= 2:
job_info['title'] = title_parts[0].strip()
job_info['company'] = title_parts[1].strip()
date_match = self.date_pattern.search(job_info['title'])
if date_match:
job_info['date'] = date_match.group()
job_info['title'] = self.date_pattern.sub('', job_info['title']).strip()
else:
job_info['date'] = datetime.now().strftime('%Y-%m-%d')
if not job_info['title'] or len(job_info['title']) < 5 or any(kw.lower() in job_info['title'].lower() for kw in ['查询', '更多', 'javascript:', '返回', '首页', 'void(0)']):
return None
return job_info
except Exception:
return None
def fetch_html(self, url: str) -> str:
resp = self.session.get(url, timeout=20)
resp.encoding = resp.apparent_encoding or "utf-8"
return resp.text
def extract_table_to_dict(self, table_tag: Tag) -> dict:
result = {}
if table_tag:
for row in table_tag.select("tr"):
headers = row.find_all("th")
values = row.find_all("td")
for i in range(min(len(headers), len(values))):
key = headers[i].get_text(strip=True).rstrip("：:")
value = values[i].get_text(" ", strip=True)
result[key] = value
return result
def parse_job_detail_page(self, html: str) -> dict:
soup = BeautifulSoup(html, "lxml")
job_info_table = soup.select_one("div.sideleft table#comTab")
job_info = self.extract_table_to_dict(job_info_table)
title_tag = soup.select_one(".banner-main > h1")
page_title = title_tag.get_text(strip=True) if title_tag else ""
return {"页面标题": page_title, "招聘信息": job_info}
def summarize_with_qwen_flash(self, parsed: dict) -> str:
api_key = os.getenv("DASHSCOPE_API_KEY", "sk-********************************")
endpoint = "https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions"
user_content = f"请用中文对以下招聘页面进行简洁总结，50-120字为宜。页面标题: {parsed.get('页面标题', '')}\n招聘信息:\n{json.dumps(parsed.get('招聘信息', {}), ensure_ascii=False)}"
headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
payload = {"model": "qwen-flash", "messages": [{"role": "user", "content": user_content}]}
try:
resp = requests.post(endpoint, headers=headers, json=payload, timeout=20)
resp.raise_for_status()
return resp.json().get("choices", [{}])[0].get("message", {}).get("content", "")
except Exception:
return ""
def build_wechat_detail_text(self, url: str) -> dict:
try:
html = self.fetch_html(url)
parsed = self.parse_job_detail_page(html)
summary = self.summarize_with_qwen_flash(parsed)
return {"summary": summary.strip()}
except Exception:
return {"summary": ""}
class AliyunOTSWriter:
def __init__(self):
self.ots_endpoint = 'https://************.cn-beijing.ots.aliyuncs.com'
self.access_key_id = '************************'
self.access_key_secret = '********************************'
self.instance_name = '************'
self.table_name = 'Tsinghua'
try:
self.client = OTSClient(self.ots_endpoint, self.access_key_id, self.access_key_secret, self.instance_name)
except Exception as e:
raise
def ensure_table_exists(self):
try:
self.client.describe_table(self.table_name)
return True
except OTSServiceError as e:
if e.get_error_code() == 'OTSObjectNotExist':
return False
else:
return False
def generate_unique_gid(self) -> int:
return int(time.time() * 1000)
def get_existing_titles_and_urls(self) -> set:
existing_items = set()
try:
inclusive_start_primary_key = [('gid', INF_MIN)]
exclusive_end_primary_key = [('gid', INF_MAX)]
columns_to_get = ['title', 'url']
_, _, row_list, _ = self.client.get_range(self.table_name, Direction.FORWARD, inclusive_start_primary_key, exclusive_end_primary_key, columns_to_get=columns_to_get, limit=None)
for row in row_list:
title, url = '', ''
for col_name, col_value, _ in row.attribute_columns:
if col_name == 'title':
title = col_value
elif col_name == 'url':
url = col_value
if title and url:
existing_items.add((title, url))
except Exception:
pass
return existing_items
主程序
main.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import logging
import sys
import os
import time
import requests
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('scraper.log', encoding='utf-8'),
logging.StreamHandler(sys.stdout)
]
)
logger = logging.getLogger(__name__)
def send_wechat_notification(job_info, webhook_url):
message = {
"msgtype": "text",
"text": {
"content": f"==========\n标题: {job_info.get('title', '')}\n公司: {job_info.get('company', '')}\n日期: {job_info.get('date', '')}\n链接: {job_info.get('url', '')}\n==========\n"
}
}
try:
response = requests.post(webhook_url, json=message)
if response.status_code == 200:
return True
return False
except Exception as e:
logger.error(f"发送企业微信通知时出错: {e}")
return False
def send_wechat_text(content: str, webhook_url: str) -> bool:
message = {"msgtype": "text", "text": {"content": content}}
try:
response = requests.post(webhook_url, json=message)
if response.status_code == 200:
return True
return False
except Exception as e:
logger.error(f"发送企业微信通知时出错: {e}")
return False
def write_to_ots(job_items, ots_config):
if not job_items:
return False
try:
from tablestore import OTSClient, Row, Condition, RowExistenceExpectation
client = OTSClient(
ots_config['endpoint'],
ots_config['access_key_id'],
ots_config['access_key_secret'],
ots_config['instance_name']
)
for i, job in enumerate(job_items):
gid = int(time.time() * 1000) + i
primary_key = [('gid', gid)]
attribute_columns = [
('title', job.get('title', '')),
('company', job.get('company', '')),
('date', job.get('date', '')),
('url', job.get('url', '')),
('description', job.get('description', '')),
('scraped_at', datetime.now().strftime('%Y-%m-%d %H:%M:%S')),
('created_at', datetime.now().strftime('%Y-%m-%d %H:%M:%S'))
]
row = Row(primary_key, attribute_columns)
try:
client.put_row(ots_config['table_name'], row, condition=Condition(RowExistenceExpectation.IGNORE))
except Exception as e:
logger.warning(f"单条写入失败: {e}")
logger.info(f"写入完成，共处理 {len(job_items)} 条数据")
return True
except Exception as e:
logger.error(f"批量写入失败: {e}")
return False
def main():
logger.info("开始执行爬虫任务...")
try:
from tsinghua_scraper import TsinghuaCareerScraper, AliyunOTSWriter
scraper = TsinghuaCareerScraper()
logger.info("开始抓取清华大学招聘信息...")
html = scraper.fetch_page()
if not html:
logger.error("无法获取网页内容，程序终止")
return
job_items = scraper.parse_content(html)
if not job_items:
logger.warning("未找到任何招聘信息")
return
logger.info(f"找到 {len(job_items)} 条招聘信息")
try:
ots_reader = AliyunOTSWriter()
existing_items = ots_reader.get_existing_titles_and_urls()
except Exception as e:
logger.error(f"初始化或读取OTS失败: {e}")
return
new_job_items = []
def fetch_detail(job):
url = job.get('url') or ''
if not url:
return job, {"summary": ""}
return job, scraper.build_wechat_detail_text(url)
with ThreadPoolExecutor(max_workers=6) as executor:
futures = {executor.submit(fetch_detail, job): job for job in job_items}
for future in as_completed(futures):
job = futures[future]
try:
_, details = future.result()
key = (job.get('title', ''), job.get('url', ''))
if key in existing_items:
continue
summary = details.get('summary', '')
msg_lines = [
f"🏷️ 标题: {job.get('title', '')}",
f"🏢 公司: {job.get('company', '')}",
f"📅 日期: {job.get('date', '')}",
f"🔗 链接: {job.get('url', '')}",
"====================",
summary
]
content = "\n".join(filter(None, msg_lines))
send_wechat_text(content, scraper.webhook_url)
new_job_items.append(job)
except Exception as e:
logger.warning(f"处理条目失败，改为发送基础通知: {e}")
send_wechat_notification(job, scraper.webhook_url)
new_job_items.append(job)
if new_job_items:
logger.info(f"发现 {len(new_job_items)} 条新数据，开始写入OTS...")
ots_config = {
'endpoint': 'https://************.cn-beijing.ots.aliyuncs.com',
'access_key_id': '************************',
'access_key_secret': '********************************',
'instance_name': '************',
'table_name': 'Tsinghua'
}
if write_to_ots(new_job_items, ots_config):
logger.info("新数据成功写入OTS")
else:
logger.error("新数据写入OTS失败")
else:
logger.info("没有新条目需要处理")
except Exception as e:
logger.error(f"执行主任务时出错: {e}")
logger.info("爬虫任务执行完毕")
if __name__ == "__main__":
main()