Python实战：从零开发天气查询工具

1. Python小练习007：从零开始的实战编程指南

Python作为当下最受欢迎的编程语言之一，其简洁的语法和强大的功能吸引了无数初学者。这个小练习系列特别适合刚入门的新手，通过循序渐进的实战项目来掌握Python核心概念。今天我们要完成的007号练习，将带你体验一个完整的Python项目开发流程。

2. 环境准备与基础配置

2.1 Python安装与验证

首先确保你的系统已安装Python 3.6或更高版本。访问python.org下载最新稳定版，推荐使用3.10+版本以获得最佳体验。安装完成后，在终端运行：

python --version # 或 python3 --version

如果看到类似"Python 3.10.8"的输出，说明安装成功。建议同时安装pip（Python包管理工具），现代Python版本通常已自带。

2.2 开发环境选择

对于初学者，我推荐以下两种开发环境方案：

1. 轻量级方案：VS Code + Python扩展

   • 安装VS Code后添加Python扩展
   • 配置Python解释器路径（Ctrl+Shift+P → Python: Select Interpreter）
   • 安装Pylance扩展获得更好的代码提示

2. 全功能方案：PyCharm Community版

   • 专为Python开发设计的IDE
   • 内置代码补全、调试器和虚拟环境管理
   • 适合中大型项目开发

提示：无论选择哪种环境，都建议创建虚拟环境隔离项目依赖。使用命令：python -m venv venv，然后激活虚拟环境。

3. 项目结构与核心代码实现

3.1 基础项目框架

我们先创建一个标准的Python项目结构：

python007/ ├── main.py # 主程序入口 ├── utils/ # 工具函数目录 │ └── helpers.py # 辅助函数 ├── requirements.txt # 依赖列表 └── README.md # 项目说明

在main.py中我们先写入基础代码框架：

def main(): print("Python练习007项目启动") # 在这里添加你的代码逻辑 if __name__ == "__main__": main()

3.2 核心功能实现

本次练习我们将实现一个简易的天气查询工具。首先安装所需依赖：

pip install requests

然后在utils/helpers.py中添加网络请求功能：

import requests def get_weather(city): """获取指定城市天气信息""" base_url = "http://wttr.in/{}?format=%C+%t" try: response = requests.get(base_url.format(city)) return response.text except Exception as e: return f"获取天气失败: {str(e)}"

更新main.py实现完整功能：

from utils.helpers import get_weather def main(): print("=== 简易天气查询 ===") while True: city = input("请输入城市名称(输入q退出): ") if city.lower() == 'q': break weather = get_weather(city) print(f"{city}的天气: {weather}") print("程序结束") if __name__ == "__main__": main()

4. 功能扩展与优化

4.1 添加异常处理

完善helpers.py中的错误处理机制：

def get_weather(city): """获取指定城市天气信息""" base_url = "http://wttr.in/{}?format=%C+%t" try: response = requests.get( base_url.format(city), headers={'User-Agent': 'curl/7.64.1'}, timeout=5 ) response.raise_for_status() return response.text.strip() except requests.exceptions.RequestException as e: return f"获取天气失败: 请检查网络连接或城市名称" except Exception as e: return f"发生未知错误: {str(e)}"

4.2 添加日志记录

在项目根目录创建config.py：

import logging def setup_logging(): logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', filename='app.log', filemode='a' )

更新main.py：

import logging from config import setup_logging def main(): setup_logging() logger = logging.getLogger(__name__) logger.info("程序启动") # ...原有代码... logger.info("程序正常退出") if __name__ == "__main__": try: main() except Exception as e: logging.error(f"程序异常终止: {str(e)}") raise

5. 项目打包与分发

5.1 使用PyInstaller打包

安装打包工具：

pip install pyinstaller

创建打包命令：

pyinstaller --onefile --name weather_app main.py

这将在dist目录生成可执行文件。添加图标：

pyinstaller --onefile --name weather_app --icon=app.ico main.py

5.2 创建安装包

使用setuptools创建标准Python包。在项目根目录创建setup.py：

from setuptools import setup, find_packages setup( name="weather_app", version="0.1", packages=find_packages(), install_requires=[ 'requests>=2.28.1', ], entry_points={ 'console_scripts': [ 'weather=main:main', ], }, )

然后可以安装到系统：

pip install .

安装后可直接通过weather命令运行程序。

6. 常见问题与解决方案

6.1 网络请求失败排查

1. 代理问题：

   • 检查系统代理设置
   • 尝试直接使用IP地址访问
   • 使用curl测试连通性

2. 超时设置：

   response = requests.get(url, timeout=(3.05, 27))

3. 重试机制：

   from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry session = requests.Session() retry = Retry(total=3, backoff_factor=1) adapter = HTTPAdapter(max_retries=retry) session.mount('http://', adapter) session.mount('https://', adapter)

6.2 打包时的常见错误

1. 模块找不到：

   • 使用--hidden-import指定隐式依赖
   • 检查虚拟环境是否激活

2. 文件路径问题：

   • 使用sys._MEIPASS处理资源文件路径
   • 在代码中添加路径检查：

   import sys import os def resource_path(relative_path): if hasattr(sys, '_MEIPASS'): return os.path.join(sys._MEIPASS, relative_path) return os.path.join(os.path.abspath("."), relative_path)

3. 杀毒软件误报：

   • 添加数字签名
   • 在打包前扫描病毒
   • 向杀毒软件厂商提交误报申请

7. 项目进阶方向

7.1 添加GUI界面

使用Tkinter创建简单界面：

import tkinter as tk from tkinter import messagebox from utils.helpers import get_weather class WeatherApp: def __init__(self, root): self.root = root self.root.title("天气查询") self.setup_ui() def setup_ui(self): tk.Label(self.root, text="城市名称:").pack(pady=5) self.city_entry = tk.Entry(self.root, width=30) self.city_entry.pack(pady=5) tk.Button( self.root, text="查询", command=self.query_weather ).pack(pady=10) self.result_label = tk.Label(self.root, text="") self.result_label.pack(pady=10) def query_weather(self): city = self.city_entry.get() if not city: messagebox.showwarning("提示", "请输入城市名称") return weather = get_weather(city) self.result_label.config(text=f"{city}天气: {weather}") if __name__ == "__main__": root = tk.Tk() app = WeatherApp(root) root.mainloop()

7.2 添加数据库支持

使用SQLite存储查询历史：

import sqlite3 from datetime import datetime def init_db(): conn = sqlite3.connect('weather.db') c = conn.cursor() c.execute('''CREATE TABLE IF NOT EXISTS queries (id INTEGER PRIMARY KEY AUTOINCREMENT, city TEXT, weather TEXT, query_time TIMESTAMP)''') conn.commit() conn.close() def save_query(city, weather): conn = sqlite3.connect('weather.db') c = conn.cursor() c.execute("INSERT INTO queries (city, weather, query_time) VALUES (?, ?, ?)", (city, weather, datetime.now())) conn.commit() conn.close() def get_history(limit=10): conn = sqlite3.connect('weather.db') c = conn.cursor() c.execute("SELECT * FROM queries ORDER BY query_time DESC LIMIT ?", (limit,)) results = c.fetchall() conn.close() return results

7.3 添加单元测试

创建tests/test_helpers.py：

import unittest from unittest.mock import patch from utils.helpers import get_weather class TestWeatherFunctions(unittest.TestCase): @patch('utils.helpers.requests.get') def test_get_weather_success(self, mock_get): mock_get.return_value.status_code = 200 mock_get.return_value.text = "Sunny +25°C" result = get_weather("Beijing") self.assertEqual(result, "Sunny +25°C") @patch('utils.helpers.requests.get') def test_get_weather_failure(self, mock_get): mock_get.side_effect = Exception("Network error") result = get_weather("Beijing") self.assertTrue("获取天气失败" in result) if __name__ == '__main__': unittest.main()

运行测试：

python -m unittest discover -s tests

8. 性能优化技巧

8.1 缓存机制实现

使用functools缓存天气数据：

from functools import lru_cache import time @lru_cache(maxsize=32) def get_weather(city): """带缓存的天气查询""" base_url = "http://wttr.in/{}?format=%C+%t" try: response = requests.get( base_url.format(city), headers={'User-Agent': 'curl/7.64.1'}, timeout=5 ) response.raise_for_status() return response.text.strip() except requests.exceptions.RequestException: return "获取天气失败"

8.2 异步请求优化

使用aiohttp实现异步请求：

import aiohttp import asyncio async def fetch_weather(session, city): url = f"http://wttr.in/{city}?format=%C+%t" try: async with session.get(url) as response: return await response.text() except Exception: return "获取天气失败" async def get_weather(cities): async with aiohttp.ClientSession() as session: tasks = [fetch_weather(session, city) for city in cities] return await asyncio.gather(*tasks) # 使用示例 cities = ["Beijing", "Shanghai", "Guangzhou"] results = asyncio.run(get_weather(cities))

8.3 内存优化技巧

对于大型数据处理，使用生成器替代列表：

def process_large_data(): """使用生成器节省内存""" for i in range(1000000): yield i * 2 # 每次只生成一个值，不占用全部内存 # 使用示例 for item in process_large_data(): print(item) # 处理每个项目

9. 项目部署方案

9.1 本地服务化

使用Flask创建Web服务：

from flask import Flask, request, jsonify from utils.helpers import get_weather app = Flask(__name__) @app.route('/weather') def weather_api(): city = request.args.get('city', '') if not city: return jsonify({"error": "Missing city parameter"}), 400 weather = get_weather(city) return jsonify({"city": city, "weather": weather}) if __name__ == '__main__': app.run(host='0.0.0.0', port=5000)

9.2 Docker容器化

创建Dockerfile：

FROM python:3.10-slim WORKDIR /app COPY . . RUN pip install --no-cache-dir -r requirements.txt EXPOSE 5000 CMD ["python", "app.py"]

构建并运行：

docker build -t weather-app . docker run -p 5000:5000 weather-app

9.3 云服务部署

以AWS Elastic Beanstalk为例：

1. 安装EB CLI：

   pip install awsebcli

2. 初始化EB应用：

   eb init -p python-3.10 weather-app --region us-west-2

3. 创建环境并部署：

   eb create weather-env eb deploy

10. 项目文档编写

良好的文档是项目成功的关键。创建完整的文档结构：

docs/ ├── README.md # 项目概述 ├── INSTALLATION.md # 安装指南 ├── USAGE.md # 使用说明 ├── DEVELOPMENT.md # 开发指南 └── CHANGELOG.md # 版本变更

示例README.md内容：

# 天气查询应用 一个简单的命令行天气查询工具，可以查询全球主要城市的当前天气情况。 ## 功能特性 - 查询指定城市天气 - 支持中英文城市名称 - 简洁的命令行界面 - 可扩展的架构设计 ## 快速开始 1. 安装依赖： ```bash pip install -r requirements.txt

2. 运行程序：

   python main.py

3. 按照提示输入城市名称查询天气

贡献指南

欢迎提交Pull Request。请确保：

• 代码符合PEP8规范
• 包含相应的单元测试
• 更新相关文档

## 11. 代码质量保障 ### 11.1 静态代码检查 使用flake8进行代码风格检查： ```bash pip install flake8 flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics

11.2 类型检查

添加类型注解并使用mypy检查：

from typing import Optional def get_weather(city: str) -> str: """获取指定城市天气信息""" base_url: str = "http://wttr.in/{}?format=%C+%t" try: response = requests.get(base_url.format(city)) return response.text.strip() except Exception as e: return f"获取天气失败: {str(e)}"

运行检查：

pip install mypy mypy .

11.3 自动化测试

配置GitHub Actions实现CI/CD。创建.github/workflows/python-app.yml：

name: Python application on: [push, pull_request] jobs: build: runs-on: ubuntu-latest steps: - uses: actions/checkout@v2 - name: Set up Python uses: actions/setup-python@v2 with: python-version: '3.10' - name: Install dependencies run: | python -m pip install --upgrade pip pip install -r requirements.txt pip install flake8 mypy pytest - name: Lint with flake8 run: | flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics - name: Run type checking run: | mypy . - name: Test with pytest run: | python -m pytest

12. 项目发布与维护

12.1 PyPI发布

1. 更新setup.py添加项目信息：

   setup( name="weather-cli", version="0.1.0", description="A simple weather query tool", long_description=open("README.md").read(), author="Your Name", author_email="your.email@example.com", url="https://github.com/yourname/weather-cli", classifiers=[ "Programming Language :: Python :: 3", "License :: OSI Approved :: MIT License", ], # ...其他配置... )

2. 构建并上传：

   pip install twine python setup.py sdist bdist_wheel twine upload dist/*

12.2 版本管理策略

采用语义化版本控制：

• MAJOR版本：不兼容的API修改
• MINOR版本：向下兼容的功能新增
• PATCH版本：向下兼容的问题修正

使用bump2version自动化版本管理：

pip install bump2version bump2version patch # 或 minor/major git push --tags

12.3 用户反馈处理

建立标准的issue模板（.github/ISSUE_TEMPLATE/bug_report.md）：

--- name: Bug报告 about: 报告项目中的问题 title: '' labels: bug assignees: '' --- **描述问题** 清晰简洁地描述问题是什么 **重现步骤** 重现问题的步骤： 1. 执行 '...' 2. 查看 '....' 3. 出现错误 '....' **预期行为** 你期望发生什么 **截图** 如果有的话，添加截图帮助说明问题 **环境信息:** - 操作系统: [如 Windows 10] - Python版本: [如 3.10.4] - 项目版本: [如 0.1.0] **附加信息** 任何其他关于问题的信息

13. 项目安全考虑

13.1 输入验证

对所有用户输入进行严格验证：

import re def validate_city(city: str) -> bool: """验证城市名称是否合法""" if not city or len(city) > 50: return False return bool(re.match(r'^[\w\s\-\.]+$', city))

13.2 敏感信息保护

使用python-dotenv管理敏感配置：

1. 安装依赖：

   pip install python-dotenv

2. 创建.env文件：

   API_KEY=your_secret_key_here DEBUG=False

3. 在代码中加载：

   from dotenv import load_dotenv import os load_dotenv() api_key = os.getenv("API_KEY")

13.3 依赖安全扫描

使用safety检查依赖漏洞：

pip install safety safety check --full-report

考虑添加到CI流程中自动检查。

14. 项目监控与日志

14.1 结构化日志

使用structlog增强日志：

import structlog structlog.configure( processors=[ structlog.processors.JSONRenderer() ], logger_factory=structlog.PrintLoggerFactory() ) log = structlog.get_logger() log.info("query_received", city=city, ip=request.remote_addr)

14.2 性能监控

使用Prometheus客户端监控性能：

from prometheus_client import start_http_server, Counter QUERY_COUNTER = Counter('weather_queries', 'Number of weather queries') def get_weather(city): QUERY_COUNTER.inc() # ...原有代码...

启动监控服务器：

start_http_server(8000)

15. 项目扩展思路

15.1 多数据源支持

抽象天气数据源接口：

from abc import ABC, abstractmethod class WeatherDataSource(ABC): @abstractmethod def get_weather(self, city: str) -> str: pass class WttrInSource(WeatherDataSource): def get_weather(self, city): # 实现wttr.in的获取逻辑 pass class OpenWeatherMapSource(WeatherDataSource): def __init__(self, api_key): self.api_key = api_key def get_weather(self, city): # 实现OpenWeatherMap的获取逻辑 pass

15.2 插件系统设计

实现简单的插件架构：

# plugins/__init__.py import importlib import pkgutil from pathlib import Path PLUGINS = {} def load_plugins(): plugin_dir = Path(__file__).parent for finder, name, _ in pkgutil.iter_modules([str(plugin_dir)]): if name.startswith('plugin_'): module = importlib.import_module(f"plugins.{name}") if hasattr(module, 'register'): module.register(PLUGINS) # plugin示例 plugins/plugin_forecast.py def register(plugins): plugins['forecast'] = get_forecast def get_forecast(city): return "3天预报功能待实现"

15.3 机器学习集成

使用scikit-learn添加天气预测：

import pandas as pd from sklearn.linear_model import LinearRegression def train_weather_model(): # 示例数据 - 实际应用中应从历史数据加载 data = { 'temperature': [25, 28, 30, 22, 20], 'humidity': [60, 65, 70, 55, 50], 'pressure': [1010, 1012, 1011, 1009, 1008] } df = pd.DataFrame(data) X = df[['humidity', 'pressure']] y = df['temperature'] model = LinearRegression() model.fit(X, y) return model def predict_weather(model, humidity, pressure): return model.predict([[humidity, pressure]])[0]

16. 跨平台兼容性

16.1 路径处理最佳实践

使用pathlib处理文件路径：

from pathlib import Path # 创建数据目录 data_dir = Path.home() / ".weather_app" / "data" data_dir.mkdir(parents=True, exist_ok=True) # 配置文件路径 config_path = data_dir / "config.ini" print(f"配置文件位于: {config_path}")

16.2 系统服务集成

Linux系统服务示例（/etc/systemd/system/weather.service）：

[Unit] Description=Weather Query Service After=network.target [Service] User=weather WorkingDirectory=/opt/weather_app ExecStart=/usr/bin/python3 /opt/weather_app/main.py Restart=always [Install] WantedBy=multi-user.target

Windows服务示例（使用pywin32）：

import win32serviceutil import win32service import win32event import servicemanager class WeatherService(win32serviceutil.ServiceFramework): _svc_name_ = "WeatherService" _svc_display_name_ = "Weather Query Service" def __init__(self, args): win32serviceutil.ServiceFramework.__init__(self, args) self.hWaitStop = win32event.CreateEvent(None, 0, 0, None) def SvcStop(self): self.ReportServiceStatus(win32service.SERVICE_STOP_PENDING) win32event.SetEvent(self.hWaitStop) def SvcDoCommand(self): servicemanager.LogMsg( servicemanager.EVENTLOG_INFORMATION_TYPE, servicemanager.PYS_SERVICE_STARTED, (self._svc_name_, '') ) self.main() def main(self): from main import main main() if __name__ == '__main__': win32serviceutil.HandleCommandLine(WeatherService)

17. 国际化支持

17.1 多语言实现

使用gettext实现国际化：

import gettext import locale from pathlib import Path locales_dir = Path(__file__).parent / "locales" lang = locale.getdefaultlocale()[0][:2] try: translation = gettext.translation( 'weather', localedir=locales_dir, languages=[lang] ) _ = translation.gettext except FileNotFoundError: _ = gettext.gettext print(_("Welcome to Weather App"))

创建翻译文件（locales/zh/LC_MESSAGES/weather.po）：

msgid "Welcome to Weather App" msgstr "欢迎使用天气查询应用"

编译翻译文件：

msgfmt weather.po -o weather.mo

17.2 时区处理

使用pytz处理时区问题：

from datetime import datetime import pytz def get_local_time(city_timezone): utc_time = datetime.utcnow().replace(tzinfo=pytz.utc) try: tz = pytz.timezone(city_timezone) return utc_time.astimezone(tz) except pytz.UnknownTimeZoneError: return utc_time

18. 用户界面增强

18.1 彩色终端输出

使用colorama添加颜色：

from colorama import init, Fore, Back, Style init(autoreset=True) def print_weather(city, weather): if "晴" in weather: color = Fore.YELLOW elif "雨" in weather: color = Fore.BLUE else: color = Fore.WHITE print(f"{city}天气: {color}{weather}{Style.RESET_ALL}")

18.2 进度显示

使用tqdm显示进度条：

from tqdm import tqdm import time def batch_query(cities): results = {} for city in tqdm(cities, desc="查询进度"): results[city] = get_weather(city) time.sleep(0.5) # 避免请求过快 return results

19. 数据持久化方案

19.1 使用SQLAlchemy ORM

定义天气数据模型：

from sqlalchemy import create_engine, Column, String, DateTime from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker Base = declarative_base() class WeatherRecord(Base): __tablename__ = 'weather_records' id = Column(Integer, primary_key=True) city = Column(String(50), nullable=False) weather = Column(String(100)) query_time = Column(DateTime, default=datetime.now) engine = create_engine('sqlite:///weather.db') Base.metadata.create_all(engine) Session = sessionmaker(bind=engine) def save_record(city, weather): session = Session() record = WeatherRecord(city=city, weather=weather) session.add(record) session.commit() session.close()

19.2 使用Redis缓存

实现Redis缓存层：

import redis import pickle from datetime import timedelta r = redis.Redis(host='localhost', port=6379, db=0) def cache_weather(city, weather, expire=3600): r.setex( f"weather:{city}", timedelta(seconds=expire), pickle.dumps(weather) ) def get_cached_weather(city): data = r.get(f"weather:{city}") return pickle.loads(data) if data else None

20. 项目总结与进阶建议

经过这个完整的Python小练习，你应该已经掌握了：

1. Python项目的基本结构和组织方式
2. 第三方库的使用和依赖管理
3. 异常处理和日志记录的最佳实践
4. 代码测试和质量保障方法
5. 项目打包和部署的基本流程

为了进一步提升Python技能，建议：

• 阅读优秀的开源项目代码（如Requests、Flask）
• 参与开源项目贡献
• 学习设计模式和架构原则
• 探索Python在数据科学、Web开发等领域的应用

记住，编程能力的提升关键在于持续实践。可以尝试扩展这个天气项目，比如添加天气预报、天气预警、历史数据查询等功能，或者将其改造成Web应用、桌面应用等不同形态。