Python Flask在线投票系统从零到上线:数据模型、防刷策略与部署全攻略
2026/9/12 10:40:00 网站建设 项目流程

简介:基于Python的在线投票网站设计源码,采用Django框架编写,面向Web开发初、中级学习者,可用于课堂作业、社团投票或小型民意调研场景。项目完整呈现了在线投票的后端业务逻辑、前端页面与数据库存储方案,能帮助读者理解用户界面、应用逻辑与数据持久化之间的协作关系。压缩包共收录39个文件,大小约693KB,包含14个Python源码文件、11个pyc字节码文件、4个XML配置文件、3个HTML页面、1个CSS样式文件以及SQLite数据库文件,并附带说明文档、许可证和Git忽略规则等配套文件,目录结构清晰,模块划分明确。已有579人学习或浏览过该资源。通过阅读这份源码,可以直观掌握Django的MVT分层思想,梳理在线投票从数据建模、页面渲染到用户交互的完整流程,同时借鉴其投票逻辑、模板组织和静态资源管理方式,为自建同类Web应用提供参考。

1. 从“能跑”到“能上线”,在线投票系统的设计分水岭

基于Python的在线投票网站设计源码,本质上是一道“全栈最小可行解”的练习题:前端收集用户选择,后端做合法性校验与计数,存储层保证结果不被篡改,最后还要面对并发投票、重复投票和结果实时性这些真实问题。很多初学者会用Flask写一个POST接口、Sqlite存计数、再用一行模板引擎渲染结果,这个链路在单机演示时确实成立,但一旦遇到“同一用户刷票”“管理员要导出报表”“刷新页面重复提交”这些场景,就会立刻暴露设计缺口。

这篇内容适合两类人:一类是想用Flask或Django完成课设/毕设的开发者,另一类是已经在写业务系统、想快速补全投票场景通用能力的后端工程师。我会从数据模型设计、防刷策略、前端实时反馈、部署安全四个维度逐步展开,每一部分都给出可以直接落地的代码片段,并解释关键参数为什么要这么设。最后收在“投票系统上线前必做的5个检查项”上,这些都是在真实项目里踩过坑之后沉淀下来的东西。

整个方案以Python 3.8+、Flask 2.x、SQLite作为默认组合,因为它们的组合最贴近“设计源码”的用途——逻辑清晰、依赖极少、即便在Windows环境也能零配置跑通。如果你更熟悉Django,本文的模型设计思路同样可以平移过去,只是ORM表述略有差异。

2. 数据模型与存储选型:单表计数为什么撑不住真实投票

2.1 投票系统的核心实体:票、选项、用户、会话

在写任何代码之前,先要把“谁投给了谁、什么时候投的、怎么证明没重复投”这三件事在数据上定义清楚。最常见的错误是只建一张表,字段长这样:vote_id、option_id、count。这种设计只能支持“把计数值加一”的演示逻辑,一旦需要查某个用户的投票历史、取消投票后回滚计数、或者统计某个时间段内的投票趋势,就只能靠全表扫描。

我一般会拆成四张表:vote(投票主题)、option(投票选项)、ballot(投票记录)、voter(投票人/会话)。其中ballot表是关键,它记录的是“一次投票行为”,而不是“一个计数结果”。这样设计的直接好处是:投票计数可以从ballot表实时聚合,而不是维护一个可能和明细不一致的冗余字段。如果需要高性能,可以在option表上加一个total_count作为缓存列,但必须通过事务保证和ballot表明细一致。

对于“谁在投票”的识别,匿名投票场景下不要用IP作为唯一凭证,因为同一个校园网出口IP会误伤大量正常用户。更稳妥的做法是生成一个随机token存入Cookie,同时记录IP和User-Agent作为辅助风控信号。一个签名后的Cookie值建议使用HMAC-SHA256,密钥从环境变量读取,不要硬编码在源码里。

2.1.1 SQLite vs MySQL:源码项目应该怎么选

SQLite的亮点是零配置、单文件、Python内置驱动,对于并发量在每秒几十次以内的课设和内部工具完全够用。它的瓶颈在写入锁是库级排他,不适合高并发写入场景。MySQL则需要额外装服务、管理账号和权限,但换来的Row-Level Locking和更成熟的连接池,会让项目在部署到公网后更有余量。

需求维度SQLiteMySQL
部署成本需安装配置
并发写低(库级锁)高(行级锁)
数据备份拷贝文件mysqldump
适合场景学习/内部/低并发公网/生产/高并发

我的建议是:源码里默认用SQLite,但DAO层做一层封装,让切换数据库时只需改一个连接字符串。后续如果投票活动上了公网、预期并发过百,再把连接串切到MySQL,不用改业务逻辑。

2.2 建表语句与ORM模型:用SQLAlchemy表达约束

我们这里使用SQLAlchemy 2.x的ORM风格,而不是裸SQL,因为对于这种多表关联的查询,ORM可以少写大量样板代码,同时通过类型注解提升可读性。下面是核心模型定义,重点关注唯一约束和索引设计。

from datetime import datetime from sqlalchemy import ( Column, Integer, String, DateTime, ForeignKey, UniqueConstraint, Index, Text, Boolean ) from sqlalchemy.orm import declarative_base, relationship Base = declarative_base() class Vote(Base): __tablename__ = "vote" id = Column(Integer, primary_key=True) title = Column(String(200), nullable=False) description = Column(Text, default="") is_multi = Column(Boolean, default=False) # 是否允许多选 max_choices = Column(Integer, default=1) # 多选时的最大勾选数 start_time = Column(DateTime, default=datetime.utcnow) end_time = Column(DateTime, nullable=True) # None 表示永不截止 is_active = Column(Boolean, default=True) options = relationship("Option", back_populates="vote", cascade="all, delete-orphan") ballots = relationship("Ballot", back_populates="vote") class Option(Base): __tablename__ = "option" id = Column(Integer, primary_key=True) vote_id = Column(Integer, ForeignKey("vote.id"), nullable=False) label = Column(String(500), nullable=False) # 选项文案 sort_order = Column(Integer, default=0) vote = relationship("Vote", back_populates="options") ballots = relationship("Ballot", back_populates="option") class Voter(Base): __tablename__ = "voter" id = Column(Integer, primary_key=True) token = Column(String(64), unique=True, nullable=False, index=True) ip_address = Column(String(64)) user_agent = Column(String(512)) created_at = Column(DateTime, default=datetime.utcnow) class Ballot(Base): __tablename__ = "ballot" id = Column(Integer, primary_key=True) vote_id = Column(Integer, ForeignKey("vote.id"), nullable=False) option_id = Column(Integer, ForeignKey("option.id"), nullable=False) voter_id = Column(Integer, ForeignKey("voter.id"), nullable=False) created_at = Column(DateTime, default=datetime.utcnow, index=True) vote = relationship("Vote", back_populates="ballots") option = relationship("Option", back_populates="ballots") voter = relationship("Voter") # 同一场投票下,同一用户只允许投一次 __table_args__ = ( UniqueConstraint("vote_id", "voter_id", name="uq_vote_voter"), Index("ix_ballot_vote_created", "vote_id", "created_at"), )

这段模型的关键点有三个。第一,UniqueConstraint("vote_id", "voter_id")是防重复投票的数据库级兜底,即使应用层漏判,数据库也会拒绝第二次插入。第二,Ballot.created_at加了索引,因为后面要做“按小时统计投票趋势”的查询,不带索引在数据量上来之后会全表扫描。第三,OptionVote之间用cascade="all, delete-orphan",这样删除投票主题时会自动清理选项,避免留下孤儿数据。

2.3 用事务保证“投票+计数”的原子性

很多源码项目会在应用代码里先执行option.total_count += 1再插入ballot记录,两步操作之间如果程序崩溃或并发冲突,就会出现计数和明细不一致的脏数据。正确的做法是把这两步放进同一个数据库事务,并且先插入ballot再更新计数,让外键约束先通过。

from sqlalchemy.exc import IntegrityError def cast_vote(db_session, vote_id, option_id, voter_token, ip, ua): # 1. 校验投票是否存在且处于进行中 vote = db_session.get(Vote, vote_id) if not vote or not vote.is_active: raise ValueError("投票不存在或已关闭") # 2. 通过token找到或创建voter voter = db_session.query(Voter).filter_by(token=voter_token).first() if not voter: voter = Voter(token=voter_token, ip_address=ip, user_agent=ua) db_session.add(voter) db_session.flush() # 提前分配voter.id # 3. 插入ballot,依赖唯一约束防重 ballot = Ballot(vote_id=vote.id, option_id=option_id, voter_id=voter.id) db_session.add(ballot) # 4. 更新计数缓存,这一步和ballot在同一事务 option = db_session.get(Option, option_id) option.total_count = (option.total_count or 0) + 1 try: db_session.commit() except IntegrityError: db_session.rollback() raise ValueError("已参与过本场投票") from None return True

这里db_session.flush()的目的不是提交事务,而是让SQLAlchemy把Voter的主键id写回到对象中,这样接下来构造Ballot时才能拿到完整的voter_id。IntegrityError捕获的是数据库唯一约束冲突,这是防重复投票的最后一道闸门。注意不要把业务校验写在commit之后,因为commit一旦执行,事务就结束了。

3. 投票业务逻辑与防刷策略:把规则前置到服务层

3.1 时间窗口校验与前端倒计时的“时间源”问题

投票活动通常有开始和结束时间,最简单的做法是每次投票时在服务端比较当前时间与配置的起止时间。但前端页面如果直接读取服务器返回的时间戳并做本地倒计时,会存在两个问题:用户修改本机时钟可以提前看到“开始投票”按钮;更严重的是,如果前端倒计时与服务器时间不同步,用户看到的窗口和真正可投票的窗口有偏差。

正确的做法是后端在渲染页面时把server_time一并传给前端,所有倒计时计算以这个时间戳为基准,并在提交投票时再次校验。不要把信任建立在客户端。

from flask import jsonify, request @app.route("/api/vote/<int:vote_id>/status") def vote_status(vote_id): vote = db_session.get(Vote, vote_id) now = datetime.utcnow() remaining = None if vote.end_time: remaining = max(0, int((vote.end_time - now).total_seconds())) return jsonify({ "server_time": now.isoformat(), "start_time": vote.start_time.isoformat(), "end_time": vote.end_time.isoformat(), "remaining_seconds": remaining, "is_active": vote.is_active and ( vote.start_time <= now and (vote.end_time is None or vote.end_time > now) ) })

返回remaining_seconds而不是直接返回“是/否开放”,是为了让前端用同一个时间基准做倒计时。每次调用投票接口时,服务端重新算一次时间窗口,不信任前端传来的任何状态。这样即便用户直接伪造POST请求,也不可能在非开放时间投票成功。

3.2 服务端幂等性:重复提交与“再想想”按钮

在真实投票场景中,用户经常点击提交后发现选错,想要取消重投。很多源码项目不允许撤销,理由是“防止刷票”,但这样体验很差。折中方案是允许在投票截止前撤销,但每次撤销和重投都要做完整审计。

3.2.1 撤销投票的事务处理

撤销操作的逻辑是:删除ballot记录,同时将对应option的total_count回退。这里必须放在同一个事务里,否则可能出现票数已回退但记录未删除的情况。

@app.route("/api/vote/<int:vote_id>/revoke", methods=["POST"]) def revote(vote_id, voter_token): voter = db_session.query(Voter).filter_by(token=voter_token).first() if not voter: return jsonify({"code": 404, "msg": "未找到投票记录"}), 404 ballot = (db_session.query(Ballot) .filter_by(vote_id=vote_id, voter_id=voter.id) .first()) if not ballot: return jsonify({"code": 404, "msg": "本场尚未投票"}), 404 option = db_session.get(Option, ballot.option_id) option.total_count = max(0, (option.total_count or 0) - 1) db_session.delete(ballot) db_session.commit() return jsonify({"code": 0, "msg": "已撤销"})

注意max(0, ...)防止并发情况下计数被扣成负数。虽然加了唯一约束后同一用户同一vote只有一条ballot,但撤销时总票数可能因为其他事务尚未提交而出现短暂不一致,加这一层保护成本极低。

3.2.2 防刷的三层结构:Cookie Token + IP频率 + 行为特征

单靠数据库唯一约束防不住攻击者写脚本遍历不同token批量投票。三层防护是常见做法:

第一层是签名Cookie。用户首次访问时种一个voter_token,这个token由服务端用HMAC密钥签名,内容包含随机数和签发时间。攻击者如果不知道密钥,就无法伪造另一个用户的token,但可以清Cookie重新拿一个新token,所以这一层只是“提升攻击成本”。

第二层是IP频率限制。同一IP在一小时内最多投10次,这个阈值根据实际活动调整。使用Redis计数器是标准方案,没有Redis时也可以用SQLite一张频率表代替,不过生产环境建议用Redis。

第三层是行为特征。投票间隔小于2秒、用户代理为空、或者提交数据中除了投票字段外还带额外隐藏参数,都视为可疑。更严谨的做法是给投票表单加一个HMAC签名的payload,把vote_id + voter_ip + timestamp加密后作为隐藏字段,提交时验证签名,可以有效拦截直接改包的重放。

3.3 多选投票的max_choices校验边界

当允许用户一次投多个选项时,最容易出问题的不是“选了超过max_choices个”,而是用户提交同一个option_id两次。比如界面限制最多选3项,攻击者直接构造JSON数组[1, 1, 2],如果服务端只判断长度≤3,就会把option 1的票数多加一次。

正确的校验方式是先去掉重复项再判断个数:

def validate_choices(option_ids, max_choices): unique_ids = set(option_ids) if len(unique_ids) > max_choices: raise ValueError(f"最多选择{max_choices}项") if len(unique_ids) != len(option_ids): raise ValueError("存在重复选项") return unique_ids

这一步放在业务层,数据库层也要相应加上UniqueConstraint("vote_id", "option_id", "voter_id")防同样的选项同一人被重复记录。注意多选投票的ballot表唯一约束不能只设vote_id + voter_id,因为那样就变成只能投一个选项了,要升级成三列联合唯一。

4. 前端交互与实时结果:WebSocket推送和异步渲染

4.1 用Fetch API提交投票并展示即时结果

后端只提供JSON接口,前端用原生Fetch提交,页面不做整页刷新。这里的关键点是加载动画与错误提示要处理到位,不能用 alert() 打断用户操作。

<form id="vote-form"> <input type="radio" name="option" value="1"> 前端 <input type="radio" name="option" value="2"> 后端 <input type="radio" name="option" value="3"> 全栈 <button type="submit">提交投票</button> </form> <div id="result-bar"></div>
document.getElementById('vote-form').addEventListener('submit', async (e) => { e.preventDefault(); const selected = document.querySelector('input[name="option"]:checked'); if (!selected) return; const resp = await fetch('/api/vote/1/cast', { method: 'POST', headers: {'Content-Type': 'application/json'}, body: JSON.stringify({ option_id: parseInt(selected.value) }) }); const data = await resp.json(); if (data.code === 0) { renderResult(data.statistics); } else { document.getElementById('error-msg').textContent = data.msg; } }); function renderResult(stats) { // stats: [{option_id, label, count, percent}] const bar = document.getElementById('result-bar'); bar.innerHTML = stats.map(s => ` <div class="vote-row"> <span>${s.label}</span> <div class="progress"> <div class="progress-fill" style="width:${s.percent}%"></div> </div> <span>${s.count}票 (${s.percent}%)</span> </div> `).join(''); }

Fetch提交要比XMLHttpRequest这类旧方案更简洁,并且支持async/await,维护起来更顺手。接口返回的统计结果最好一次性返回所有选项的计数和百分比,避免前端一次刷新多次请求。

4.2 短轮询 vs WebSocket的选择

如果只是展示“当前票数”,短轮询就够用,每10秒请求一次/api/vote/1/result。这种方式实现简单,和后端语言无关,但缺点是没有实时性且浪费请求。对于投票这种低频更新场景,10秒刷新几乎无感。

如果预期用户数量大、需要实时看到票数增长动画,用WebSocket更合适。Flask-SocketIO可以把Python后端变成WebSocket服务端,但目前主流部署方式是用gunicorn + eventletgevent运行。

from flask_socketio import SocketIO, emit socketio = SocketIO(app, cors_allowed_origins="*", async_mode="eventlet") @socketio.on("join_vote_room") def on_join(data): vote_id = data["vote_id"] join_room(f"vote_{vote_id}") emit("joined", {"vote_id": vote_id}, room=f"vote_{vote_id}") @socketio.on("cast_ballot") def on_cast(data): # 调用和HTTP接口相同的服务层函数 result = cast_vote(data) socketio.emit("vote_updated", result.statistics, room=f"vote_{vote_id}")

这里要注意,WebSocket事件函数里不能再依赖Flask的request上下文,需要把tokenvote_id等参数显式通过事件数据传进来。socket.io的room机制允许把不同投票主题隔离在不同的频道里,避免用户收到不相关投票的刷新事件。

4.3 防止用户重复投票的前端禁用逻辑

即使服务端有完整校验,前端也应当在用户投票成功后立刻禁用提交按钮,减少无效请求。同时要确保这种“禁用”是临时性的,在页面刷新后由服务端告知是否可投票,而不是硬编码在前端状态里。

if (data.code === 0) { submitBtn.disabled = true; submitBtn.textContent = '已投票'; localStorage.setItem(`voted_${vote_id}`, '1'); } else if (data.code === 4001) { // 重复投票专用错误码 submitBtn.disabled = true; submitBtn.textContent = '您已参与过'; }

错误码区分业务逻辑很重要。4001代表“重复投票”,4002代表“投票已关闭”,4003代表“参数不合法”。前端可以根据错误码执行不同的UI反馈,而不是统一渲染一个“失败”字符串。

5. 部署与安全加固:从localhost到公网服务器

5.1 用Waitress或Gunicorn替换开发服务器

Flask自带的开发服务器会在控制台打印Debug信息,并且并发能力极弱,只适合本地调试。部署到公网时要换用生产级WSGI服务器。Windows环境推荐waitress,Linux环境用gunicorn

# Windows pip install waitress waitress-serve --host 0.0.0.0 --port 5000 app:app # Linux pip install gunicorn gunicorn -w 4 -b 0.0.0.0:5000 -k eventlet app:app

-w 4指4个worker进程,每个worker独享Python解释器。CPU核数为N时,worker数建议设为2N+1,但需要注意如果使用了SQLite,多worker并发写会触发database is locked错误,所以生产环境还是建议切换到MySQL。-k eventlet配合Flask-SocketIO时是必须的,否则WebSocket无法工作。

5.2 Nginx反向代理与HTTPS强制跳转

用Nginx处理静态文件和TLS证书,把动态请求反向代理给后端的Waitress/Gunicorn。配置要点是设置正确的X-Forwarded-Proto头,否则Flask无法区分请求是HTTP还是HTTPS。

server { listen 443 ssl; server_name vote.example.com; ssl_certificate /etc/letsencrypt/live/vote.example.com/fullchain.pem; ssl_certificate_key /etc/letsencrypt/live/vote.example.com/privkey.pem; location / { proxy_pass http://127.0.0.1:5000; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_set_header X-Forwarded-Proto $scheme; } }

在Flask侧,要开启ProxyFix中间件让request.remote_addr读取真实IP,否则所有来源IP都会变成Nginx地址,导致IP限流失效。

from werkzeug.middleware.proxy_fix import ProxyFix app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1, x_proto=1, x_host=1)

5.3 结果防篡改:哈希链与定时快照

自助部署的投票系统最容易受质疑的点是“管理员能不能偷偷改票数”。如果这是正式选举或具有纪念意义的投票,建议在每次投票落库后,把(vote_id, option_id, voter_token, created_at)拼接成一条字符串,再用SHA-256迭代计算哈希值,写入一个只追加的审计文件。这个文件不放在数据库里,而是用独立目录存储,管理员甚至无法直接修改。

认真做的话可以设计成类似区块链的哈希链:每条记录包含前一条记录哈希,这样篡改任何历史记录都会导致后续哈希全部失效。

import hashlib import os AUDIT_FILE = "/var/vote_audit/audit.log" def append_audit(record: str, prev_hash: str) -> str: payload = f"{prev_hash}|{record}" cur_hash = hashlib.sha256(payload.encode()).hexdigest() with open(AUDIT_FILE, "a", encoding="utf-8") as f: f.write(f"{cur_hash} {record}\n") return cur_hash

每次投票、撤销、管理员导出的操作都追加一条记录。安全强度不需要达到加密级别,哈希链的意义在于“事后审计时可验证完整性”,这比单一哈希能暴露篡改位置的能力强很多。

6. 上线前必做的5个检查项

投票系统上线前,按以下顺序逐个检查,可以避开大部分翻车现场。

6.1 检查一:时间边界测试

创建一场“开始时间距今2分钟、结束时间距今5分钟”的测试投票,在开始前1秒调用投票接口应返回“未开始”,结束后1秒调用应返回“已关闭”。重点检查时区问题——代码里所有时间用datetime.utcnow(),前端展示时再转换时区,不要混用本地时间和UTC时间。

6.2 检查二:并发投票压测

ablocust模拟50个并发同时投同一个选项,观察是否有IntegrityError或计数异常。如果使用SQLite,这一步非常容易触发database is locked,不要惊慌,这是预期行为——如果业务预期并发超过30,直接换MySQL再压测。

ab -n 100 -c 20 -p payload.json -T application/json \ http://127.0.0.1:5000/api/vote/1/cast

6.3 检查三:数据库备份恢复演练

每5分钟自动备份SQLite文件到异地目录,备份命令很简单:cp vote.db vote_$(date +\%Y\%m\%d_\%H\%M).db。但更要紧的是验证备份能正常恢复——在备机上执行sqlite3 backup.db ".integrity_check",确认返回ok。不要假设备份文件一定有效。

6.4 检查四:管理员接口的权限控制

后台管理接口不能只靠隐藏URL实现安全。所有修改投票状态、删除选项、导出结果的接口必须校验管理员Session,同时为每个操作记录操作日志。最简单的方案是用Flask-Login加@admin_required装饰器,同时开启CSRF Protection。

6.5 检查五:结果导出的可读性

用户和管理员查看结果时,如果票数是0,除数为0会让前端显示NaN。后端返回统计数据时对percent字段做防御,保留一位小数并限制范围为0~100。

percent = round(count / total * 100, 1) if total else 0.0 percent = max(0.0, min(percent, 100.0))

最后一个检查项看似不起眼,但我见过不止一次因为空结果集导致后端JSON里出现NaN字符串,前端解析时直接报错。投票系统上线前跑一遍这五步,至少能过滤掉九成的低级问题。

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

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询