"""Reproduce chapter 2 using only the Python standard library and in-memory SQLite.""" import json import sqlite3 from collections import Counter, defaultdict from datetime import datetime from decimal import Decimal from pathlib import Path def yuan(cents): return format(Decimal(cents) / 100, '.2f') def create_database(folder, exercise=False, missing_details=False): connection = sqlite3.connect(':memory:') connection.row_factory = sqlite3.Row names = ['seed.sql', 'chapter-02-details.sql'] if exercise: names.append('chapter-02-equal-amount-exercise.sql') for name in names: connection.executescript((folder / name).read_text(encoding='utf-8')) if missing_details: connection.execute( 'INSERT INTO invoices VALUES (?, ?, ?, ?, ?)', (1100, 5, '2025-06-28 09:00:00', 100000, 'ISSUED'), ) connection.execute('PRAGMA query_only = ON') return connection def baseline(connection): # 直接读取原始记录，独立按客户累计发票金额与明细条数。 invoices = connection.execute('SELECT * FROM invoices ORDER BY invoice_id').fetchall() lines = connection.execute('SELECT * FROM invoice_lines ORDER BY line_id').fetchall() amounts = defaultdict(int) detail_counts = Counter() line_amounts = defaultdict(int) line_counts = Counter() for line in lines: line_amounts[line['invoice_id']] += line['line_amount_cents'] line_counts[line['invoice_id']] += 1 for invoice in invoices: if line_amounts[invoice['invoice_id']] != invoice['total_amount_cents']: raise RuntimeError('样例的发票总额与明细金额不一致') date = datetime.fromisoformat(invoice['invoice_date']) if invoice['status'] == 'ISSUED' and (date.year, date.month) == (2025, 6): amounts[invoice['customer_id']] += invoice['total_amount_cents'] detail_counts[invoice['customer_id']] += line_counts[invoice['invoice_id']] return dict(amounts), dict(detail_counts) def query_rows(connection, query): return [dict(row) for row in connection.execute(query).fetchall()] def amount_map(rows): return {row['customer_id']: format(row['sales_amount'], '.2f') for row in rows} def main(): folder = Path(__file__).resolve().parent wrong = (folder / 'chapter-02-wrong-join.sql').read_text(encoding='utf-8') correct = (folder / 'chapter-02-correct-join.sql').read_text(encoding='utf-8') distinct = wrong.replace('SUM(i.total_amount_cents)', 'SUM(DISTINCT i.total_amount_cents)') connection = create_database(folder) try: expected_amounts, expected_counts = baseline(connection) if expected_amounts != {1: 550000, 2: 420000, 3: 300000, 4: 90000}: raise RuntimeError('原始金额基线变化') if expected_counts != {1: 4, 2: 1, 3: 4, 4: 2}: raise RuntimeError('原始明细条数基线变化') wrong_rows = query_rows(connection, wrong) correct_rows = query_rows(connection, correct) if amount_map(wrong_rows) != {1: '11000.00', 3: '6600.00', 2: '4200.00', 4: '1800.00'}: raise RuntimeError('错误关联没有复现预期的重复累计') if [row['customer_id'] for row in wrong_rows] != [1, 3, 2, 4]: raise RuntimeError('错误查询的排名未复现') if amount_map(correct_rows) != {key: yuan(value) for key, value in expected_amounts.items()}: raise RuntimeError('修正 SQL 与原始金额基线不同') if {row['customer_id']: row['detail_line_count'] for row in correct_rows} != expected_counts: raise RuntimeError('修正 SQL 的明细条数不正确') if [row['customer_id'] for row in correct_rows] != [1, 2, 3, 4]: raise RuntimeError('修正后的排名不正确') if amount_map(query_rows(connection, distinct)) != amount_map(correct_rows): raise RuntimeError('基础样例的 DISTINCT 演示与预期不同') bad_date_query = correct.replace( "i.invoice_date < '2025-07-01'", "i.invoice_date <= '2025-06-30'" ) date_rows = query_rows(connection, bad_date_query) if amount_map(date_rows)[3] != '1800.00': raise RuntimeError('月末漏数未复现') expanded_invoice = query_rows(connection, ''' SELECT i.invoice_id, d.line_id, i.total_amount_cents, d.line_amount_cents FROM invoices i JOIN invoice_lines d ON d.invoice_id = i.invoice_id WHERE i.invoice_id = 1002 ORDER BY d.line_id ''') finally: connection.close() exercise_connection = create_database(folder, exercise=True) try: exercise_amounts, _ = baseline(exercise_connection) exercise_correct = query_rows(exercise_connection, correct) exercise_distinct = query_rows(exercise_connection, distinct) if exercise_amounts[1] != 850000 or amount_map(exercise_correct)[1] != '8500.00': raise RuntimeError('同额发票的正确金额核对失败') if amount_map(exercise_distinct)[1] != '5500.00': raise RuntimeError('SUM DISTINCT 的漏算未复现') finally: exercise_connection.close() missing_connection = create_database(folder, missing_details=True) try: missing_rows = query_rows(missing_connection, correct) missing_row = next(row for row in missing_rows if row['customer_id'] == 5) if format(missing_row['sales_amount'], '.2f') != '1000.00' or missing_row['detail_line_count'] != 0: raise RuntimeError('没有明细的发票未正确保留') finally: missing_connection.close() result = { 'status': 'passed', 'sqlite_version': sqlite3.sqlite_version, 'wrong_join_rows': wrong_rows, 'correct_join_rows': correct_rows, 'expanded_invoice_1002': expanded_invoice, 'wrong_date_customer_3_amount_yuan': '1800.00', 'equal_amount_exercise': { 'customer_id': 1, 'correct_amount_yuan': '8500.00', 'sum_distinct_amount_yuan': '5500.00', }, 'invoice_without_details': missing_row, 'checks': [ '基础样例的每张发票明细金额与头金额一致', '错误 JOIN 复现重复累计与排名变化', '修正查询的金额与逐笔基线一致', '修正查询的明细条数与原始明细一致', '月末日期条件漏掉 1200 元的记录', '同额发票复现 SUM DISTINCT 漏算', '无明细发票保留金额且明细条数为零', ], } (folder / 'chapter-02-results.json').write_text( json.dumps(result, ensure_ascii=False, indent=2) + '\n', encoding='utf-8' ) print(json.dumps(result, ensure_ascii=False, indent=2)) if __name__ == '__main__': main()