档案线上培训价格算法实现与代码落地指南
一、系统架构与需求拆解
在开发档案线上培训系统时，价格计算是核心业务逻辑。为了实现灵活的定价策略，我们需要构建一个支持多种维度（课程类型、用户身份、购买数量）的定价引擎。本文将使用 Python Flask 框架配合 SQLite 数据库，手把手带你搭建一个完整的后端服务。
我们将实现以下具体定价规则：
- 基础定价：根据课程类目（如“档案数字化”、“档案保护技术”）设定不同底价。
- 身份折扣：企业用户购买“档案管理基础”课程享受 8.5 折。
- 阶梯优惠：单次购买课程数量超过 10 门，总价立减 200 元。
二、开发环境初始化
我们需要准备标准的 Python 开发环境。为了避免依赖冲突，建议创建一个新的虚拟环境。请直接复制以下命令在终端执行：
```bash 创建项目目录 mkdir archive_training_system cd archive_training_system 创建并激活虚拟环境 python3 -m venv venv source venv/bin/activate Windows系统使用 venv\Scripts\activate 安装核心依赖包 pip install flask flask-sqlalchemy ```安装完成后，创建项目的基础文件结构。我们需要建立 app.py 作为主程序入口，models.py 存放数据库模型，pricing_service.py 存放核心定价算法。
三、数据库模型设计
我们需要设计三张核心表：Courses（课程表）、UserProfiles（用户表）和 Orders（订单表）。打开 models.py，写入以下完整代码：
这段代码定义了数据结构。注意 base_price 字段，它是我们后续计算价格的基准。接下来，我们需要在 app.py 中初始化数据库并预置一些测试数据。
四、核心定价算法实现
这是本指南最关键的部分。我们将创建一个 PricingEngine 类，封装所有价格计算逻辑。这种设计模式便于后续扩展新的规则（如节日促销）。创建 pricing_service.py 并填入以下代码：
```python from models import Course, UserProfile class PricingEngine: def __init__(self, db_session): self.db = db_session def calculate_price(self, user_id, course_ids): """ 计算订单最终价格 :param user_id: 用户ID :param course_ids: 课程ID列表 :return: 原价总金额, 优惠金额, 最终应付金额 """ 1. 获取用户信息 user = self.db.session.query(UserProfile).get(user_id) if not user: raise ValueError("用户不存在") 2. 获取课程信息并计算基础总价 courses = self.db.session.query(Course).filter(Course.id.in_(course_ids)).all() if len(courses) != len(course_ids): raise ValueError("部分课程ID无效") original_total = sum(course.base_price for course in courses) discount_amount = 0.0 current_total = original_total 规则1: 企业用户购买“档案管理基础”类课程享受 8.5 折 假设“档案管理基础”的 category 为 'basic_archives' if user.role == 'enterprise': basic_courses = [c for c in courses if c.category == 'basic_archives'] if basic_courses: basic_total = sum(c.base_price for c in basic_courses) discount_on_basic = basic_total 0.15 15% off discount_amount += discount_on_basic current_total -= discount_on_basic 规则2: 单次购买超过 10 门，总价立减 200 元 if len(course_ids) > 10: discount_amount += 200 current_total -= 200 防止价格为负数 if current_total < 0: current_total = 0 discount_amount = original_total 优惠不能超过原价 return { "original_total": round(original_total, 2), "discount_amount": round(discount_amount, 2), "final_price": round(current_total, 2), "item_count": len(course_ids) } ```请注意代码中的逻辑细节：先计算身份折扣，再计算阶梯优惠。顺序很重要，通常业务规则会指定优惠叠加顺序。这里我们严格遵循了先折扣后立减的逻辑。
五、API接口开发
现在我们需要将算法通过 HTTP 暴露给前端。在 app.py 中编写路由和初始化逻辑。为了确保你复制后能直接运行，我包含了数据库初始化和种子数据插入的代码：
```python from flask import Flask, request, jsonify from models import db, Course, UserProfile from pricing_service import PricingEngine app = Flask(__name__) app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///archive_training.db' app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False db.init_app(app) 初始化数据库和测试数据 with app.app_context(): db.create_all() 仅当表为空时插入数据，防止重复插入 if Course.query.count() == 0: print("初始化测试数据...") 添加课程 c1 = Course(name="档案管理基础入门", category="basic_archives", base_price=200.0) c2 = Course(name="档案数字化高级实务", category="digital_archives", base_price=500.0) c3 = Course(name="档案保护技术", category="protection", base_price=350.0) db.session.add_all([c1, c2, c3]) 添加用户 u1 = UserProfile(username="张三", role="individual") u2 = UserProfile(username="某科技公司", role="enterprise") db.session.add_all([u1, u2]) db.session.commit() pricing_engine = PricingEngine(db) @app.route('/api/calculate', methods=['POST']) def calculate_order_price(): """ 接口入参示例: { "user_id": 1, "course_ids": [1, 2] } """ data = request.get_json() 参数校验 if not data or 'user_id' not in data or 'course_ids' not in data: return jsonify({"error": "缺少必要参数 user_id 或 course_ids"}), 400 try: user_id = data['user_id'] course_ids = data['course_ids'] 调用核心引擎 result = pricing_engine.calculate_price(user_id, course_ids) return jsonify({ "code": 200, "message": "计算成功", "data": result }) except ValueError as e: return jsonify({"error": str(e)}), 404 except Exception as e: return jsonify({"error": f"服务器内部错误: {str(e)}"}), 500 if __name__ == '__main__': app.run(debug=True, port=5000) ```上述代码实现了一个完整的闭环。当服务启动时，它会自动检测数据库是否存在，如果不存在则创建并插入两个用户和三门课程。这解决了“空数据库无法测试”的痛点。
六、接口测试与验证
启动服务后，我们需要验证定价逻辑是否正确。请确保你的终端位于项目目录下，并启动服务：
```bash python app.py ```服务启动后，我们使用 curl 命令进行测试。你也可以使用 Postman，但为了演示方便，这里提供命令行直接复制的方案。
场景一：个人用户购买两门课程
用户 ID 1 是个人用户，购买 ID 1 (200元) 和 ID 2 (500元)。预期原价 700，无折扣，实付 700。
```bash curl -X POST http://127.0.0.1:5000/api/calculate \ -H "Content-Type: application/json" \ -d '{"user_id": 1, "course_ids": [1, 2]}' ```预期返回结果：
```json { "code": 200, "data": { "discount_amount": 0.0, "final_price": 700.0, "item_count": 2, "original_total": 700.0 }, "message": "计算成功" } ```场景二：企业用户购买基础课程（触发身份折扣）
用户 ID 2 是企业用户，购买 ID 1 (200元, 类别为 basic_archives)。预期原价 200，折扣 15% (30元)，实付 170。
```bash curl -X POST http://127.0.0.1:5000/api/calculate \ -H "Content-Type: application/json" \ -d '{"user_id": 2, "course_ids": [1]}' ```预期返回结果：
```json { "code": 200, "data": { "discount_amount": 30.0, "final_price": 170.0, "item_count": 1, "original_total": 200.0 }, "message": "计算成功" } ```场景三：批量购买触发阶梯优惠
为了测试“购买超过10门减200元”的规则，我们需要构造一个包含 11 个课程 ID 的列表。这里我们重复购买 ID 3 (350元) 11次。原价应为 3850，优惠 200，实付 3650。
```bash 构造包含11个3的JSON数组 curl -X POST http://127.0.0.1:5000/api/calculate \ -H "Content-Type: application/json" \ -d '{"user_id": 1, "course_ids": [3,3,3,3,3,3,3,3,3,3,3]}' ```预期返回结果：
```json { "code": 200, "data": { "discount_amount": 200.0, "final_price": 3650.0, "item_count": 11, "original_total": 3850.0 }, "message": "计算成功" } ```七、常见问题排查
在实操过程中，你可能会遇到以下具体问题，请对照排查：
- ImportError: No module named 'flask'：请确认你是否在虚拟环境内安装了依赖，并且运行 python 命令的是同一个虚拟环境下的 python。
- sqlite3.OperationalError: no such table：这通常发生在你手动删除了 db 文件但未重启 app。由于我们的初始化代码在
with app.app_context():内，重启 flask 服务即可自动重建表。 - 返回 404 User Not Found：请检查数据库初始化代码是否执行成功。查看终端启动日志中是否有“初始化测试数据...”的字样。
至此，一个完整的档案线上培训价格计算系统已经落地。该代码结构清晰，逻辑与数据分离，你可以直接在此基础上扩展更复杂的业务规则，例如增加“VIP会员专属价”或“限时秒杀”逻辑，只需在 PricingEngine 类中增加新的判断分支即可。
相关文章
