用Python实现一个电影订票系统！

昨天晚上十一点多吧，我在公司楼下抽烟，手机一震，我们组小李在群里说“东哥你不是老说架构嘛，能不能整一个电影订票系统的demo，我周末想拿去练手”，我当时就…困死了，但烟也没抽完，就顺手在脑子里过了一遍：这玩意儿说白了就是三件事，电影/场次/座位，外加一个“别超卖”的底线，别的都是花活儿。

我先说个很真实的坑哈，订票最容易翻车的不是页面好不好看，是两个用户同时点了同一个座位，你要是直接 UPDATE seats set status='sold' 这种写法，线上一抖就给你卖出去两张票了。你们知道吧，秒杀那味儿就出来了。所以我这次就故意用 SQLite 写个小而硬的版本，CLI 的，跑起来就能订，关键是把“并发下不超卖”这个点做对。

我随手在电脑上敲了个最小可用的结构，别嫌丑哈，语音转文字一样的效果：先建三张表，shows 场次，seats 座位，orders 订单。座位用 UNIQUE(show_id, seat_no)，订单里也加个 UNIQUE(show_id, seat_no)，这样哪怕你代码写烂了，数据库也能最后兜个底。

import sqlite3
from dataclasses import dataclass
from datetime import datetime
import uuid
DB = "cinema.db"
defget_conn():
    conn = sqlite3.connect(DB, timeout=5.0, isolation_level=None)  # autocommit off by explicit BEGIN
    conn.row_factory = sqlite3.Row
return conn
definit_db():
    conn = get_conn()
    cur = conn.cursor()
    cur.executescript("""
    PRAGMA foreign_keys = ON;
    CREATE TABLE IF NOT EXISTS shows (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        movie TEXT NOT NULL,
        hall TEXT NOT NULL,
        start_time TEXT NOT NULL,
        price_cents INTEGER NOT NULL
    );
    CREATE TABLE IF NOT EXISTS seats (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        show_id INTEGER NOT NULL,
        seat_no TEXT NOT NULL,
        status TEXT NOT NULL CHECK(status IN ('free','held','sold')),
        hold_token TEXT,
        hold_expire_at TEXT,
        UNIQUE(show_id, seat_no),
        FOREIGN KEY(show_id) REFERENCES shows(id) ON DELETE CASCADE
    );
    CREATE TABLE IF NOT EXISTS orders (
        id TEXT PRIMARY KEY,
        show_id INTEGER NOT NULL,
        seat_no TEXT NOT NULL,
        user TEXT NOT NULL,
        amount_cents INTEGER NOT NULL,
        status TEXT NOT NULL CHECK(status IN ('created','paid','canceled')),
        created_at TEXT NOT NULL,
        UNIQUE(show_id, seat_no),
        FOREIGN KEY(show_id) REFERENCES shows(id) ON DELETE CASCADE
    );
    """
)
    conn.close()
defseed_demo():
    conn = get_conn()
    cur = conn.cursor()
    cur.execute("SELECT COUNT(*) AS c FROM shows")
if cur.fetchone()["c"] > 0:
        conn.close()
return
    cur.execute(
"INSERT INTO shows(movie, hall, start_time, price_cents) VALUES (?,?,?,?)",
        ("流浪地球之我又加班了", "1号厅", "2026-02-01 19:30", 4500),
    )
    show_id = cur.lastrowid
# 5排 * 8列
for r in range(1, 6):
for c in range(1, 9):
            seat_no = f"{r}排{c}座"
            cur.execute(
"INSERT INTO seats(show_id, seat_no, status) VALUES (?,?, 'free')",
                (show_id, seat_no),
            )
    conn.commit()
    conn.close()
deflist_shows():
    conn = get_conn()
    cur = conn.cursor()
    rows = cur.execute("SELECT * FROM shows ORDER BY id").fetchall()
    conn.close()
return rows
defcleanup_expired_holds(conn):
# 把过期的 held 座位释放掉
    now = datetime.utcnow().isoformat()
    conn.execute("""
        UPDATE seats
        SET status='free', hold_token=NULL, hold_expire_at=NULL
        WHERE status='held' AND hold_expire_at IS NOT NULL AND hold_expire_at < ?
    """, (now,))
defshow_seat_map(show_id: int):
    conn = get_conn()
    cur = conn.cursor()
    rows = cur.execute("""
        SELECT seat_no, status FROM seats
        WHERE show_id=?
        ORDER BY id
    """, (show_id,)).fetchall()
    conn.close()
# 简单打印一下
    grid = {}
for r in rows:
        grid[r["seat_no"]] = r["status"]
for rr in range(1, 6):
        line = []
for cc in range(1, 9):
            sn = f"{rr}排{cc}座"
            st = grid.get(sn, "free")
            mark = "□"if st == "free"else ("△"if st == "held"else"■")
            line.append(mark)
        print(f"{rr}排: " + " ".join(line))
    print("提示：□可选，△有人占座(未支付)，■已售")
defhold_seat(show_id: int, seat_no: str, hold_minutes=5) -> str:
    conn = get_conn()
    cur = conn.cursor()
try:
        cur.execute("BEGIN IMMEDIATE")  # 关键：拿写锁，避免并发下两人同时hold成功
        cleanup_expired_holds(conn)
        row = cur.execute("""
            SELECT status FROM seats
            WHERE show_id=? AND seat_no=?
        """, (show_id, seat_no)).fetchone()
ifnot row:
raise ValueError("座位不存在…你是不是打错了")
if row["status"] != "free":
raise ValueError("这个座位不空了，别犟…换一个")
        token = uuid.uuid4().hex
        expire = (datetime.utcnow().timestamp() + hold_minutes * 60)
        expire_iso = datetime.utcfromtimestamp(expire).isoformat()
        cur.execute("""
            UPDATE seats
            SET status='held', hold_token=?, hold_expire_at=?
            WHERE show_id=? AND seat_no=? AND status='free'
        """, (token, expire_iso, show_id, seat_no))
if cur.rowcount != 1:
raise ValueError("占座失败(可能被人抢先一步)，再试试")
        conn.commit()
return token
except:
        conn.rollback()
raise
finally:
        conn.close()
defcreate_and_pay(show_id: int, seat_no: str, user: str, hold_token: str):
    conn = get_conn()
    cur = conn.cursor()
try:
        cur.execute("BEGIN IMMEDIATE")
        cleanup_expired_holds(conn)
        seat = cur.execute("""
            SELECT status, hold_token, hold_expire_at FROM seats
            WHERE show_id=? AND seat_no=?
        """, (show_id, seat_no)).fetchone()
ifnot seat or seat["status"] != "held":
raise ValueError("这座位没被占住，或者过期了…你点慢了")
if seat["hold_token"] != hold_token:
raise ValueError("token不对，别闹…这不是你占的座")
        show = cur.execute("SELECT price_cents FROM shows WHERE id=?", (show_id,)).fetchone()
ifnot show:
raise ValueError("场次没了…这就很离谱")
        order_id = uuid.uuid4().hex
        now = datetime.utcnow().isoformat()
# 先落订单(可扩展为未支付状态)，这里直接模拟支付成功
        cur.execute("""
            INSERT INTO orders(id, show_id, seat_no, user, amount_cents, status, created_at)
            VALUES (?,?,?,?,?,'paid',?)
        """, (order_id, show_id, seat_no, user, show["price_cents"], now))
# 再把座位变 sold
        cur.execute("""
            UPDATE seats
            SET status='sold', hold_token=NULL, hold_expire_at=NULL
            WHERE show_id=? AND seat_no=? AND hold_token=?
        """, (show_id, seat_no, hold_token))
if cur.rowcount != 1:
raise ValueError("出票失败…一般是并发/数据被改了")
        conn.commit()
return order_id
except sqlite3.IntegrityError:
        conn.rollback()
raise ValueError("这个座位已经被别的订单占了(唯一约束兜底)，别硬顶")
except:
        conn.rollback()
raise
finally:
        conn.close()
defmain():
    init_db()
    seed_demo()
    print("== 场次列表 ==")
    shows = list_shows()
for s in shows:
        print(f"[{s['id']}] {s['movie']} | {s['hall']} | {s['start_time']} | ￥{s['price_cents']/100:.2f}")
    show_id = int(input("选场次id：").strip())
    user = input("你叫啥：").strip() or"匿名"
whileTrue:
        print("\n== 座位图 ==")
        show_seat_map(show_id)
        seat_no = input("输入座位号(比如 3排5座)，或 q 退出：").strip()
if seat_no.lower() == "q":
            print("行，那就不看了…")
break
try:
            token = hold_seat(show_id, seat_no)
            print(f"占座成功，给你5分钟，token={token[:8]}...（别截图发群里啊）")
            ok = input("现在支付？y/n：").strip().lower()
if ok == "y":
                order_id = create_and_pay(show_id, seat_no, user, token)
                print(f"支付成功！订单号={order_id[:10]}... 取票码自己记一下")
else:
                print("你不付那就等过期自动释放…")
except Exception as e:
            print("出事了：", e)
if __name__ == "__main__":
    main()

你看这个 demo，表面上就是个“选场次-看座位-占座-支付”，但里面我刻意塞了两个很“工程味”的小动作：一个是 BEGIN IMMEDIATE，别小看它，SQLite 这一下等于你把“关键区”锁住了，俩人同时点同一座，后来的那个会被卡住或失败；另一个是 UNIQUE(show_id, seat_no)，这属于最后一道防线，万一你写错逻辑，数据库也会用 IntegrityError 把你拦下来，别让你默默超卖。

然后占座不是永远占着，我搞了 hold_expire_at，每次下单/占座前顺手 cleanup_expired_holds，过期就释放。你们后面要升级成 Web 版本的话，其实也是这个套路，只不过清理可以用定时任务、或者每次请求都顺带清一下，反正别让“△”一直占着把厅卡死就行。

有人可能要杠一句“东哥你这支付也太假了”，对对对，就是假，我就留一个口子：orders.status 现在直接写 paid，你要真做支付就分两步：先 created，等支付回调再把订单改 paid，再把座位 sold。注意哈，这里就会出现另一个坑：支付回调重复、网络抖动重复通知，你得做幂等。最土但好用的办法就是用订单号当幂等键，重复回调就直接返回成功，别二次扣款别二次出票，这些你做过一次就懂了…哎我以前也踩过，真不想回忆。

反正这个小系统你跑起来以后，周末就能拿去扩展：比如加会员、加退票（退票就把订单 canceled，座位从 sold 变 free，但要校验开场时间）、加选座规则（情侣座不能拆）、加推荐（这个就别搞太重）。你要我说，练手最值钱的不是页面，是把“状态机”搞顺：座位 free/held/sold，订单 created/paid/canceled，这两条线别打架，别互相写穿了。

行了我这会儿又困了，刚刚小李又@我说他数据库连不上，我估计又是路径写错了还是权限没给…我先去看一眼，等下再说哈。