2026.11.06 AI学习

AI编程（六十）：完整的数据管理系统 — 会记、会查、会汇总的”账本大脑”

作者：climbing

阶段5的毕业作品来了。这一篇不学新概念——第49到59篇的手艺你已经全学完了：建表（第50篇）、增删改查（第51~52篇）、排序分页（第53篇）、多表关联（第54篇）、Python 接管数据库（第55篇）、网页操作数据库（第56篇）、备份（第57篇）、CSV 搬家（第58篇）、索引提速（第59篇）。今天干一件更大的事：把散装手艺攒成一台会转的机器。还记得第22篇那个简易记账本吗？今天给它换一颗”数据库心脏”：能记账、能改账、能按分类看汇总、网页上点一点就能用，老账本还能整箱搬进新家。做完这篇，”数据管理”这件事你就正式出师了。

一、真实场景：记账本的”中年危机”

第22篇的记账本，用到现在应该积了不少账了。当时它还是个 CSV 文件，几百条账目用着挺顺，可账目一上千，它就开始犯难：

  • 想查”10月买菜花了多少”？得把整个文件读进来，一行行筛；
  • 发现上周有笔账记错了？得把整份文件重写一遍；
  • 想看看每月总支出的走势？二十行循环起步；
  • 想在手机上随手记一笔？CSV 文件表示做不到。

这不是 CSV 的错——它只是个表格文件，本来就不该干这些活。数据一多、要求一高，就该搬进数据库了。 这正是第49篇开头讲过的那句话，今天轮到亲手兑现它。

不过今天不只是”搬个家”，而是把这一个阶段学的所有数据手艺整合起来，做一个完整的数据管理系统——一个”账本大脑”：数据有结构、查账有目录、改账不动全家、坏了有备份、网页上就能用。

写程序的高手和新手，差距往往不在”会多少招”，而在”能不能把招连成套路”。今天练的就是连招。

二、项目蓝图：先画图纸，再动工

老规矩，动手前先画图纸（这个习惯从第48篇就开始养了）。一个完整的数据管理系统 = 表结构 + 功能清单 + 文件分工。

表结构：两张表，一次关联

账目和分类是”多对一”的关系：很多笔账属于同一个分类。所以拆两张表（第54篇的思路）：

  • categories 分类表：id、name（买菜、交通、房租……）——分类名只存这一份；
  • entries 账目表：id、date、category_id（指向分类表）、amount、note——每笔账用一个 id 指向自己的分类。

为什么非要拆？把分类名直接写进每行账目当然也能跑，但哪天想把”买菜”改叫”食品”，一千行账目就得改一千次；拆开之后分类只存一份，改名只动一行，账目用 id 指过去就行。

功能清单：这就是今天的施工范围

功能 用到的手艺 出处

|——|———–|——|

建库建表 + 索引 CREATE TABLE / CREATE INDEX 第50、59篇
记一笔 / 改一笔 / 删一笔 INSERT / UPDATE / DELETE 第52篇
查账单（倒序 + 分页） SELECT / ORDER BY / LIMIT 第51、53篇
按分类汇总 JOIN + Python 累加 第54、55篇
老 CSV 账本搬家 csv 读取 + 批量插入 第25、58篇
网页记账 Flask + SQLite 第56篇
备份 + 校验 文件复制 + 行数对账 第57篇

文件分工：一个项目不止一个文件

  • setup_db.py —— 地基：建库、建表、预置分类；
  • ledger.py —— 手脚：记账、查账、改账、删账；
  • migrate.py —— 搬家：把老 CSV 账本导进来；
  • backup.py —— 后手：备份加校验；
  • app.py —— 门面：网页界面。

每个文件只管一摊，出问题好定位，改起来不互相踩——这比把几百行代码塞一个文件里，是”正经软件”和”随手脚本”的分水岭。

三、动手：一步步把账本大脑搭起来

第 1 步：建库建表（setup_db.py）

# setup_db.py — 账本大脑的"地基"：建库、建表、预置分类
import sqlite3

DB = "ledger.db"

def init_db():
    conn = sqlite3.connect(DB)
    cur = conn.cursor()

    # 分类表：账目的"户口本"，分类名只存这一份
    cur.execute("""
        CREATE TABLE IF NOT EXISTS categories (
            id   INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL UNIQUE
        )
    """)

    # 账目表：每笔账用 category_id 指向分类表（多对一）
    cur.execute("""
        CREATE TABLE IF NOT EXISTS entries (
            id          INTEGER PRIMARY KEY AUTOINCREMENT,
            date        TEXT NOT NULL,
            category_id INTEGER NOT NULL,
            amount      REAL NOT NULL,
            note        TEXT DEFAULT '',
            FOREIGN KEY (category_id) REFERENCES categories(id)
        )
    """)

    # 索引：查账最常按日期查，给它开本"目录"（第59篇手艺）
    cur.execute("""
        CREATE INDEX IF NOT EXISTS idx_entries_date ON entries(date)
    """)

    # 预置几个常用分类（UNIQUE + IGNORE：跑几遍都不会重复）
    for name in ["买菜", "交通", "房租", "娱乐", "医疗", "其他"]:
        cur.execute("INSERT OR IGNORE INTO categories (name) VALUES (?)", (name,))

    conn.commit()
    conn.close()
    print("账本大脑就绪：ledger.db 已建好")

if __name__ == "__main__":
    init_db()

第 2 步：核心增删改查（ledger.py）

# ledger.py — 账本大脑的"手脚"：记一笔、查账、改账、删账
import sqlite3

DB = "ledger.db"

def get_conn():
    conn = sqlite3.connect(DB)
    conn.execute("PRAGMA foreign_keys = ON")  # 坑点疫苗1：外键不开等于白建
    return conn

def add_entry(date, category_name, amount, note=""):
    """记一笔：分类没有就顺手建一个"""
    conn = get_conn()
    cur = conn.cursor()
    cur.execute("SELECT id FROM categories WHERE name = ?", (category_name,))
    row = cur.fetchone()
    if row:
        category_id = row[0]
    else:
        cur.execute("INSERT INTO categories (name) VALUES (?)", (category_name,))
        category_id = cur.lastrowid
    cur.execute(
        "INSERT INTO entries (date, category_id, amount, note) VALUES (?, ?, ?, ?)",
        (date, category_id, amount, note),
    )
    conn.commit()
    conn.close()
    print(f"记上了：{date} {category_name} {amount} 元 {note}")

def list_entries(page=1, per_page=10):
    """查账单：倒序 + 分页，分类名用 JOIN 带出来"""
    conn = get_conn()
    cur = conn.cursor()
    offset = (page - 1) * per_page
    cur.execute("""
        SELECT e.id, e.date, c.name, e.amount, e.note
        FROM entries e
        JOIN categories c ON e.category_id = c.id
        ORDER BY e.date DESC, e.id DESC
        LIMIT ? OFFSET ?
    """, (per_page, offset))
    rows = cur.fetchall()
    conn.close()
    for r in rows:
        print(f"[{r[0]}] {r[1]} {r[2]} {r[3]} 元 {r[4]}")
    return rows

def update_entry(entry_id, amount=None, note=None):
    """改一笔：只改传进来的字段"""
    conn = get_conn()
    cur = conn.cursor()
    if amount is not None:
        cur.execute("UPDATE entries SET amount = ? WHERE id = ?", (amount, entry_id))
    if note is not None:
        cur.execute("UPDATE entries SET note = ? WHERE id = ?", (note, entry_id))
    conn.commit()
    conn.close()
    print(f"改好了：第 {entry_id} 号账目")

def delete_entry(entry_id):
    """删一笔"""
    conn = get_conn()
    conn.execute("DELETE FROM entries WHERE id = ?", (entry_id,))
    conn.commit()
    conn.close()
    print(f"删掉了：第 {entry_id} 号账目")

if __name__ == "__main__":
    # 快速自测：记两笔、翻一页、改一笔、删一笔
    add_entry("2026-11-06", "买菜", 35.5, "早市")
    add_entry("2026-11-06", "交通", 2, "公交")
    list_entries()
    update_entry(1, amount=33.5, note="早市（讲价成功）")
    list_entries()
    delete_entry(2)
    list_entries()

第 3 步：老账本搬家（migrate.py）

第58篇的手艺今天正式派上用场：把第22篇攒下的 CSV 账本整箱搬进数据库。

# migrate.py — 老账本搬家：把 CSV 记账本整箱搬进数据库
import csv
import sqlite3

DB = "ledger.db"
CSV_FILE = "记账本.csv"  # 第22篇格式：日期,分类,金额,备注

def migrate():
    conn = sqlite3.connect(DB)
    conn.execute("PRAGMA foreign_keys = ON")
    cur = conn.cursor()

    with open(CSV_FILE, encoding="utf-8") as f:
        reader = csv.DictReader(f)
        count = 0
        for row in reader:
            cur.execute("SELECT id FROM categories WHERE name = ?", (row["分类"],))
            r = cur.fetchone()
            if r:
                category_id = r[0]
            else:
                cur.execute("INSERT INTO categories (name) VALUES (?)", (row["分类"],))
                category_id = cur.lastrowid
            cur.execute(
                "INSERT INTO entries (date, category_id, amount, note) VALUES (?, ?, ?, ?)",
                (row["日期"], category_id, float(row["金额"]), row.get("备注", "")),
            )
            count += 1

    conn.commit()
    conn.close()
    print(f"搬家完成：{count} 笔账目已入库")

if __name__ == "__main__":
    migrate()

如果你第22篇的 CSV 列名跟这里写的不一样（比如叫”时间”而不是”日期”），把你的表头发给 AI：”这是我的 CSV 表头和两行样例，帮我改 migrate.py 里的字段映射”——30 秒的事，别硬改半小时。

第 4 步：网页门面（app.py）

第56篇的手艺：让网页替你操作数据库。手机、电脑打开就能记账。

# app.py — 账本大脑的"门面"：网页上记账、查账、删账
from flask import Flask, request, redirect, render_template_string
import sqlite3

DB = "ledger.db"
app = Flask(__name__)

PAGE = """

账本大脑


  日期 
  分类 
  金额 
  备注 
  


  {% for e in entries %}
  
  {% endfor %}


  
日期分类金额备注

    {{ e[1] }}{{ e[2] }}{{ e[3] }}{{ e[4] }}
    
  



  {% if page > 1 %}上一页{% endif %}
  第 {{ page }} 页
  下一页
   |  按分类看汇总


"""

SUMMARY = """

按分类汇总


  {% for name, total in rows %}
  
  {% endfor %}


  
分类合计
{{ name }}{{ total }} 元


回账本

"""

def query(sql, args=()):
    conn = sqlite3.connect(DB)
    conn.execute("PRAGMA foreign_keys = ON")
    cur = conn.cursor()
    cur.execute(sql, args)
    rows = cur.fetchall()
    conn.close()
    return rows

@app.route("/")
def index():
    page = int(request.args.get("page", 1))
    entries = query("""
        SELECT e.id, e.date, c.name, e.amount, e.note
        FROM entries e JOIN categories c ON e.category_id = c.id
        ORDER BY e.date DESC, e.id DESC LIMIT 10 OFFSET ?
    """, ((page - 1) * 10,))
    return render_template_string(PAGE, entries=entries, page=page)

@app.route("/add", methods=["POST"])
def add():
    date = request.form["date"]
    category = request.form["category"]
    amount = float(request.form["amount"])
    note = request.form.get("note", "")
    conn = sqlite3.connect(DB)
    conn.execute("PRAGMA foreign_keys = ON")
    cur = conn.cursor()
    cur.execute("SELECT id FROM categories WHERE name = ?", (category,))
    row = cur.fetchone()
    if row:
        category_id = row[0]
    else:
        cur.execute("INSERT INTO categories (name) VALUES (?)", (category,))
        category_id = cur.lastrowid
    cur.execute(
        "INSERT INTO entries (date, category_id, amount, note) VALUES (?, ?, ?, ?)",
        (date, category_id, amount, note),
    )
    conn.commit()
    conn.close()
    return redirect("/")

@app.route("/delete/", methods=["POST"])
def delete(entry_id):
    conn = sqlite3.connect(DB)
    conn.execute("DELETE FROM entries WHERE id = ?", (entry_id,))
    conn.commit()
    conn.close()
    return redirect("/")

@app.route("/summary")
def summary():
    # 汇总交给 Python 累加（第08篇的字典手艺），不用新概念
    rows = query("""
        SELECT c.name, e.amount
        FROM entries e JOIN categories c ON e.category_id = c.id
    """)
    totals = {}
    for name, amount in rows:
        totals[name] = totals.get(name, 0) + amount
    summary_rows = sorted(totals.items(), key=lambda kv: kv[1], reverse=True)
    return render_template_string(SUMMARY, rows=summary_rows)

if __name__ == "__main__":
    app.run(debug=True, port=5000)

第 5 步：备份留后手（backup.py）

第57篇的铁律今天再强调一次：备份没验证过 = 没有备份。

# backup.py — 给账本留后手：带日期的副本 + 行数校验
import shutil
import sqlite3
from datetime import datetime

DB = "ledger.db"

def backup():
    stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
    target = f"ledger-{stamp}.db"
    shutil.copy2(DB, target)  # 先复制，再校验

    # 校验：备份文件能打开、行数对得上，才算备份成功
    src = sqlite3.connect(DB)
    dst = sqlite3.connect(target)
    n1 = src.execute("SELECT COUNT(*) FROM entries").fetchone()[0]
    n2 = dst.execute("SELECT COUNT(*) FROM entries").fetchone()[0]
    src.close()
    dst.close()

    if n1 == n2:
        print(f"备份成功：{target}（{n1} 笔账目，行数对账通过）")
    else:
        print("备份可疑！行数不一致，原件千万别删")

if __name__ == "__main__":
    backup()

第 6 步：跑起来看效果

按顺序来，每步都该看到对应的输出：

$ python setup_db.py
账本大脑就绪：ledger.db 已建好

$ python migrate.py
搬家完成：1043 笔账目已入库

$ python ledger.py
记上了：2026-11-06 买菜 35.5 元 早市
记上了：2026-11-06 交通 2 元 公交
[2] 2026-11-06 交通 2 元 公交
[1] 2026-11-06 买菜 35.5 元 早市
改好了：第 1 号账目
[2] 2026-11-06 交通 2 元 公交
[1] 2026-11-06 买菜 33.5 元 早市（讲价成功）
删掉了：第 2 号账目
[1] 2026-11-06 买菜 33.5 元 早市（讲价成功）

$ python backup.py
备份成功：ledger-20261106-120000.db（1044 笔账目，行数对账通过）

$ python app.py
 * Running on http://127.0.0.1:5000

浏览器打开 http://127.0.0.1:5000：记一笔、翻一页、删一笔、点”按分类看汇总”——十一篇手艺在这一刻全部咬合转动。这就是”完整的数据管理系统”：不是新招，是连招。

四、改进：跟AI一起升级

系统能跑了，下一步是”越用越顺手”。还是那套学习法：跟AI说需求 → 读懂它的方案 → 提出改进。比如把这段发给 AI：

“这是一个记账系统的表结构和核心代码（贴 setup_db.py + ledger.py）。我想加两个功能：1）按月汇总，输出’2026-10：买菜 812 元、交通 156 元……’；2）在网页上按关键词搜备注。请先给我改进方案（动哪些函数、加什么查询），确认后再给代码。”

先看方案，再拿代码。 读懂它打算”多查一次”还是”多算一遍”，比直接复制粘贴值钱十倍。方案里它大概率会给两个方向：

改进1：搜索账目。 在 ledger.py 里加一个 search(keyword)，查询就一句 WHERE e.note LIKE ?，参数用 f"%{keyword}%"。注意：中间通配的 LIKE 走不了索引（第59篇坑点疫苗1讲过），账目几万条之前不用管它，知道边界就行。

改进2：按月汇总。 你会看到两种写法：一种是 Python 字典累加（就是 app.py 里那种），一种是一个新句式 GROUP BY——”按分类分组、每组求和”，SQL 一句话干完 Python 十行的活。让 AI 把两种写法并排讲给你听，看懂之后选一种。这就是”改进”的意义：不是加功能，是借着功能把理解再往前推一步。

五、四个必踩的坑（先打疫苗）

  1. 外键不开等于白建。 SQLite 默认不检查外键，FOREIGN KEY 写了也白写——每条连接开头都要 PRAGMA foreign_keys = ON（代码里每处 get_conn() 都加了，不是啰嗦，是 SQLite 的规矩）。不开的后果很隐蔽：分类删了，账目还指着一个不存在的 id，JOIN 时这笔账直接”人间蒸发”。
  1. 金额别拿浮点数较真。 0.1 + 0.2 在计算机里不等于 0.3，是浮点数的老毛病。记账本用 REAL 存”元”，教学和个人记账够用；但你心里要有数：正经的财务系统都用整数存”分”（33.5 元存 3350），彻底绕开误差。哪天这个账本要管真钱，第一个要改的就是它。
  1. SQL 一律用 ? 占位符，不拼字符串。 'INSERT ... VALUES (' + amount + ')' 这种写法，金额里混进一个引号就是一场小灾难（注水攻击叫 SQL 注入）。哪怕工具只有自己用，习惯从第一天养——这些代码以后是要复制进”正经软件”的。
  1. 搬家前先备份，防”搬两次记两遍”。 migrate.py 再跑一次，一千多笔账就变成两千多笔——它只管搬，不管查重。搬家前 python backup.py 走一遍（顺手也练了第57篇），万一重复了，删库重来就是。另外 CSV 里的日期格式（”10/5″还是”2026-10-05″）最好统一，不然按月汇总时它会被当成两个月——搬家清单里的”格式核对”，跟搬家公司是一个道理。

六、跟AI协作的正确姿势（项目特供版）

  1. 先给图纸，再要代码。 把表结构（建表语句）和功能清单贴给 AI：”按这个蓝图施工”。不给图纸，AI 只能凭感觉盖房子，盖出来的多半不是你要的户型。
  1. 分文件逐个要，每要一个跑通一个。 “帮我写整个项目”一口气拿回五文件几百行，出错根本没法定位。正确的节奏是：setup_db.py 跑通了，再要 ledger.py，再要 app.py——每一步都验证，问题永远只藏在最后那一步里。
  1. 让AI列”改进清单”，而不是东改西改。 “这个系统还能怎么改进？按’实用性高低’列五条”——从清单里挑着做。功能加着加着失控，是所有小项目烂尾的第一步；蓝图内的克制，比代码量更像”正经软件”。

七、今日小练习

练习1：给账本加”收入”。 现在只能记支出，让工资也能进账。两种设计摆在你面前：给 entries 表加一列 type（支出/收入），或者约定”金额为负 = 支出”。把两种方案的取舍发给 AI 让它讲透，选一种实现，并让分类汇总能分出”收入合计、支出合计”（提示：改 add_entry 加个参数，summary 里按类型分开累加；想想哪种设计以后加”转账”类型时更省事）。

练习2：做一份月度汇总报告。 输出”2026-10：买菜 812 元、交通 156 元、合计 xxx 元”这样的按月报表。先用 Python 字典累加实现（提示：键用”年月”字符串如 row[1][:7]，字典套字典或元组当键都行）；然后让 AI 教你 GROUP BY 版本，对比两种写法的行数和速度——看得懂两种，才算真会。

练习3：来一次备份恢复演习。 备份 → 故意 DELETE 掉三条账目 → 写几行代码从备份文件把数据”捞”回来（提示：连上备份文件把 entries 里的行读出来再插回主库；捞完 COUNT(*) 对账）。想想第57篇那句话：没验证过恢复的备份，不算备份——今天就验证它。

下篇预告

下一篇（第61篇）：画第一张图 — matplotlib 基础。 阶段6″可视化”正式开张：数据躺在表里，看半天没感觉；画成图，一眼就懂。先画出人生第一张图，再一步步学会折线图、柱状图、饼图、散点图……让数据”开口说话”。你的账本大脑，马上就要配上一块”会画图的屏幕”了。