当前位置：首页 > news >正文

Python实战：基于控制台与MySQL的电影票预订系统开发指南

news 2025/5/6 6:10:40

前言

在当今数字化时代，电影票预订系统已经成为影院管理的核心工具。本文将通过Python与MySQL的结合，带你从零开始开发一个功能完整的控制台版电影票预订系统。这个项目不仅涵盖了Python基础语法和MySQL数据库操作，还涉及面向对象编程、异常处理等核心概念，是提升Python实战能力的绝佳练习。

一、系统设计与功能规划

1.1 系统架构设计

我们的电影票预订系统将采用三层架构：

表示层：控制台界面（后期可扩展为Web或GUI）
业务逻辑层：处理核心业务规则
数据访问层：负责与MySQL数据库交互

1.2 核心功能模块

1. 用户管理- 注册/登录/注销- 个人信息管理2. 电影管理- 电影信息查询- 电影排期查看3. 票务管理- 选座购票- 订单查询- 退票改签4. 管理员功能- 电影信息维护- 场次安排- 销售统计

二、数据库设计与实现

2.1 MySQL表结构设计

用户表(users)：

CREATE TABLE users (user_id INT AUTO_INCREMENT PRIMARY KEY,username VARCHAR(50) NOT NULL UNIQUE,password VARCHAR(100) NOT NULL,real_name VARCHAR(50),phone VARCHAR(20),email VARCHAR(100),is_admin BOOLEAN DEFAULT FALSE,created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

电影表(movies)：

CREATE TABLE movies (movie_id INT AUTO_INCREMENT PRIMARY KEY,title VARCHAR(100) NOT NULL,director VARCHAR(50),actors VARCHAR(200),genre VARCHAR(50),duration INT COMMENT '分钟',release_date DATE,description TEXT,poster_url VARCHAR(255)
);

放映场次表(screenings)：

CREATE TABLE screenings (screening_id INT AUTO_INCREMENT PRIMARY KEY,movie_id INT NOT NULL,theater_id INT NOT NULL,start_time DATETIME NOT NULL,end_time DATETIME NOT NULL,price DECIMAL(10,2) NOT NULL,available_seats INT NOT NULL,FOREIGN KEY (movie_id) REFERENCES movies(movie_id),FOREIGN KEY (theater_id) REFERENCES theaters(theater_id)
);

订单表(orders)：

CREATE TABLE orders (order_id INT AUTO_INCREMENT PRIMARY KEY,user_id INT NOT NULL,screening_id INT NOT NULL,seat_count INT NOT NULL,total_amount DECIMAL(10,2) NOT NULL,status ENUM('pending', 'paid', 'cancelled') DEFAULT 'pending',created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,FOREIGN KEY (user_id) REFERENCES users(user_id),FOREIGN KEY (screening_id) REFERENCES screenings(screening_id)
);

2.2 数据库连接工具类

import mysql.connector
from mysql.connector import Errorclass Database:def __init__(self, host='localhost', database='movie_booking', user='root', password='password'):self.host = hostself.database = databaseself.user = userself.password = passwordself.connection = Nonedef connect(self):try:self.connection = mysql.connector.connect(host=self.host,database=self.database,user=self.user,password=self.password)return Trueexcept Error as e:print(f"数据库连接失败: {e}")return Falsedef disconnect(self):if self.connection and self.connection.is_connected():self.connection.close()def execute_query(self, query, params=None, fetch=False):cursor = Nonetry:cursor = self.connection.cursor(dictionary=True)cursor.execute(query, params or ())if fetch:return cursor.fetchall()else:self.connection.commit()return cursor.rowcountexcept Error as e:print(f"数据库操作失败: {e}")return Nonefinally:if cursor: cursor.close()

三、核心功能实现

3.1 用户认证模块

import hashlibclass UserService:def __init__(self, db):self.db = dbdef register(self, username, password, **kwargs):# 检查用户名是否存在if self.get_user_by_username(username):return False, "用户名已存在"# 密码加密hashed_pwd = self._hash_password(password)# 插入用户数据query = """INSERT INTO users (username, password, real_name, phone, email) VALUES (%s, %s, %s, %s, %s)"""params = (username, hashed_pwd, kwargs.get('real_name'), kwargs.get('phone'), kwargs.get('email'))if self.db.execute_query(query, params):return True, "注册成功"return False, "注册失败"def login(self, username, password):user = self.get_user_by_username(username)if not user:return None, "用户不存在"if self._verify_password(password, user['password']):return user, "登录成功"return None, "密码错误"def get_user_by_username(self, username):query = "SELECT * FROM users WHERE username = %s"result = self.db.execute_query(query, (username,), fetch=True)return result[0] if result else Nonedef _hash_password(self, password):return hashlib.sha256(password.encode()).hexdigest()def _verify_password(self, input_pwd, hashed_pwd):return self._hash_password(input_pwd) == hashed_pwd

3.2 电影管理模块

class MovieService:def __init__(self, db):self.db = dbdef add_movie(self, title, director, actors, genre, duration, release_date, description, poster_url):query = """INSERT INTO movies (title, director, actors, genre, duration, release_date, description, poster_url)VALUES (%s, %s, %s, %s, %s, %s, %s, %s)"""params = (title, director, actors, genre, duration, release_date, description, poster_url)return self.db.execute_query(query, params)def get_all_movies(self):query = "SELECT * FROM movies ORDER BY release_date DESC"return self.db.execute_query(query, fetch=True)def get_movie_by_id(self, movie_id):query = "SELECT * FROM movies WHERE movie_id = %s"result = self.db.execute_query(query, (movie_id,), fetch=True)return result[0] if result else Nonedef search_movies(self, keyword):query = """SELECT * FROM movies WHERE title LIKE %s OR director LIKE %s OR actors LIKE %s"""params = (f"%{keyword}%", f"%{keyword}%", f"%{keyword}%")return self.db.execute_query(query, params, fetch=True)

3.3 票务管理模块

from datetime import datetimeclass BookingService:def __init__(self, db):self.db = dbdef get_available_screenings(self, movie_id=None):query = """SELECT s.*, m.title, m.duration, t.name as theater_nameFROM screenings sJOIN movies m ON s.movie_id = m.movie_idJOIN theaters t ON s.theater_id = t.theater_idWHERE s.start_time > %s AND s.available_seats > 0"""params = [datetime.now()]if movie_id:query += " AND s.movie_id = %s"params.append(movie_id)query += " ORDER BY s.start_time"return self.db.execute_query(query, params, fetch=True)def book_tickets(self, user_id, screening_id, seat_count):# 检查场次和座位screening = self.get_screening_by_id(screening_id)if not screening or screening['available_seats'] < seat_count:return False, "座位不足"# 计算总价total_amount = screening['price'] * seat_count# 创建订单order_query = """INSERT INTO orders (user_id, screening_id, seat_count, total_amount)VALUES (%s, %s, %s, %s)"""order_params = (user_id, screening_id, seat_count, total_amount)# 更新场次座位screening_query = """UPDATE screenings SET available_seats = available_seats - %sWHERE screening_id = %s"""screening_params = (seat_count, screening_id)# 事务处理try:self.db.connection.start_transaction()self.db.execute_query(order_query, order_params)self.db.execute_query(screening_query, screening_params)self.db.connection.commit()return True, "订票成功"except Exception as e:self.db.connection.rollback()return False, f"订票失败: {e}"def get_user_orders(self, user_id):query = """SELECT o.*, m.title, s.start_time, t.name as theater_nameFROM orders oJOIN screenings s ON o.screening_id = s.screening_idJOIN movies m ON s.movie_id = m.movie_idJOIN theaters t ON s.theater_id = t.theater_idWHERE o.user_id = %sORDER BY o.created_at DESC"""return self.db.execute_query(query, (user_id,), fetch=True)def cancel_order(self, order_id):# 获取订单信息order_query = """SELECT o.*, s.screening_id, o.seat_countFROM orders oJOIN screenings s ON o.screening_id = s.screening_idWHERE o.order_id = %s AND o.status = 'paid'"""order = self.db.execute_query(order_query, (order_id,), fetch=True)if not order:return False, "订单不存在或不可取消"order = order[0]# 事务处理try:self.db.connection.start_transaction()# 更新订单状态update_order = """UPDATE orders SET status = 'cancelled' WHERE order_id = %s"""self.db.execute_query(update_order, (order_id,))# 恢复座位update_screening = """UPDATE screenings SET available_seats = available_seats + %sWHERE screening_id = %s"""self.db.execute_query(update_screening, (order['seat_count'], order['screening_id']))self.db.connection.commit()return True, "退票成功"except Exception as e:self.db.connection.rollback()return False, f"退票失败: {e}"

四、控制台界面实现

4.1 主菜单与导航

class ConsoleUI:def __init__(self):self.db = Database()if not self.db.connect():raise Exception("无法连接数据库")self.user_service = UserService(self.db)self.movie_service = MovieService(self.db)self.booking_service = BookingService(self.db)self.current_user = Nonedef run(self):while True:if not self.current_user:self.show_main_menu()else:if self.current_user['is_admin']:self.show_admin_menu()else:self.show_user_menu()def show_main_menu(self):print("\n=== 电影票预订系统 ===")print("1. 登录")print("2. 注册")print("3. 查看电影")print("0. 退出")choice = input("请选择: ")if choice == "1":self.login()elif choice == "2":self.register()elif choice == "3":self.show_movies()elif choice == "0":exit()else:print("无效选择")def show_user_menu(self):print(f"\n=== 欢迎, {self.current_user['username']} ===")print("1. 查看电影")print("2. 查询场次")print("3. 我的订单")print("4. 个人信息")print("0. 注销")choice = input("请选择: ")if choice == "1":self.show_movies()elif choice == "2":self.show_screenings()elif choice == "3":self.show_user_orders()elif choice == "4":self.show_user_profile()elif choice == "0":self.current_user = Noneelse:print("无效选择")def show_admin_menu(self):print(f"\n=== 管理员面板 ===")print("1. 电影管理")print("2. 场次管理")print("3. 销售统计")print("0. 注销")choice = input("请选择: ")if choice == "1":self.manage_movies()elif choice == "2":self.manage_screenings()elif choice == "3":self.view_sales_report()elif choice == "0":self.current_user = Noneelse:print("无效选择")

4.2 用户交互功能实现

    def login(self):print("\n=== 用户登录 ===")username = input("用户名: ")password = input("密码: ")user, message = self.user_service.login(username, password)if user:self.current_user = userprint(f"登录成功，欢迎 {user['username']}!")else:print(f"登录失败: {message}")def register(self):print("\n=== 用户注册 ===")username = input("用户名: ")password = input("密码: ")real_name = input("真实姓名(可选): ")phone = input("电话(可选): ")email = input("邮箱(可选): ")success, message = self.user_service.register(username, password, real_name=real_name, phone=phone, email=email)print(message)def show_movies(self):movies = self.movie_service.get_all_movies()print("\n=== 电影列表 ===")for idx, movie in enumerate(movies, 1):print(f"{idx}. {movie['title']} ({movie['release_date'].year})")print(f"   导演: {movie['director']} 类型: {movie['genre']}")print(f"   时长: {movie['duration']}分钟")choice = input("\n输入电影编号查看详情(0返回): ")if choice.isdigit() and int(choice) > 0:movie = movies[int(choice)-1]self.show_movie_detail(movie['movie_id'])def show_movie_detail(self, movie_id):movie = self.movie_service.get_movie_by_id(movie_id)if not movie:print("电影不存在")returnprint(f"\n=== {movie['title']} ===")print(f"导演: {movie['director']}")print(f"主演: {movie['actors']}")print(f"类型: {movie['genre']} 时长: {movie['duration']}分钟")print(f"上映日期: {movie['release_date']}")print("\n剧情简介:")print(movie['description'])screenings = self.booking_service.get_available_screenings(movie_id)if screenings:print("\n=== 放映场次 ===")for idx, s in enumerate(screenings, 1):print(f"{idx}. {s['start_time']} {s['theater_name']} "f"¥{s['price']} 剩余座位: {s['available_seats']}")if self.current_user:choice = input("\n输入场次编号订票(0返回): ")if choice.isdigit() and int(choice) > 0:self.book_ticket(screenings[int(choice)-1]['screening_id'])def book_ticket(self, screening_id):screening = self.booking_service.get_screening_by_id(screening_id)if not screening:print("场次不存在")returnprint(f"\n=== 订票: {screening['title']} ===")print(f"时间: {screening['start_time']}")print(f"影院: {screening['theater_name']}")print(f"价格: ¥{screening['price']}/张 剩余座位: {screening['available_seats']}")while True:seat_count = input("请输入购票数量: ")if seat_count.isdigit() and 0 < int(seat_count) <= screening['available_seats']:breakprint("输入无效或座位不足")success, message = self.booking_service.book_tickets(self.current_user['user_id'], screening_id, int(seat_count))print(message)