大家好，5月份的技术汇总贴它虽迟但到。本次带来了关于MYSQL,Linux,puthon,Java,Vue,GO语言,Android等等多方面知识，量大管饱，希望可以帮助到大家。1.Mysql之存储过程-转载https://bbs.huaweicloud.com/forum/thread-0290181146500803029-1-1.html2.Linux 指令初探：开启终端世界的大门-转载https://bbs.huaweicloud.com/forum/thread-0254181146580042030-1-1.html3.Flink 的 source、transformations、sink的详细示例-转载https://bbs.huaweicloud.com/forum/thread-0290181146683326030-1-1.html4.vue 实现纯 web H5 打印功能https://bbs.huaweicloud.com/forum/thread-0278181376093532038-1-1.html5.Java内存分析全面指南https://bbs.huaweicloud.com/forum/thread-0221182242802337058-1-1.html6.Go语言中获取函数实体的四种方法小结【转】https://bbs.huaweicloud.com/forum/thread-0290183611684337158-1-1.html7.Python numpy数组元素的访问及数组的增删改查功能详解【转】https://bbs.huaweicloud.com/forum/thread-0259183611770450178-1-1.html8.SpringMVC 通过ajax 实现文件上传的步骤【转】https://bbs.huaweicloud.com/forum/thread-0261183611880824110-1-1.html9.SpringMVC 如何使用thymeleaf 进行数据展示【转】https://bbs.huaweicloud.com/forum/thread-0275183611920676184-1-1.html10.Maven Profile高级策略与冲突解决方案【转】https://bbs.huaweicloud.com/forum/thread-0261183611970508111-1-1.html11.IDEA中为SpringBoot项目接入MySQL数据库的详细指南【转】https://bbs.huaweicloud.com/forum/thread-0221183612078346159-1-1.html12.判断Android APP使用了Tauri 这个跨端框架的操作方法【转】https://bbs.huaweicloud.com/forum/thread-0221183612125526160-1-1.html13.git如何查看提交行数、删除行数【转】https://bbs.huaweicloud.com/forum/thread-0275183612159112185-1-1.html14.Python自动化之pynput键盘鼠标操作全攻略【转】https://bbs.huaweicloud.com/forum/thread-0278183612235224170-1-1.html15.Python结合DeepSeek实现自动生成测试用例的示例代码【转】https://bbs.huaweicloud.com/forum/thread-0261183612276691112-1-1.html16.使用Python开发一个Word文档批量合并工具【转】https://bbs.huaweicloud.com/forum/thread-0275183612333980186-1-1.html17.AJAX get请求提交数据的具体步骤【转】https://bbs.huaweicloud.com/forum/thread-0261183612688189113-1-1.html18.mybatis in foreach双层嵌套的问题解决【转】https://bbs.huaweicloud.com/forum/thread-0261183612884484114-1-1.html19.MyBatis防止SQL注入攻击的有效方法【转】https://bbs.huaweicloud.com/forum/thread-0278183612934320172-1-1.html20.Java内存区域与内存溢出异常分析与解决【转】https://bbs.huaweicloud.com/forum/thread-0221183612977820161-1-1.html

基于华为开发者空间使用 FastAPI 构建 MCP 天气查询服务案例介绍本案例将指导您在华为云开发者空间中使用 FastAPI 框架构建一个功能完整的 MCP（Model Context Protocol）天气查询服务。通过本案例，您将学会如何将传统的 REST API 转换为 MCP 工具，让 AI 模型能够实时获取天气信息并与用户进行智能交互。技术栈后端框架: FastAPI协议标准: MCP (Model Context Protocol)天气数据: 和风天气 API认证方式: JWT (JSON Web Token)客户端: Cherry Studio部署环境: 华为云开发者空间学习目标掌握 MCP 协议的基本概念和实现方式实现 AI 模型与外部服务的实时数据交互体验华为云开发者空间的开发便利性以及 Maas DeepSeek 模型服务的无缝集成案例内容1. 概述1.1 案例背景在传统的 AI 对话系统中，模型只能基于训练数据回答问题，无法获取实时信息。MCP 协议的出现解决了这一痛点，它允许 AI 模型通过标准化的接口调用外部工具和服务，获取实时数据。本案例选择天气查询作为示例，因为：实时性强: 天气数据需要实时获取，体现 MCP 的价值应用广泛: 天气查询是用户常见需求技术完整: 涵盖 API 调用、数据处理、错误处理等完整流程易于扩展: 可以基于此案例扩展更多功能1.2 适用对象企业用户：关注业务自动化、提升开发效率、集成AI能力，适合希望快速构建智能工具、提升企业数字化水平的团队。个人开发者：具备一定Python基础(或能借助 AI 辅助编程)，关注技术创新与个人成长，适合希望掌握MCP协议、API开发与AI集成的开发者。高校学生：有编程基础(或能借助 AI 辅助编程)，关注AI应用实践与创新能力培养，适合希望参与AI+云平台项目实训、提升综合开发能力的学生群体。1.3 案例时间准备阶段: 30 分钟（环境配置、账号注册）开发阶段: 20 分钟（代码编写、调试）测试阶段: 10 分钟（功能验证、MCP 集成）总计时间: 约 60 分钟1.4 案例流程环境准备API 注册项目初始化依赖安装核心代码开发JWT 认证实现天气接口开发MCP 集成本地测试客户端配置端到端验证1.5 资源总览本案例预计花费0元。资源类型名称/说明免费额度/价格备注开发资源华为云开发者空间0元免费注册使用Maas DeepSeek 模型服务0元免费额度，按需扩展和风天气开发者账号0元免费额度，1000次/天Cherry Studio 客户端0元开源免费API 配额和风天气免费版0元1000次/天Maas DeepSeek 模型服务0元免费额度注意事项：本案例基于免费额度，如需更高配额请自行申请。API 可能存在调用次数限制，超出免费额度需付费或申请扩容。2. 环境配置2.1 开发者空间配置步骤 1: 登录华为云领取免费云主机访问 华为云官网，注册并登录华为云账号点击“开发者->开发者空间”，进入。如果没有领取过云主机可以点击"立即体验"进行开通。步骤 2: 配置云主机访问开发者空间工作台, 点击"配置云主机"配置基础资源： CPU架构为 ARM，规格为 4 vCPUs 8 GB，操作系统为 Ubuntu ，镜像为 Ubuntu 24.04 Server定制版等待环境初始化完成后点击"打开云主机-进入桌面"，打开云主机桌面2.2 MaaS 领取 DeepSeek步骤 1: 访问华为云 MaaS 服务在华为云控制台搜索"MaaS"，点击进入 Maas 控制台区域选择西南-贵阳一，在左侧菜单栏，选择在线推理 > 预置服务 > 免费服务，选择DeepSeek-R1-32K模型，点击领取额度，领取200万免费token。步骤 2: 获取API地址和模型名称领取后点击调用说明*，可以获取到对应的API地址、模型名称。步骤 3: 获取 API 密钥点击“API Key管理-创建API Key“”，自定义标签和描述，点击确定创建API Key。（⚠️注意：API Key仅会在新建后显示一次，请及时复制并妥善保存，若API Key丢失，请新建API Key。）至此，我们获取到了所需的 API 地址、模型名称和 API 密钥。2.3 安装 Cherry Studio下载安装:使用云主机中的Firefox访问 [Cherry Studio 官网] https://cherry-studio.com根据操作系统下载对应版本，如 **Cherry-Studio-1.3.12-arm64.AppImage给下载的文件添加执行权限chmod +x Cherry-Studio-1.3.12-arm64.AppImage安装运行相关依赖sudo apt update sudo apt install zlib1g zlib1g-dev sudo add-apt-repository universe sudo apt install libfuse2运行安装包./Cherry-Studio-1.3.12-arm64.AppImage --no-sandbox基础配置:配置 DeepSeek 模型连接测试模型连接是否正常2.4 和风天气 API 注册（可选）💡 提示: 教程代码中已包含测试用的 API 配置，如无法使用请自行注册注册流程:访问 和风天气开发者平台 https://id.qweather.com/注册开发者账号并完成邮箱验证进入开发服务控制台创建项目并获取必要的认证信息获取配置信息:PROJECT_ID: 项目标识符KEY_ID: JWT 密钥标识PRIVATE_KEY: JWT 签名私钥WEATHER_API_HOST: API 服务地址访问[https://id.qweather.com/] https://id.qweather.com/ 注册账号,初次使用需绑定手机号和邮箱。注册成功后，点击菜单中的 开发服务控制台，进入开发控制台。点击项目管理-创建项目，创建一个项目，如 fastapi， 创建之后此步骤获得PROJECT_ID。点击项目如fastapi，进入项目详情页并点击创建凭据开始创建 JWT 凭据，需先在本地生成公私钥对再创建，创建之后此步骤获得PRIVATE_KEY和KEY_ID。（PS: 虽然 APIKey 更简单，但推荐使用JSON Web Token (JWT)的认证方式获得更高等级的安全性以及不受限的API请求）点击设置, 从开发者信息中获取 API Host，也就是WEATHER_API_HOST至此，我们天气查询 MCP 中需要用到的配置信息已经全部获取。3. 项目搭建3.1 项目初始化使用CodeArts IDE for Python创建项目工程或者使用脚本创建项目目录:mkdir qweather-mcp cd qweather-mcp安装依赖包:pip install fastapi uvicorn fastapi-mcp pyjwt cryptography requests -i https://repo.huaweicloud.com/repository/pypi/simple创建项目结构:qweather-mcp/ ├── main.py # 主程序文件 ├── requirements.txt # 依赖列表 ├── config.py # 配置文件（可选） └── README.md # 项目说明3.2 核心代码实现详细的代码实现包括：FastAPI 应用初始化JWT 认证机制天气 API 封装MCP 协议集成错误处理机制3.3 关键功能模块JWT 认证模块: 实现安全的 API 访问控制天气查询模块: 封装和风天气 API 调用MCP 集成模块: 将 REST API 转换为 MCP 工具数据模型模块: 定义请求和响应的数据结构3.4 代码实现我们创建 main.py 文件，并编写代码。3.4.1 引入依赖from fastapi import FastAPI, HTTPException, Depends, Query, status from fastapi.security import HTTPBearer from pydantic import BaseModel, Field from typing import List, Optional, Dict, Any import jwt import time import requests from datetime import datetime import base64 from cryptography.hazmat.primitives import serialization from cryptography.hazmat.backends import default_backend # 引入 fastapi_mcp from fastapi_mcp import FastApiMCP, AuthConfig3.4.2 配置天气服务 APIapp = FastAPI(title="天气查询API") # 配置项 - 在实际应用中应从环境变量或配置文件中读取 # PRIVATE_KEY = """YOUR_PRIVATE_KEY""" # PROJECT_ID = "YOUR_PROJECT_ID" # KEY_ID = "YOUR_KEY_ID" # WEATHER_API_HOST = "your_api_host" PRIVATE_KEY = "MC4CAQAwBQYDK2VwBCIEIG2wMZga50X1YDHmR8jkE5TGKNXpriFPXXCT/kgwZkcT" PROJECT_ID = "3MKPRE78RN" KEY_ID = "T8B7JMU7AY" WEATHER_API_HOST = "ky6mte27bb.re.qweatherapi.com" 3.4.3 定义响应 Model# 响应模型 class Location(BaseModel): name: str id: str lat: str lon: str adm2: str adm1: str country: str tz: str utcOffset: str isDst: str type: str rank: str fxLink: str class LocationResponse(BaseModel): code: str location: List[Location] class WeatherNow(BaseModel): obsTime: str temp: str feelsLike: str icon: str text: str wind360: str windDir: str windScale: str windSpeed: str humidity: str precip: str pressure: str vis: str cloud: str dew: str class ReferSources(BaseModel): sources: List[str] license: List[str] class WeatherResponse(BaseModel): code: str updateTime: str fxLink: str now: WeatherNow refer: ReferSources class JWTRequest(BaseModel): expiry_seconds: Optional[int] = 900 # 默认15分钟 custom_claims: Optional[dict] = None # 允许添加自定义声明 class WeatherQuery(BaseModel): city: str = Field(..., description="城市名，如：北京") 3.4.4 JWT 令牌生成# 解码并加载私钥 try: _der_key_bytes = base64.b64decode(PRIVATE_KEY) PRIVATE_KEY = serialization.load_der_private_key( _der_key_bytes, password=None, backend=default_backend() ) except Exception as e: print(f"关键错误：无法从配置加载EdDSA私钥。错误: {e}") PRIVATE_KEY = None if PRIVATE_KEY is None: raise RuntimeError(f"关键错误：EdDSA私钥加载失败，应用无法启动。错误: {e}") # 令牌缓存（简单实现，生产环境应使用Redis等缓存系统） token_cache = { "token": None, "expires_at": 0 } # Bearer Token 安全头 token_auth_scheme = HTTPBearer() EXPECTED_BEARER_TOKEN = "huaweicloud" # 定义期望的固定Token # Bearer Token 验证依赖 async def verify_bearer_token(token_payload = Depends(token_auth_scheme)): if token_payload.credentials != EXPECTED_BEARER_TOKEN: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="无效的认证凭据", headers={"WWW-Authenticate": "Bearer"}, ) return token_payload.credentials # 生成JWT令牌 def generate_jwt(expiry_seconds: int = 900): current_time = int(time.time()) # 构建标准JWT载荷 payload = { 'iat': current_time - 30, # 颁发时间（提前30秒，避免时钟偏差问题） 'exp': current_time + expiry_seconds, # 过期时间 'sub': PROJECT_ID # 主题（项目ID） } # JWT头部 headers = { 'kid': KEY_ID # 密钥ID } if PRIVATE_KEY is None: raise ValueError("JWT生成失败: 私钥未初始化或加载失败。") try: # 生成JWT encoded_jwt = jwt.encode(payload, PRIVATE_KEY, algorithm='EdDSA', headers=headers) # 更新缓存 token_cache["token"] = encoded_jwt token_cache["expires_at"] = current_time + expiry_seconds - 60 # 提前1分钟过期，确保安全 return encoded_jwt except Exception as e: raise ValueError(f"JWT生成失败: {str(e)}") # 获取有效的JWT令牌（如果缓存中有有效令牌则使用缓存，否则生成新令牌） def get_valid_token(): current_time = int(time.time()) # 检查缓存中的令牌是否有效 if token_cache["token"] and token_cache["expires_at"] > current_time: return token_cache["token"] # 生成新令牌 return generate_jwt() @app.post("/generate-jwt", operation_id="generate_jwt", tags=["JWT"]) async def create_jwt( request: JWTRequest = JWTRequest() ): """ 生成JWT令牌 - 使用EdDSA算法签名 - 默认有效期为15分钟 - 可以添加自定义声明 """ try: encoded_jwt = generate_jwt(request.expiry_seconds) current_time = int(time.time()) return { "jwt": encoded_jwt, "expires_at": datetime.fromtimestamp(current_time + request.expiry_seconds).isoformat(), "issued_at": datetime.fromtimestamp(current_time - 30).isoformat(), "valid_for_seconds": request.expiry_seconds } except Exception as e: raise HTTPException(status_code=500, detail=str(e)) 3.4.5 天气查询接口# 发送HTTP请求到天气API，并处理gzip压缩 def fetch_weather_api(endpoint: str, params: Dict[str, Any]): # 获取有效的JWT令牌 token = get_valid_token() headers = { 'Authorization': f'Bearer {token}', 'Accept-Encoding': 'gzip' # 我们请求gzip压缩 } url = f"https://{WEATHER_API_HOST}/{endpoint}" try: response = requests.get(url, headers=headers, params=params) # 检查响应状态 if response.status_code != 200: # 如果是401或403，尝试刷新令牌并重试 if response.status_code in [401, 403]: # 强制生成新令牌 new_token = generate_jwt() # 更新请求头 headers['Authorization'] = f'Bearer {new_token}' # 重试请求 response = requests.get(url, headers=headers, params=params) # 如果还是失败，则抛出异常 if response.status_code != 200: raise HTTPException( status_code=response.status_code, detail=f"天气API请求失败: HTTP {response.status_code}" ) else: raise HTTPException( status_code=response.status_code, detail=f"天气API请求失败: HTTP {response.status_code}" ) # 依赖 requests 库自动处理Gzip解压缩，并直接解析JSON # 旧的Gzip处理逻辑已被移除 return response.json() except requests.exceptions.JSONDecodeError as e: # 如果响应不是有效的JSON（即使在解压缩后），则捕获此特定错误 raise HTTPException(status_code=500, detail=f"天气API响应解析失败: 无效的JSON内容 - {str(e)}") except HTTPException: # 重新抛出已捕获的HTTPException，以便FastAPI处理 raise except Exception as e: # 捕获其他潜在错误 raise HTTPException(status_code=500, detail=f"天气API请求时发生未知错误: {str(e)}") @app.get("/city/lookup", response_model=LocationResponse, operation_id="lookup_city", tags=["天气查询"]) async def lookup_city( location: str = Query(..., description="城市名称，如：北京") ): """ 根据城市名称查询位置ID - 返回城市的详细信息和位置ID - 位置ID用于后续天气查询 """ try: data = fetch_weather_api("geo/v2/city/lookup", {"location": location}) return data except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.get("/weather/now", response_model=WeatherResponse, operation_id="get_weather_now", tags=["天气查询"]) async def get_weather_now( location: str = Query(..., description="位置ID，如：101010100") ): """ 获取指定位置的实时天气 - 需要提供位置ID - 返回当前天气详情 """ try: data = fetch_weather_api("v7/weather/now", {"location": location}) return data except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.post("/weather/by-city", response_model=WeatherResponse, operation_id="get_weather_by_city", tags=["天气查询"]) async def get_weather_by_city( query: WeatherQuery ): """ 一站式查询城市天气 - 只需提供城市名 - 自动查询位置ID并获取天气 """ try: # 先查询城市ID location_data = fetch_weather_api("geo/v2/city/lookup", {"location": query.city}) # 检查是否找到城市 if location_data.get("code") != "200" or not location_data.get("location"): raise HTTPException(status_code=404, detail=f"找不到城市: {query.city}") # 获取第一个匹配城市的ID location_id = location_data["location"][0]["id"] # 查询天气 weather_data = fetch_weather_api("v7/weather/now", {"location": location_id}) return weather_data except HTTPException: raise except Exception as e: raise HTTPException(status_code=500, detail=str(e)) 3.4.6 主程序及 MCP 集成@app.get("/") async def root(): """主页 - 提供API简介""" return { "message": "天气查询API", "endpoints": [ "/generate-jwt - 生成JWT令牌", "/city/lookup - 根据城市名查询位置ID", "/weather/now - 根据位置ID查询当前天气", "/weather/by-city - 一站式查询城市天气" ], "docs": "/docs 查看完整API文档" } # mcp 实现 mcp = FastApiMCP( app, name="My Weather MCP", description="天气查询API", include_operations=["get_weather_by_city"], # 只公开 get_weather_by_city 接口作为 MCP tool auth_config=AuthConfig(dependencies=[Depends(verify_bearer_token)]) # MCP 使用 Bearer Token 验证 ) mcp.mount() # 启动服务器的命令（在命令行中运行）: # uvicorn main:app --reload if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8009) 4. 测试验证4.1 本地服务测试启动服务:python main.pyAPI 文档验证:访问 http://localhost:8009/docs测试各个接口功能验证数据格式正确性我们可以使用浏览器访问 http://localhost:8009/docs 来查看并测试API。4.2 MCP 客户端集成Cherry Studio 配置:服务名称: FastAPI 天气查询连接类型: 服务器发送事件（SSE）服务地址: http://localhost:8009/mcp认证头: Authorization=Bearer huaweicloud功能验证:配置 MCP 服务连接验证工具列表显示测试天气查询功能对比启用前后的效果在 Cherry Studio 中配置好的 MCP 服务，我们可以直接在 MCP 客户端中使用 get_weather_by_city 接口:名称：FastAPI 天气查询类型：服务器发送事件（SSE）URL：http://localhost:8009/mcp请求头：Authorization=Bearer zishu.co在工具中，我们可以看到get_weather_by_city接口已经添加到MCP服务中。新建一个话题，先不开启 MCP 服务器设置，直接输入北京天气怎么样，我们会发现大模型无法实时获取天气。接着我们启用 MCP 服务，并输入北京天气怎么样，我们会发现大模型可以实时获取天气。get_weather_by_city tool 被成功调用，并返回了天气信息。5. 总结5.1 技术收获通过本案例，您将掌握：MCP 协议: 理解 AI 模型与外部工具的交互机制FastAPI 开发: 快速构建高性能 Web APIJWT 认证: 实现安全的 API 访问控制第三方 API 集成: 学会封装和调用外部服务错误处理: 构建健壮的服务应用5.2 应用价值实时数据获取: AI 模型可以获取最新的天气信息用户体验提升: 提供准确、及时的天气查询服务技术架构: 为更复杂的 MCP 应用奠定基础可扩展性: 可以轻松添加更多天气相关功能如果想要拓展更多功能，您可以动手试试看！

案例介绍本实践中使用鲲鹏DevKit系统性能分析工具对业务中使用Python进行字符串拼接接口执行系统全景分析，应用热点函数分析，找到性能瓶颈点，并根据分析结果进行优化修改，从而实现使用Python进行字符串拼接性能增强。案例内容1 概述1.1 实验介绍鲲鹏DevKit系统性能分析是针对基于鲲鹏的服务器的性能分析工具，能收集服务器的处理器硬件、操作系统、进程/线程、函数等各层次的性能数据，分析出系统性能指标，定位到系统瓶颈点及热点函数，给出优化建议。该工具可以辅助用户快速定位和处理软件性能问题。本实验选择Python进行字符串拼接性能分析作为示例，并借助开发者空间云主机提供的鲲鹏沙箱资源进行安装、配置，直观地展示Devkit中系统性能分析能力在实际应用开发中为开发者带来的便利。1.2 实验对象企业个人开发者高校学生1.3 实验时间本次实验总时长预计40分钟。1.4 实验流程说明：自动部署鲲鹏云服务器；安装鲲鹏Devkit插件；通过浏览器访问，添加IP节点，以配置在线分析环境；安装Python3;使用Python3命令执行代码；在线分析，通过全景分析和进程分析查看CPU负载和使用率情况进行对比分析；修改执行命令，再次分析。1.5 实验资源本次实验预计花费总计0元资源名称规格单价（元）时长（h）云主机4vCPUs | 8GB |ARM | Ubuntu免费12 鲲鹏DevKit之Python字符串拼接系统性能分析2.1 自动部署鲲鹏云服务器在云主机桌面右键选择“Open Terminal Here”，打开命令终端窗口。执行自动部署命令如下，该命令会自动部署鲲鹏云服务器。hcd deploy --password abcd@1234 --time 3600–password 待部署项目所在ECS的root用户密码（至少8个字符），如果不修改部署命令，鲲鹏云服务器密码就是abcd@1234。–time 待部署资源的保留期（单位为秒，至少600秒，默认600秒）。当前实验预估需要40分钟，可以配置time为1小时保留期。看到“application is running, service addr: https://xxx.xxx.xxx.xxx:8084”表示部署成功，记录部署远端服务器公网IP，如截图中对应的就是：115.175.25.9 。自动部署的鲲鹏云务器已经预装了鲲鹏DevKit插件，该案例会用到DevKit的系统性能分析能力。2.2 添加IP节点在云主机桌面右键选择“Open Terminal Here”，打开命令终端窗口。通过ssh连接云服务器，然后输入“密码”，出现“Welcome to XXX”代表连接成功。ssh root@云主机IP登录鲲鹏云服务器查询鲲鹏云服务器内网IP地址。ifconfig通过浏览器访问鲲鹏云服务器，添加目标节点，以配置在线分析环境。打开浏览器，输入“https://xxx.xxx.xxx.xxx:8086”（IP为2.1自动部署的鲲鹏云服务器IP），如果提示风险，点击接受并继续。首次登录需要设置密码，设置后登录。登录完成后添加节点：点击“调优”，选择通用分析下面的“新建”按钮，右边窗口选择“管理节点”。进入管理界面后，点击“添加节点”，勾选协议，弹窗节点IP输入上面ifconfig查询到的鲲鹏云服务器内网IP，输入“密码”，点击“确定”。选择“继续添加”。添加成功如下图所示，点击左上角“DevKit Tools”返回上一级菜单。上一级菜单如下图所示，可以看到节点添加成功。2.3 执行Python代码回到云主机桌面打开终端，创建concatenate_string.py文件。vim concatenate_string.py回车进入，点击“i”进入编辑模式。concatenate_string.py代码如下（concatenate_string.py 代码仓地址请至案例原文获取）：import sysBASE_STRING = "Hello world"LOOP_TIMES = 100000000STRING_LIST = [BASE_STRING] * LOOP_TIMESdef string_plus(): new_string = '' for string in STRING_LIST: new_string += string print(len(new_string)) def string_join(): new_string = ''.join(STRING_LIST) print(len(new_string)) def main(): function_map = { 'string_plus': string_plus, 'string_join': string_join, } print(sys.argv) _, function = sys.argv if function in function_map: function_map[function]() else: print('function {} not exist'.format(function)) if __name__ == "__main__": main()代码复制完毕，键盘点击“esc”，然后输入“:wq”保存退出。执行以下指令，可以代码看到执行成功，该代码中使用“+”在for循环中拼接字符串，运行时间为10s。time python3 concatenate_string.py string_plus2.4 在线分析2.4.1 创建全景分析任务回到浏览器，可以看到2.2添加的节点界面。分析类型选择“全景分析”，采样类型全部勾选，采样时长设置为“20”，其他默认，点击“确定”，跳出弹窗点击“确定”。出现如下界面时，点击“开始分析”，等待分析进度条完成。采样完成后，点击“系统性能”，打开“CPU负载”，可以看到数据显示CPU负载值。2.4.2 创建进程线程分析任务回到执行Python终端，如连接超时重新连接即可。再次执行Python3命令。time python3 concatenate_string.py string_plus执行成功返回浏览器。点击左上角系统性能分析后的新建任务图标，选择“通用分析”。分析类型选择“进程分析”，采样类型全部勾选，采样时长设置为“20”，其他默认，点击“确定”，跳出弹窗点击“确定”出现如下界面时，点击“开始分析”。等待分析进度条完成。分析完成后查看进程性能数据，CPU使用率也比其他大部分进程高。2.5 修改执行命令并再次分析回到Python终端，将执行命令进行修改，使用join拼接字符串，执行如下命令，可以看到运行时间变为3s。time python3 concatenate_string.py string_join再次创建全景分析，参数与2.4.1保持一致，分析成功后查看结果。再次创建进程分析，参数与2.4.2保持一致，分析成功后查看结果。时间对比：字符串拼接方式运行时长使用“+”拼接10.850s使用“join”拼接3.112s总结使用鲲鹏DevKit系统性能分析工具查看性能数据，通过系统全景分析和进程分析的数据采集，可以分析出性能瓶颈点，Python中的string是不可变对象，循环中使用+进行大量字符串拼接时，会频繁的进行内存的申请、分配以及字符串的复制，导致性能低下。将代码进行修改，使用join拼接字符串，执行时间减少，性能明显提升。至此本次实验全部内容完成。更多案例戳→https://devstation.connect.huaweicloud.com/space/devportal/casecenter

痛点识别在日常工作中，文档合并的常见痛点包括：格式丢失：复制粘贴时原有的字体、样式、表格格式经常丢失文档标识困难：合并后难以区分内容来源效率低下：处理大量文档时手工操作耗时巨大顺序混乱：缺乏直观的排序机制解决方案设计基于这些痛点，我们的解决方案需要具备：直观的图形界面：降低使用门槛拖拽排序功能：便于调整文档顺序格式保持能力：确保合并后格式不丢失文档标识机制：自动标注每个文档的来源批量处理能力：支持一次处理多个文档技术栈选择与架构设计核心技术栈GUI框架：wxPython跨平台兼容性好原生界面外观丰富的控件支持文档处理：python-docx专业的Word文档处理库支持格式保持和复制良好的表格处理能力并发处理：threading避免界面冻结提供实时进度反馈架构设计思路采用经典的MVC模式变体：1234┌─────────────────┐    ┌──────────────────┐    ┌─────────────────┐│   View Layer    │    │  Control Layer   │    │   Model Layer   ││  (wxPython UI)  │◄──►│ (Event Handlers) │◄──►│ (Document API)  │└─────────────────┘    └──────────────────┘    └─────────────────┘核心功能实现详解1. 用户界面设计界面采用垂直布局，分为四个主要区域：123456789# 创建主面板和布局管理器panel = wx.Panel(self)main_sizer = wx.BoxSizer(wx.VERTICAL) # 文件列表区域 - 使用ListCtrl展示文档self.file_list = wx.ListCtrl(panel, style=wx.LC_REPORT | wx.LC_SINGLE_SEL)self.file_list.AppendColumn("序号", width=60)self.file_list.AppendColumn("文件名", width=300)self.file_list.AppendColumn("路径", width=400)设计亮点：使用wx.ListCtrl提供表格式文件列表支持单选和多列显示清晰的视觉层次划分2. 文件管理机制实现了完整的文件生命周期管理：123456789101112def on_add_files(self, event):    """添加Word文档 - 支持多选"""    wildcard = "Word文档 (*.docx;*.doc)|*.docx;*.doc"    dlg = wx.FileDialog(self, "选择Word文档", wildcard=wildcard,                        style=wx.FD_OPEN | wx.FD_MULTIPLE)         if dlg.ShowModal() == wx.ID_OK:        paths = dlg.GetPaths()        for path in paths:            if path not in self.file_paths:  # 防止重复添加                self.file_paths.append(path)                self.update_file_list()技术特点：文件格式过滤确保只选择Word文档重复文件检测避免冗余实时更新界面显示3. 文档排序功能提供直观的上移下移操作：123456789def on_move_up(self, event):    """上移选中文件"""    selected = self.file_list.GetFirstSelected()    if selected > 0:        # 交换列表中的位置        self.file_paths[selected], self.file_paths[selected-1] = \            self.file_paths[selected-1], self.file_paths[selected]        self.update_file_list()        self.file_list.Select(selected-1)  # 保持选中状态用户体验优化：边界检查防止越界操作操作后保持选中状态实时视觉反馈4. 核心合并算法这是整个工具的核心，需要处理格式保持和文档标识：123456789101112131415161718192021222324252627282930def merge_documents_thread(self, output_path):    """文档合并的核心逻辑"""    try:        merged_doc = Document()                 for i, file_path in enumerate(self.file_paths):            filename = os.path.splitext(os.path.basename(file_path))[0]                         # 添加分页符（除第一个文档外）            if i > 0:                merged_doc.add_page_break()                         # 添加文档标识标题            title_paragraph = merged_doc.add_paragraph()            title_run = title_paragraph.add_run(f"【{filename}】")            title_run.font.bold = True            title_run.font.color.rgb = RGBColor(0, 0, 139)  # 深蓝色            title_paragraph.alignment = WD_PARAGRAPH_ALIGNMENT.CENTER                         # 处理源文档内容            source_doc = Document(file_path)                         # 复制段落并保持格式            for paragraph in source_doc.paragraphs:                new_paragraph = merged_doc.add_paragraph()                                 for run in paragraph.runs:                    new_run = new_paragraph.add_run(run.text)                    # 格式属性复制                    self.copy_run_format(run, new_run)技术亮点：格式保持：逐个复制文本运行的格式属性文档标识：自动插入带格式的文档名称标题分页控制：合理的分页符插入策略表格处理：完整的表格结构和内容复制5. 异步处理与进度反馈使用多线程避免界面冻结：1234567891011121314151617def on_merge_documents(self, event):    """启动合并操作"""    # 参数验证...         # 禁用合并按钮，重置进度    self.merge_btn.Enable(False)    self.progress.SetValue(0)         # 在新线程中执行合并    thread = threading.Thread(target=self.merge_documents_thread, args=(full_path,))    thread.start() def update_progress(self, current, total, status):    """线程安全的进度更新"""    progress = int((current / total) * 100)    self.progress.SetValue(progress)    self.status_text.SetLabel(status)并发处理优势：UI响应性保持实时进度显示用户体验友好跨平台兼容性处理在开发过程中遇到的一个典型问题是wxPython版本差异导致的API不兼容：123456789101112def get_default_path(self):    """兼容不同版本的路径获取"""    try:        std_paths = wx.StandardPaths.Get()        if hasattr(std_paths, 'GetDocumentsDir'):            return std_paths.GetDocumentsDir()        elif hasattr(std_paths, 'GetDesktopDir'):            return std_paths.GetDesktopDir()        else:            return os.path.expanduser("~/Desktop")    except:        return os.path.expanduser("~")兼容性策略：运行时API检测多级后备方案优雅降级处理性能优化与错误处理内存管理12345678910# 及时释放文档对象try:    source_doc = Document(file_path)    # 处理文档...except Exception as e:    wx.CallAfter(self.show_error, f"处理文件 {file_path} 时出错: {str(e)}")    continuefinally:    # 确保资源释放    source_doc = None错误恢复机制单个文档错误不影响整体流程详细的错误信息反馈用户友好的错误提示部署与分发依赖管理1234567# 创建虚拟环境python -m venv word_merger_envsource word_merger_env/bin/activate  # Linux/Mac# word_merger_env\Scripts\activate  # Windows # 安装依赖pip install wxpython python-docx打包分发可以使用PyInstaller创建独立可执行文件：12pip install pyinstallerpyinstaller --windowed --onefile word_merger.py

一、DeepSeek+Python：自动生成测试用例分工1.1 DeepSeek：作为自动生成测试用例的 大脑DeepSeek 作为一款强大的 AI工具，它能理解软件的功能需求，自动生成测试用例框架。就像一个经验丰富的测试专家，能快速梳理出各种可能的测试场景。比如，对于一个用户登录功能，它能想到正常登录、密码错误、用户名不存在、账号被锁定等多种情况，并生成相应的测试用例结构，为后续测试提供基础。1.2 Python：灵活实现测试的万能 钥匙Python 作为一种功能强大且灵活的编程语言，在测试领域应用广泛。结合 pytest 框架，它能将 DeepSeek 生成的测试用例框架转化为可执行的测试脚本。利用pytest框架不仅能方便地组织和运行测试用例，还提供了丰富的插件和功能，像参数化测试、测试夹具等，让测试工作更加高效和便捷。二、具体实现步骤与代码示例2.1 环境搭建首先，确保你已经安装了 Python 环境。然后，通过 pip 安装 pytest 和 deepseek 库（假设 deepseek 库已存在并可安装）。如果要进行覆盖率分析，还需安装 coverage.py ，命令如下：1pip install pytest deepseek pytest-cov2.2 使用 DeepSeek 生成基础测试用例以测试一个简单的用户登录接口为例，假设接口地址为https://api.example.com/login，接收 JSON 格式的用户名和密码，返回登录结果。代码如下：1234567891011121314151617181920212223import pytestfrom deepseek import DeepSeek  import requests # 初始化 DeepSeekdeepseek = DeepSeek() def test_user_login():    """测试用户登录接口"""    # 利用 DeepSeek 生成用户登录接口的测试用例    test_cases = deepseek.generate_test_cases('user_login')         for case in test_cases:        # 发送 POST 请求至登录接口        response = requests.post("https://api.example.com/login", json=case['input'])                 # 断言响应状态码与预期相符        assert response.status_code == case['expected']['status_code'], \            f"Expected status code {case['expected']['status_code']}, but got {response.status_code}"                 # 断言响应内容与预期一致        assert response.json() == case['expected']['response'], \            f"Expected response {case['expected']['response']}, but got {response.json()}"2.3 扩展测试用例：自动生成边界值测试用例为了提高测试用例的覆盖率，我们可以结合 Python 手动扩展一些边界值测试用例。比如，对用户名和密码的长度限制进行测试：12345678910111213141516171819202122232425def generate_boundary_test_cases():    """生成用户登录接口的边界值测试用例"""    return [        # 输入用户名过短、密码过短的边界情况        {'input': {'username': 'a', 'password': '1'}, 'expected': {'status_code': 400, 'response': {'error': 'Invalid username or password'}}},                 # 输入用户名过长、密码过长的边界情况        {'input': {'username': 'abcdefghijklmnopqrstuvwxyz', 'password': '1234567890'}, 'expected': {'status_code': 400, 'response': {'error': 'Username too long'}}},                 # 输入正常用户名、密码的常规情况        {'input': {'username': 'validUser', 'password': '123456'}, 'expected': {'status_code': 200, 'response': {'message': 'Login successful'}}}    ] @pytest.mark.parametrize("test_case", generate_boundary_test_cases())def test_user_login_boundary(test_case):    """测试用户登录接口的边界值"""    response = requests.post("https://api.example.com/login", json=test_case['input'])         # 断言响应状态码与预期相符    assert response.status_code == test_case['expected']['status_code'], \        f"Expected status code {test_case['expected']['status_code']}, but got {response.status_code}"         # 断言响应内容与预期一致    assert response.json() == test_case['expected']['response'], \        f"Expected response {test_case['expected']['response']}, but got {response.json()}"2.4 用例覆盖率分析：精准评估测试效果为了精准评估测试用例对代码的覆盖程度，我们可以借助 coverage.py 这一实用工具。它能够清晰地呈现哪些代码路径已被测试覆盖，哪些尚未涉及。12345678910111213141516# 安装 coverage.py 及其 pytest 插件# pip install pytest-cov # 运行测试并生成覆盖率报告# pytest --cov=your_module_name tests/ def test_user_login():    """测试用户登录接口"""    test_cases = deepseek.generate_test_cases('user_login')    for case in test_cases:        response = requests.post("https://api.example.com/login", json=case['input'])        assert response.status_code == case['expected']['status_code']        assert response.json() == case['expected']['response'] if __name__ == "__main__":    pass  # pytest 会自动发现并运行测试函数也可以直接在在命令行运行以下命令：1pytest --cov=your_module_name tests/其中，your_module_name是你要测试的模块名，tests/是测试脚本所在的目录。这会生成一份覆盖率报告，帮助你找出哪些代码路径还没有被测试覆盖，从而针对性地补充测试用例。

键盘操作需要执行 press 和 release 来输入按键。 当然，也可以单独执行 press123456789101112131415from pynput.keyboard import Key, Controllerfrom time import sleep # 初始化键盘控制器keyboard = Controller() sleep(3) # 模拟按下和释放 A 键keyboard.press('a')keyboard.release('a') # 模拟按下和释放 Enter 键keyboard.press(Key.enter)keyboard.release(Key.enter)鼠标操作123456789101112131415from pynput.mouse import Button,Controllerfrom time import sleep # 初始化鼠标控制器mouse = Controller() sleep(3) # 模拟鼠标移动到 (100, 100) 位置mouse.position = (100, 100) mouse.click(Button.left,1) # 模拟鼠标右键点击mouse.click(Button.right, 1)注意：这里引入的是 pynput.mouse 的 Controller，与键盘处的名称相同，但是作用不同，需要区分。推荐使用别名的方式将 Controller 命名为 MouseController。下面的方法也会用别名的方式进行区分。1from pynput.mouse import Controller as MouseController键盘监听1234567891011121314from pynput.keyboard import Listener as KeyboardListener def on_press(key):    print(f'Alphanumeric key pressed: {key}') def on_release(key):    print(f'Alphanumeric key released: {key}')    if key == Key.esc:        # 停止监听        return False   # 设置键盘监听with KeyboardListener(on_press=on_press, on_release=on_release) as listener:    listener.join()该方法输出用户按下的按键，当用户按下 Esc 时停止。全局热键监听12345678910111213141516def keyboard_hotkey():    keyboard_controller = Controller()       def on_activate_f():        print('全局热键 F 被激活')        keyboard_controller.type('你按了热键F')     def on_activate_shift_a():        print('全局热键 Shift+A 被激活')        keyboard_controller.type('你按了热键Shift+A')     # 创建热键    with keyboard.GlobalHotKeys({            'f': on_activate_f,            '<shift>+a': on_activate_shift_a}) as h:        h.join()鼠标监听1234567891011121314151617181920212223242526272829303132from pynput.mouse import Listener as MouseListener def mouse_listener():    # 添加左键点击计数器    left_click_count = 0       def on_move(x, y):        print(f'Mouse moved to ({x}, {y})')       def on_click(x, y, button, pressed):        nonlocal left_click_count        print(f'Mouse clicked at ({x}, {y}) with {button}')        # 检查是否是左键点击且按下状态        if button == Button.left and pressed:            left_click_count += 1            print(f'左键点击次数: {left_click_count}')             # 如果左键点击次数超过三次，停止监听            if left_click_count >= 3:                print('左键点击已达到3次，停止监听')                return False     def on_scroll(x, y, dx, dy):        print(f'Scrolled at ({x}, {y})')         # 设置鼠标监听    with MouseListener(on_move=on_move, on_click=on_click, on_scroll=on_scroll) as listener:        listener.join() mouse_listener()该示例记录鼠标信息，当鼠标左键点击三次停止。知识补充如何通过pynput与日志记录实现键盘、鼠标的监听行为？将相关的模块导入到代码块中。123456# 鼠标键盘监控from pynput import keyboard, mouse# 日志处理from loguru import logger# 多线程处理from threading import Thread将目标日志记录文件加入到logger代码处理器中。1logger.add('lister.log')编写鼠标被按下时的处理函数。1234567def on_keyboard_press(key):    '''    按键时记录所按下的键    :param key:    :return:    '''    logger.debug(f'{key} :被按下了')编写鼠标抬起时的处理函数。12345678def on_keyboard_release(key):    '''    释放按键处理函数    :param key:    :return:    '''    if key == keyboard.Key.esc:        return False编写鼠标单击、右击以及滑轮的处理函数。12345678def on_mouse_click(x, y, click, pressed):    if click == mouse.Button.left:        logger.debug('鼠标左键按下了')    elif click == mouse.Button.right:        logger.debug('鼠标右键按下了')        return False    else:        logger.debug('中间滚轮按下了')将键盘相关的处理函数加入到键盘事件的监听中。1234567def func_keyboard():    '''    键盘的按下/释放的监听    :return:    '''    with keyboard.Listener(on_press=on_keyboard_press, on_release=on_keyboard_release) as keyboard_listener:        keyboard_listener.join()将鼠标相关的处理函数加入到鼠标事件的监听中。1234567def func_keyboard():    '''    键盘的按下/释放的监听    :return:    '''    with keyboard.Listener(on_press=on_keyboard_press, on_release=on_keyboard_release) as keyboard_listener:        keyboard_listener.join()在main()的处理函数中将鼠标监听、键盘监听分别作为两个线程启动。1234567891011if __name__ == '__main__':    '''    执行线程    '''    # 定义键盘监听线程    thread_keyboard = Thread(target=func_keyboard)    # 定义鼠标监听线程    thread_mouse = Thread(target=func_mouse_click)    # 分别启动线程    thread_keyboard.start()    thread_mouse.start()最后可以得到一份键盘、鼠标历史执行状态记录的日志文件，通过NLTK语言的处理从而逆向推出电脑的操作内容，这一部分就不演示了。

ndarray数组元素的访问🌺 ndarray 切片操作的基本语法和列表类似，为 arr[start:stop:step]，其中 star、stop 和 step 是可选参数。🌺 对于多维数组，每个维度都可以单独指定切片范围，语法为 arr[dim1_start:dim1_stop:dim1_step, dim2_start:dim2_stop:dim2_step, …]。🌺 start：切片起始索引，包含该索引对应的元素，默认值为 0。负数索引表示从数组末尾开始计数。🌺 stop：切片结束索引，不包含该索引对应的元素，默认值为数组的长度。负数索引表示从数组末尾开始计数。🌺 step：切片步长，默认值为 1。正数表示从左到右选取元素，负数表示从右到左选取元素。🌺 得到的结果仍是ndarray。一维ndarray的元素访问🍁 0个冒号的，stop123456code:import numpy as npnd1 = np.array(range(8))print("nd1[3]: %d, nd1[-2]: %d" % (nd1[3], nd1[-2]))result:nd1[3]: 3, nd1[-2]: 6🍁 1个冒号的，start and stop123456code:import numpy as npnd1 = np.array(range(8))print("nd1[:3]: %s, nd1[2:4]: %s, nd1[4:]: %s, nd1[:]: %s" % (nd1[:3], nd1[2:4], nd1[4:], nd1[:]))result:nd1[:3]: [0 1 2], nd1[2:4]: [2 3], nd1[4:]: [4 5 6 7], nd1[:]: [0 1 2 3 4 5 6 7]🍁 2个冒号的，start，stop，step12345678910code:import numpy as npnd1 = np.array(range(8))import numpy as npnd1 = np.array(range(8))print("nd1[::3]: %s, nd1[2::4]: %s, nd1[:4:]: %s, nd1[:6:2]: %s,"% (nd1[::3], nd1[2::4], nd1[:4:], nd1[:6:2]))print("nd1[2:5:]: %s,  nd1[1::]: %s, nd1[1:7:3]: %s, nd1[::]: %s" % (nd1[2:5:],  nd1[1::], nd1[1:7:3], nd1[::]))result:nd1[::3]: [0 3 6], nd1[2::4]: [2 6], nd1[:4:]: [0 1 2 3], nd1[:6:2]: [0 2 4],nd1[2:5:]: [2 3 4],  nd1[1::]: [1 2 3 4 5 6 7], nd1[1:7:3]: [1 4], nd1[::]: [0 1 2 3 4 5 6 7]二维ndarray的元素访问🌹访问单个元素，[行，列]指明要访问的元素的行列123456code:import numpy as npnd1 = np.array([[1, 2, 3], [4, 5, 6]])print("nd1[1,2]: %s" % nd1[1, 2])result:nd1[1,2]: 6🌹[]访问一行或者多行🌹[]中一个数指行，有逗号，逗号前指行，后指列🌹行和列分别可以用切片，列表标识多行或者多列1234567891011121314code:import numpy as npnd1 = np.array([[1, 2, 3], [4, 5, 6], [7, 8, 9]])print("nd1[1]: %s", nd1[1])  # 一个参数指行print("nd1[1, :]", nd1[1, :])  # 用中间逗号的方式指定行和列print("nd1[0:2, :]", nd1[0:2, :])  # 用切片的方式指定多行print("nd1[[0,1], :]", nd1[0:2, :])  # 用列表的方式指定多行result:nd1[1]: [4 5 6]nd1[1, :]: [4 5 6]nd1[0:2, :]: [[1 2 3] [4 5 6]]nd1[[0,1], :]: [[1 2 3] [4 5 6]]🌹访问一列或者多列🌹访问某一列[:, 列号]🌹访问某一列[:, 切片或者列表]访问多列1234567891011121314code:import numpy as npnd1 = np.array([[1, 2, 3], [4, 5, 6], [7, 8, 9]])print("nd1[:, :]:", nd1[:, :])  # 所有行, 所有列print("nd1[:, 1]:", nd1[:, 1])  # 第1列print("nd1[1, 0:2]:", nd1[1, 0:2])  # 第一行的0列和1列print("nd1[[0,1], [0, 2]]:", nd1[[0, 1], [0, 2]])  # 第一行的0列和1列result:nd1[:, :]: [[1 2 3] [4 5 6] [7 8 9]]nd1[:, 1]: [2 5 8]nd1[1, 0:2]: [4 5]nd1[[0,1], [0, 2]]: [1 6]ndarray数组元素的增删改查ndarray数组元素的增加 np.append-🌻np.append(arr, values, axis=None)，values为要添加的数组，注意要和arr的shape匹配才能添加。🌻生成新的数据，原数据不变，values为要添加的数组，注意要和arr的shape匹配才能添加。1234567891011121314151617code:import numpy as npnd1 = np.array([[1, 2, 3], [4, 5, 6], [7, 8, 9]])nd2 = [[11, 22, 33]]  # 上下方向添加, 要是一行数据的形式nd3 = [[44, 444], [55, 555], [66, 666]]new_nd1 = np.append(nd1, nd2, axis=0)  # 上下方向添加print(new_nd1)new_nd2 = np.append(nd1, nd3, axis=1)  # 左右方向添加print(new_nd2)result:[[ 1  2  3] [ 4  5  6] [ 7  8  9] [11 22 33]][[  1   2   3  44 444] [  4   5   6  55 555] [  7   8   9  66 666]]np.insert-🌻 np.insert() 函数可以在数组的指定位置插入值，返回一个新的数组。-🌻 np.insert(arr, obj, values, axis=None)123456789101112131415161718192021code:import numpy as npnd1 = np.array([[1, 2, 3], [4, 5, 6], [7, 8, 9]])# insert(arr, obj, values, axis=None)nd2 = np.insert(nd1, 1, [11, 22, 33], axis=0)nd3 = np.insert(nd2, 3, [[44, 55, 66]], axis=0)nd4 = np.insert(nd1, 1, [11, 22, 33], axis=1)nd5 = np.insert(nd4, 3, [[44, 55, 66]], axis=1)print("nd3:\n", nd3)print("nd5:\n", nd5)result:nd3: [[ 1  2  3] [11 22 33] [ 4  5  6] [44 55 66] [ 7  8  9]]nd5: [[ 1 11  2 44  3] [ 4 22  5 55  6] [ 7 33  8 66  9]]np.concatenate🪻 np.concatenate() 函数可以将多个数组沿着指定的轴连接在一起，实现数据的合并。🪻返回一个新的数组，原数组不会被修改。12345678910111213141516171819202122232425262728code:import numpy as np# 一维数组拼接arr_1d_1 = np.array([1, 2, 3])arr_1d_2 = np.array([4, 5, 6])new_arr_1d = np.concatenate((arr_1d_1, arr_1d_2))  # 一维数字组拼接不用指定axisprint("一维数组拼接后的结果：", new_arr_1d)# 二维数组拼接arr_2d_1 = np.array([[1, 2], [3, 4]])arr_2d_2 = np.array([[5, 6], [7, 8]])# 垂直拼接new_arr_2d_vertical = np.concatenate((arr_2d_1, arr_2d_2), axis=0)  # 垂直方向上拼接print("\n二维数组垂直拼接后的结果：")print(new_arr_2d_vertical)# 水平拼接new_arr_2d_horizontal = np.concatenate((arr_2d_1, arr_2d_2), axis=1)  # 水平方向上拼接print("\n二维数组水平拼接后的结果：")print(new_arr_2d_horizontal)result:一维数组拼接后的结果： [1 2 3 4 5 6]二维数组垂直拼接后的结果：[[1 2] [3 4] [5 6] [7 8]]二维数组水平拼接后的结果：[[1 2 5 6] [3 4 7 8]]np.hstack和np.vstack☘️ np.hstack(tup1,tup2,…)，注意参数的类型。☘️ 参数tup是一个元组、列表或其他可迭代对象，其中包含了要进行水平拼接的多个数组。☘️ np.vstack(tup1,tup2)，和np.hstack不同的是拼接的方向不同。12345678910111213141516code:import numpy as npnd1 = np.array([[1, 2], [3, 4], [5, 6]])nd2 = np.array([[11, 22], [33, 44], [55, 66]])print("hstack: ", np.hstack((nd1, nd2)))print("vstack: ", np.vstack((nd1, nd2)))result:hstack:  [[ 1  2 11 22] [ 3  4 33 44] [ 5  6 55 66]]vstack:  [[ 1  2] [ 3  4] [ 5  6] [11 22] [33 44] [55 66]]ndarray数组元素的删除np.delete()删除指定位置的元素-🪻 NumPy 数组是固定大小的，直接删除元素通常会创建一个新数组。-🪻 删除一维数组元素，np.delete(arr1, idx)，指定位置为idx。-🪻 删除二维数组元素，np.delete(arr1, idx, axis=0/1)，指定位置为idx，按行或者列要指定axis。12345678910111213141516code:import numpy as nparr1 = np.array(range(5))arr2 = np.delete(arr1, [0,4])arr3 = np.delete(arr1, 2)print(arr1)print(arr2)print(arr3)result:import numpy as nparr1 = np.array(range(5))arr2 = np.delete(arr1, [0,4])arr3 = np.delete(arr1, 2)print(arr1)print(arr2)print(arr3)1234567891011121314151617code:arr1 = np.array([[1, 2, 3], [4, 5, 6], [7, 8, 9]])arr2 = np.delete(arr1, 1, axis=0)  # 删除第2行（索引1）arr3 = np.delete(arr1, [1,2], axis=0)  # 删除第2行（索引1）print("arr1:\n",arr1)print("arr2:\n",arr2)print("arr3:\n",arr3)result:arr1: [[1 2 3] [4 5 6] [7 8 9]]arr2: [[1 2 3] [7 8 9]]arr3: [[1 2 3]]根据条件删除元素（过滤）🍅 使用布尔索引创建符合条件的新数组🍅 类似于(arr1>1) & (arr1<5)的表达式生成一个布尔数组。1234567891011121314code:import numpy as nparr1 = np.array([1, 2, 3, 4, 5, 6])condition = (arr1>1) & (arr1<5)arr2 = arr1[condition]print("arr1\n", arr1)print(condition)print("arr2\n", arr2)result:arr1 [1 2 3 4 5 6][False  True  True  True False False]arr2 [2 3 4]np.unique删除重复元素🍉 删除重复元素，并重新排序1234567code:import numpy as nparr = np.array([1, 2, 2, 3, 3, 3, 5, 5, 4, 4])unique_arr = np.unique(arr)print(type(unique_arr), unique_arr)result:<class 'numpy.ndarray'> [1 2 3 4 5]修改ndarray中的元素直接索引赋值💖 支持一维和多维123456789101112131415code:import numpy as nparr1 = np.array([10, 20, 30, 40])arr1[2] = 300arr1[-1] = 400print("arr1:\n", arr1)arr2 = np.array([[1, 2], [3, 4]])arr2[0, 1] = 20print("arr2:\n", arr2)result:arr1: [ 10  20 300 400]arr2: [[ 1 20] [ 3  4]]切片赋值💖 通过切片选择多个元素，批量修改值。💖 赋值时需要确保右侧值的形状与切片形状兼容。1234567891011121314151617181920code:import numpy as nparr1 = np.array([1, 2, 3, 4, 5])arr1[1:4] = [20, 30, 40]print("arr1:\n", arr1)arr2 = np.array([1, 2, 3, 4, 5])arr2[1:4] = 0  # 赋值为同一个数print("arr2:\n", arr2)arr3 = np.array([[1, 2, 3], [4, 5, 6], [7, 8, 9]])arr3[:, 2] = [20, 50, 80]  # 修改第2列print("arr3:\n", arr3)result:arr1: [ 1 20 30 40  5]arr2: [1 0 0 0 5]arr3: [[ 1  2 20] [ 4  5 50] [ 7  8 80]]布尔索引赋值💞 通过布尔掩码选择符合条件的元素，批量修改。12345678code:import numpy as nparr1 = np.array([1, 2, 6, 4, 10, 5, 8])arr1[(arr1 > 3) & (arr1 < 8) ] = 0print("arr1:\n", arr1)result:arr1: [ 1  2  0  0 10  0  8]使用 np.put() 按索引批量赋值-💘 np.put(arr, [索引]，[对应索引的修改值]）1234567code:import numpy as nparr = np.array([10, 20, 30, 40, 50])np.put(arr, [0, 2, 4], [100, 300, 500])  # 修改索引0、2、4的元素print(arr)result:[100  20 300  40 500]使用 np.where() 条件替换-☘️ np.where(condition)，返回满足条件的元素的索引。1234567891011121314151617code:import numpy as nparr1 = np.array([1, 2, 3, 4, 5, 6])indices = np.where(arr1 > 3)print("indices:\n", indices)print("arr1[indices]:\n", arr1[indices])# np.where对于二维数组, 返回的是行索引和列索引列表, 满足条件的是2, 3,4这两个元素, 即为# arr2[0,1], arr2[1,0], arr2[1,1],它们行索引形成[0,1,1], 列索引形成[1,0,1]arr2 = np.array([[1, 2], [3, 4]])row_idx, col_idx = np.where(arr2 > 1)print(f"row_idx：{row_idx}, col_idx： {col_idx}")result:indices: (array([3, 4, 5], dtype=int64),)arr1[indices]: [4 5 6]row_idx：[0 1 1], col_idx： [1 0 1]-☘️ np.where(condition, x, y)，当 condition 为 True 时，取 x 的值，否则取y的值。-☘️ 算是3目运算符。12345678910111213141516171819code:import numpy as npdt = np.dtype([    ('name', 'U3'),  # Unicode字符串，最大长度3    ('age', np.int32),    ('height', float)])arr = np.array([    ('Alice', 25, 1.65),    ('Bob', 30, 1.80),    ('Charlie', 22, 1.75)], dtype=dt)print(arr)print(arr[0]["name"])print(arr["age"])result:[('Ali', 25, 1.65) ('Bob', 30, 1.8 ) ('Cha', 22, 1.75)]Ali[25 30 22]修改为特殊值🏵️ 如果要赋值为nan，必须dtarray为float型，先要进行数据类型的转换。numpy中的裁剪函数Clip🌺Clip函数用于将数组中的元素限制在指定的最小值和最大值之间。🌺 np.clip(arr, a_min, a_max, out=None), arr：输入的数组。a_min：最小值（低于此值的元素将被替换为 a_min, a_max：最大值（高于此值的元素将被替换为 a_max）。out（可选）：指定输出结果的数组，如果指定为arr，相当于原地修改，若不提供则返回新数组。1234567891011code:import numpy as nparr2d = np.array([[1, 5], [10, 15]])clipped = np.clip(arr2d, 3, 12, out=arr2d)print(clipped)print(arr2d)result:[[ 3  5] [10 12]][[ 3  5] [10 12]]

TVT写了好几天呢,问群内好几次了,一点消息没有

在 FastAPI 中实现函数结果缓存，可以使用 cachetools 库或者 FastAPI 内置的 Depends 结合缓存机制。下面我将展示两种实现方式：方法一：使用 cachetools 库首先安装依赖：pip install cachetools fastapi然后实现缓存：from fastapi import FastAPI from cachetools import cached, TTLCache from datetime import datetime, timedelta app = FastAPI() # 创建缓存，最大100个条目，每个条目存活1800秒（30分钟） cache = TTLCache(maxsize=100, ttl=1800) @app.get("/expensive-operation/{param}") @cached(cache) async def expensive_operation(param: str): """ 这个函数会被缓存，30分钟内相同参数的调用会直接返回缓存结果 """ # 模拟耗时操作 print(f"执行耗时操作，参数: {param}") result = {"data": f"处理结果 for {param}", "timestamp": datetime.now().isoformat()} return result方法二：自定义缓存装饰器如果你需要更多控制，可以自己实现一个缓存装饰器：from fastapi import FastAPI from functools import wraps from datetime import datetime, timedelta from typing import Any, Callable, Dict, Tuple, Optional app = FastAPI() # 缓存字典 function_cache: Dict[Tuple, Dict[str, Any]] = {} def cache_result(ttl: int = 1800): """ 缓存装饰器，默认30分钟(1800秒)有效期 """ def decorator(func: Callable): @wraps(func) async def wrapper(*args, **kwargs): # 创建缓存键（基于参数） cache_key = (func.__name__, args, tuple(sorted(kwargs.items()))) # 检查缓存中是否有结果 now = datetime.now() if cache_key in function_cache: cached_result, cached_time = function_cache[cache_key] if (now - cached_time).total_seconds() <= ttl: print(f"返回缓存结果 for {cache_key}") return cached_result # 执行函数并缓存结果 print(f"执行函数 for {cache_key}") result = await func(*args, **kwargs) function_cache[cache_key] = (result, now) return result return wrapper return decorator @app.get("/another-expensive-operation/{param}") @cache_result() # 默认30分钟 async def another_expensive_operation(param: str): """ 使用自定义缓存装饰器的函数 """ # 模拟耗时操作 result = {"data": f"另一个处理结果 for {param}", "timestamp": datetime.now().isoformat()} return result方法三：使用 FastAPI 的 Depends 和状态对于更复杂的应用，可以结合 FastAPI 的依赖注入系统：from fastapi import FastAPI, Depends, HTTPException from datetime import datetime, timedelta from typing import Dict, Optional app = FastAPI() class CacheManager: def __init__(self): self._cache: Dict[str, Dict[str, Any]] = {} async def get_or_set(self, key: str, func: Callable, *args, **kwargs) -> Any: # 检查缓存 if key in self._cache: cached_data, expires_at = self._cache[key] if datetime.now() < expires_at: return cached_data # 执行函数 result = await func(*args, **kwargs) # 设置缓存（30分钟） expires_at = datetime.now() + timedelta(minutes=30) self._cache[key] = { "data": result, "expires_at": expires_at } return result cache_manager = CacheManager() def get_cache_key(param: str) -> str: return f"expensive_op_{param}" @app.get("/dependent-operation/{param}") async def dependent_operation(param: str, cache: CacheManager = Depends()): cache_key = get_cache_key(param) def fetch_data(): # 模拟耗时操作 print(f"执行耗时操作，参数: {param}") return {"data": f"依赖注入缓存结果 for {param}", "timestamp": datetime.now().isoformat()} result = await cache.get_or_set(cache_key, fetch_data) return result注意事项缓存键：确保缓存键能唯一标识函数调用，通常基于函数名和参数线程安全：如果使用多线程，确保缓存实现是线程安全的内存限制：对于生产环境，考虑使用 Redis 等分布式缓存替代内存缓存敏感数据：不要缓存敏感数据或包含用户特定信息的数据缓存失效：考虑如何处理数据更新时的缓存失效问题对于生产环境，建议使用专业的缓存系统如 Redis，可以通过 aioredis 或 redis 库实现分布式缓存。

“随着异构计算需求的增长，ARM架构服务器在数据中心领域的应用日益广泛。本文深入探讨了华为鲲鹏处理器在Python科学计算生态中的应用实践，分析了DevLit优化框架在鲲鹏平台上的部署与性能优化方法，并通过实验数据验证了该方案在大规模数据处理场景下的效能提升。研究表明，通过针对性优化，鲲鹏架构能够为Python科学计算工作负载提供显著的性能与能效优势。”1. 背景计算架构多样化已成为当今高性能计算领域的重要趋势。华为鲲鹏处理器作为基于ARM架构的高性能计算平台，以其卓越的能效比和可扩展性，正在数据中心和科学计算领域获得广泛应用。Python作为科学计算和人工智能领域最流行的编程语言之一，其在新兴计算架构上的性能优化成为研究热点。本文旨在探讨基于华为鲲鹏架构的Python科学计算生态系统构建，以及DevLit优化框架在此平台上的应用实践。通过分析鲲鹏架构的技术特性，结合Python计算框架的优化方法，为高性能计算应用提供参考方案。2. 鲲鹏架构技术特性分析2.1 鲲鹏处理器架构概述华为鲲鹏处理器基于ARM v8架构设计，采用多核心设计理念，具有以下关键特性：高核心密度：单芯片最高支持64核心，提供强大的并行计算能力大缓存架构：三级缓存结构，L3缓存容量达128MB，有效减少内存访问延迟优化的向量计算单元：支持SVE（Scalable Vector Extension）指令集，适合科学计算工作负载创新的内存子系统：支持DDR4-3200内存，带宽可达192GB/s华为鲲鹏920处理器采用7nm工艺制程，在保持高性能的同时，能效比相比同类x86处理器提升30%以上，这一特性使其在数据中心应用中具有明显优势。2.2 与传统x86架构的差异分析鲲鹏架构与x86架构在指令集、内存访问模式和缓存策略等方面存在显著差异：指令集特性：ARM指令集相比x86更为精简，指令译码效率更高内存访问模式：ARM架构采用不同的内存一致性模型，影响并行程序性能向量计算能力：SVE指令集提供可扩展的向量长度，相比x86的AVX-512具有更好的灵活性这些差异对Python科学计算库的性能有直接影响，需要针对性优化才能发挥鲲鹏架构的最大潜力。3. Python在鲲鹏平台的适配与优化3.1 Python核心解释器优化针对鲲鹏架构特性，Python解释器的优化包括：字节码执行引擎优化：利用ARM架构的寄存器优势，减少内存访问内存管理优化：根据鲲鹏内存子系统特性，调整垃圾回收策略JIT编译优化：针对热点代码路径，生成ARM原生指令优化后的Python解释器在鲲鹏平台上基准测试性能提升20%-35%，显著改善了CPU密集型任务的执行效率。3.2 科学计算库适配主流Python科学计算库在鲲鹏平台的适配情况：表1: 科学计算库在鲲鹏平台的适配状态+------------+---------------+---------------------+| 库名称 | 版本号 | 性能状态 |+------------+---------------+---------------------+| NumPy | 1.21.0+ | 完全优化，性能+40% || SciPy | 1.7.0+ | 基本优化，性能+25% || Pandas | 1.3.0+ | 部分优化，性能+15% || scikit-learn| 1.0.0+ | 基本优化，性能+20% |+------------+---------------+---------------------+核心优化策略包括：利用BLAS/LAPACK优化实现（OpenBLAS-ARM）向量化计算利用SVE指令集内存访问模式优化，减少缓存失效多线程并行计算框架调整4. DevLit优化框架概述与应用4.1 DevLit框架介绍DevLit是一个专为异构计算环境设计的高性能计算优化框架，具有以下特点：抽象计算图表示：将计算任务抽象为计算图，便于分析和优化自动调度策略：根据硬件特性自动选择最优执行策略算子融合技术：减少内存访问，提高计算密度硬件感知优化：针对不同硬件平台生成优化代码在鲲鹏平台上，DevLit通过深度整合ARM计算库和编译技术，为Python应用提供了强大的性能提升能力。4.2 DevLit在鲲鹏平台的部署方案DevLit在鲲鹏平台的部署架构如图1所示： +------------------+ | Python应用层 | +------------------+ | +------------------+ | DevLit优化层 | +------------------+ / \ +--------------+ +--------------+ | 计算图优化器 | | 代码生成引擎 | +--------------+ +--------------+ \ / +------------------+ | 鲲鹏硬件抽象层 | +------------------+ | +------------------+ | 鲲鹏处理器 | +------------------+部署流程包括：安装鲲鹏优化版Python开发环境部署DevLit优化框架核心组件配置硬件特性感知模块安装优化后的科学计算库应用程序适配与调优本文分享自华为云社区博客《鲲鹏架构的Python高性能计算实践与DevLit优化框架应用【华为根技术】》，作者： i-WIFI。

大家好，4月份干货合集又来了，这次主要带来的时候python相关的知识点，希望可以帮到大家。1.Ruby 迭代器及文件的输入与输出【转】https://bbs.huaweicloud.com/forum/thread-0224180932715256027-1-1.html2.Perl进行错误处理和创建子程序的示例【转】https://bbs.huaweicloud.com/forum/thread-0275180932589341012-1-1.html3.Perl 特殊变量详解【转】https://bbs.huaweicloud.com/forum/thread-0254180932469815022-1-1.html4.Perl语言的循环实现方法小结【转】https://bbs.huaweicloud.com/forum/thread-0221180932411580018-1-1.html    5.Perl语言的语法糖示例详解【转】https://bbs.huaweicloud.com/forum/thread-0278180932366717020-1-1.html6.perl 交叉编译详解【转】https://bbs.huaweicloud.com/forum/thread-0259180932299618019-1-1.html7.带你搞懂Python中__init__.py到底是什么【转】https://bbs.huaweicloud.com/forum/thread-0224180932100743026-1-1.html8.Python利用ElementTree实现快速解析XML文件【转】https://bbs.huaweicloud.com/forum/thread-0224180932024344025-1-1.html9.几种查看PyTorch、cuda 和 Python 版本方法小结【转】https://bbs.huaweicloud.com/forum/thread-02127180931934706017-1-1.html10.conda将python低版本环境升级到高版本的完整步骤【转】https://bbs.huaweicloud.com/forum/thread-0278180931858775019-1-1.html11.Python脚本实现删除谷歌浏览器历史记录【转】https://bbs.huaweicloud.com/forum/thread-0254180931766384021-1-1.html12.Python包管理利器之pip命令的安装与使用入门指南【转】https://bbs.huaweicloud.com/forum/thread-02127180931638312016-1-1.html13.Python常见数据类型处理详细代码示例【转】https://bbs.huaweicloud.com/forum/thread-02127180931580960015-1-1.html14.Python中解包操作的性能实测与最佳实践【转】https://bbs.huaweicloud.com/forum/thread-0254180931412732020-1-1.html15.Python字符串len()、split()、join()深度解析【转】https://bbs.huaweicloud.com/forum/thread-0278180931322530017-1-1.html

系统版本为openEuler 22.03  python为3.10  cann为8.0.rc3  npu-smi为24.1.rc3  torch为2.1.0  torch 2.1.0.post8  执行x = torch.randn(2,2).npu() 长时间不返回   如何排查错误？执行python3 -c "import torch;import torch_npu;print(torch_npu.npu.is_available())"返回为True

先搞懂 Python 模块（module）在聊 __init__.py 之前，我们得先弄清楚 Python 里的 模块 和 包 这两个概念。模块（module）：简单来说，就是一个 .py 文件，里面写了一些函数、类或者变量。比如，有个叫 math_tools.py 的文件，里面有一堆数学工具函数，那它就是个模块。123456# math_tools.pydef add(a, b):    return a + b def subtract(a, b):    return a - b然后，我们可以在别的 Python 文件里这样用它：123import math_tools print(math_tools.add(3, 5))  # 输出 8这就是 模块的基本用法，没啥难的，对吧？Python 包（package）是啥？如果你写的模块越来越多，代码量越来越大，就得想办法组织它们。这时候，Python 里的 包（package） 就派上用场了。包（package）：一个 文件夹，里面包含多个模块（.py 文件）。在 Python 3.3 之前，如果要让一个目录被识别为 Python 包，必须在里面创建 __init__.py 文件。但从 Python 3.3 开始，即使没有 __init__.py，Python 也能识别它是一个包（称为“命名空间包”）。不过，大部分实际项目 依然建议添加 __init__.py，因为它可以：✅ 明确这个文件夹是一个包，避免某些工具（如打包工具）识别错误。✅ 允许在包初始化时执行特定代码，比如自动导入子模块。✅ 让导入行为更加可控，避免意外的命名冲突。比如，咱们有个 math_utils 目录，里面放了几个数学相关的模块：math_utils/     # 这个文件夹就是一个包│── __init__.py│── basic.py│── advanced.py其中，basic.py 和 advanced.py 分别是两个模块，而 __init__.py 可以用来 自定义包的导入行为。那么 __init__.py 到底是干嘛的？虽然 __init__.py 不再是创建包的 必需 条件，但它依然是 Python 项目里一个重要的组件。它的主要作用有 两个：1.明确标记目录为 Python 包如果 __init__.py 存在，Python 解析器就会知道：“这个目录是个 Python 包，而不是普通文件夹。”即使 Python 3.3+ 之后不强制要求 __init__.py，但加上它可以：✅ 避免 Python 解释器在某些情况下误认为这是普通目录。✅ 兼容旧版本 Python，让代码能在不同环境中运行得更稳定。✅ 让某些工具（如 pytest、mypy）更好地识别项目结构。2.让包能像模块一样被导入如果 __init__.py 里什么都不写，那它的作用只是个“标志”。但如果我们在 __init__.py 里加点代码，它就能 自定义包的导入行为。示例 1：让包直接暴露子模块123# math_utils/__init__.pyfrom .basic import add, subtractfrom .advanced import power这样，我们就可以直接 import 整个 math_utils，而不需要写 .basic 或 .advanced 了：1234import math_utils print(math_utils.add(2, 3))  # 输出 5print(math_utils.power(2, 3))  # 假设 advanced 里有个 power 函数等于说，__init__.py 让 包变得像一个大模块 一样，外部不需要知道里面的模块结构，直接用就行。示例 2：包初始化操作__init__.py 还能在包被导入时执行一些初始化操作，比如加载配置、设置日志等：12# math_utils/__init__.pyprint("数学工具包加载成功！")  # 只要 import 这个包，就会执行这行代码__init__.py 还能干点啥大厂的 Python 项目里，__init__.py 还经常被用来做这些事：1. 动态导入子模块在大型 Python 项目中，随着模块越来越多，手动维护__init__.py将变得特别复杂还容易出错，这时候动态导入子模块就成了香饽饽了。假设我们不知道 math_utils 里具体有哪些模块，可以让 __init__.py 在导入时动态扫描并加载：123456789101112# math_utils/__init__.pyimport osimport importlib # 获取当前包的路径package_path = os.path.dirname(__file__) # 遍历当前目录下的所有 .py 文件（不包括 __init__.py 本身）for module in os.listdir(package_path):    if module.endswith(".py") and module != "__init__.py":        module_name = module[:-3]  # 去掉 .py 后缀        importlib.import_module(f"{__name__}.{module_name}")  # 动态导入模块效果： 这样，当你在别的地方写 import mypackage，所有 mypackage 里的 .py 文件都会自动加载，不用再手动 import 了！没加动态导入要这么写：1234import math_utils.basicprint(math_utils.basic.add(1,2)) #如果直接 import math_utils 会报错AttributeError: module 'math_utils' has no attribute 'basic'加了动态导入可以这么写：12import math_utilsprint(math_utils.basic.add(1,2))2. 控制对外暴露的模块有时候，我们不想让 所有 子模块都被自动导入，而是只暴露一部分给外部用。这时候可以用 __all__ 来 手动控制 允许被 from mypackage import * 访问的模块。123456789101112# math_utils/__init__.pyimport osimport importlib package_path = os.path.dirname(__file__)__all__ = [] for module in os.listdir(package_path):    if module.endswith(".py") and module != "__init__.py":        module_name = module[:-3]        __all__.append(module_name)  # 只暴露在 __all__ 里的模块        importlib.import_module(f"{__name__}.{module_name}")效果：123from math_utils import * print(basic)  # 只有在 __all__ 里的模块能被导入 3. 懒加载（Lazy Import）如果某些模块比较大，加载它们会影响性能，那可以用 懒加载（lazy import）技术，在需要时才导入，而不是在 import mypackage 时一次性全加载。1234567# math_utils/__init__.pyimport importlib def lazy_import(name):    return importlib.import_module(f"{__name__}.{name}") module1 = lazy_import("basic")效果：这样，basic 只有在第一次被使用时才会真正导入，提高了性能！4. 做版本控制__init__.py 还能给包加上版本号，让外部代码可以访问：12# math_utils/__init__.py__version__ = "1.0.0"然后，在别的地方可以这样用：123import math_utils print(math_utils.__version__)  # 输出 "1.0.0"5. 隐藏内部实现有些模块是“内部用”的，不想让外部访问，怎么办？可以在 __init__.py 里手动控制 对外暴露的内容：1234# math_utils/__init__.pyfrom .basic import add, subtract __all__ = ["add", "subtract"]  # advanced.py 里的东西就不会被直接 import这样，外部只能用 math_utils.add()，但 math_utils.advanced 就不让直接访问了。

一、XML文件解析到底有多重要假设你收到一个这样的XML文件：<bookstore>  <book category="编程">    <title>Python从入门到精通</title>    <author>张伟</author>    <year>2023</year>  </book>  <book category="小说">    <title>三体</title>    <author>刘慈欣</author>    <year>2008</year>  </book></bookstore>需要提取所有书名和作者信息，你会怎么做？手动复制粘贴？当文件有几百MB时这显然行不通！Python的ElementTree模块就是为解决这类问题而生的。二、ElementTree快速入门1. 加载XML的两种方式方式1：直接解析字符串123456789101112import xml.etree.ElementTree as ET xml_string = """<bookstore>  <book category="编程">    <title>Python从入门到精通</title>    <author>张伟</author>  </book></bookstore>""" root = ET.fromstring(xml_string)  # 从字符串加载方式2：读取XML文件12tree = ET.parse('books.xml')  # 从文件加载root = tree.getroot()2. 遍历XML节点获取所有book节点：12345for book in root.findall('book'):    print("找到一本书：")    print(f"类别：{book.get('category')}")    print(f"书名：{book.find('title').text}")    print(f"作者：{book.find('author').text}")输出结果：找到一本书：类别：编程书名：Python从入门到精通作者：张伟找到一本书：类别：小说书名：三体作者：刘慈欣三、ElementTree核心操作详解1. 查找元素的三种方法12345678# 查找第一个匹配的节点first_book = root.find('book') # 查找所有匹配节点all_books = root.findall('book') # 用XPath查找（更强大）titles = root.findall('.//title')  # 查找所有title节点2. 获取节点属性与文本12345678910# 获取属性category = book.get('category') # 获取文本内容title = book.find('title').text # 处理可能不存在的节点year = book.find('year')if year is not None:    print(year.text)3. 处理命名空间遇到带命名空间的XML怎么办？123<ns:book xmlns:ns="http://example.com">  <ns:title>XML解析指南</ns:title></ns:book>解析方法：12ns = {'ns': 'http://example.com'}title = root.find('ns:title', ns).text四、实战：解析真实场景XML假设要处理一个RSS订阅源（实际就是XML格式）：12345678910import requests url = "https://example.com/rss"response = requests.get(url)root = ET.fromstring(response.content) for item in root.findall('.//item'):    print(f"标题：{item.find('title').text}")    print(f"链接：{item.find('link').text}")    print("----")五、性能优化技巧当处理大型XML文件时（比如几百MB）：1. 使用迭代解析1234for event, elem in ET.iterparse('big_file.xml'):    if elem.tag == 'book':        print(elem.find('title').text)        elem.clear()  # 及时清理内存2. 使用lxml加速12345from lxml import etree  # 需要安装：pip install lxml # 比标准库快3-5倍parser = etree.XMLParser(remove_blank_text=True)tree = etree.parse('books.xml', parser)六、常见问题解决方案问题1：编码错误怎么办？12with open('data.xml', 'r', encoding='utf-8') as f:    tree = ET.parse(f)问题2：处理特殊字符12from xml.sax.saxutils import escapesafe_text = escape('文本&特殊字符<>"')问题3：美化输出123from xml.dom import minidomxml_str = ET.tostring(root)pretty_xml = minidom.parseString(xml_str).toprettyxml()七、完整代码示例12345678910111213141516171819202122import xml.etree.ElementTree as ET def parse_xml(file_path):    tree = ET.parse(file_path)    root = tree.getroot()         results = []    for book in root.findall('book'):        data = {            'category': book.get('category'),            'title': book.find('title').text,            'author': book.find('author').text,            'year': book.find('year').text if book.find('year') is not None else None        }        results.append(data)         return results # 使用示例books = parse_xml('books.xml')for book in books:    print(f"{book['title']}（{book['year']}）")

方法 1：直接访问属性（原始代码）12345import torchimport sys print("PyTorch Version: {}".format(torch.__version__))print("Python Version: {}".format(sys.version))特点：简单直接，无需额外依赖。适用于快速检查版本信息。方法 2：通过命令行工具如果希望在脚本外部检查版本，可以直接使用命令行工具。Python 版本123python --version# 或python -VPyTorch 版本1python -c "import torch; print(torch.__version__)"特点：适用于脚本外部检查，无需编写 Python 代码。可以集成到 CI/CD 流程中。方法 3：使用 torch.version 模块PyTorch 提供了一个 torch.version 模块，可以获取更详细的版本信息。12345678910import torchimport sys # 获取 PyTorch 版本信息print("PyTorch Version: {}".format(torch.version.__version__))  # 或直接使用 torch.__version__print("PyTorch CUDA Version: {}".format(torch.version.cuda))  # 获取 CUDA 版本print("PyTorch cuDNN Version: {}".format(torch.backends.cudnn.version()))  # 获取 cuDNN 版本 # Python 版本print("Python Version: {}".format(sys.version))特点：可以获取 CUDA 和 cuDNN 的版本信息，对于调试 GPU 环境非常有用。torch.version 提供了更细粒度的版本控制。方法 4：使用 pkg_resourcespkg_resources 是 setuptools 提供的一个工具，可以查询已安装包的版本信息。123456789101112import pkg_resources # 获取 PyTorch 版本try:    pytorch_version = pkg_resources.get_distribution("torch").version    print("PyTorch Version: {}".format(pytorch_version))except pkg_resources.DistributionNotFound:    print("PyTorch is not installed.") # Python 版本仍通过 sys 模块import sysprint("Python Version: {}".format(sys.version))特点：可以查询任何已安装包的版本，而不仅仅是 PyTorch。如果包未安装，会捕获 DistributionNotFound 异常。方法 5：使用 platform 模块（补充 Python 信息）虽然 sys.version 已经提供了 Python 版本信息，但 platform 模块可以提供更详细的系统信息。123456import torchimport platform print("PyTorch Version: {}".format(torch.__version__))print("Python Version: {}".format(platform.python_version()))print("Platform: {}".format(platform.platform()))特点：platform.platform() 提供了操作系统的详细信息。适用于需要记录系统环境的场景。方法 6：结合 subprocess 调用命令行如果需要在 Python 脚本中调用外部命令行工具，可以使用 subprocess 模块。123456789101112import subprocess def get_python_version():    result = subprocess.run(["python", "--version"], capture_output=True, text=True)    return result.stdout.strip() def get_pytorch_version():    result = subprocess.run(["python", "-c", "import torch; print(torch.__version__)"], capture_output=True, text=True)    return result.stdout.strip() print("Python Version: {}".format(get_python_version()))print("PyTorch Version: {}".format(get_pytorch_version()))特点：适用于需要从外部命令行获取信息的场景。可以灵活地调用其他命令行工具。方法 7：使用 torch.utils.collect_envPyTorch 提供了一个 torch.utils.collect_env 工具，可以收集详细的系统环境信息，包括 PyTorch、Python、CUDA、cuDNN 等。1234import torch env_info = torch.utils.collect_env()print(env_info)特点：提供全面的环境信息，适合用于调试和问题报告。输出格式为字典，可以进一步处理。