企业官网建设流程全解析

简介：这套资料面向计算机与软件工程专业毕业生、Python Web开发与数据分析初学者，提供豆瓣电影数据分析及可视化系统的完整实现，覆盖网络爬虫采集、pandas数据清洗、MySQL数据库存储、Flask后端接口与ECharts多角度图表展示的完整链路，可用于毕业设计选题、课程项目复现和全栈入门练习。压缩包共1054个文件，约30.07MB，包含688个py源码文件、157个pyc缓存文件、15个html页面、11个js脚本和1个sql建库脚本，并配有图片、说明文档及依赖文件，目录按前端页面、后端逻辑、数据处理和运行环境组织，便于检索与二次修改。已有257人学习或下载，说明项目具备一定的参考与复现基础。包内提供可运行源码、数据库脚本和论文文档，读者可据此理解电影数据爬取、清洗分析与动态图表实现，对照项目结构完成环境配置、功能梳理和扩展开发。

1. 从一份爬虫脚本到一个能交付的系统

答辩现场被问得最多的一句不是代码怎么写，而是这些图说明了什么。这份 python 毕业设计做的正是把某影视评分社区的公开电影数据，从采集、清洗、入库一路做到 python 数据分析与 python 数据可视化：爬虫用 requests 加 BeautifulSoup，存储落 MySQL 并配了完整 sql 脚本，分析层用 pandas 算指标，Flask 提供页面渲染和 JSON 接口，前端用 ECharts 画图，另外还带了论文。它适合三类人——赶毕设需要一套链路完整、能讲清楚每一环的代码；想练手数据从原始页面到图表的全过程；或者拿它当模板换成自己课题。项目不追求并发和性能，追求的是每一层都能打开看，每张图的数据都能顺着接口追回数据库。

2. 架构与目录：Flask 蓝图、pandas 分析层和静态前端怎么分层

拿到这份源码，第一件事不是 pip install，而是先看它怎么分层。毕设项目最容易翻车的地方是写成一个大 app.py，所有路由、SQL、统计逻辑全塞在一起，答辩老师随便指一个图问「这个数怎么算的」，人就卡住了。这份代码把链路切成了四段：采集、存储、分析、展示，段与段之间用普通的 Python 函数和 dict 传递，没有引入消息队列、任务调度这类对毕设来说过重的东西。

2.1 四条链路：采集、存储、分析、展示

采集层放在spider/里，职责非常窄：给定页码和目标站点地址，返回一个结构化字典列表。它不知道数据库长什么样，也不知道前端要画几个图。这样做的好处是换数据源时只改这一层，后面的分析代码一行不动。

存储层就是 MySQL，一张movie主表承担了绝大部分查询。有人会问为什么不拆成电影表、类型表、演员表的标准三范式。实际取舍是：类型是斜杠分隔的多值字段，拆表之后每次按类型统计都要 join 一次，对几万条数据量来说完全没必要，用explode在 pandas 里拆反而更直观，答辩时也更好讲。

分析层在analysis/下，核心就两个文件：loader.py负责把表读成 DataFrame 并做类型修正，metrics.py负责产出图表需要的指标。这两个文件里没有 Flask 的任何 import，意味着你完全可以脱离 Web 跑一段脚本验证某个统计口径对不对，这一点在调试阶段非常省事。

展示层拆成两半：views/page.py走 Jinja2 模板渲染 HTML 骨架，views/api.py走蓝图返回 JSON。前端不参与任何计算，只负责把 JSON 塞进 ECharts 的 option 里。这种前后端职责清晰的写法，比用 pyecharts 直接在后端生成 HTML 片段要灵活得多——同一个接口的数据可以同时给柱状图和饼图用。

2.2 目录树与每个文件的职责

实际目录大致是这样组织的，每个人命名习惯不同，但主干跑不出这个范围：

movie_analysis/ ├── app.py # 应用入口：创建 app、注册蓝图、加载配置 ├── config.py # 数据库、分页、缓存的配置集中在这里 ├── requirements.txt # 依赖清单 ├── spider/ │ ├── base.py # Session 构建、限速重试的公共方法 │ ├── movie_spider.py # 列表页 + 详情页的解析逻辑 │ └── pipeline.py # 清洗、去重、批量入库 ├── analysis/ │ ├── loader.py # read_sql -> DataFrame，含类型修正 │ └── metrics.py # 各类统计指标的计算函数 ├── views/ │ ├── page.py # 首页、排行榜、详情页的模板路由 │ └── api.py # /api/* 的 JSON 接口 ├── static/ │ ├── js/echarts.min.js # 本地放一份，别依赖 CDN │ ├── js/charts.js # 各图表的 option 与取数逻辑 │ └── css/style.css ├── templates/ │ ├── base.html # 公共 head、导航、脚本引入 │ ├── index.html # 总览页 │ ├── rank.html # 高分榜与类型榜 │ └── detail.html # 单部影片详情 └── sql/ ├── schema.sql # 建库建表 └── init_data.sql # 初始化样例数据

app.py里最典型的一段是应用工厂，把配置、蓝图、错误处理一次性装配好：

from flask import Flask, render_template from config import Config from views.page import page_bp from views.api import api_bp def create_app(): app = Flask(__name__) app.config.from_object(Config) # 蓝图拆分：页面路由和接口路由互不干扰 app.register_blueprint(page_bp) app.register_blueprint(api_bp) @app.errorhandler(404) def not_found(e): return render_template("404.html"), 404 return app app = create_app() if __name__ == "__main__": # debug 只在本地开，部署时用 gunicorn 之类托管 app.run(host="0.0.0.0", port=5000, debug=True)

这段代码的关键点在于create_app工厂函数：测试时可以传入不同配置创建互不影响的实例，也避免了循环 import。register_blueprint的顺序无所谓，但两个蓝图必须设置不同的url_prefix（接口蓝图设成/api），否则页面路由和接口会撞在一起。host="0.0.0.0"是为了让同一局域网内的手机或另一台机器能访问，只在本机调试的话写127.0.0.1更安全。

config.py我是习惯把配置抽出来单独放，密钥和密码不进 Git：

import os from dotenv import load_dotenv load_dotenv() # 从项目根目录的 .env 读取，方便换机器不用改代码 class Config: DB_HOST = os.getenv("DB_HOST", "127.0.0.1") DB_PORT = int(os.getenv("DB_PORT", 3306)) DB_USER = os.getenv("DB_USER", "root") DB_PASS = os.getenv("DB_PASS", "123456") DB_NAME = os.getenv("DB_NAME", "movie_analysis") # pandas 的 read_sql 需要 SQLAlchemy 风格的连接串，charset 必须显式写 SQLALCHEMY_DATABASE_URI = ( f"mysql+pymysql://{DB_USER}:{DB_PASS}@{DB_HOST}:{DB_PORT}/{DB_NAME}" f"?charset=utf8mb4" ) PAGE_SIZE = 20 # 列表页每页条数 CACHE_TTL = 300 # 统计接口内存缓存秒数 SECRET_KEY = os.getenv("SECRET_KEY", "dev-only-change-me")

charset=utf8mb4这一段是很多人抄代码时漏掉的，漏了之后中文标题会变成问号或者直接抛Incorrect string value，后面第 5 章会具体讲。CACHE_TTL留着是因为统计接口每次都跑一遍全表聚合有点浪费，缓存五分钟对毕设演示足够了。

2.3 依赖清单与版本约定

把依赖版本写清楚，能省掉一半「在我电脑上能跑」的纠纷。常见的组合是这样：

这里有个容易忽略的版本坑：pandas 2.x 对read_sql的入参收紧了，直接传 pymysql 的 connection 对象会报类型错误或警告，必须传 SQLAlchemy 的Engine或Connection。所以requirements.txt里 SQLAlchemy 不能省，哪怕你一次 ORM 都不用。另一个坑是 Windows 上装 lxml 偶尔需要编译环境，装不上就退一步用html.parser，速度慢一点但不影响跑通。

3. 采集与入库：从分页列表到 MySQL 表的完整过程

爬虫部分是这份资源里最值得逐行看的地方，因为它不是那种「一口气抓完再存」的脚本，而是分成了「列表页落队列」和「详情页补字段」两段。这样设计的原因很实际：列表页一页能拿到二三十条片名和评分，翻五十页也就五十个请求；详情页每部片一个请求，量级直接放大三十倍。两段分开之后，中间任何一步中断，都能从上次的位置续上，不至于从头再来。

3.1 会话、限速与重试

先看请求的基础封装：

# spider/base.py import random import time import requests from requests.adapters import HTTPAdapter HEADERS = { "User-Agent": ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/120.0 Safari/537.36" ), "Accept-Language": "zh-CN,zh;q=0.9", "Referer": "https://demo-movie.invalid/", # 换成目标站首页 } def build_session(retries=3): """复用连接 + 自动重试，避免每条请求都重新握手""" session = requests.Session() session.headers.update(HEADERS) adapter = HTTPAdapter(max_retries=retries, pool_connections=10, pool_maxsize=10) session.mount("http://", adapter) session.mount("https://", adapter) return session def polite_get(session, url, params=None, min_sleep=1.5, max_sleep=3.5): """每次请求前随机停顿，失败由 adapter 接管重试""" time.sleep(random.uniform(min_sleep, max_sleep)) resp = session.get(url, params=params, timeout=10) resp.raise_for_status() # 响应头没声明编码时，requests 会退回 ISO-8859-1，中文必乱 if not resp.encoding or resp.encoding.lower() == "iso-8859-1": resp.encoding = resp.apparent_encoding return resp

Session的价值在于连接复用和 header 复用，几百次请求下来比裸requests.get快不少。HTTPAdapter(max_retries=3)只对连接错误和 5xx 生效，4xx 不重试——因为 4xx 通常是地址写错或权限问题，重试只是浪费时间。random.uniform的随机停顿比固定sleep(2)更自然，固定间隔的请求序列在服务端日志里一眼就能看出来。apparent_encoding会用 chardet 猜编码，对没有声明 charset 的页面是必要的兜底，代价是每个响应多花几毫秒做探测。

列表页和详情页的解析函数分开写：

# spider/movie_spider.py from urllib.parse import urljoin from bs4 import BeautifulSoup def parse_list(html, base_url): """列表页：只拿 id、片名、评分、详情链接""" soup = BeautifulSoup(html, "lxml") rows = [] for node in soup.select("div.item"): link = node.select_one("a.nbg") if link is None or not link.get("href"): continue rating_node = node.select_one("span.rating_nums") rows.append({ # 详情页地址末段当业务主键，比片名可靠 "source_id": link["href"].strip("/").split("/")[-1], "title": link.get("title", "").strip(), "detail_url": urljoin(base_url, link["href"]), "rating": rating_node.get_text(strip=True) if rating_node else None, }) return rows def parse_detail(html): """详情页：补导演、主演、类型、地区、时长等字段""" soup = BeautifulSoup(html, "lxml") info_text = soup.select_one("#info").get_text(" ", strip=True) if soup.select_one("#info") else "" return { "info": info_text, "genres": "".join( span.get_text(strip=True) + "/" for span in soup.select("span[property='v:genre']") ).rstrip("/"), "summary": soup.select_one("span[property='v:summary']").get_text(strip=True) if soup.select_one("span[property='v:summary']") else "", }

解析函数只做「HTML 到 dict」的机械转换，不做任何清洗和格式化，这一步的边界要守住。原因很实际：选择器写错的时候，你希望问题定位在解析层，而不是混在清洗逻辑里。urljoin处理相对路径，避免href="/subject/123/"拼出错误地址。source_id取路径末段，比用片名当主键稳得多，同名电影不会互相覆盖。

3.2 脏数据清洗口径

原始页面上的字段非常自由：年份可能写在简介里，时长可能带「分钟」也可能带「min」，评分人数可能写着「12345人评价」，类型之间用斜杠也可能用空格。清洗要统一口径，不然后面 pandas 聚合的时候一堆object类型没法算。

# spider/pipeline.py import re YEAR_RE = re.compile(r"(19|20)\d{2}") RUNTIME_RE = re.compile(r"(\d+)\s*(?:分钟|min)") DATE_RE = re.compile(r"(\d{4}-\d{2}-\d{2})") def clean_row(raw): """把原始 dict 转成可直接入库的类型""" info = raw.get("info", "") year_match = YEAR_RE.search(info) runtime_match = RUNTIME_RE.search(info) date_match = DATE_RE.search(info) # 评分人数：去掉所有非数字字符，"12345人评价" -> 12345 votes_text = re.sub(r"\D", "", raw.get("votes") or "0") rating = raw.get("rating") try: rating = round(float(rating), 1) if rating else None except ValueError: rating = None # 出现"暂无评分"这类文本时直接置空 genres = raw.get("genres", "") return { "source_id": raw["source_id"], "title": re.sub(r"\s+", " ", raw.get("title", "")).strip()[:120], "director": (raw.get("director") or "").strip()[:120], "actors": (raw.get("actors") or "").strip()[:255], "year": int(year_match.group(0)) if year_match else None, "region": (raw.get("region") or "").replace(" ", "")[:60], "genres": "/".join(g.strip() for g in re.split(r"[/、,]", genres) if g.strip())[:120], "runtime": int(runtime_match.group(1)) if runtime_match else None, "release_date": date_match.group(1) if date_match else None, "rating": rating, "votes": int(votes_text) if votes_text else 0, "summary": (raw.get("summary") or "").strip(), }

几个参数值得单独说明。YEAR_RE用的是(19|20)\d{2}而不是\d{4}，因为简介里会出现片长、票房之类的四位数，加前缀能过滤掉大部分误匹配。rating用try/except而不是float()直接转，是因为页面上确实存在「暂无评分」这种文本，一处异常就能让整批数据中断。字段都做了截断（[:120]、[:255]），这是为了防止个别超长简介把VARCHAR撑爆报Data too long，这个错误在批量插入时特别难定位是哪一条出的问题。

清洗完之后还有一步去重。用source_id做集合过滤，同一批里出现的重复直接丢掉；跨批次重复交给数据库的唯一索引兜底。

3.3 建表 SQL 与批量写入

表结构决定了后面所有查询的上限，字段类型选错了，改起来要重建表：

-- sql/schema.sql CREATE DATABASE IF NOT EXISTS movie_analysis DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci; USE movie_analysis; CREATE TABLE IF NOT EXISTS movie ( id INT UNSIGNED NOT NULL AUTO_INCREMENT COMMENT '自增主键', source_id VARCHAR(20) NOT NULL COMMENT '来源站点业务ID', title VARCHAR(120) NOT NULL COMMENT '片名', director VARCHAR(120) DEFAULT NULL COMMENT '导演', actors VARCHAR(255) DEFAULT NULL COMMENT '主演', year SMALLINT DEFAULT NULL COMMENT '上映年份', region VARCHAR(60) DEFAULT NULL COMMENT '国家/地区', genres VARCHAR(120) DEFAULT NULL COMMENT '类型，斜杠分隔', runtime SMALLINT DEFAULT NULL COMMENT '片长（分钟）', release_date DATE DEFAULT NULL COMMENT '上映日期', rating DECIMAL(3,1) DEFAULT NULL COMMENT '评分，一位小数', votes INT UNSIGNED DEFAULT 0 COMMENT '评价人数', summary TEXT COMMENT '简介', crawl_time DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT '入库时间', PRIMARY KEY (id), UNIQUE KEY uk_source (source_id), KEY idx_year (year), KEY idx_rating (rating), KEY idx_region (region) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci COMMENT='电影主表';

rating用DECIMAL(3,1)而不是FLOAT，因为浮点数在排序和分组时会出8.699999这种鬼东西，做成排行榜的时候名次会玄学变化。uk_source这个唯一键是幂等写入的基础，重跑爬虫不会产生重复行。三个普通索引对应的就是最常见的三个筛选维度：年份趋势、评分排行、地区分布。summary用TEXT而不是VARCHAR(2000)，是因为部分影片简介确实超过两千字，TEXT不占行内空间，查询时不选它就没开销。

写入用批量executemany加冲突更新：

# spider/pipeline.py import pymysql FIELDS = ("source_id", "title", "director", "actors", "year", "region", "genres", "runtime", "release_date", "rating", "votes", "summary") INSERT_SQL = """ INSERT INTO movie (source_id, title, director, actors, year, region, genres, runtime, release_date, rating, votes, summary) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE rating = VALUES(rating), votes = VALUES(votes), summary = VALUES(summary), crawl_time = CURRENT_TIMESTAMP """ def save_rows(conn, rows, batch_size=200): """分批提交，避免单条事务过大把连接撑爆""" if not rows: return 0 with conn.cursor() as cur: for i in range(0, len(rows), batch_size): chunk = rows[i:i + batch_size] cur.executemany(INSERT_SQL, [tuple(r[k] for k in FIELDS) for r in chunk]) conn.commit() return len(rows)

ON DUPLICATE KEY UPDATE只更新评分、人数、简介三个字段，是有意为之：片名、导演这些字段不会变，更新它们纯属浪费；而评分和评价人数会随站点数据变化，属于必须刷新的。batch_size=200是个经验值，太小提交次数多，太大单条 SQL 超过max_allowed_packet会断，几百条的量级比较稳。conn.commit()放在每个批次后面，中途失败最多丢一批，不会全部回滚。

采集主流程串起来大概二十行：

def run(start=0, pages=50): session = build_session() conn = pymysql.connect(host=Config.DB_HOST, port=Config.DB_PORT, user=Config.DB_USER, password=Config.DB_PASS, database=Config.DB_NAME, charset="utf8mb4") try: for page in range(pages): resp = polite_get(session, LIST_URL, params={"start": start + page * 25}) for item in parse_list(resp.text, BASE_URL): detail = polite_get(session, item["detail_url"]) merged = {**item, **parse_detail(detail.text)} save_rows(conn, [clean_row(merged)], batch_size=1) finally: conn.close()

这里batch_size=1是简化写法，实际跑的时候应该先把一整页的详情都抓完，攒够一批再统一写，能少几十次提交。finally里关连接是硬性要求，脚本跑到一半报错但连接没关，MySQL 那边会留下大量Sleep状态的连接，久了就Too many connections。

4. 分析与展示：pandas 指标、Flask 接口和 ECharts 配置怎么串起来

这一层是整份代码里最容易被低估的部分。很多人以为数据分析就是df.groupby().count()一行搞定，实际上真正决定成败的是「口径」——同一个「类型分布」，按影片数算和按评价人数加权算，出来的图完全不同。代码里每个图都对应一个明确的口径，写论文的时候这些口径就是方法论章节的素材。

4.1 六个图的指标口径

先看全景，再逐个拆代码：

高分榜那张表必须加votes >= 1000这个门槛，这是血泪经验。不加的话榜单头部会出现一堆只有几十个人打分、分数 9.5 的冷门片，看起来漂亮但完全没有分析意义。这个阈值在论文里正好可以写成一节「样本量过滤策略」，说明为什么小样本评分不可靠。

4.2 explode 与 cut：类型和评分怎么算

读数据和基础清洗放一起，避免每个指标函数都重复处理：

# analysis/loader.py import pandas as pd from sqlalchemy import create_engine from config import Config _engine = None def get_engine(): """全局单例，连接池复用，不要每次统计都新建""" global _engine if _engine is None: _engine = create_engine(Config.SQLALCHEMY_DATABASE_URI, pool_pre_ping=True) return _engine def load_movie_df(): sql = "SELECT * FROM movie WHERE rating IS NOT NULL AND year IS NOT NULL" df = pd.read_sql(sql, get_engine()) # DECIMAL 读出来可能是 Decimal，统一转成 float 才能做分桶 df["rating"] = pd.to_numeric(df["rating"], errors="coerce") df["votes"] = pd.to_numeric(df["votes"], errors="coerce").fillna(0).astype(int) df["year"] = df["year"].astype(int) df = df.dropna(subset=["rating"]) return df

pool_pre_ping=True会在取连接前先 ping 一下，防止 MySQL 那边超时断开后前端报Lost connection——本地开发长时间不操作再刷新页面，这个错误几乎必现。pd.to_numeric(..., errors="coerce")把无法转换的值变成NaN而不是抛异常，配合后面的dropna就是一条完整的清洗链。

类型统计是这套代码里最有代表性的一段：

# analysis/metrics.py import pandas as pd def genre_rank(df, top=10): """类型分布：先把斜杠串拆成 list，再 explode 成多行""" series = df["genres"].fillna("").str.split("/") long_df = df.assign(genre=series).explode("genre") long_df = long_df[long_df["genre"].str.strip() != ""] counts = long_df["genre"].value_counts().head(top) return {"x": counts.index.tolist(), "y": counts.values.tolist()} def rating_dist(df): """评分分布：左闭右开分桶，空区间也要占位""" bins = [0, 6, 7, 7.5, 8, 8.5, 9, 10.01] labels = ["6分以下", "6-7", "7-7.5", "7.5-8", "8-8.5", "8.5-9", "9分以上"] bucket = pd.cut(df["rating"], bins=bins, labels=labels, right=False) counts = bucket.value_counts().reindex(labels, fill_value=0) return {"labels": labels, "values": counts.tolist()}

explode的前置条件是列里必须是 list 或可迭代对象，所以必须先str.split("/")。如果忘了这一步直接 explode，pandas 会把字符串当序列拆成单个字符，结果图上一堆「动」「作」这种单字，第一次跑出来我自己都愣了一下。assign(genre=series)比直接改原 DataFrame 好，因为函数可能被别的指标复用，不能污染入参。

pd.cut的bins最后一个写10.01而不是10，是因为默认right=True时 10 分本身会落在区间外变成NaN；这里配了right=False左闭右开，理论上10够用，但评分是DECIMAL(3,1)，万一出现10.0之外的浮点误差就漏了，多给一点余量更保险。reindex(labels, fill_value=0)保证某个区间一部片都没有时也返回 0，否则 ECharts 的 x 轴刻度会和数据错位，图看起来就「缺一块」。

趋势和散点的实现更简单：

def year_trend(df): counts = df.groupby("year").size().sort_index() return {"x": [int(y) for y in counts.index], "y": counts.values.tolist()} def rating_vs_votes(df, limit=800): """散点数据：[评分, 人数, 片名, 片长]""" sample = df[df["votes"] > 0].nlargest(limit, "votes") return [ [float(r.rating), int(r.votes), r.title, int(r.runtime or 0)] for r in sample.itertuples() ]

散点图限制 800 个点是有意为之。全部几万条数据扔上去，浏览器渲染会明显卡顿，而且点叠在一起根本看不出分布，nlargest取人气最高的那批反而更能说明问题。返回的每个点是个四元组数组，比返回对象数组省一半字节，前端按下标取值也快。

4.3 蓝图与统一返回结构

接口层最关键的不是代码多少，而是返回结构要统一。前端拿到{code, msg, data}这套结构，错误处理就能写一次管所有接口：

# views/api.py import time from functools import wraps from flask import Blueprint, jsonify, request from analysis.loader import load_movie_df from analysis import metrics from config import Config api_bp = Blueprint("api", __name__, url_prefix="/api") _cache = {} def cached(key, ttl=None): """手写内存缓存，key 里必须带上所有查询参数""" ttl = ttl or Config.CACHE_TTL now = time.time() if key in _cache and now - _cache[key][0] < t <p> <a href="https://download.csdn.net/download/caofeng891102/91297907" style="color:#ec7500;font-size:14px;"> 本文还有配套的精品资源，点击获取 </a> <img alt="menu-r.4af5f7ec.gif" src="https://csdnimg.cn/release/wenkucmsfe/public/img/menu-r.4af5f7ec.gif" style="width:16px;margin-left:4px;vertical-align:text-bottom;cursor:text;"> </p>