Python项目实战教程（第三阶段）#

欢迎来到Python项目实战教程的第三阶段！在前面的教程中，我们学习了Python的基础知识和高级特性。在本阶段中，我们将通过实际项目来应用这些知识，学习Python在各个领域的实际应用场景，帮助你成为一名真正的Python开发者。

1. Web应用开发实战#

1.1 Flask项目实战 - 个人博客系统#

项目概述：创建一个功能完整的个人博客系统，包括用户注册、登录、文章发布、评论等功能。

技术栈：

Flask：Web框架
Flask-SQLAlchemy：ORM数据库工具
Flask-Login：用户认证
Flask-WTF：表单处理
Bootstrap：前端框架

项目结构：

1
blog/
2
├── app/
3
│   ├── __init__.py
4
│   ├── routes.py
5
│   ├── models.py
6
│   ├── forms.py
7
│   ├── templates/
8
│   │   ├── base.html
9
│   │   ├── index.html
10
│   │   ├── login.html
11
│   │   ├── register.html
12
│   │   ├── post.html
13
│   │   └── create_post.html
14
│   └── static/
15
│       ├── css/
16
│       └── js/
17
├── config.py
18
├── requirements.txt
19
└── run.py

核心代码：

app/__init__.py：

1
from flask import Flask
2
from flask_sqlalchemy import SQLAlchemy
3
from flask_login import LoginManager
4
from config import Config
5

6
app = Flask(__name__)
7
app.config.from_object(Config)
8
db = SQLAlchemy(app)
9
login = LoginManager(app)
10
login.login_view = 'login'
11

12
from app import routes, models

app/models.py：

1
from datetime import datetime
2
from app import db, login
3
from flask_login import UserMixin
4
from werkzeug.security import generate_password_hash, check_password_hash
5

6
class User(UserMixin, db.Model):
7
    id = db.Column(db.Integer, primary_key=True)
8
    username = db.Column(db.String(64), index=True, unique=True)
9
    email = db.Column(db.String(120), index=True, unique=True)
10
    password_hash = db.Column(db.String(128))
11
    posts = db.relationship('Post', backref='author', lazy='dynamic')
12

13
    def set_password(self, password):
14
        self.password_hash = generate_password_hash(password)
15

16
    def check_password(self, password):
17
        return check_password_hash(self.password_hash, password)
18

19
class Post(db.Model):
20
    id = db.Column(db.Integer, primary_key=True)
21
    title = db.Column(db.String(140))
22
    content = db.Column(db.Text)
23
    timestamp = db.Column(db.DateTime, index=True, default=datetime.utcnow)
24
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'))
25
    comments = db.relationship('Comment', backref='post', lazy='dynamic')
26

27
class Comment(db.Model):
28
    id = db.Column(db.Integer, primary_key=True)
29
    content = db.Column(db.Text)
30
    timestamp = db.Column(db.DateTime, index=True, default=datetime.utcnow)
31
    post_id = db.Column(db.Integer, db.ForeignKey('post.id'))
32
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'))
33

34
@login.user_loader
35
def load_user(id):
36
    return User.query.get(int(id))

app/forms.py：

1
from flask_wtf import FlaskForm
2
from wtforms import StringField, PasswordField, SubmitField, TextAreaField
3
from wtforms.validators import DataRequired, Email, EqualTo, Length
4

5
class LoginForm(FlaskForm):
6
    username = StringField('Username', validators=[DataRequired()])
7
    password = PasswordField('Password', validators=[DataRequired()])
8
    submit = SubmitField('Sign In')
9

10
class RegistrationForm(FlaskForm):
11
    username = StringField('Username', validators=[DataRequired(), Length(min=2, max=64)])
12
    email = StringField('Email', validators=[DataRequired(), Email()])
13
    password = PasswordField('Password', validators=[DataRequired()])
14
    password2 = PasswordField('Repeat Password', validators=[DataRequired(), EqualTo('password')])
15
    submit = SubmitField('Register')
16

17
class PostForm(FlaskForm):
18
    title = StringField('Title', validators=[DataRequired(), Length(max=140)])
19
    content = TextAreaField('Content', validators=[DataRequired()])
20
    submit = SubmitField('Post')
21

22
class CommentForm(FlaskForm):
23
    content = TextAreaField('Comment', validators=[DataRequired()])
24
    submit = SubmitField('Submit')

app/routes.py：

1
from flask import render_template, redirect, url_for, flash, request
2
from flask_login import current_user, login_user, logout_user, login_required
3
from app import app, db
4
from app.models import User, Post, Comment
5
from app.forms import LoginForm, RegistrationForm, PostForm, CommentForm
6

7
@app.route('/')
8
@app.route('/index')
9
def index():
10
    posts = Post.query.order_by(Post.timestamp.desc()).all()
11
    return render_template('index.html', title='Home', posts=posts)
12

13
@app.route('/login', methods=['GET', 'POST'])
14
def login():
15
    if current_user.is_authenticated:
16
        return redirect(url_for('index'))
17
    form = LoginForm()
18
    if form.validate_on_submit():
19
        user = User.query.filter_by(username=form.username.data).first()
20
        if user is None or not user.check_password(form.password.data):
21
            flash('Invalid username or password')
22
            return redirect(url_for('login'))
23
        login_user(user)
24
        return redirect(url_for('index'))
25
    return render_template('login.html', title='Sign In', form=form)
26

27
@app.route('/logout')
28
def logout():
29
    logout_user()
30
    return redirect(url_for('index'))
31

32
@app.route('/register', methods=['GET', 'POST'])
33
def register():
34
    if current_user.is_authenticated:
35
        return redirect(url_for('index'))
36
    form = RegistrationForm()
37
    if form.validate_on_submit():
38
        user = User(username=form.username.data, email=form.email.data)
39
        user.set_password(form.password.data)
40
        db.session.add(user)
41
        db.session.commit()
42
        flash('Congratulations, you are now a registered user!')
43
        return redirect(url_for('login'))
44
    return render_template('register.html', title='Register', form=form)
45

46
@app.route('/post/<int:post_id>', methods=['GET', 'POST'])
47
def post(post_id):
48
    post = Post.query.get_or_404(post_id)
49
    form = CommentForm()
50
    if form.validate_on_submit() and current_user.is_authenticated:
51
        comment = Comment(content=form.content.data, post_id=post_id, user_id=current_user.id)
52
        db.session.add(comment)
53
        db.session.commit()
54
        flash('Your comment has been added!')
55
        return redirect(url_for('post', post_id=post_id))
56
    comments = post.comments.order_by(Comment.timestamp.desc()).all()
57
    return render_template('post.html', title=post.title, post=post, form=form, comments=comments)
58

59
@app.route('/create_post', methods=['GET', 'POST'])
60
@login_required
61
def create_post():
62
    form = PostForm()
63
    if form.validate_on_submit():
64
        post = Post(title=form.title.data, content=form.content.data, author=current_user)
65
        db.session.add(post)
66
        db.session.commit()
67
        flash('Your post has been created!')
68
        return redirect(url_for('index'))
69
    return render_template('create_post.html', title='Create Post', form=form)

config.py：

1
import os
2

3
class Config:
4
    SECRET_KEY = os.environ.get('SECRET_KEY') or 'you-will-never-guess'
5
    SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or 'sqlite:///blog.db'
6
    SQLALCHEMY_TRACK_MODIFICATIONS = False

run.py：

1
from app import app, db
2
from app.models import User, Post, Comment
3

4
@app.shell_context_processor
5
def make_shell_context():
6
    return {'db': db, 'User': User, 'Post': Post, 'Comment': Comment}
7

8
if __name__ == '__main__':
9
    app.run(debug=True)

requirements.txt：

1
Flask
2
Flask-SQLAlchemy
3
Flask-Login
4
Flask-WTF
5
Werkzeug

运行项目：

1
# 安装依赖
2
pip install -r requirements.txt
3

4
# 初始化数据库
5
flask shell
6
>>> from app import db
7
>>> db.create_all()
8
>>> exit()
9

10
# 运行项目
11
flask run

1.2 Django项目实战 - 电商网站#

项目概述：创建一个简单的电商网站，包括商品展示、购物车、订单管理等功能。

技术栈：

Django：Web框架
Django ORM：数据库工具
Django Authentication：用户认证
Bootstrap：前端框架

项目结构：

1
ecommerce/
2
├── ecommerce/
3
│   ├── __init__.py
4
│   ├── settings.py
5
│   ├── urls.py
6
│   └── wsgi.py
7
├── shop/
8
│   ├── __init__.py
9
│   ├── admin.py
10
│   ├── apps.py
11
│   ├── models.py
12
│   ├── views.py
13
│   ├── urls.py
14
│   └── templates/
15
│       └── shop/
16
│           ├── base.html
17
│           ├── index.html
18
│           ├── product_detail.html
19
│           ├── cart.html
20
│           └── checkout.html
21
├── manage.py
22
└── requirements.txt

核心代码：

shop/models.py：

1
from django.db import models
2
from django.contrib.auth.models import User
3

4
class Category(models.Model):
5
    name = models.CharField(max_length=200, db_index=True)
6
    slug = models.SlugField(max_length=200, unique=True)
7

8
    class Meta:
9
        ordering = ('name',)
10
        verbose_name = 'category'
11
        verbose_name_plural = 'categories'
12

13
    def __str__(self):
14
        return self.name
15

16
class Product(models.Model):
17
    category = models.ForeignKey(Category, related_name='products', on_delete=models.CASCADE)
18
    name = models.CharField(max_length=200, db_index=True)
19
    slug = models.SlugField(max_length=200, db_index=True)
20
    image = models.ImageField(upload_to='products/%Y/%m/%d', blank=True)
21
    description = models.TextField(blank=True)
22
    price = models.DecimalField(max_digits=10, decimal_places=2)
23
    available = models.BooleanField(default=True)
24
    created = models.DateTimeField(auto_now_add=True)
25
    updated = models.DateTimeField(auto_now=True)
26

27
    class Meta:
28
        ordering = ('name',)
29
        index_together = (('id', 'slug'),)
30

31
    def __str__(self):
32
        return self.name
33

34
class CartItem(models.Model):
35
    user = models.ForeignKey(User, related_name='cart_items', on_delete=models.CASCADE)
36
    product = models.ForeignKey(Product, related_name='cart_items', on_delete=models.CASCADE)
37
    quantity = models.PositiveIntegerField(default=1)
38
    created = models.DateTimeField(auto_now_add=True)
39

40
    def __str__(self):
41
        return f'{self.quantity} x {self.product.name}'
42

43
    def get_total_price(self):
44
        return self.quantity * self.product.price
45

46
class Order(models.Model):
47
    user = models.ForeignKey(User, related_name='orders', on_delete=models.CASCADE)
48
    created = models.DateTimeField(auto_now_add=True)
49
    updated = models.DateTimeField(auto_now=True)
50
    paid = models.BooleanField(default=False)
51

52
    class Meta:
53
        ordering = ('-created',)
54

55
    def __str__(self):
56
        return f'Order {self.id}'
57

58
    def get_total_cost(self):
59
        return sum(item.get_cost() for item in self.items.all())
60

61
class OrderItem(models.Model):
62
    order = models.ForeignKey(Order, related_name='items', on_delete=models.CASCADE)
63
    product = models.ForeignKey(Product, related_name='order_items', on_delete=models.CASCADE)
64
    price = models.DecimalField(max_digits=10, decimal_places=2)
65
    quantity = models.PositiveIntegerField(default=1)
66

67
    def __str__(self):
68
        return f'{self.quantity} x {self.product.name}'
69

70
    def get_cost(self):
71
        return self.price * self.quantity

shop/views.py：

1
from django.shortcuts import render, get_object_or_404, redirect
2
from django.contrib.auth.decorators import login_required
3
from django.views.decorators.http import require_POST
4
from .models import Category, Product, CartItem, Order, OrderItem
5

6
@login_required
7
def index(request):
8
    categories = Category.objects.all()
9
    products = Product.objects.filter(available=True)
10
    return render(request, 'shop/index.html', {'categories': categories, 'products': products})
11

12
@login_required
13
def product_detail(request, id, slug):
14
    product = get_object_or_404(Product, id=id, slug=slug, available=True)
15
    return render(request, 'shop/product_detail.html', {'product': product})
16

17
@login_required
18
def cart(request):
19
    cart_items = CartItem.objects.filter(user=request.user)
20
    total = sum(item.get_total_price() for item in cart_items)
21
    return render(request, 'shop/cart.html', {'cart_items': cart_items, 'total': total})
22

23
@login_required
24
@require_POST
25
def add_to_cart(request, product_id):
26
    product = get_object_or_404(Product, id=product_id)
27
    cart_item, created = CartItem.objects.get_or_create(
28
        user=request.user,
29
        product=product,
30
        defaults={'quantity': 1}
31
    )
32
    if not created:
33
        cart_item.quantity += 1
34
        cart_item.save()
35
    return redirect('cart')
36

37
@login_required
38
@require_POST
39
def remove_from_cart(request, cart_item_id):
40
    cart_item = get_object_or_404(CartItem, id=cart_item_id, user=request.user)
41
    cart_item.delete()
42
    return redirect('cart')
43

44
@login_required
45
def checkout(request):
46
    cart_items = CartItem.objects.filter(user=request.user)
47
    if not cart_items:
48
        return redirect('cart')
49

50
    if request.method == 'POST':
51
        order = Order.objects.create(user=request.user)
52
        for item in cart_items:
53
            OrderItem.objects.create(
54
                order=order,
55
                product=item.product,
56
                price=item.product.price,
57
                quantity=item.quantity
58
            )
59
        cart_items.delete()
60
        return redirect('order_success')
61

62
    total = sum(item.get_total_price() for item in cart_items)
63
    return render(request, 'shop/checkout.html', {'cart_items': cart_items, 'total': total})
64

65
@login_required
66
def order_success(request):
67
    return render(request, 'shop/order_success.html')

shop/urls.py：

1
from django.urls import path
2
from . import views
3

4
urlpatterns = [
5
    path('', views.index, name='index'),
6
    path('product/<int:id>/<slug:slug>/', views.product_detail, name='product_detail'),
7
    path('cart/', views.cart, name='cart'),
8
    path('add-to-cart/<int:product_id>/', views.add_to_cart, name='add_to_cart'),
9
    path('remove-from-cart/<int:cart_item_id>/', views.remove_from_cart, name='remove_from_cart'),
10
    path('checkout/', views.checkout, name='checkout'),
11
    path('order-success/', views.order_success, name='order_success'),
12
]

ecommerce/urls.py：

1
from django.contrib import admin
2
from django.urls import path, include
3
from django.contrib.auth import views as auth_views
4

5
urlpatterns = [
6
    path('admin/', admin.site.urls),
7
    path('shop/', include('shop.urls')),
8
    path('login/', auth_views.LoginView.as_view(template_name='shop/login.html'), name='login'),
9
    path('logout/', auth_views.LogoutView.as_view(), name='logout'),
10
]

运行项目：

1
# 安装依赖
2
pip install -r requirements.txt
3

4
# 迁移数据库
5
python manage.py makemigrations
6
python manage.py migrate
7

8
# 创建超级用户
9
python manage.py createsuperuser
10

11
# 运行项目
12
python manage.py runserver

2. 数据分析项目实战#

2.1 数据分析项目 - 销售数据可视化#

项目概述：分析销售数据，生成可视化报表，帮助企业了解销售趋势和产品表现。

技术栈：

Python
Pandas：数据处理
Matplotlib：数据可视化
Seaborn：高级数据可视化
Jupyter Notebook：交互式开发环境

项目结构：

1
sales-analysis/
2
├── data/
3
│   └── sales_data.csv
4
├── notebooks/
5
│   └── sales_analysis.ipynb
6
└── requirements.txt

销售数据示例：

OrderID	Product	Category	Price	Quantity	Date	Region
1	Product A	Electronics	100	2	2026-01-01	North
2	Product B	Clothing	50	1	2026-01-02	South
3	Product C	Electronics	150	1	2026-01-03	East
4	Product D	Clothing	30	3	2026-01-04	West
5	Product A	Electronics	100	1	2026-01-05	North

核心代码：

1
import pandas as pd
2
import matplotlib.pyplot as plt
3
import seaborn as sns
4

5
# 设置中文字体
6
plt.rcParams['font.sans-serif'] = ['SimHei']  # 用来正常显示中文标签
7
plt.rcParams['axes.unicode_minus'] = False  # 用来正常显示负号
8

9
# 读取数据
10
df = pd.read_csv('data/sales_data.csv')
11

12
# 数据清洗
13
print("数据基本信息：")
14
print(df.info())
15

16
print("\n数据前5行：")
17
print(df.head())
18

19
print("\n数据统计摘要：")
20
print(df.describe())
21

22
# 计算销售额
23
df['Sales'] = df['Price'] * df['Quantity']
24

25
# 按日期分析销售趋势
26
df['Date'] = pd.to_datetime(df['Date'])
27
df.set_index('Date', inplace=True)
28

29
# 月度销售趋势
30
monthly_sales = df.resample('M').sum()['Sales']
31

32
plt.figure(figsize=(12, 6))
33
plt.plot(monthly_sales.index, monthly_sales.values)
34
plt.title('月度销售趋势')
35
plt.xlabel('月份')
36
plt.ylabel('销售额')
37
plt.grid(True)
38
plt.show()
39

40
# 按产品类别分析
41
category_sales = df.groupby('Category').sum()['Sales']
42

43
plt.figure(figsize=(10, 6))
44
sns.barplot(x=category_sales.index, y=category_sales.values)
45
plt.title('各产品类别销售额')
46
plt.xlabel('产品类别')
47
plt.ylabel('销售额')
48
plt.show()
49

50
# 按地区分析
51
region_sales = df.groupby('Region').sum()['Sales']
52

53
plt.figure(figsize=(10, 6))
54
plt.pie(region_sales.values, labels=region_sales.index, autopct='%1.1f%%')
55
plt.title('各地区销售占比')
56
plt.show()
57

58
# 产品销售排行
59
top_products = df.groupby('Product').sum()['Sales'].sort_values(ascending=False).head(10)
60

61
plt.figure(figsize=(12, 6))
62
sns.barplot(x=top_products.values, y=top_products.index)
63
plt.title('产品销售排行前10')
64
plt.xlabel('销售额')
65
plt.ylabel('产品')
66
plt.show()
67

68
# 销售数据相关性分析
69
corr = df[['Price', 'Quantity', 'Sales']].corr()
70

71
plt.figure(figsize=(8, 6))
72
sns.heatmap(corr, annot=True, cmap='coolwarm')
73
plt.title('销售数据相关性分析')
74
plt.show()
75

76
# 总结
77
print("\n销售数据分析总结：")
78
print(f"总销售额：{df['Sales'].sum():.2f}")
79
print(f"平均销售额：{df['Sales'].mean():.2f}")
80
print(f"最高销售额：{df['Sales'].max():.2f}")
81
print(f"最低销售额：{df['Sales'].min():.2f}")
82
print(f"销售最好的产品类别：{category_sales.idxmax()}")
83
print(f"销售最好的地区：{region_sales.idxmax()}")
84
print(f"销售最好的产品：{top_products.idxmax()}")

运行项目：

1
# 安装依赖
2
pip install -r requirements.txt
3

4
# 运行Jupyter Notebook
5
jupyter notebook
6

7
# 在浏览器中打开notebooks/sales_analysis.ipynb并运行所有单元格

3. 爬虫项目实战#

3.1 爬虫项目 - 电影信息爬取#

项目概述：爬取电影网站的电影信息，包括电影名称、评分、导演、演员、上映日期等，存储到数据库中，并生成可视化报表。

技术栈：

Python
requests：网络请求
BeautifulSoup：HTML解析
SQLite：数据库存储
Matplotlib：数据可视化

核心代码：

1
import requests
2
from bs4 import BeautifulSoup
3
import sqlite3
4
import matplotlib.pyplot as plt
5

6
# 设置中文字体
7
plt.rcParams['font.sans-serif'] = ['SimHei']
8
plt.rcParams['axes.unicode_minus'] = False
9

10
# 创建数据库连接
11
conn = sqlite3.connect('movies.db')
12
cursor = conn.cursor()
13

14
# 创建电影表
15
cursor.execute('''
16
CREATE TABLE IF NOT EXISTS movies (
17
    id INTEGER PRIMARY KEY AUTOINCREMENT,
18
    title TEXT NOT NULL,
19
    rating REAL,
20
    directors TEXT,
21
    actors TEXT,
22
    release_date TEXT,
23
    genre TEXT,
24
    duration TEXT,
25
    description TEXT
26
)
27
''')
28
conn.commit()
29

30
# 爬取电影信息
31
def crawl_movies(url):
32
    headers = {
33
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
34
    }
35
    response = requests.get(url, headers=headers)
36
    soup = BeautifulSoup(response.text, 'html.parser')
37

38
    # 解析电影列表
39
    movie_list = soup.find_all('div', class_='movie-item')
40

41
    for movie in movie_list:
42
        # 电影标题
43
        title = movie.find('h2', class_='movie-title').text.strip()
44

45
        # 电影评分
46
        rating = movie.find('span', class_='rating').text.strip()
47
        rating = float(rating) if rating else None
48

49
        # 电影信息
50
        info = movie.find('div', class_='movie-info').text.strip()
51

52
        # 提取导演、演员、上映日期等信息
53
        directors = None
54
        actors = None
55
        release_date = None
56
        genre = None
57
        duration = None
58

59
        # 这里需要根据实际网站的HTML结构进行调整
60
        # 示例代码，实际使用时需要根据目标网站的结构修改
61
        if '导演:' in info:
62
            directors = info.split('导演:')[1].split('主演:')[0].strip()
63
        if '主演:' in info:
64
            actors = info.split('主演:')[1].split('上映日期:')[0].strip()
65
        if '上映日期:' in info:
66
            release_date = info.split('上映日期:')[1].split('类型:')[0].strip()
67
        if '类型:' in info:
68
            genre = info.split('类型:')[1].split('片长:')[0].strip()
69
        if '片长:' in info:
70
            duration = info.split('片长:')[1].strip()
71

72
        # 电影描述
73
        description = movie.find('p', class_='movie-description').text.strip() if movie.find('p', class_='movie-description') else None
74

75
        # 插入数据库
76
        cursor.execute('''
77
        INSERT INTO movies (title, rating, directors, actors, release_date, genre, duration, description)
78
        VALUES (?, ?, ?, ?, ?, ?, ?, ?)
79
        ''', (title, rating, directors, actors, release_date, genre, duration, description))
80

81
    conn.commit()
82
    print(f"成功爬取 {len(movie_list)} 部电影信息")
83

84
# 爬取多个页面
85
for page in range(1, 11):  # 爬取前10页
86
    url = f'https://example.com/movies?page={page}'  # 替换为实际的电影网站URL
87
    print(f"爬取第 {page} 页...")
88
    crawl_movies(url)
89

90
# 分析电影数据
91
def analyze_movies():
92
    # 读取数据库
93
    cursor.execute('SELECT * FROM movies')
94
    movies = cursor.fetchall()
95

96
    print(f"\n数据库中共有 {len(movies)} 部电影")
97

98
    # 按评分统计
99
    cursor.execute('SELECT rating FROM movies WHERE rating IS NOT NULL')
100
    ratings = [row[0] for row in cursor.fetchall()]
101

102
    if ratings:
103
        avg_rating = sum(ratings) / len(ratings)
104
        print(f"平均评分：{avg_rating:.2f}")
105

106
        # 评分分布
107
        plt.figure(figsize=(10, 6))
108
        plt.hist(ratings, bins=10, range=(0, 10))
109
        plt.title('电影评分分布')
110
        plt.xlabel('评分')
111
        plt.ylabel('电影数量')
112
        plt.show()
113

114
    # 按类型统计
115
    cursor.execute('SELECT genre FROM movies WHERE genre IS NOT NULL')
116
    genres = [row[0] for row in cursor.fetchall()]
117

118
    genre_counts = {}
119
    for genre in genres:
120
        if genre:
121
            for g in genre.split('/'):
122
                g = g.strip()
123
                genre_counts[g] = genre_counts.get(g, 0) + 1
124

125
    if genre_counts:
126
        top_genres = sorted(genre_counts.items(), key=lambda x: x[1], reverse=True)[:10]
127

128
        plt.figure(figsize=(12, 6))
129
        plt.bar([g[0] for g in top_genres], [g[1] for g in top_genres])
130
        plt.title('电影类型分布前10')
131
        plt.xlabel('类型')
132
        plt.ylabel('电影数量')
133
        plt.xticks(rotation=45)
134
        plt.show()
135

136
    # 评分最高的电影
137
    cursor.execute('SELECT title, rating FROM movies WHERE rating IS NOT NULL ORDER BY rating DESC LIMIT 10')
138
    top_rated = cursor.fetchall()
139

140
    print("\n评分最高的10部电影：")
141
    for movie in top_rated:
142
        print(f"{movie[0]} - {movie[1]}")
143

144
# 运行分析
145
analyze_movies()
146

147
# 关闭数据库连接
148
conn.close()

运行项目：

1
# 安装依赖
2
pip install requests beautifulsoup4 matplotlib
3

4
# 运行爬虫
5
python movie_crawler.py

4. 自动化脚本开发#

4.1 自动化脚本 - 文件整理工具#

项目概述：开发一个自动化脚本，根据文件类型自动整理文件夹中的文件，提高文件管理效率。

核心代码：

1
import os
2
import shutil
3
import argparse
4

5
# 文件类型映射
6
extension_map = {
7
    'images': ['.jpg', '.jpeg', '.png', '.gif', '.bmp', '.tiff', '.webp'],
8
    'documents': ['.pdf', '.doc', '.docx', '.txt', '.rtf', '.odt'],
9
    'spreadsheets': ['.xls', '.xlsx', '.csv', '.ods'],
10
    'presentations': ['.ppt', '.pptx', '.odp'],
11
    'archives': ['.zip', '.rar', '.7z', '.tar', '.gz'],
12
    'audio': ['.mp3', '.wav', '.flac', '.aac', '.ogg'],
13
    'video': ['.mp4', '.avi', '.mov', '.wmv', '.flv', '.mkv'],
14
    'code': ['.py', '.java', '.cpp', '.js', '.html', '.css', '.php'],
15
    'executables': ['.exe', '.msi', '.app', '.dmg']
16
}
17

18
# 反向映射，扩展名到文件夹名
19
extension_to_folder = {}
20
for folder, extensions in extension_map.items():
21
    for ext in extensions:
22
        extension_to_folder[ext.lower()] = folder
23

24
def organize_files(directory):
25
    """根据文件类型整理文件夹中的文件"""
26
    if not os.path.exists(directory):
27
        print(f"错误：目录 {directory} 不存在")
28
        return
29

30
    # 遍历目录中的所有文件
31
    for filename in os.listdir(directory):
32
        filepath = os.path.join(directory, filename)
33

34
        # 跳过文件夹
35
        if os.path.isdir(filepath):
36
            continue
37

38
        # 获取文件扩展名
39
        _, ext = os.path.splitext(filename)
40
        ext = ext.lower()
41

42
        # 确定目标文件夹
43
        if ext in extension_to_folder:
44
            target_folder = os.path.join(directory, extension_to_folder[ext])
45
        else:
46
            target_folder = os.path.join(directory, 'others')
47

48
        # 创建目标文件夹（如果不存在）
49
        if not os.path.exists(target_folder):
50
            os.makedirs(target_folder)
51

52
        # 移动文件
53
        try:
54
            target_filepath = os.path.join(target_folder, filename)
55

56
            # 如果文件已存在，添加编号
57
            counter = 1
58
            while os.path.exists(target_filepath):
59
                name, ext = os.path.splitext(filename)
60
                target_filepath = os.path.join(target_folder, f"{name}_{counter}{ext}")
61
                counter += 1
62

63
            shutil.move(filepath, target_filepath)
64
            print(f"移动：{filename} -> {extension_to_folder.get(ext, 'others')}/")
65
        except Exception as e:
66
            print(f"错误移动文件 {filename}：{e}")
67

68
def main():
69
    parser = argparse.ArgumentParser(description='根据文件类型自动整理文件夹中的文件')
70
    parser.add_argument('directory', help='要整理的文件夹路径')
71
    args = parser.parse_args()
72

73
    print(f"开始整理文件夹：{args.directory}")
74
    organize_files(args.directory)
75
    print("整理完成！")
76

77
if __name__ == "__main__":
78
    main()

运行脚本：

1
# 运行脚本整理指定文件夹
2
python file_organizer.py /path/to/directory

4.2 自动化脚本 - 网络监控工具#

项目概述：开发一个网络监控工具，定期检查网站是否可访问，记录响应时间和状态码，并在网站不可用时发送通知。

核心代码：

1
import requests
2
import time
3
import json
4
import smtplib
5
from email.mime.text import MIMEText
6
from email.mime.multipart import MIMEMultipart
7
import argparse
8

9
# 配置文件
10
CONFIG_FILE = 'config.json'
11
LOG_FILE = 'monitor.log'
12

13
# 加载配置
14
def load_config():
15
    try:
16
        with open(CONFIG_FILE, 'r') as f:
17
            return json.load(f)
18
    except Exception as e:
19
        print(f"错误加载配置文件：{e}")
20
        return {
21
            'websites': [
22
                {'url': 'https://www.google.com', 'name': 'Google'},
23
                {'url': 'https://www.baidu.com', 'name': 'Baidu'}
24
            ],
25
            'interval': 60,  # 检查间隔（秒）
26
            'timeout': 10,  # 请求超时（秒）
27
            'email': {
28
                'enabled': False,
29
                'smtp_server': 'smtp.example.com',
30
                'smtp_port': 587,
31
                'smtp_user': 'user@example.com',
32
                'smtp_password': 'password',
33
                'from_email': 'user@example.com',
34
                'to_email': 'user@example.com'
35
            }
36
        }
37

38
# 保存配置
39
def save_config(config):
40
    try:
41
        with open(CONFIG_FILE, 'w') as f:
42
            json.dump(config, f, indent=4)
43
    except Exception as e:
44
        print(f"错误保存配置文件：{e}")
45

46
# 发送邮件通知
47
def send_email(config, subject, message):
48
    if not config['email']['enabled']:
49
        return
50

51
    try:
52
        msg = MIMEMultipart()
53
        msg['From'] = config['email']['from_email']
54
        msg['To'] = config['email']['to_email']
55
        msg['Subject'] = subject
56
        msg.attach(MIMEText(message, 'plain'))
57

58
        server = smtplib.SMTP(config['email']['smtp_server'], config['email']['smtp_port'])
59
        server.starttls()
60
        server.login(config['email']['smtp_user'], config['email']['smtp_password'])
61
        server.send_message(msg)
62
        server.quit()
63
        print("邮件通知已发送")
64
    except Exception as e:
65
        print(f"错误发送邮件：{e}")
66

67
# 记录日志
68
def log_message(message):
69
    timestamp = time.strftime('%Y-%m-%d %H:%M:%S')
70
    log_entry = f"[{timestamp}] {message}"
71
    print(log_entry)
72
    try:
73
        with open(LOG_FILE, 'a') as f:
74
            f.write(log_entry + '\n')
75
    except Exception as e:
76
        print(f"错误写入日志：{e}")
77

78
# 监控网站
79
def monitor_website(url, name, timeout):
80
    try:
81
        start_time = time.time()
82
        response = requests.get(url, timeout=timeout)
83
        end_time = time.time()
84
        response_time = (end_time - start_time) * 1000  # 转换为毫秒
85

86
        if response.status_code == 200:
87
            status = "正常"
88
            log_message(f"{name} ({url}) - 状态：{status}，响应时间：{response_time:.2f}ms，状态码：{response.status_code}")
89
        else:
90
            status = "异常"
91
            message = f"{name} ({url}) - 状态：{status}，响应时间：{response_time:.2f}ms，状态码：{response.status_code}"
92
            log_message(message)
93
            return False, message
94

95
        return True, None
96
    except Exception as e:
97
        status = "错误"
98
        message = f"{name} ({url}) - 状态：{status}，错误信息：{str(e)}"
99
        log_message(message)
100
        return False, message
101

102
# 主函数
103
def main():
104
    parser = argparse.ArgumentParser(description='网站监控工具')
105
    parser.add_argument('--config', help='配置文件路径', default=CONFIG_FILE)
106
    args = parser.parse_args()
107

108
    global CONFIG_FILE
109
    CONFIG_FILE = args.config
110

111
    config = load_config()
112
    log_message("开始监控网站...")
113

114
    # 初始化状态
115
    website_status = {site['url']: True for site in config['websites']}
116

117
    try:
118
        while True:
119
            for site in config['websites']:
120
                success, message = monitor_website(site['url'], site['name'], config['timeout'])
121

122
                # 检查状态变化
123
                if not success and website_status[site['url']]:
124
                    # 网站从正常变为异常
125
                    website_status[site['url']] = False
126
                    send_email(config, f"【警告】网站 {site['name']} 不可访问", message)
127
                elif success and not website_status[site['url']]:
128
                    # 网站从异常变为正常
129
                    website_status[site['url']] = True
130
                    send_email(config, f"【恢复】网站 {site['name']} 已恢复访问", f"网站 {site['name']} ({site['url']}) 已恢复正常访问")
131

132
            time.sleep(config['interval'])
133
    except KeyboardInterrupt:
134
        log_message("监控已停止")
135

136
if __name__ == "__main__":
137
    main()

运行脚本：

1
# 运行网络监控工具
2
python network_monitor.py
3

4
# 使用自定义配置文件
5
python network_monitor.py --config my_config.json

5. 机器学习入门项目#

5.1 机器学习项目 - 图像分类#

项目概述：使用Python和机器学习库构建一个图像分类模型，能够识别不同类别的图像。

技术栈：

Python
TensorFlow/Keras：深度学习框架
NumPy：数值计算
Matplotlib：数据可视化
scikit-learn：机器学习工具

核心代码：

1
import tensorflow as tf
2
from tensorflow.keras import layers, models
3
import numpy as np
4
import matplotlib.pyplot as plt
5
from sklearn.model_selection import train_test_split
6
from tensorflow.keras.preprocessing.image import ImageDataGenerator
7

8
# 加载数据集
9
# 这里使用CIFAR-10数据集作为示例
10
(x_train, y_train), (x_test, y_test) = tf.keras.datasets.cifar10.load_data()
11

12
# 数据预处理
13
x_train = x_train.astype('float32') / 255.0
14
x_test = x_test.astype('float32') / 255.0
15

16
# 标签编码
17
y_train = tf.keras.utils.to_categorical(y_train, 10)
18
y_test = tf.keras.utils.to_categorical(y_test, 10)
19

20
# 数据集标签
21
class_names = ['airplane', 'automobile', 'bird', 'cat', 'deer', 'dog', 'frog', 'horse', 'ship', 'truck']
22

23
# 数据增强
24
datagen = ImageDataGenerator(
25
    rotation_range=15,
26
    width_shift_range=0.1,
27
    height_shift_range=0.1,
28
    horizontal_flip=True,
29
    zoom_range=0.1
30
)
31

32
datagen.fit(x_train)
33

34
# 构建模型
35
model = models.Sequential([
36
    layers.Conv2D(32, (3, 3), activation='relu', input_shape=(32, 32, 3)),
37
    layers.MaxPooling2D((2, 2)),
38
    layers.Conv2D(64, (3, 3), activation='relu'),
39
    layers.MaxPooling2D((2, 2)),
40
    layers.Conv2D(64, (3, 3), activation='relu'),
41
    layers.Flatten(),
42
    layers.Dense(64, activation='relu'),
43
    layers.Dense(10, activation='softmax')
44
])
45

46
# 编译模型
47
model.compile(optimizer='adam',
48
              loss='categorical_crossentropy',
49
              metrics=['accuracy'])
50

51
# 查看模型结构
52
model.summary()
53

54
# 训练模型
55
history = model.fit(datagen.flow(x_train, y_train, batch_size=64),
56
                    epochs=50,
57
                    validation_data=(x_test, y_test))
58

59
# 评估模型
60
test_loss, test_acc = model.evaluate(x_test, y_test, verbose=2)
61
print(f"\n测试准确率：{test_acc:.4f}")
62

63
# 绘制训练过程
64
plt.figure(figsize=(12, 4))
65

66
# 准确率
67
plt.subplot(1, 2, 1)
68
plt.plot(history.history['accuracy'], label='训练准确率')
69
plt.plot(history.history['val_accuracy'], label='验证准确率')
70
plt.title('准确率')
71
plt.xlabel(' epoch')
72
plt.ylabel('准确率')
73
plt.legend()
74

75
# 损失
76
plt.subplot(1, 2, 2)
77
plt.plot(history.history['loss'], label='训练损失')
78
plt.plot(history.history['val_loss'], label='验证损失')
79
plt.title('损失')
80
plt.xlabel('epoch')
81
plt.ylabel('损失')
82
plt.legend()
83

84
plt.show()
85

86
# 预测示例
87
predictions = model.predict(x_test)
88

89
# 显示预测结果
90
plt.figure(figsize=(15, 15))
91
for i in range(25):
92
    plt.subplot(5, 5, i+1)
93
    plt.xticks([])
94
    plt.yticks([])
95
    plt.grid(False)
96
    plt.imshow(x_test[i])
97
    predicted_label = np.argmax(predictions[i])
98
    true_label = np.argmax(y_test[i])
99
    if predicted_label == true_label:
100
        color = 'green'
101
    else:
102
        color = 'red'
103
    plt.xlabel(f"{class_names[predicted_label]} ({class_names[true_label]})", color=color)
104
plt.show()
105

106
# 保存模型
107
model.save('image_classification_model.h5')
108
print("模型已保存为 image_classification_model.h5")

运行项目：

1
# 安装依赖
2
pip install tensorflow numpy matplotlib scikit-learn
3

4
# 运行项目
5
python image_classification.py

6. 部署和运维#

6.1 Docker部署Python应用#

项目概述：使用Docker容器化部署Python应用，提高应用的可移植性和可维护性。

核心代码：

Dockerfile：

1
# 使用官方Python镜像作为基础镜像
2
FROM python:3.9-slim
3

4
# 设置工作目录
5
WORKDIR /app
6

7
# 复制requirements.txt文件
8
COPY requirements.txt .
9

10
# 安装依赖
11
RUN pip install --no-cache-dir -r requirements.txt
12

13
# 复制应用代码
14
COPY . .
15

16
# 暴露端口
17
EXPOSE 5000
18

19
# 运行应用
20
CMD ["python", "app.py"]

docker-compose.yml：

1
version: '3'
2
services:
3
  web:
4
    build: .
5
    ports:
6
      - "5000:5000"
7
    environment:
8
      - FLASK_ENV=production
9
    restart: always

app.py：

1
from flask import Flask
2

3
app = Flask(__name__)
4

5
@app.route('/')
6
def hello():
7
    return "Hello, Docker!"
8

9
if __name__ == '__main__':
10
    app.run(host='0.0.0.0', port=5000)

requirements.txt：

1
Flask

部署步骤：

1
# 构建Docker镜像
2
docker build -t python-app .
3

4
# 运行Docker容器
5
docker run -p 5000:5000 python-app
6

7
# 或使用docker-compose
8
docker-compose up -d
9

10
# 查看运行状态
11
docker ps
12

13
# 访问应用
14
# 打开浏览器访问 http://localhost:5000

6.2 CI/CD流水线配置#

项目概述：配置CI/CD流水线，实现代码提交后自动构建、测试和部署。

GitHub Actions配置：

.github/workflows/ci-cd.yml：

1
name: CI/CD Pipeline
2

3
on:
4
  push:
5
    branches: [ main ]
6
  pull_request:
7
    branches: [ main ]
8

9
jobs:
10
  build:
11
    runs-on: ubuntu-latest
12

13
    steps:
14
    - uses: actions/checkout@v2
15

16
    - name: Set up Python
17
      uses: actions/setup-python@v2
18
      with:
19
        python-version: '3.9'
20

21
    - name: Install dependencies
22
      run: |
23
        python -m pip install --upgrade pip
24
        pip install -r requirements.txt
25
        pip install pytest
26

27
    - name: Run tests
28
      run: |
29
        pytest
30

31
    - name: Build Docker image
32
      run: |
33
        docker build -t python-app .
34

35
    - name: Deploy to server
36
      if: github.event_name == 'push' && github.ref == 'refs/heads/main'
37
      run: |
38
        # 这里可以添加部署脚本，例如使用SSH连接服务器并部署
39
        echo "Deploying to server..."

GitLab CI/CD配置：

.gitlab-ci.yml：

1
stages:
2
  - test
3
  - build
4
  - deploy
5

6
test:
7
  stage: test
8
  image: python:3.9-slim
9
  script:
10
    - pip install -r requirements.txt
11
    - pip install pytest
12
    - pytest
13

14
build:
15
  stage: build
16
  image: docker:latest
17
  services:
18
    - docker:dind
19
  script:
20
    - docker build -t python-app .
21

22
deploy:
23
  stage: deploy
24
  image: alpine:latest
25
  script:
26
    - echo "Deploying to server..."
27
  only:
28
    - main

7. 学习资源推荐#

7.1 Python官方文档#

7.2 B站Python视频（播放量高于50W）#

7.3 推荐书籍#

《Python编程：从入门到实践》
《流畅的Python》
《Python Cookbook》
《Effective Python》
《Python数据科学手册》
《Django实战》
《Flask Web开发》
《Python网络爬虫开发实战》
《Python自动化运维》
《机器学习实战》
《Docker实战》

8. 总结#

本教程介绍了Python的实际应用场景和项目实战，包括：

Web应用开发实战（Flask博客系统、Django电商网站）
数据分析项目实战（销售数据可视化）
爬虫项目实战（电影信息爬取）
自动化脚本开发（文件整理工具、网络监控工具）
机器学习入门项目（图像分类）
部署和运维（Docker部署、CI/CD流水线）

通过本教程的学习，你应该已经掌握了Python的实际应用技能，可以开始开发自己的Python项目了。在实际开发中，你可能会遇到各种问题，但是通过查阅官方文档、参考书籍和在线资源，你一定能够解决这些问题。

Python是一种功能强大、简单易学的编程语言，它的应用领域非常广泛。希望本教程能够帮助你更好地理解和应用Python，让你在编程的道路上越走越远。

祝你学习愉快！

提示：本教程是Python学习的第三阶段，后续将推出Python高级主题教程，包括性能优化、安全编程、分布式系统等内容。

私のブログ

Python项目实战教程（第三阶段）#

1. Web应用开发实战#

1.1 Flask项目实战 - 个人博客系统#

1.2 Django项目实战 - 电商网站#

2. 数据分析项目实战#

2.1 数据分析项目 - 销售数据可视化#

3. 爬虫项目实战#

3.1 爬虫项目 - 电影信息爬取#

4. 自动化脚本开发#

4.1 自动化脚本 - 文件整理工具#

4.2 自动化脚本 - 网络监控工具#

5. 机器学习入门项目#

5.1 机器学习项目 - 图像分类#

6. 部署和运维#

6.1 Docker部署Python应用#

6.2 CI/CD流水线配置#

7. 学习资源推荐#

7.1 Python官方文档#

7.2 B站Python视频（播放量高于50W）#

7.3 推荐书籍#

8. 总结#