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的输出说明安装成功。建议同时安装pipPython包管理工具现代Python版本通常已自带。2.2 开发环境选择对于初学者我推荐以下两种开发环境方案轻量级方案VS Code Python扩展安装VS Code后添加Python扩展配置Python解释器路径CtrlShiftP → Python: Select Interpreter安装Pylance扩展获得更好的代码提示全功能方案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}, timeout5 ) 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.pyimport logging def setup_logging(): logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, filenameapp.log, filemodea )更新main.pyimport 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)}) raise5. 项目打包与分发5.1 使用PyInstaller打包安装打包工具pip install pyinstaller创建打包命令pyinstaller --onefile --name weather_app main.py这将在dist目录生成可执行文件。添加图标pyinstaller --onefile --name weather_app --iconapp.ico main.py5.2 创建安装包使用setuptools创建标准Python包。在项目根目录创建setup.pyfrom setuptools import setup, find_packages setup( nameweather_app, version0.1, packagesfind_packages(), install_requires[ requests2.28.1, ], entry_points{ console_scripts: [ weathermain:main, ], }, )然后可以安装到系统pip install .安装后可直接通过weather命令运行程序。6. 常见问题与解决方案6.1 网络请求失败排查代理问题检查系统代理设置尝试直接使用IP地址访问使用curl测试连通性超时设置response requests.get(url, timeout(3.05, 27))重试机制from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry session requests.Session() retry Retry(total3, backoff_factor1) adapter HTTPAdapter(max_retriesretry) session.mount(http://, adapter) session.mount(https://, adapter)6.2 打包时的常见错误模块找不到使用--hidden-import指定隐式依赖检查虚拟环境是否激活文件路径问题使用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)杀毒软件误报添加数字签名在打包前扫描病毒向杀毒软件厂商提交误报申请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(pady5) self.city_entry tk.Entry(self.root, width30) self.city_entry.pack(pady5) tk.Button( self.root, text查询, commandself.query_weather ).pack(pady10) self.result_label tk.Label(self.root, text) self.result_label.pack(pady10) def query_weather(self): city self.city_entry.get() if not city: messagebox.showwarning(提示, 请输入城市名称) return weather get_weather(city) self.result_label.config(textf{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(limit10): 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 results7.3 添加单元测试创建tests/test_helpers.pyimport 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 tests8. 性能优化技巧8.1 缓存机制实现使用functools缓存天气数据from functools import lru_cache import time lru_cache(maxsize32) 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}, timeout5 ) 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 fhttp://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(host0.0.0.0, port5000)9.2 Docker容器化创建DockerfileFROM 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-app9.3 云服务部署以AWS Elastic Beanstalk为例安装EB CLIpip install awsebcli初始化EB应用eb init -p python-3.10 weather-app --region us-west-2创建环境并部署eb create weather-env eb deploy10. 项目文档编写良好的文档是项目成功的关键。创建完整的文档结构docs/ ├── README.md # 项目概述 ├── INSTALLATION.md # 安装指南 ├── USAGE.md # 使用说明 ├── DEVELOPMENT.md # 开发指南 └── CHANGELOG.md # 版本变更示例README.md内容# 天气查询应用 一个简单的命令行天气查询工具可以查询全球主要城市的当前天气情况。 ## 功能特性 - 查询指定城市天气 - 支持中英文城市名称 - 简洁的命令行界面 - 可扩展的架构设计 ## 快速开始 1. 安装依赖 bash pip install -r requirements.txt运行程序python main.py按照提示输入城市名称查询天气贡献指南欢迎提交Pull Request。请确保代码符合PEP8规范包含相应的单元测试更新相关文档## 11. 代码质量保障 ### 11.1 静态代码检查 使用flake8进行代码风格检查 bash pip install flake8 flake8 . --count --selectE9,F63,F7,F82 --show-source --statistics11.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.ymlname: Python application on: [push, pull_request] jobs: build: runs-on: ubuntu-latest steps: - uses: actions/checkoutv2 - name: Set up Python uses: actions/setup-pythonv2 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 --selectE9,F63,F7,F82 --show-source --statistics - name: Run type checking run: | mypy . - name: Test with pytest run: | python -m pytest12. 项目发布与维护12.1 PyPI发布更新setup.py添加项目信息setup( nameweather-cli, version0.1.0, descriptionA simple weather query tool, long_descriptionopen(README.md).read(), authorYour Name, author_emailyour.emailexample.com, urlhttps://github.com/yourname/weather-cli, classifiers[ Programming Language :: Python :: 3, License :: OSI Approved :: MIT License, ], # ...其他配置... )构建并上传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 --tags12.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管理敏感配置安装依赖pip install python-dotenv创建.env文件API_KEYyour_secret_key_here DEBUGFalse在代码中加载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_factorystructlog.PrintLoggerFactory() ) log structlog.get_logger() log.info(query_received, citycity, iprequest.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的获取逻辑 pass15.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(fplugins.{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(parentsTrue, exist_okTrue) # 配置文件路径 config_path data_dir / config.ini print(f配置文件位于: {config_path})16.2 系统服务集成Linux系统服务示例/etc/systemd/system/weather.service[Unit] DescriptionWeather Query Service Afternetwork.target [Service] Userweather WorkingDirectory/opt/weather_app ExecStart/usr/bin/python3 /opt/weather_app/main.py Restartalways [Install] WantedBymulti-user.targetWindows服务示例使用pywin32import 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, localedirlocales_dir, languages[lang] ) _ translation.gettext except FileNotFoundError: _ gettext.gettext print(_(Welcome to Weather App))创建翻译文件locales/zh/LC_MESSAGES/weather.pomsgid Welcome to Weather App msgstr 欢迎使用天气查询应用编译翻译文件msgfmt weather.po -o weather.mo17.2 时区处理使用pytz处理时区问题from datetime import datetime import pytz def get_local_time(city_timezone): utc_time datetime.utcnow().replace(tzinfopytz.utc) try: tz pytz.timezone(city_timezone) return utc_time.astimezone(tz) except pytz.UnknownTimeZoneError: return utc_time18. 用户界面增强18.1 彩色终端输出使用colorama添加颜色from colorama import init, Fore, Back, Style init(autoresetTrue) 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 results19. 数据持久化方案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_keyTrue) city Column(String(50), nullableFalse) weather Column(String(100)) query_time Column(DateTime, defaultdatetime.now) engine create_engine(sqlite:///weather.db) Base.metadata.create_all(engine) Session sessionmaker(bindengine) def save_record(city, weather): session Session() record WeatherRecord(citycity, weatherweather) session.add(record) session.commit() session.close()19.2 使用Redis缓存实现Redis缓存层import redis import pickle from datetime import timedelta r redis.Redis(hostlocalhost, port6379, db0) def cache_weather(city, weather, expire3600): r.setex( fweather:{city}, timedelta(secondsexpire), pickle.dumps(weather) ) def get_cached_weather(city): data r.get(fweather:{city}) return pickle.loads(data) if data else None20. 项目总结与进阶建议经过这个完整的Python小练习你应该已经掌握了Python项目的基本结构和组织方式第三方库的使用和依赖管理异常处理和日志记录的最佳实践代码测试和质量保障方法项目打包和部署的基本流程为了进一步提升Python技能建议阅读优秀的开源项目代码如Requests、Flask参与开源项目贡献学习设计模式和架构原则探索Python在数据科学、Web开发等领域的应用记住编程能力的提升关键在于持续实践。可以尝试扩展这个天气项目比如添加天气预报、天气预警、历史数据查询等功能或者将其改造成Web应用、桌面应用等不同形态。