当前位置：网站首页 > 技术资源 > 正文

Python进阶-day21:复习与小项目

off999 2025-05-21 15:46 128 浏览 0 评论

学习目标

复习内容：巩固OOP、异常处理、文件操作、模块化等知识。
高级概念： 设计模式：单例模式（确保账户唯一性）、工厂模式（创建交易对象）。 上下文管理：管理文件操作和数据库连接。 元编程：动态添加方法或属性。 多线程：异步保存数据和统计分析。
小项目：开发一个功能强大的命令行个人财务管理系统，支持持久化、并发操作和动态扩展。
实践技能：通过项目综合应用高级Python技术，提升代码设计能力。

时间分配（总计：5-7小时）

1小时：复习本周内容
4-5小时：开发个人财务管理系统项目（融入高级概念）
1小时：项目测试、优化与总结
你已掌握Python基础、OOP、文件操作、模块化编程。
本周学习了OOP、异常处理、文件操作、模块化、数据结构等。
熟悉标准库（如json、datetime、os）。

一、复习内容（1小时）

复习重点

面向对象编程（OOP）：类、继承、封装、多态私有属性、属性装饰器（@property）
异常处理： try-except-else-finally 自定义异常类
文件操作：读写文本、JSON、CSV with语句管理资源
模块与包：模块导入、包组织 init.py与相对导入
数据结构：列表、字典、集合推导式、高阶函数（map、filter）

复习方法

快速回顾笔记（20分钟）：
复习笔记或代码，重点关注OOP、异常处理、文件操作。
写下每个知识点的核心代码示例。例如：python
# OOP 示例：使用属性装饰器 class Account: def __init__(self): self._balance = 0 @property def balance(self): return self._balance @balance.setter def balance(self, value): if value < 0: raise ValueError("Balance cannot be negative") self._balance = value
练习题（30分钟）：
完成以下练习：定义一个Transaction类，使用@property管理金额属性，确保金额为正。编写一个函数，从transactions.json读取交易记录，处理文件不存在的异常。创建一个模块utils.py，实现一个高阶函数，过滤金额大于某个阈值的交易。使用try-except处理用户输入，验证日期格式（YYYY-MM-DD）。
总结问题（10分钟）：
记录复习中的薄弱点，留待项目中实践。

二、小项目：命令行个人财务管理系统（4-5小时）

项目概述

开发一个命令行个人财务管理系统，支持：

记录收入和支出（金额、类别、日期、备注）。
查看账户余额、交易历史。
按类别或日期筛选交易。
统计某类别的总收入/支出。
数据持久化（JSON文件）。
并发保存数据和统计分析。

融入高级概念

设计模式： 单例模式：确保全局只有一个Account实例。 工厂模式：通过工厂类创建不同类型的交易（收入、支出）。
上下文管理：使用contextlib管理文件操作，确保资源正确关闭。实现自定义上下文管理器，记录操作日志。
元编程：使用metaclass动态为Transaction类添加审计方法。使用setattr动态添加统计方法。
多线程：使用threading异步保存数据到文件。使用线程池执行统计分析。

项目功能需求

主菜单：添加收入/支出查看余额、交易历史筛选交易（按类别或日期）统计分析退出并保存
数据存储：使用JSON文件存储交易和余额。异步保存数据。
输入验证：验证金额、日期、类别。
统计功能：异步计算类别统计。
日志：使用上下文管理器记录操作日志。

项目结构

personal_finance/
├── main.py           # 主程序入口
├── account.py        # 账户和交易逻辑（单例模式、工厂模式、元编程）
├── storage.py        # 文件操作（上下文管理、多线程）
├── menu.py           # 用户交互界面
├── logger.py         # 日志管理（上下文管理）
└── data.json         # 数据存储文件

项目实现步骤

1. 日志管理（logger.py）

实现上下文管理器，记录操作日志。

python

# logger.py
from contextlib import contextmanager
import logging
from datetime import datetime

# 配置日志
logging.basicConfig(
    filename="finance.log",
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)

@contextmanager
def operation_logger(operation_name):
    """上下文管理器，记录操作开始和结束"""
    logging.info(f"Starting operation: {operation_name}")
    start_time = datetime.now()
    try:
        yield
        logging.info(f"Completed operation: {operation_name}, Duration: {datetime.now() - start_time}")
    except Exception as e:
        logging.error(f"Failed operation: {operation_name}, Error: {str(e)}")
        raise

2. 账户与交易逻辑（account.py）

实现单例模式、工厂模式和元编程。

python

# account.py
from datetime import datetime
import uuid
from abc import ABC, ABCMeta, abstractmethod
from logger import operation_logger  # Import operation_logger

class TransactionMeta(ABCMeta):
    """Custom metaclass for Transaction, inheriting from ABCMeta to resolve metaclass conflict"""
    def __new__(cls, name, bases, attrs):
        # Dynamically add an audit method to the class
        def audit(self):
            return f"Audit: Transaction {self.id} created at {self.created_at}"
        attrs['audit'] = audit
        return super().__new__(cls, name, bases, attrs)

class Transaction(ABC, metaclass=TransactionMeta):
    """Abstract base class for transactions"""
    def __init__(self, amount, category, date=None, note=""):
        if amount <= 0:
            raise ValueError("Amount must be positive")
        self.id = str(uuid.uuid4())[:8]  # Unique transaction ID
        self.amount = amount
        self.category = category
        self.date = date or datetime.now().strftime("%Y-%m-%d")
        self.note = note
        self.created_at = datetime.now()

    @abstractmethod
    def apply(self, balance):
        """Abstract method to apply transaction to balance"""
        pass

class IncomeTransaction(Transaction):
    """Concrete class for income transactions"""
    def apply(self, balance):
        return balance + self.amount

class ExpenseTransaction(Transaction):
    """Concrete class for expense transactions"""
    def apply(self, balance):
        if balance < self.amount:
            raise ValueError("Insufficient balance")
        return balance - self.amount

class TransactionFactory:
    """Factory class to create transaction objects"""
    @staticmethod
    def create_transaction(trans_type, amount, category, date=None, note=""):
        if trans_type == "income":
            return IncomeTransaction(amount, category, date, note)
        elif trans_type == "expense":
            return ExpenseTransaction(amount, category, date, note)
        else:
            raise ValueError("Invalid transaction type")

class Account:
    """Singleton class for managing the account"""
    _instance = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super(Account, cls).__new__(cls)
            cls._instance.balance = 0
            cls._instance.transactions = []
        return cls._instance

    def add_transaction(self, trans_type, amount, category, date=None, note=""):
        """Add a transaction to the account"""
        with operation_logger(f"Add {trans_type} transaction"):
            transaction = TransactionFactory.create_transaction(trans_type, amount, category, date, note)
            self.balance = transaction.apply(self.balance)
            self.transactions.append(transaction)
            print(f"Transaction audit: {transaction.audit()}")  # Call dynamically added audit method
            return transaction

    def get_balance(self):
        """Return the current balance"""
        return self.balance

    def get_transactions(self, category=None, start_date=None, end_date=None):
        """Filter transactions by category, start date, or end date"""
        result = self.transactions
        if category:
            result = [t for t in result if t.category == category]
        if start_date:
            result = [t for t in result if t.date >= start_date]
        if end_date:
            result = [t for t in result if t.date <= end_date]
        return result

    def get_category_summary(self, category):
        """Calculate total income and expense for a category"""
        total_income = sum(t.amount for t in self.transactions
                          if t.category == category and isinstance(t, IncomeTransaction))
        total_expense = sum(t.amount for t in self.transactions
                           if t.category == category and isinstance(t, ExpenseTransaction))
        return {"income": total_income, "expense": total_expense}

3. 文件操作（storage.py）

实现上下文管理和多线程保存。

python

# storage.py
import json
import os
from contextlib import contextmanager
from threading import Thread
from logger import operation_logger

class Storage:
    def __init__(self, filename="data.json"):
        self.filename = filename

    @contextmanager
    def file_handler(self, mode):
        """上下文管理器：管理文件操作"""
        file = open(self.filename, mode)
        try:
            yield file
        finally:
            file.close()

    def save_async(self, account):
        """异步保存数据"""
        def save_task():
            with operation_logger("Save data"):
                data = {
                    "balance": account.balance,
                    "transactions": [
                        {
                            "id": t.id,
                            "amount": t.amount,
                            "category": t.category,
                            "trans_type": "income" if isinstance(t, IncomeTransaction) else "expense",
                            "date": t.date,
                            "note": t.note
                        } for t in account.transactions
                    ]
                }
                with self.file_handler("w") as f:
                    json.dump(data, f, indent=4)
        thread = Thread(target=save_task)
        thread.start()
        return thread

    def load(self):
        """加载数据"""
        if not os.path.exists(self.filename):
            return None
        with operation_logger("Load data"):
            with self.file_handler("r") as f:
                return json.load(f)

4. 用户界面（menu.py）

实现交互界面，支持异步操作。

python

# menu.py
from account import Account, TransactionFactory, IncomeTransaction, ExpenseTransaction  # Added imports
from storage import Storage
from logger import operation_logger
from datetime import datetime
from concurrent.futures import ThreadPoolExecutor

def validate_date(date_str):
    """验证日期格式"""
    try:
        datetime.strptime(date_str, "%Y-%m-%d")
        return True
    except ValueError:
        return False

def async_summary(account, category):
    """异步计算类别统计"""
    return account.get_category_summary(category)

def main_menu():
    account = Account()
    storage = Storage()

    # 加载数据
    with operation_logger("Initialize account"):
        data = storage.load()
        if data:
            account.balance = data["balance"]
            for t in data["transactions"]:
                account.add_transaction(
                    t["trans_type"], t["amount"], t["category"], t["date"], t["note"]
                )

    # 动态添加统计方法（元编程）
    def monthly_summary(self, year, month):
        """动态添加：按月统计"""
        total_income = sum(t.amount for t in self.transactions
                           if t.date.startswith(f"{year}-{month:02d}") and isinstance(t, IncomeTransaction))
        total_expense = sum(t.amount for t in self.transactions
                            if t.date.startswith(f"{year}-{month:02d}") and isinstance(t, ExpenseTransaction))
        return {"income": total_income, "expense": total_expense}

    setattr(Account, "monthly_summary", monthly_summary)

    with ThreadPoolExecutor(max_workers=2) as executor:
        while True:
            print("\\n=== Personal Finance Manager ===")
            print("1. Add Income")
            print("2. Add Expense")
            print("3. View Balance")
            print("4. View Transactions")
            print("5. Filter Transactions")
            print("6. Category Summary")
            print("7. Monthly Summary")
            print("8. Exit")
            choice = input("Enter choice (1-8): ")

            try:
                if choice in ["1", "2"]:
                    with operation_logger(f"Add {'income' if choice == '1' else 'expense'}"):
                        amount = float(input("Enter amount: "))
                        category = input("Enter category (e.g., Salary, Food): ")
                        date = input("Enter date (YYYY-MM-DD, press Enter for today): ")
                        if date and not validate_date(date):
                            print("Invalid date format!")
                            continue
                        note = input("Enter note (optional): ")
                        trans_type = "income" if choice == "1" else "expense"
                        account.add_transaction(trans_type, amount, category, date or None, note)
                        print(f"{trans_type.capitalize()} added successfully!")

                elif choice == "3":
                    print(f"Current Balance: ${account.get_balance():.2f}")

                elif choice == "4":
                    transactions = account.get_transactions()
                    if not transactions:
                        print("No transactions found.")
                    for t in transactions:
                        print(f"ID: {t.id}, Type: {'Income' if isinstance(t, IncomeTransaction) else 'Expense'}, "
                              f"Amount: ${t.amount:.2f}, Category: {t.category}, Date: {t.date}, Note: {t.note}")

                elif choice == "5":
                    category = input("Enter category to filter (press Enter to skip): ")
                    start_date = input("Enter start date (YYYY-MM-DD, press Enter to skip): ")
                    if start_date and not validate_date(start_date):
                        print("Invalid start date!")
                        continue
                    end_date = input("Enter end date (YYYY-MM-DD, press Enter to skip): ")
                    if end_date and not validate_date(end_date):
                        print("Invalid end date!")
                        continue
                    transactions = account.get_transactions(category or None, start_date or None, end_date or None)
                    if not transactions:
                        print("No transactions found.")
                    for t in transactions:
                        print(f"ID: {t.id}, Type: {'Income' if isinstance(t, IncomeTransaction) else 'Expense'}, "
                              f"Amount: ${t.amount:.2f}, Category: {t.category}, Date: {t.date}, Note: {t.note}")

                elif choice == "6":
                    category = input("Enter category: ")
                    with operation_logger(f"Category summary for {category}"):
                        future = executor.submit(async_summary, account, category)
                        summary = future.result()
                        print(f"Category: {category}")
                        print(f"Total Income: ${summary['income']:.2f}")
                        print(f"Total Expense: ${summary['expense']:.2f}")

                elif choice == "7":
                    year = int(input("Enter year (e.g., 2025): "))
                    month = int(input("Enter month (1-12): "))
                    with operation_logger(f"Monthly summary for {year}-{month:02d}"):
                        summary = account.monthly_summary(year, month)
                        print(f"Monthly Summary ({year}-{month:02d})")
                        print(f"Total Income: ${summary['income']:.2f}")
                        print(f"Total Expense: ${summary['expense']:.2f}")

                elif choice == "8":
                    with operation_logger("Exit and save"):
                        storage.save_async(account).join()  # 等待保存完成
                        print("Data saved. Goodbye!")
                        break

                else:
                    print("Invalid choice!")

            except ValueError as e:
                print(f"Error: {e}")
            except Exception as e:
                print(f"An unexpected error occurred: {e}")

5. 主程序（main.py）

启动程序。

python

# main.py
from menu import main_menu

if __name__ == "__main__":
    main_menu()

6. 测试与调试（1小时）

测试用例：添加收入（1000，Salary，2025-04-22，"Monthly salary"）。添加支出（200，Food，2025-04-22，"Dinner"）。查看余额（应为800）。查看所有交易。筛选Food类别的交易。查看Salary类别的统计（异步）。查看2025年4月月度统计。退出并检查data.json和finance.log。
调试：检查单例模式（多次实例化Account应返回同一对象）。验证异步保存（data.json是否正确更新）。测试元编程（audit方法和monthly_summary是否正常工作）。确保上下文管理器记录所有操作日志。
优化：使用argparse支持命令行参数。添加删除交易功能。使用tabulate美化输出。

三、项目测试、优化与总结（1小时）

测试

运行程序，逐一测试所有功能。
故意输入错误数据（负金额、无效日期），验证异常处理。
检查finance.log，确保所有操作被记录。
重启程序，确认数据正确加载。

优化建议

添加数据库支持（sqlite3）。
使用asyncio替代threading，提升异步性能。
实现更多设计模式（如观察者模式，用于交易通知）。
添加单元测试（使用unittest或pytest）。

总结

知识点回顾： 设计模式：单例模式（Account）、工厂模式（TransactionFactory）。 上下文管理：文件操作（Storage.file_handler）、日志（operation_logger）。 元编程：动态添加audit和monthly_summary方法。 多线程：异步保存（save_async）、异步统计（async_summary）。 OOP：继承、抽象类、封装。 异常处理：输入验证、文件操作错误。 模块化：逻辑分离到多个文件。
问题与解决方案：记录调试中遇到的问题（如线程同步、元类调试）。总结解决方案（如使用Thread.join()确保保存完成）。
扩展方向：添加GUI（tkinter）。集成pandas进行数据分析。实现REST API（使用FastAPI）。

学习成果