xtl/python----
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 添加多个功能模块和工具脚本

Browse Source 
- 新增websocket客户端和服务端实现
- 添加图片压缩工具和快速压缩脚本
- 实现学生信息处理相关API
- 添加MQTT客户端和消息处理功能
- 更新.gitignore忽略更多文件类型
- 添加数据库操作工具和示例
- 实现多个测试脚本和工具类
This commit is contained in:
admin
2026-04-13 14:47:50 +08:00
parent 2cfd7c5a08
commit aac9f5934d
87 changed files with 216279 additions and 6519 deletions
Download Patch File Download Diff File Expand all files Collapse all files
北川/上传人脸照片_修复版.py
+277
View File
@@ -0,0 +1,277 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
上传人脸照片修复版 - 解决文件太大问题
自动压缩超过3MB的照片后再上传
"""
from requests import get, post
import os
from PIL import Image
import io
def compress_image_for_upload(image_path, max_size_mb=3):
"""
压缩图片用于上传
Args:
image_path: 图片路径
max_size_mb: 最大文件大小(MB)
Returns:
bytes: 压缩后的图片数据，如果失败返回None
"""
try:
# 检查文件大小
original_size = os.path.getsize(image_path)
original_size_mb = original_size / (1024 * 1024)
print(f"原始文件大小: {original_size_mb:.2f}MB")
# 如果文件已经小于目标大小，直接读取返回
if original_size_mb <= max_size_mb:
with open(image_path, 'rb') as f:
return f.read()
# 打开图片进行压缩
with Image.open(image_path) as img:
# 转换为RGB模式
if img.mode != 'RGB':
img = img.convert('RGB')
width, height = img.size
print(f"原始尺寸: {width}x{height}")
# 如果图片很大，先调整尺寸
if max(width, height) > 2000:
scale = 2000 / max(width, height)
new_width = int(width * scale)
new_height = int(height * scale)
img = img.resize((new_width, new_height), Image.Resampling.LANCZOS)
print(f"调整尺寸至: {new_width}x{new_height}")
# 尝试不同质量等级进行压缩
for quality in [85, 75, 65, 55, 45, 35, 25]:
output_buffer = io.BytesIO()
img.save(output_buffer, format='JPEG', quality=quality, optimize=True)
compressed_data = output_buffer.getvalue()
compressed_size_mb = len(compressed_data) / (1024 * 1024)
print(f"质量{quality}%: {compressed_size_mb:.2f}MB")
if compressed_size_mb <= max_size_mb:
compression_ratio = (original_size - len(compressed_data)) / original_size * 100
print(f"✓ 压缩成功！压缩率: {compression_ratio:.1f}%")
return compressed_data
output_buffer.close()
print("✗ 无法压缩到目标大小")
return None
except Exception as e:
print(f"压缩失败: {str(e)}")
return None
def upload_face_photo(file_path, server_url="http://10.143.25.3:8090", max_size_mb=3):
"""
上传单张人脸照片（带压缩）
Args:
file_path: 照片文件路径
server_url: 服务器地址
max_size_mb: 最大文件大小(MB)
Returns:
bool: 上传是否成功
str: 结果信息
"""
try:
filename = os.path.basename(file_path)
print(f"\n处理照片: {filename}")
# 检查文件是否存在
if not os.path.exists(file_path):
return False, f"文件不存在: {file_path}"
# 检查文件格式
if not file_path.lower().endswith(('.jpg', '.jpeg', '.png')):
return False, f"不支持的文件格式: {file_path}"
# 压缩图片
print("正在压缩图片...")
compressed_data = compress_image_for_upload(file_path, max_size_mb)
if compressed_data is None:
return False, "图片压缩失败"
# 准备上传数据
files = {
"file": (filename, compressed_data, "image/jpeg"),
}
# 获取认证token
print("正在获取认证token...")
param = {
"username": "tangchao",
"password": "123456a",
"client_id": "client",
"grant_type": "password",
"client_secret": "123456",
}
try:
token_response = get(url=f"{server_url}/oauth/token", params=param, timeout=10)
token_data = token_response.json()
token = token_data["access_token"]
except Exception as e:
return False, f"获取token失败: {e}"
# 设置请求头
headers = {
"authorization": f"Bearer {token}",
}
# 上传文件
print("正在上传文件...")
try:
response = post(
url=f"{server_url}/school/stuface/upload",
headers=headers,
files=files,
timeout=60 # 60秒超时
)
if response.status_code == 200:
result = response.json()
if result.get('success'):
return True, f"上传成功: {filename}"
else:
error_msg = result.get('msg', '未知错误')
return False, f"上传失败: {error_msg}"
else:
return False, f"HTTP错误: {response.status_code}"
except Exception as e:
return False, f"上传请求失败: {e}"
except Exception as e:
return False, f"处理过程出错: {e}"
def batch_upload_faces(students_dir="./students/", max_size_mb=3):
"""
批量上传学生人脸照片
Args:
students_dir: 学生照片目录路径
max_size_mb: 最大文件大小(MB)
"""
if not os.path.exists(students_dir):
print(f"目录不存在: {students_dir}")
return
print("开始批量上传学生人脸照片...")
print("=" * 50)
success_count = 0
failed_count = 0
failed_files = []
# 支持的图片格式
valid_extensions = {'.jpg', '.jpeg', '.png'}
# 获取所有图片文件
image_files = []
for root, dirs, files in os.walk(students_dir):
for file in files:
file_ext = os.path.splitext(file)[1].lower()
if file_ext in valid_extensions:
image_files.append(os.path.join(root, file))
print(f"找到 {len(image_files)} 张照片")
print("-" * 50)
# 处理每张照片
for i, file_path in enumerate(image_files, 1):
print(f"\n[{i}/{len(image_files)}] 处理: {os.path.basename(file_path)}")
success, message = upload_face_photo(file_path, max_size_mb=max_size_mb)
if success:
success_count += 1
print(f"✓ {message}")
else:
failed_count += 1
failed_files.append(os.path.basename(file_path))
print(f"✗ {message}")
# 短暂延迟，避免服务器压力过大
import time
time.sleep(0.5)
# 打印统计结果
print("\n" + "=" * 50)
print("上传完成！")
print(f"成功: {success_count}")
print(f"失败: {failed_count}")
print(f"总计: {len(image_files)}")
if failed_files:
print(f"\n失败的文件:")
for failed_file in failed_files:
print(f" - {failed_file}")
def main():
"""
主函数
"""
import sys
print("上传人脸照片修复版")
print("=" * 30)
print("自动压缩超过3MB的照片后再上传")
print()
if len(sys.argv) > 1:
# 命令行模式
path = sys.argv[1]
if os.path.exists(path):
if os.path.isfile(path):
# 上传单张照片
success, message = upload_face_photo(path)
print(message)
elif os.path.isdir(path):
# 批量上传
batch_upload_faces(path)
else:
print(f"路径不存在: {path}")
else:
# 交互模式
print("选择操作:")
print("1. 批量上传学生照片目录")
print("2. 上传单张照片")
print("3. 自定义目录上传")
choice = input("\n请输入选择(1-3): ").strip()
if choice == "1":
batch_upload_faces()
elif choice == "2":
file_path = input("请输入照片文件路径: ").strip()
if os.path.exists(file_path):
success, message = upload_face_photo(file_path)
print(message)
else:
print(f"文件不存在: {file_path}")
elif choice == "3":
dir_path = input("请输入照片目录路径: ").strip()
if os.path.exists(dir_path):
batch_upload_faces(dir_path)
else:
print(f"目录不存在: {dir_path}")
else:
print("无效选择")
if __name__ == "__main__":
main()