企业官网建设流程全解析

简介：面向计算机专业期末大作业与项目实战的Boss直聘在线爬虫及数据分析可视化系统完整源码，附详细文档说明，适合需要完成课程设计、毕业设计或学习Python爬虫与数据分析的读者。项目基于Django与Django REST framework搭建后端，涵盖爬虫设计、数据清洗、存储及可视化展示全流程，并借助pandas、matplotlib、seaborn等库输出图表，代码注释详细，可运行。资源包共665个文件，约7.29MB，以Python脚本、HTML模板、JS交互逻辑、CSS样式及配置文件为主，另有文档、图片与演示素材，目录结构清晰便于按模块学习。已有95人学习下载，除源码外还包含环境搭建说明、使用手册及附赠内容，可帮助理解从爬取到分析展示的完整链路。

1. 从输入职位关键词到一张可视化大屏：这套系统把Boss直聘数据变成了什么

如果你想要投递某个城市的Java岗位，却只看到首页前两三页的职位，你根本不知道这个市场真实的薪资分布、经验要求集中在哪一段、哪些公司招人最猛。Boss直聘在线爬虫及数据分析可视化系统，就是解决这个问题的：用爬虫把职位列表、公司信息、薪资范围、经验学历要求采集下来，落进数据库，再通过数据分析把原始字段变成“城市薪资中位数”“岗位需求量Top10公司”“经验要求分布”这类结论，最后用可视化大屏呈现在浏览器里。整套链路适合两类人：一类是想把Python爬虫、SQLAlchemy存储、Pandas清洗、ECharts展示串成一个完整项目的开发者，另一类是做招聘市场研究但苦于数据量不够的运营或HR。它的核心价值不是“爬得有多快”，而是“把爬下来的数据真正用起来”。

2. 系统架构与选型：抓取、入库、分析、展示四层怎么分才不返工

2.1 四层架构：为什么先定边界再写第一行代码

我经手过的爬虫项目，凡是写到一半推倒重来的，基本都是因为没在开头把“抓回来的数据长什么样”想清楚。这个系统的数据量不是海量，单城市、单岗位关键词抓几千条，撑死几十MB，完全不需要Hadoop、Spark这套大数据体系。常见做法是拆成四层：

• 采集层：requests处理JSON接口，Selenium做兜底。Python生态里没有比这两个更稳的组合。
• 存储层：MySQL或PostgreSQL做库，SQLAlchemy做ORM。选SQLAlchemy不只是因为写代码方便，它的连接池、批量插入、事务控制都是爬虫场景刚需。
• 分析层：Pandas做清洗和透视，SQL做聚合查询。数据量小，内存计算完全够用。
• 展示层：FastAPI提供数据接口，前端用ECharts渲染图表。

这样分层最大的好处是：每一层都能独立测试。采集层挂了不影响已经入库的数据，分析层写得再烂也毁不掉原始数据。如果你把爬虫逻辑和展示逻辑写在一个脚本里，后期每改一个图表需求都要重新跑一遍全量抓取，那是给自己挖坑。

2.2 存储模型设计：职位表、公司表、快照表分开，别把所有字段塞一张表

我第一次做类似系统时，把所有字段塞进了一张表：职位名、公司名、薪资、经验、学历、城市、发布时间、抓取时间全混在一起。结果想分析“最近一周新增岗位数”时，发现抓取时间覆盖了原始发布时间，数据完全没法用。建议至少拆三张表：

position表里同时存salary_text和salary_min/max有点冗余，但值得：原始文本留着做校验，解析后的数值用于聚合。publish_time是职位在原平台上的发布时间，crawl_time在快照表里，两个时间字段必须分清，否则后面做“新增岗位趋势”时一定会算错。

SQLAlchemy里定义模型时，常见做法是这样：

from sqlalchemy import Column, Integer, String, DateTime, ForeignKey, create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker, relationship from datetime import datetime Base = declarative_base() class Position(Base): __tablename__ = 'position' id = Column(Integer, primary_key=True, autoincrement=True) job_name = Column(String(128), nullable=False) # 职位名称 salary_text = Column(String(64)) # 原始薪资文本，如 "15-20K·13薪" salary_min = Column(Integer) # 解析后的最低月薪，单位K salary_max = Column(Integer) # 解析后的最高月薪，单位K experience = Column(String(32)) # 经验要求，如 "1-3年" education = Column(String(32)) # 学历要求，如 "本科" company_id = Column(Integer, ForeignKey('company.id')) city = Column(String(32)) publish_time = Column(DateTime) # 平台上显示的发布时间 url = Column(String(256), unique=True) # 职位链接，天然唯一键 company = relationship("Company", back_populates="positions") class Company(Base): __tablename__ = 'company' id = Column(Integer, primary_key=True, autoincrement=True) company_name = Column(String(128), nullable=False) industry = Column(String(64)) # 所在行业 financing_stage = Column(String(64)) # 融资阶段 size = Column(String(64)) # 公司规模 address = Column(String(256)) class CrawlSnapshot(Base): __tablename__ = 'crawl_snapshot' id = Column(Integer, primary_key=True, autoincrement=True) position_id = Column(Integer, ForeignKey('position.id')) crawl_time = Column(DateTime, default=datetime.now) data_source = Column(String(32)) # 标记是列表页还是详情页采集

代码逻辑说明：Position和Company通过company_id建立外键关系，目的是分析“哪些公司同一职位重复发布”时可以直接JOIN。position表里给url加了unique=True，这是爬虫去重的根基。CrawlSnapshot表单独记录抓取批次，后面做增量更新全靠它。

参数说明：url字段长度设256够用，真要超长就改Text类型。publish_time存DateTime而不是字符串，不然Pandas做时间序列分析还得先转类型。autoincrement=True是MySQL的常规写法，SQLite也兼容。

2.3 调度与采集频率：单账号低频还是多账号散点采集

很多人一开始就想用几十个账号轮询，把抓取速度拉满。我的建议是：单账号低频起步，数据缺口比速度更致命。Boss直聘的岗位数据是动态的，同一个关键词在不同时间翻页，结果会有差异，所以“抓到的数据能持续稳定”比“一次性抓全”重要得多。

单账号低频的典型参数：每个请求之间sleep 5到10秒，每次抓取任务间隔至少1小时，单次任务翻页不超过5页。这样一小时能拿到大约300条职位记录，跑一天就有大几千条，足够做绝大多数市场分析。

如果确实需要多账号，也要控制在三五个以内，并且做好账号状态监控。账号一旦被限制访问，不只是当前任务失败，之前用这个账号抓的数据都可能是残缺样本。真正的工程重点不是提高并发，而是保证采集过程的连续性。

3. 爬虫模块落地：requests登录态、岗位列表解析与SQLAlchemy增量入库

3.1 登录态管理：手动扫码一次，Cookie串重复用

Boss直聘的职位详情和部分搜索结果需要登录态才能完整返回。每次启动都重新扫码登录不现实，常见做法是：第一次手动登录，把Cookie持久化到本地文件，后续运行直接从文件加载。

import json import time from pathlib import Path from selenium import webdriver from selenium.webdriver.chrome.options import Options COOKIE_FILE = Path("boss_cookies.json") def save_cookies(driver, path=COOKIE_FILE): """保存当前浏览器会话的Cookie到本地文件""" cookies = driver.get_cookies() with open(path, "w", encoding="utf-8") as f: json.dump(cookies, f, ensure_ascii=False, indent=2) print(f"已保存 {len(cookies)} 条Cookie") def load_cookies(driver, path=COOKIE_FILE): """从本地文件加载Cookie并注入浏览器""" if not path.exists(): return False with open(path, "r", encoding="utf-8") as f: cookies = json.load(f) for cookie in cookies: driver.add_cookie(cookie) return True # 首次登录：打开网页，人工扫码 options = Options() options.add_argument("--disable-blink-features=AutomationControlled") options.add_experimental_option("excludeSwitches", ["enable-automation"]) driver = webdriver.Chrome(options=options) driver.get("https://www.zhipin.com/") time.sleep(20) # 等待人工扫码完成 save_cookies(driver) driver.quit()

逻辑说明：load_cookies之前必须先driver.get到目标域名，否则Cookie无法注入。Cookie保存成JSON而不是直接用浏览器Profile，好处是跨机器迁移容易，而且你能肉眼看到里面有哪些字段，排查问题方便。

参数说明：sleep(20)是扫码等待时间，网络差可以加长。add_experimental_option("excludeSwitches", ["enable-automation"])用来去掉浏览器顶部的“Chrome正在受到自动软件控制”提示，这个提示本身不危险，但会影响页面渲染时的JS执行分支。

3.2 岗位列表请求与解析：JSON接口的翻页参数里藏着反爬线索

Boss直聘网页端的职位列表不是纯HTML渲染，它通过一个JSON接口返回结构化数据。直接请求这个接口比解析HTML稳定得多，字段就是现成的JSON，不用写一堆XPath解析。

import requests import json import time SESSION = requests.Session() SESSION.headers.update({ "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Referer": "https://www.zhipin.com/web/geek/job", "Accept": "application/json, text/plain, */*" }) # 这里从之前保存的Cookie文件中读取，注入Session with open("boss_cookies.json", "r", encoding="utf-8") as f: cookies = json.load(f) for cookie in cookies: if "domain" in cookie and "zhipin.com" in cookie["domain"]: SESSION.cookies.set(cookie["name"], cookie["value"]) def fetch_job_list(keyword: str, city_code: str, page: int = 1, page_size: int = 15) -> list: """ 抓取Boss直聘职位列表 :param keyword: 搜索关键词，比如 "Java开发" :param city_code: 城市编码，101010100代表北京 :param page: 页码，从1开始 :param page_size: 每页条数，建议保持默认15 """ url = "https://www.zhipin.com/wapi/zpgeek/search/joblist.json" params = { "query": keyword, "city": city_code, "page": page, "pageSize": page_size, } resp = SESSION.get(url, params=params, timeout=10) # 接口返回JSON，但有时会带JSONP前缀，先清理 text = resp.text.strip() if text.startswith("(") and text.endswith(")"): text = text[1:-1] data = json.loads(text) # 业务状态码和HTTP状态码是两回事，必须分开判断 if data.get("code") != 0: print(f"接口业务错误: {data.get('message')}") return [] job_list = data.get("zpData", {}).get("jobList", []) return job_list # 演示：抓取北京Java岗位第一页 jobs = fetch_job_list("Java开发", "101010100", page=1) print(f"本页返回职位数: {len(jobs)}") if jobs: sample = jobs[0] print("示例字段:", sample["jobName"], sample.get("salaryDesc"), sample.get("jobExperience"))

逻辑说明：SESSION复用TCP连接，比每次requests.get快不少，也能减少服务端看到的异常连接特征。接口返回的JSON外层有个code字段，0表示业务成功，非0时要打印message排错。有的接口会把JSONP外壳包在外面，所以加了去括号的前置处理。

参数说明：timeout=10是请求超时，Boss直聘接口偶尔会卡顿，设太小容易误杀。city_code用的是平台定义的编码，101010100是北京，101020100是上海，这个编码可以从网页地址栏里抄。page_size保持15不要改大，改大容易触发风控，而且翻页本身就能拿到足够数据。

3.3 SQLAlchemy增量入库：用唯一键去重而不是先查后插

爬虫最常见的低效操作是：每抓到一条数据，先去数据库查一下有没有，没有才插入。这样每条数据都要走一次网络往返，几千条下来慢得不可接受。更好的方式是利用数据库的唯一键，做插入或忽略。

from sqlalchemy.dialects.mysql import insert as mysql_insert from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker engine = create_engine( "mysql+pymysql://user:password@localhost:3306/boss_db?charset=utf8mb4", pool_size=5, pool_recycle=3600, pool_pre_ping=True ) SessionLocal = sessionmaker(bind=engine) def save_job_list(jobs: list): """批量入库，url重复的自动跳过""" session = SessionLocal() try: for job in jobs: stmt = mysql_insert(Position).values( job_name=job.get("jobName"), salary_text=job.get("salaryDesc", ""), experience=job.get("jobExperience", ""), education=job.get("jobDegree", ""), city=job.get("cityName", ""), url=job.get("encryptJobId", ""), publish_time=parse_publish_time(job.get("publishTime", "")), ).on_duplicate_key_update( # 如果url重复，说明已抓过，只更新时间戳 url=stmt.inserted.url ) session.execute(stmt) session.commit() print(f"入库完成，共处理 {len(jobs)} 条") except Exception as e: session.rollback() print(f"入库失败: {e}") finally: session.close()

逻辑说明：on_duplicate_key_update是MySQL的方言能力，SQLAlchemy通过mysql dialect暴露出来。重复时执行url=stmt.inserted.url，等于什么都不改，但能把冲突吞掉。这样批量插入时不需要先查数据库，速度提升非常明显。注意这段代码依赖MySQL，SQLite不支持这个语法。

参数说明：pool_size=5是连接池大小，单线程爬虫5个连接绝对够。pool_recycle=3600表示连接超过1小时自动重建，规避MySQL的wait_timeout断连问题。pool_pre_ping=True每次取连接前先ping一下，连接断了自动重连。这三个参数是我每写一个SQLAlchemy爬虫项目都要带的标配，少了任何一个都会在长跑任务里偶尔报错。

3.4 Selenium作为兜底：动态渲染字段和登录态失效时的备用通道

JSON接口虽然快，但不是所有信息都在里面。比如公司详情页的某些字段、职位的技能标签，经常是页面加载后由JS异步渲染出来的。这时候requests拿不到，只能用Selenium渲染页面再提取。

from selenium.webdriver.common.by import By from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC def fetch_job_detail_with_selenium(driver, url: str) -> dict: """通过浏览器渲染抓取职位详情页的动态字段""" driver.get(url) wait = WebDriverWait(driver, timeout=15) result = {} try: # 等待薪资元素出现，出现说明页面主体渲染完成 salary_elem = wait.until( EC.presence_of_element_located((By.CLASS_NAME, "salary")) ) result["salary_text"] = salary_elem.text.strip() job_desc_elem = driver.find_element(By.CLASS_NAME, "job-sec-text") result["job_desc"] = job_desc_elem.text.strip()[:500] # 只存前500字 except Exception as e: print(f"详情页解析失败: {url} -> {e}") result["error"] = str(e) return result

逻辑说明：selenium的定位策略里，CLASS_NAME是最直观的，但页面结构调整时容易失效。wait.until配合presence_of_element_located，是为了等异步渲染完成再取数据，直接find_element遇到慢网速大概率拿不到节点。

参数说明：timeout=15是最大等待秒数，超过就抛异常。job_desc截取前500字，是因为分析阶段用不到完整JD，只用来做关键词词频统计，存太多反而让表变臃肿。这个兜底通道不需要全量启用，只在JSON接口缺字段时针对少量职位调用，否则Selenium的开销会让任务时间成倍拉长。

4. 数据分析与可视化大屏：从薪资字段清洗到ECharts图表联动

4.1 薪资区间文本转数值：15-20K·13薪这种字符串怎么拆

数据库中存的是原始文本“15-20K·13薪”，这种字符串没法直接做均值、中位数计算。Pandas清洗时，需要把它拆成最低薪资和最高薪资两个数值字段。这个步骤看起来简单，实际坑很多：有“5-8K”，有“20-35K·14薪”，还有“薪资面议”，甚至“300-500元/天”。

import pandas as pd import re def parse_salary_text(text: str) -> tuple: """ 解析薪资文本，返回 (最低月薪K, 最高月薪K) "15-20K·13薪" -> (15.0, 20.0) "300-500元/天" -> (6.0, 10.0) # 按21个工作日折算成月薪 "薪资面议" -> (None, None) """ if not text or "面议" in text: return None, None # 先处理按天计薪的 day_match = re.search(r"(\d+)-(\d+)元/天", text) if day_match: low, high = int(day_match.group(1)), int(day_match.group(2)) return round(low * 21 / 1000, 1), round(high * 21 / 1000, 1) # 处理按K计薪的 k_match = re.search(r"(\d+)-(\d+)K", text, re.IGNORECASE) if k_match: return float(k_match.group(1)), float(k_match.group(2)) # 处理单一薪资 single_match = re.search(r"(\d+)K", text, re.IGNORECASE) if single_match: val = float(single_match.group(1)) return val, val return None, None # 应用到DataFrame df = pd.read_sql("SELECT salary_text FROM position", engine) df[["salary_min", "salary_max"]] = df["salary_text"].apply( lambda x: pd.Series(parse_salary_text(x)) ) print(df[["salary_text", "salary_min", "salary_max"]].head(10))

逻辑说明：正则的匹配顺序很重要，我把“元/天”的判断放在“K”之前，因为“300-500元/天”里也包含数字，但不包含K，顺序反了会漏。按月薪折算时按21个工作日计算，是保守估计，比按30天折更接近实际情况。

参数说明：re.IGNORECASE是为了兼容“15-20k”这类小写写法。round(..., 1)保留一位小数，避免后续聚合出现长尾巴。这个清洗函数要放到一个独立的模块里，因为后面每次跑分析都要用它，写重了维护成本高。

4.2 分析维度设计：岗位数、薪资中位数、经验学历交叉

数据入表后，先别急着做图形。先把分析维度定下来，常见的看板维度有三个：城市薪资中位数排行、岗位需求量的行业分布、经验要求与薪资的关系。每个维度对应一条SQL，SQL跑通了图表才有数据源头。

from sqlalchemy import text def query_salary_median_by_city(): """按城市统计岗位薪资中位数""" sql = text(""" SELECT city, COUNT(*) AS job_count, ROUND(AVG((salary_min + salary_max) / 2), 1) AS avg_salary, ROUND(MIN(salary_min), 1) AS min_salary, ROUND(MAX(salary_max), 1) AS max_salary FROM position WHERE salary_min IS NOT NULL GROUP BY city ORDER BY avg_salary DESC LIMIT 20 """) with engine.connect() as conn: df = pd.read_sql(sql, conn) return df def query_experience_salary_cross(): """交叉分析：经验要求对应的薪资分布""" sql = text(""" SELECT experience, COUNT(*) AS job_count, ROUND(AVG(salary_min), 1) AS avg_min_salary, ROUND(AVG(salary_max), 1) AS avg_max_salary FROM position WHERE salary_min IS NOT NULL GROUP BY experience ORDER BY job_count DESC """) with engine.connect() as conn: df = pd.read_sql(sql, conn) return df

逻辑说明：这里用了AVG((salary_min + salary_max) / 2)而不是直接AVG(salary_min)或AVG(salary_max)，是因为中位数区间值更接近招聘方的真实预算。LIMIT 20只取前20个城市，否则图表X轴太长根本没法看。

参数说明：WHERE salary_min IS NOT NULL这个条件很重要，它把“薪资面议”的记录全过滤掉了，不然AVG计算会把Null跳过但COUNT会计入，导致图表上的岗位数虚高。分组字段是离散值，聚合函数是连续值，这种“维度+度量”的结构，是后续所有GIF图表能正常渲染的前提。

4.3 FastAPI提供数据接口：别把数据直接写死在HTML里

可视化大屏的网页如果直接把数据写死在HTML里，每跑一次分析就要重新生成一次页面，根本没法维护。常见做法是后端用FastAPI暴露几个只读接口，前端ECharts通过fetch异步取数。这样后端更新数据后，前端刷新页面就能看到新图表。

from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel app = FastAPI() # 开发阶段允许前端任意地址访问，上线前必须收紧 app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["GET"], allow_headers=["*"], ) class SalaryRankItem(BaseModel): city: str job_count: int avg_salary: float min_salary: float max_salary: float @app.get("/api/salary_rank", response_model=list[SalaryRankItem]) def get_salary_rank(): df = query_salary_median_by_city() return df.to_dict(orient="records") @app.get("/api/experience_cross") def get_experience_cross(): df = query_experience_salary_cross() return df.to_dict(orient="records")

逻辑说明：response_model用Pydantic定义，FastAPI会自动校验返回字段的类型和个数，前端少一个字段直接报错，而不是渲染一张不完整的图。CORS中间件设置为允许所有来源，是为了让前端静态页面调试方便，上线前要改成具体域名，不然任何人都能调用你的数据接口。

参数说明：allow_methods=["GET"]已经够用，不需要PUT和POST。df.to_dict(orient="records")把DataFrame转成[{列名: 值}, ...]结构，JSON序列化不会出问题，这是Pandas转API响应最省事的做法。

4.4 ECharts大屏组装：地图、柱状图、折线图怎么联动

标题里带着“可视化”三个字，核心展示端我用ECharts。它的中文文档全、社区案例多、对地图的支持也好，比Tableau这类商业工具更适合嵌入自己的系统。大屏布局一般做成左边柱状图、中间核心指标卡片、右边折线图，下面的地图放城市分布。

<!DOCTYPE html> <html> <head> <meta charset="utf-8"> <title>Boss直聘数据分析大屏</title> <script src="https://cdn.jsdelivr.net/npm/echarts@5.4.3/dist/echarts.min.js"></script> </head> <body> <div id="salary_bar" style="width:33%;height:400px;float:left;"></div> <div id="trend_line" style="width:33%;height:400px;float:left;"></div> <script> const barChart = echarts.init(document.getElementById('salary_bar')); const lineChart = echarts.init(document.getElementById('trend_line')); // 柱状图：城市平均薪资Top20 fetch('/api/salary_rank') .then(res => res.json()) .then(data => { const cities = data.map(item => item.city); const avgSalaries = data.map(item => item.avg_salary); barChart.setOption({ title: { text: '城市平均薪资Top20' }, tooltip: { trigger: 'axis' }, xAxis: { type: 'category', data: cities, axisLabel: { rotate: 30 } }, yAxis: { type: 'value', name: '月薪(K)' }, series: [{ type: 'bar', data: avgSalaries, itemStyle: { color: '#5470c6' } }] }); }); // 折线图：经验要求与平均薪资关系 fetch('/api/experience_cross') .then(res => res.json()) .then(data => { const labels = data.map(item => item.experience); const avgMin = data.map(item => item.avg_min_salary); const avgMax = data.map(item => item.avg_max_salary); lineChart.setOption({ title: { text: '经验要求与薪资区间' }, tooltip: { trigger: 'axis' }, legend: { data: ['最低薪资', '最高薪资'] }, xAxis: { type: 'category', data: labels }, yAxis: { type: 'value', name: '月薪(K)' }, series: [ { name: '最低薪资', type: 'line', data: avgMin }, { name: '最高薪资', type: 'line', data: avgMax } ] }); }); </script> </body> </html>

逻辑说明：两个图表分别对应两个数据接口，ECharts的setOption只需要给xAxis的category数组和series的data数组，数据的映射关系在JavaScript里完成。折线图同时画两条线，是为了展示“经验要求越高，薪资下限和上限同时抬升”的趋势，这个图比单条曲线信息量大。

参数说明：axisLabel的rotate: 30是因为城市名太长横排会重叠，旋转30度是常见折中值。分析数据的工具有很多选择，我在Python侧用Pandas和SQL交替处理，前端固定ECharts，这套组合在“单机数据量在GB以下”的看板场景里基本是最省心的方案，不用引入太重的东西。

5. Boss直聘爬虫高频踩坑：数据为空、账号受限、连接池耗尽的三类现场

5.1 响应内容全是HTML的壳，关键数据不在里面

现象：requests请求joblist.json，返回200，但解析出来的jobList是空数组，打印全文发现返回的是一段HTML结构。

原因：接口地址变了。Boss直聘的接口路径不是永久稳定的，每年会调整几次。你在网上找到的旧教程里的接口地址，大概率已经失效。HTTP请求被重定向到了网页端页面，所以返回HTML而不是JSON。

解决：用浏览器开发者工具的网络面板，手动搜索一次“Java开发”，看真实发出的XHR请求路径和参数，替换掉代码里的URL。这件事没有捷径，接口地址只能靠实时抓包确认，不要盲目信任任何教程里写死的路径。我的习惯是每次跑批前先打印一次响应头里的x-b3-traceid，确认走到了正确的接口。

5.2 爬着爬着接口开始返回验证码和空数据

现象：任务跑到第200条左右，接口开始返回“验证码校验”或zpData为null，用浏览器打开同页面也出现滑块验证。

原因：请求频率超过阈值。Boss直聘对同一Session的请求频率有隐式限制，低频请求没事，一旦短时间翻页超过某个量级，风控就会介入。另一个诱因是Cookie过期，登录态失效后接口直接拒绝返回数据。

解决：两条腿走路。第一，把请求间隔从3秒调整到8到12秒，单次任务控制在50页以内。第二，启动任务前增加Cookie有效性检查，用第一个请求的返回码判断是否需要重新扫码。如果已经被验证码卡住，不要试图用程序自动过验证码，直接停下任务，人工扫码刷新Cookie再继续。验证码识别这条路不要碰，既容易把账号搞挂，工程投入也不划算。

5.3 入库越来越慢，最后MySQL连接池报错

现象：任务跑到一半，控制台出现Can't connect to MySQL server (timeout)，之前一直很正常的入库操作突然全部失败。

原因：SQLAlchemy连接池里的连接被MySQL服务端断开了。MySQL默认的wait_timeout是8小时，如果连接池里的连接空闲超过这个时间，服务端会把连接关掉，而客户端还持有旧连接，一用就报错。

解决：连接配置里加上pool_recycle=3600和pool_pre_ping=True。pool_recycle让连接在1小时未用后自动重建，pool_pre_ping让每次取连接前发送一个SELECT 1验证连接是否可用。这两个参数组合起来，基本能覆盖长跑任务的断连问题。另外，如果任务之间间隔很久，更推荐每次任务创建新的Session而不是复用全局Session。

5.4 可视化图表数据和数据库对不上：时区与编码

现象：前端图表显示的岗位数比数据库直接COUNT(*)的结果少了一半，而且中文城市名在图表上显示为乱码。

原因：两个问题叠加。第一，连接数据库时charset参数没设utf8mb4，中文读出来就是乱码。第二，FastAPI接口对DataFrame的空值处理方式与数据库不同，某些计数口径不一致，比如前端读取的接口做了WHERE salary_min IS NOT NULL过滤，而数据库COUNT(*)没过滤。

解决：连接串固定带上charset=utf8mb4，这是MySQL中文场景的基本配置。清洗阶段把每条数据的salary_min和salary_max都补齐，即使解析不出来也写成None，而不是留着空字符串。接口联调时，用一个固定样本量的页面做基准：数据库查一次、接口拉一次、页面渲染一次，三个数对不上就逐个排查，不要直接信任中间某一步。

6. 进阶：增量更新、断点续爬和定时调度，让数据管道自己跑起来

基础版本能全量抓取就够用了，但要让它长期不维护也能稳定产出，还需要三件事：增量更新、断点续爬、定时调度。

增量更新的核心是依赖publish_time。每次入库前，先取数据库里最大的publish_time，只抓比它晚的职位。这样当天新发的岗位会进来，历史数据不会重复占用空间。断点续爬则是把“上次任务抓到了第几页”记录在一个state.json文件里，任务中断后重启，直接跳回上次的页数继续，不用从头刷一遍。这两个机制加起来，能让数据管道每天只抓新增部分，跑一次不到5分钟。

定时调度我习惯用APScheduler，不需要额外起服务，在采集脚本里就能完成任务编排：

from apscheduler.schedulers.blocking import BlockingScheduler from datetime import datetime def daily_crawl_job(): print(f"[{datetime.now()}] 开始每日抓取任务") jobs = fetch_job_list("Java开发", "101010100", page=1) save_job_list(jobs) print(f"[{datetime.now()}] 当日任务完成") if __name__ == "__main__": scheduler = BlockingScheduler() # 每天早上8点执行一次 scheduler.add_job(daily_crawl_job, 'cron', hour=8, minute=0) scheduler.start()

参数说明：hour=8, minute=0是每天早8点，这个时间点岗位更新基本完成，而且服务器访问量还没到高峰。BlockingScheduler会阻塞主进程，适合跑在云服务器上配合systemd管理，异常退出时自动重启。

最后收个尾：这套系统做完之后，我最怀念的不是某张图表有多炫，而是每天早上一睁眼，数据库里又多了一天的真实岗位数据。那种持续积累的确定性，比任何一次全量抓取都踏实。后来我给它加了一个简单的数据质量检查：每天随机抽10条记录，人工核对一遍薪资和岗位名称是否对应。数据管道跑得再顺，也要定期回看原始文本，这个习惯救了我很多次——有一次接口调整字段顺序，图表看着正常，抽查才发现城市字段全串列了。希望帮到你。

本文还有配套的精品资源，点击获取