admin
aac9f5934d
- 新增websocket客户端和服务端实现 - 添加图片压缩工具和快速压缩脚本 - 实现学生信息处理相关API - 添加MQTT客户端和消息处理功能 - 更新.gitignore忽略更多文件类型 - 添加数据库操作工具和示例 - 实现多个测试脚本和工具类
277 lines
8.7 KiB
Python
277 lines
8.7 KiB
Python
#!/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() |