这个应用和上个程序类似,但是是通过人物姿势来查找图片的。

启动py程序后,在浏览器访问,上传对应人物照片,就可以在本地文件夹搜索查找相同或相似姿势的人物照片了。(代码依然在结尾处)

也可以勾选不同的关注部位,来匹配姿势

单击上方这一排上传到图片和搜索出的图片,可以将图片对应的姿势分析并绘制出来

单击下方第二排的图片,可以将绘制会骨骼姿势的图片放大,并支持滚轮缩放;

程序启动后,在浏览器访问http://127.0.0.1:5000即可,访问后将本地需要查找的路径粘贴到2的位置,然后创建索引,首次创建会消耗比较多的时间,索引创建完成后,就可以上传需要查找的人物姿势照片进行搜索了

可以设置不同的相似度进行搜图,建议0.8;

以下为代部分(千问及百度提供技术支持,图片由百度AI生成,避免肖像及版权风险):

#索引通过SQLite实现
# pose_match_web.py
import os
import cv2
import sqlite3
from PIL import Image
import mediapipe as mp
import numpy as np
import pickle
#from sklearn.metrics.pairwise import cosine_similarity
from sklearn.preprocessing import MinMaxScaler
from pathlib import Path
import logging
from flask import Flask, render_template_string, send_file,request, jsonify, send_from_directory, url_for
from werkzeug.utils import secure_filename
from urllib.parse import quote, unquote
import traceback
import base64
from scipy.spatial import procrustes

# -----------------------------
# 配置
# -----------------------------
INDEX_DB = 'index.db' #索引库文件
UPLOAD_FOLDER = 'uploads'
DEFAULT_SIMILARITY_THRESHOLD = 0.8  # ✅ 降低阈值,更容易匹配
DEFAULT_RESULTS_COUNT = 5
os.makedirs(UPLOAD_FOLDER, exist_ok=True)

# 初始化 MediaPipe
mp_pose = mp.solutions.pose
mp_drawing = mp.solutions.drawing_utils

# -----------------------------
# 初始化 Flask
# Flask App
# -----------------------------
app = Flask(__name__)
app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER

def create_pose_estimator():
    return mp_pose.Pose(
        static_image_mode=True,
        model_complexity=2,
        enable_segmentation=False,
        min_detection_confidence=0.5
    )

# 初始化索引库
def init_database(db_folder):
    """处理 SQLite 数据库文件夹路径"""
    if not isinstance(db_folder, Path):
        db_folder = Path(db_folder)
    # 确保文件夹存在
    db_folder.mkdir(parents=True, exist_ok=True)
    
    # 构建数据库文件的完整路径
    db_path = db_folder / INDEX_DB
	
    """初始化 SQLite 数据库"""
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS image_index (
            id INTEGER PRIMARY KEY,
            filename TEXT UNIQUE NOT NULL,
            file_path TEXT NOT NULL,
            file_size INTEGER NOT NULL,
            file_mtime REAL NOT NULL,
            features BLOB NOT NULL  -- 存储序列化的特征向量
        )
    ''')
    
    # 创建索引,加速查询
    cursor.execute('CREATE INDEX IF NOT EXISTS idx_filename ON image_index(filename)')
    cursor.execute('CREATE INDEX IF NOT EXISTS idx_mtime ON image_index(file_mtime)')
    
    conn.commit()
    conn.close()
    print("✅ 数据库初始化完成")

# -----------------------------
# 构建索引 → 保存在目标文件夹下
# -----------------------------
def build_or_update_index(image_folder='images'):
    """构建或更新 SQLite 索引"""
    image_folder = Path(image_folder)
    if not image_folder.exists():
        raise FileNotFoundError(f"图片文件夹不存在: {image_folder}")
    
    image_files = [f for f in image_folder.iterdir() 
                   if f.is_file() and f.suffix.lower() in SUPPORTED_EXTENSIONS]
    
    pose = create_pose_estimator()
	# 构建数据库文件的完整路径
    index_path = image_folder / INDEX_DB
    conn = sqlite3.connect(index_path)
    cursor = conn.cursor()
    
    processed = 0
    reused = 0
    
    for img_path in image_files:
        stat = img_path.stat()
        file_size = stat.st_size
        file_mtime = stat.st_mtime
        filename = img_path.name
        
        # 检查是否已存在且未修改
        cursor.execute(
            "SELECT file_mtime, features FROM image_index WHERE filename = ?", 
            (filename,)
        )
        row = cursor.fetchone()
        
        if row and abs(row[0] - file_mtime) < 1:  # 时间差小于1秒视为未修改
            reused += 1
            continue  # 跳过,复用旧特征
        
        # 否则重新提取特征
        try:
            pil_img = Image.open(img_path).convert('RGB')
            rgb_img = np.array(pil_img)
            results = pose.process(cv2.cvtColor(rgb_img, cv2.COLOR_BGR2RGB))
            
            if not results.pose_landmarks:
                print(f"⚠️ 未检测到姿态,跳过: {filename}")
                continue  # 跳过这张图
                
            features = extract_shape_features(results.pose_landmarks,DEFAULT_ALL_SET)
            if features is None:
                print(f"⚠️ 特征提取失败,跳过: {filename}")
                continue
            
			 # ✅ 安全:确保 features 是 list 或 np.array
            if not isinstance(features, (list, np.ndarray)):
                print(f"❌ 特征格式错误: {type(features)}")
                continue 	

            # 序列化特征
            features_blob = pickle.dumps(features)
            
            # 插入或替换
            cursor.execute('''
                INSERT OR REPLACE INTO image_index 
                (filename, file_path, file_size, file_mtime, features)
                VALUES (?, ?, ?, ?, ?)
            ''', (filename, str(img_path), file_size, file_mtime, features_blob))
            
            processed += 1
            
        except Exception as e:
            print(f"❌ 处理失败 {img_path}: {e}")
            continue  # 出错也跳过,不要中断整个索引过程
    
    conn.commit()
    conn.close()
    pose.close()
    
    print(f"✅ 索引更新完成 | 新增/更新: {processed}, 复用: {reused}, 总计: {processed + reused}")
    return processed,reused,processed + reused,index_path


def load_index_from_sqlite(db_path):
    """
    从 SQLite 数据库加载索引数据,格式与旧 pkl 一致
    返回: [{'filename': ..., 'file_size': ..., 'file_mtime': ..., 'features': [...], 'image_path': ...}, ...]
    """
    try:
        conn = sqlite3.connect(str(db_path))
        cursor = conn.cursor()
        
        cursor.execute("SELECT filename, file_path, file_size, file_mtime, features FROM image_index")
        rows = cursor.fetchall()
        conn.close()
        
        index_data = []
        for row in rows:
            filename, file_path, file_size, file_mtime, features_blob = row
            try:
                features = pickle.loads(features_blob)
                index_data.append({
                    'filename': filename,
                    'image_path': file_path,
                    'file_size': file_size,
                    'file_mtime': file_mtime,
                    'features': features
                })
            except Exception as e:
                print(f"❌ 解析特征失败 {filename}: {e}")
                continue
        
        return index_data
    
    except Exception as e:
        raise RuntimeError(f"读取 SQLite 索引失败: {e}")

# -----------------------------
# 角度计算(保持不变)
# -----------------------------
def calculate_angle(a, b, c):
    ba = a - b
    bc = c - b
    cosine_angle = np.dot(ba, bc) / (np.linalg.norm(ba) * np.linalg.norm(bc))
    angle = np.arccos(np.clip(cosine_angle, -1.0, 1.0))
    return np.degrees(angle)

# -----------------------------
# 特征提取:提取指定关键点的坐标(形状特征)
# -----------------------------
def extract_shape_features(landmarks, indices):
    """
    提取指定关键点的 (x, y) 坐标作为形状特征
    """
    points = []
    for idx in indices:
        if idx < len(landmarks.landmark):
            lm = landmarks.landmark[idx]
            points.append([lm.x, lm.y])
        else:
            return None  # 缺失任意点则跳过
    return np.array(points)

# -----------------------------
# 提取特征 
# -----------------------------
def extract_pose_from_image(image_path, pose_estimator,selectedPart):
    """
    使用 PIL 支持 .webp、.jpg、.png 等多种格式
    """
    try:
        # 使用PIL读取图片
        image = Image.open(image_path).convert('RGB')  # 自动支持 webp, png等
        image_np = np.array(image)
        image_rgb = cv2.cvtColor(image_np, cv2.COLOR_RGB2BGR)  # PIL 是 RGB,OpenCV 是 BGR
    except Exception as e:
        logging.warning(f"无法读取图片 (PIL): {image_path}, 错误: {e}")
        return None

    if image_rgb is None or image_rgb.size == 0:
        logging.warning(f"图片为空或解码失败: {image_path}")
        return None

    results = pose_estimator.process(cv2.cvtColor(image_rgb, cv2.COLOR_BGR2RGB))

    if not results.pose_landmarks:
        logging.warning(f"未检测到人体: {image_path}")
        return None

    features = extract_shape_features(results.pose_landmarks,selectedPart)
    return {
        'features': features,
        'image_path': str(image_path)
    }

SUPPORTED_EXTENSIONS = {'.jpg', '.jpeg', '.png', '.bmp', '.webp', '.tiff', '.tif'}



# HTML 模板(内嵌)
HTML_TEMPLATE = '''
<!DOCTYPE html>
<html lang="zh">
<head>
    <meta charset="UTF-8">
    <title>姿势匹配服务</title>
    <style>
        body { font-family: Arial, sans-serif; margin: 20px;}
        .all {display: flex; gap: 20px; /* 现代浏览器支持的间距属性 */ }
		.resultsAll {display: flex; gap: 20px; /* 现代浏览器支持的间距属性 */ align-items: flex-start;}
		.upload, .query, .index { margin: 20px 0; padding: 15px;width: 400px; border: 1px solid #ddd; border-radius: 5px; }
        .results { display: flex; flex-wrap: wrap; gap: 10px; margin-top: 20px; }
        .result-item { text-align: center; max-width: 150px; }
        .result-item img { width: 100%; height: auto; }
        input[type="text"], input[type="number"] { width: 200px; padding: 5px; margin: 5px 0; }
        button { padding: 8px 16px; margin: 5px; }
		
		/* 图片缩放功能 */ 
		.gallery { display: flex; flex-wrap: wrap; gap: 15px; justify-content: center; }
		.gallery img { width: 200px; height: 200px; object-fit: cover; border: 3px solid #ddd; border-radius: 8px; cursor: pointer; transition: transform 0.2s, box-shadow 0.2s; }
		.gallery img:hover { transform: scale(1.05); box-shadow: 0 4px 10px rgba(0,0,0,0.2); }
		.modal { display: none; position: fixed; top: 0; left: 0; width: 100%; height: 100%; background-color: rgba(0,0,0,0.9); justify-content: center; align-items: center; z-index: 1000; animation: fadeIn 0.3s ease-in-out; overflow: hidden; }
		.modal img { max-width: 90vw; max-height: 90vh; border: 4px solid white; border-radius: 6px; box-shadow: 0 0 20px rgba(255,255,255,0.1); animation: zoomIn 0.3s ease-out; transition: transform 0.2s; cursor: grab; }
		.modal img:active { cursor: grabbing; }
		@keyframes fadeIn { from { opacity: 0; } to { opacity: 1; } }
		@keyframes zoomIn { from { transform: scale(0.8); opacity: 0; } to { transform: scale(1); opacity: 1; } }
	</style>
</head>
<body>
    <h1>📷 姿势匹配服务</h1>
	<div class = "all">
		<div class="upload">
			<h2>上传查询图片</h2>
			<form method="post" enctype="multipart/form-data" action="/upload" onsubmit="handleUpload(event)">
				<input onchange="handleUpload(event)" type="file" name="file" accept="image/*" required>
				<button type="submit">上传</button>
			</form>
			<p><strong>当前图片:</strong> <span id="currentFile">无</span></p>
		</div>

		<div class="index">
			<h2>创建/更新索引</h2>
			<form method="post" action="/create_index" onsubmit="updateIndex(event)">
				<label>目标文件夹路径(如:images, yoga_poses 等):</label><br>
				<input type="text" id = "folder1" name="folder" value="images" style="width: 350px;"><br>
				<button type="submit">🔍 创建/更新索引</button>
			</form>
			<p><strong>状态:</strong> <span id="indexStatus">未创建</span></p>
		</div>

		<div class="query">
			<h2>姿势搜图</h2>
			<label>相似度阈值(0.0~1.0,默认 0.6):</label><br>
			<input type="number" id="similarity" step="0.01" min="0" max="1" value=""><br>
			<label>显示数量(默认 5):</label><br>
			<input type="number" id="count" min="1" value=""><br>

			<label>关注部位:</label><br>
			  <div class="controls">
				<button id="selectAll">全选/取消</button>
				<label><input type="checkbox" id="left_hand" checked> 左手</label>&nbsp;
				<label><input type="checkbox" id="right_hand" checked> 右手</label>&nbsp;
				<label><input type="checkbox" id="left_leg" checked> 左腿</label>&nbsp;
				<label><input type="checkbox" id="right_leg" checked> 右腿</label>&nbsp;
			  </div>
			<button onclick="findSimilar()">🔎 姿势搜图</button>
		</div>
    </div>
	<div class = "resultsAll">
		<h2>上传图片</h2>
		<div id="uploadSource" class="results"></div>
		
		<h2>匹配结果</h2>
		<div id="results" class="results"></div>
	</div>
    
    <div class="modal" id="modal" onclick="closeModal(event)">
        <img id="modal-img" />
    </div>
    
	<script>
		document.addEventListener('DOMContentLoaded', function () {
			// DOM加载完成后执行的代码 而不需要等待样式表、图片和子框架的加载
            fetch('/getArgs', {
                method: 'POST'
            })
            .then(response => response.json())
            .then(data => {
                document.getElementById('similarity').value = data.similarity;
                document.getElementById('count').value = data.count;
                
            });
		});

        let currentFile = null;
        function handleUpload(event) {
            const fileInput = event.target;
            const file = fileInput.files[0];
            
            if (!file) {
                return;
            }
			document.getElementById('currentFile').textContent = file.name;
            // 创建FormData对象并添加文件
            const formData = new FormData();
            formData.append('file', file);

            fetch('/upload', {
                method: 'POST',
                body: formData
            })
            .then(response => response.json())
            .then(data => {
                if (data.filename) {
                    currentFile = data.filename;
                    document.getElementById('currentFile').textContent = currentFile;
                    
					const uploadedUrl = data.uploaded_img_url;
					const names = "up_"+data.filename;
					document.getElementById('uploadSource').innerHTML = 
                    `<div class="result-item">
						<img src="${uploadedUrl}" class="result-item" onclick="showImage('${data.filename}','${names}','${data.upload_path}')" ><br>
						<p>展示:<span id = "${names}"/></p>
						<img id = "img_${names}" src="" onclick = zoomImg(event) >
					  </div>`
					document.getElementById('results').innerHTML = '';

                } else {
                    alert('上传失败: ' + (data.error || '未知错误'));
                }
            });
        }

        function updateIndex(event) {
            document.getElementById('indexStatus').textContent = '更新中...';
			event.preventDefault();
            const formData = new FormData(event.target);
            fetch('/create_index', {
                method: 'POST',
                body: formData
            })
            .then(response => response.json())
            .then(data => {
                if (data.error) {
                    document.getElementById('indexStatus').textContent = '错误: ' + data.error;
                } else {
                    document.getElementById('indexStatus').textContent = 
                        `✅ 索引已更新,新增 ${data.processed},
						更新 ${data.reused},共计${data.allcount}:张图片,索引路径${data.index_path}`;
                }
            });
        }

        function findSimilar() {
            if (!currentFile) {
                alert('请先上传一张图片!');
                return;
            }
            document.getElementById('results').innerHTML = '🔍 匹配中...';

            const similarity = document.getElementById('similarity').value;
            const count = document.getElementById('count').value;
			const selected = getSelected();
			//const  folder1 = document.getElementById('folder1').value;
			const folder = document.getElementsByName('folder')[0].value;
			//alert(folder)
            fetch('/find_similar?filename=' + encodeURIComponent(currentFile) + 
                  '&similarity=' + similarity + '&count=' + count + '&folder=' + folder + '&selected=' + selected.join(','))
            .then(response => response.json())
            .then(data => {
                const resultsDiv = document.getElementById('results');
                resultsDiv.innerHTML = '';
                if (data.error) {
                    resultsDiv.innerHTML = '<p style="color:red;">' + data.error + '</p>';
                    return;
                }
                if (data.length === 0) {
                    resultsDiv.innerHTML = '<p>未找到相似姿势</p>';
                    return;
                }

                data.forEach(item => {
                    resultsDiv.innerHTML += `
                        <div class="result-item">
                            <img src="${item.img}" alt="match" onclick="showImage('${item.image_name}','${item.image_name}','null')">
							<p>名称:${item.image_name}</p>
                            <p>相似度: ${item.similarity.toFixed(3)}</p>
							<p>展示:<span id = "${item.image_name}"/></p>
							<img id = "img_${item.image_name}" src="" onclick = zoomImg(event) >
                        </div>`;
                });
            });
         }

        function showImage(name,names,folder) {
			if (folder == "null"){
				folder = document.getElementsByName('folder')[0].value;
			}
			types = name.split('.').pop();
			const selected = getSelected();
			const spanEle = document.getElementById(names);
			const imgEle = document.getElementById('img_'+names);
			spanEle.innerHTML = '';
			imgEle.src = '';
			const url = '/show_image?folder=' + folder + '&name=' + name + '&types=' + types + '&selected=' + selected.join(',');
            fetch(url, {
                method: 'POST',
				headers: {
					'Content-Type': 'application/x-www-form-urlencoded',
				},
            })
            .then(response => response.json())
            .then(data => {
                
                
                if (data.error) {
                    spanEle.innerHTML = 'data.error';
                    return;
                }

                spanEle.innerHTML = '√';
				//imgEle.src=`"data:image/${data.type};base64,${data.str}"` 
				const blobUrl = base64ToBlobUrl(`image/${data.type}`, data.str);
				imgEle.src = blobUrl;
				
				//alert(imgEle.src)
            });
			URL.revokeObjectURL(blobUrl);
         }
	
		function base64ToBlobUrl(mimeType,base64Str) {
			// 将 base64 转换为二进制数据
			const binaryString = atob(base64Str);
			const bytes = new Uint8Array(binaryString.length);
			for (let i = 0; i < binaryString.length; i++) {
				bytes[i] = binaryString.charCodeAt(i);
			}
			
			// 创建 Blob 和 URL
			const blob = new Blob([bytes], { type: mimeType });
			return URL.createObjectURL(blob);
		}

		document.getElementById('selectAll').addEventListener('click', function() {
		  const checkboxes = document.querySelectorAll('.controls input[type="checkbox"]');
		  const currentStatus = Array.from(checkboxes).some(cb => !cb.checked);
		  checkboxes.forEach(cb => cb.checked = currentStatus);
		});
		
		function getSelected() {
		  const selected = [];
		  if (document.getElementById('left_hand').checked) selected.push('0');
		  if (document.getElementById('right_hand').checked) selected.push('1');
		  if (document.getElementById('left_leg').checked) selected.push('2');
		  if (document.getElementById('right_leg').checked) selected.push('3');

		  if (selected.length === 0) {
			//alert("请至少选择一个部位");
		  }
		  return selected;
		}

		//图片缩放函数------start
		const gallery = document.getElementById('gallery');
		const modal = document.getElementById('modal');
		const modalImg = document.getElementById('modal-img');
		
		let scale = 1;
		let isDragging = false;
		let startPos = { x: 0, y: 0 };
		let currentTranslate = { x: 0, y: 0 };

		// 打开模态框
		function zoomImg(e) {
		  if (e.target.tagName === 'IMG') {
			modalImg.src = e.target.src;
			modal.style.display = 'flex';
			document.body.style.overflow = 'hidden';
			scale = 1;
			currentTranslate = { x: 0, y: 0 };
			modalImg.style.transform = `scale(${scale}) translate(0px, 0px)`;
		  }
		}

		// 关闭模态框(直接绑定在HTML上)
		function closeModal(e) {
		  if (e.target === modal || e.target === modalImg) {
			modal.style.display = 'none';
			document.body.style.overflow = 'auto';
			// 重置状态
			isDragging = false;
			scale = 1;
			currentTranslate = { x: 0, y: 0 };
		  }
		}

		// 滚轮缩放
		modalImg.addEventListener('wheel', function(e) {
		  e.preventDefault();
		  const delta = e.deltaY > 0 ? -0.1 : 0.1;
		  scale = Math.min(Math.max(scale + delta, 0.5), 5); // 支持放大到5倍
		  applyTransform();
		});

		// 鼠标按下:开始拖拽
		modalImg.addEventListener('mousedown', function(e) {
		  if (scale <= 1) return; // 仅在放大时允许拖拽
		  isDragging = true;
		  startPos = { x: e.clientX - currentTranslate.x, y: e.clientY - currentTranslate.y };
		  modalImg.style.cursor = 'grabbing';
		});

		// 鼠标移动:实时更新位置
		document.addEventListener('mousemove', function(e) {
		  if (!isDragging) return;
		  currentTranslate = {
			x: e.clientX - startPos.x,
			y: e.clientY - startPos.y
		  };
		  applyTransform();
		});

		// 鼠标松开:结束拖拽
		document.addEventListener('mouseup', function() {
		  isDragging = false;
		  modalImg.style.cursor = 'grab';
		});

		// 应用缩放和平移变换
		function applyTransform() {
		  modalImg.style.transform = `scale(${scale}) translate(${-currentTranslate.x}px, ${-currentTranslate.y}px)`;
		}
		//图片缩放函数------end

	</script>
</body>
</html>
'''

@app.route('/')
def index():
    return render_template_string(HTML_TEMPLATE)

@app.route('/upload', methods=['POST'])
def upload():
    if 'file' not in request.files:
        return jsonify({'error': '未选择文件'})
    file = request.files['file']
    if file.filename == '':
        return jsonify({'error': '文件名为空'})

    filename = secure_filename(file.filename)
    upload_path = os.path.join(app.config['UPLOAD_FOLDER'], filename)
    file.save(upload_path)

    # 返回上传图的 URL
    uploaded_img_url = f"/uploads/{filename}"

    return jsonify({
        'filename': filename,
        'uploaded_img_url': uploaded_img_url,
		"upload_path":app.config['UPLOAD_FOLDER']
    })

@app.route('/uploads/<filename>')
def uploaded_file(filename):
    return send_from_directory(UPLOAD_FOLDER, filename)

@app.route('/create_index', methods=['POST'])
def create_index():
    folder = request.form.get('folder', '').strip()
    if not folder:
        return jsonify({'error': '请填写图片文件夹路径'})

    try:
        print(f"=====目标路径文件夹{folder}")
        image_folder = Path(folder)
        if not image_folder.is_absolute():
            image_folder = Path.cwd() / image_folder
		
		# 1. 初始化数据库
        init_database(folder)
    
		# 2. 构建或更新索引
		#build_or_update_index(str(image_folder))
		
        processed, reused,allcount,index_path = build_or_update_index(str(folder))
        return jsonify({
            'status': 'success',
			'processed': processed,
			'reused': reused,
            'allcount': allcount,
            'index_path': str(index_path)
        })
    except Exception as e:
        traceback.print_exc()#打印追踪信息
        return jsonify({'error': str(e)})

		
@app.route('/getArgs', methods=['POST'])
def getArgs():
	return jsonify({
		'similarity': DEFAULT_SIMILARITY_THRESHOLD,
		'count': DEFAULT_RESULTS_COUNT,
	})


@app.route('/find_similar')
def find_similar():
    filename = request.args.get('filename')
    if not filename:
        return jsonify({'error': '缺少查询图片'})

    query_path = os.path.join(UPLOAD_FOLDER, filename)
    if not os.path.exists(query_path):
        return jsonify({'error': f'查询图片不存在: {query_path}'})

    try:
        similarity_threshold = float(request.args.get('similarity', 0.1))
        results_count = int(request.args.get('count', 5))
    except:
        return jsonify({'error': '参数错误'})

    # 获取目标文件夹(默认 images)
    folder = request.args.get('folder', '').strip()
    image_folder = Path(folder)
    if not image_folder.is_absolute():
        image_folder = Path.cwd() / image_folder

    index_path = image_folder / INDEX_DB
    if not index_path.exists():
        return jsonify({'error': f'索引文件不存在: {index_path}'})

    try:
        #index_data = joblib.load(str(index_path))
		# ✅ 使用 SQLite 加载索引
        index_data = load_index_from_sqlite(index_path)
        print(f"🔍 加载索引: {len(index_data)} 张图片")  # 调试
    except Exception as e:
        return jsonify({'error': f'加载索引失败: {str(e)}'})

    # 提取上传图片查询特征
    pose_estimator = create_pose_estimator()
    selectedPart = getSelectedPart()
    query_data = extract_pose_from_image(query_path, pose_estimator,selectedPart)
    
    if not query_data or query_data['features'] is None:
        return jsonify({'error': '无法提取查询图片特征', 'debug': '可能是无人体或关键点检测失败'})

    query_vec = query_data['features']
    print(f"✅ 查询图特征: {query_vec}")  # 调试输出

    similarities = []
    for idx, item in enumerate(index_data):
        db_vec = getDbSelectVec(item['features'],selectedPart)
        print(f"📁 库中图 {idx + 1} 特征: {db_vec}")  # 调试输出

        # 确保都是 2D 数组
        if query_vec.ndim == 1:
            query_vec = query_vec.reshape(1, -1)
        if db_vec.ndim == 1:
            db_vec = db_vec.reshape(1, -1)

        try:
            sim = shape_similarity_score(query_vec, db_vec)
            sim = float(1 - min(sim, 1.0))  # 转为 [0,1] 范围
            print(f"📊 相似度得分: {sim},设定值:{similarity_threshold}")  # 调试
            if sim >= similarity_threshold: #(越小越严格)
                similarities.append((item, sim))
        except Exception as e:
            print(f"❌ 计算相似度失败: {e}")
            similarities.append((item, 0.0))

    # 按相似度排序
    similarities.sort(key=lambda x: x[1], reverse=True)
    matches = []
    for item,sim in similarities[:results_count]:
        img_path = Path(item['image_path'])
        if img_path.exists():
			# 使用 URL 编码路径,防止特殊字符
            img_url = f"/image_proxy/{quote(str(img_path))}"
            matches.append({
			    'image_name': img_path.name,
                'img': img_url,
                'similarity': sim
            })

    print(f"🎯 返回匹配数: {len(matches)}")
    return jsonify(matches)

# 处理数据库索引,使点数对齐
def getDbSelectVec(dbFeatures,selectedPart):
	reFeatures = dbFeatures
	if len(dbFeatures) != len(selectedPart):
		lisTemp = []
		for i in selectedPart:
			lisTemp.append(dbFeatures[i])
		reFeatures = np.array(lisTemp)
	return reFeatures


# -----------------------------
# 相似度匹配:使用 Procrustes 对齐 + 欧氏距离
# -----------------------------
def shape_similarity_score(points1, points2):
    """
    计算两个点集的形状相似度(越小越相似)
    """
    if points1 is None or points2 is None or len(points1) != len(points2):
        print("点数不同,无法计算相似度:",len(points1), len(points2))
        return float('inf')
    try:
        # Procrustes 分析:对齐两个形状(去除平移、旋转、缩放)
        mtx1, mtx2, disparity = procrustes(points1, points2)
        # 计算对齐后点集的欧氏距离
        dist = np.sum((mtx1 - mtx2) ** 2)
        return np.sqrt(dist)
    except Exception as e:
        traceback.print_exc()#打印追踪信息
        return float('inf')  # 对齐失败则视为不相似

KEYPOINTS_OF_INTEREST = [
    mp_pose.PoseLandmark.NOSE,                    # 0: 鼻子
    mp_pose.PoseLandmark.LEFT_EYE_INNER,         # 1: 左眼内眼角
    mp_pose.PoseLandmark.LEFT_EYE,               # 2: 左眼中心
    mp_pose.PoseLandmark.LEFT_EYE_OUTER,         # 3: 左眼外眼角
    mp_pose.PoseLandmark.RIGHT_EYE_INNER,        # 4: 右眼内眼角
    mp_pose.PoseLandmark.RIGHT_EYE,              # 5: 右眼中心
    mp_pose.PoseLandmark.RIGHT_EYE_OUTER,        # 6: 右眼外眼角
    mp_pose.PoseLandmark.LEFT_EAR,               # 7: 左耳
    mp_pose.PoseLandmark.RIGHT_EAR,              # 8: 右耳
    mp_pose.PoseLandmark.MOUTH_LEFT,             # 9: 嘴巴左侧
    mp_pose.PoseLandmark.MOUTH_RIGHT,            # 10: 嘴巴右侧
    mp_pose.PoseLandmark.LEFT_SHOULDER,          # 11: 左肩
    mp_pose.PoseLandmark.RIGHT_SHOULDER,         # 12: 右肩
    mp_pose.PoseLandmark.LEFT_ELBOW,             # 13: 左肘
    mp_pose.PoseLandmark.RIGHT_ELBOW,            # 14: 右肘
    mp_pose.PoseLandmark.LEFT_WRIST,             # 15: 左手腕
    mp_pose.PoseLandmark.RIGHT_WRIST,            # 16: 右手腕
    mp_pose.PoseLandmark.LEFT_PINKY,             # 17: 左小指指尖
    mp_pose.PoseLandmark.RIGHT_PINKY,            # 18: 右小指指尖
    mp_pose.PoseLandmark.LEFT_INDEX,             # 19: 左食指指尖
    mp_pose.PoseLandmark.RIGHT_INDEX,            # 20: 右食指指尖
    mp_pose.PoseLandmark.LEFT_THUMB,             # 21: 左拇指尖
    mp_pose.PoseLandmark.RIGHT_THUMB,            # 22: 右拇指尖
    mp_pose.PoseLandmark.LEFT_HIP,               # 23: 左髋部(臀部)
    mp_pose.PoseLandmark.RIGHT_HIP,              # 24: 右髋部(臀部)
    mp_pose.PoseLandmark.LEFT_KNEE,              # 25: 左膝
    mp_pose.PoseLandmark.RIGHT_KNEE,             # 26: 右膝
    mp_pose.PoseLandmark.LEFT_ANKLE,             # 27: 左踝
    mp_pose.PoseLandmark.RIGHT_ANKLE,            # 28: 右踝
    mp_pose.PoseLandmark.LEFT_HEEL,              # 29: 左脚后跟
    mp_pose.PoseLandmark.RIGHT_HEEL,             # 30: 右脚后跟
    mp_pose.PoseLandmark.LEFT_FOOT_INDEX,        # 31: 左脚大脚趾
    mp_pose.PoseLandmark.RIGHT_FOOT_INDEX        # 32: 右脚大脚趾
]


LH_SET = {11,13,15,17,19,21}#左手
RH_SET = {12,14,16,18,20,22}#右手
LL_SET = {23,25,27,29,31}#右腿
RL_SET = {24,26,28,30,32}#右腿
ALL_LIST = [LH_SET,RH_SET,LL_SET,RL_SET]
DEFAULT_ALL_SET = {kp.value for kp in KEYPOINTS_OF_INTEREST}# 转换为索引集合,便于快速查找

def getSelectedPart():
	#selected = request.args.getlist('selected')
	selected = request.args.get('selected').split(',')
	#处理选中部位
	selectedPart = DEFAULT_ALL_SET
	if len(selected) > 0 and selected[0] != "":
		selectedPart = set() #注意 selectedPart = {} 这会创建一个空字典,而不是空集合
		for i in selected:
			selectedPart.update(ALL_LIST[int(i)])
	print("---------",selectedPart)
	return selectedPart

@app.route('/show_image', methods=['POST'])
def show_image():
	folder = request.args.get('folder')
	fileName = request.args.get('name')
	typeArg = request.args.get('types')
	selectedPart = getSelectedPart()
	image_file = os.path.join(folder, fileName)

	mp_pose = mp.solutions.pose
	mp_drawing = mp.solutions.drawing_utils
	mp_drawing_styles = mp.solutions.drawing_styles
	
	""" 参数说明:
	mp_pose = mp.solutions.pose.Pose(#人体姿态估计函数
		static_image_mode=False, #bool, 默认: False-连续视频流 True-不连续单张图片
		model_complexity=1,      #int, 默认: 1 选择姿态检测模型的复杂度,影响推理速度和准确性 0-轻量,最快 1-中等,平衡 2-重型,最慢,精度最高
		smooth_landmarks=True,   # 是否对关键点坐标进行平滑处理 False:不进行平滑处理,每帧都独立输出原始结果,可能导致抖动。
		enable_segmentation=False, # bool, 默认: False 是否启用身体分割功能,输出一个掩码(mask),标识出人体区域。
		smooth_segmentation=True,  #bool, 默认: True 当 enable_segmentation=True 时,是否对分割掩码进行跨帧平滑处理。
		min_detection_confidence=0.5, #float ∈ [0,1], 默认: 0.5 只有当模型检测到的姿态置信度高于此值时,才会返回结果。设置过高可能导致漏检;过低可能导致误检。在 static_image_mode=True 时使用此阈值。
		min_tracking_confidence=0.5  #float ∈ [0,1], 默认: 0.5 姿态跟踪的最小置信度阈值。当 static_image_mode=False(即视频模式)时,系统尝试基于前一帧结果进行跟踪。如果跟踪置信度低于此值,则重新运行完整检测模型。较高的值确保跟踪质量,但可能增加计算开销。
	)
	"""

	with mp_pose.Pose(static_image_mode=True,model_complexity=2,enable_segmentation=False,min_detection_confidence=0.1) as pose:
		image = cv2.imread(image_file)
		if image is None:
			print(f"⚠️ 无法读取图片: {image_file}")
			return
			
		results = pose.process(cv2.cvtColor(image, cv2.COLOR_BGR2RGB))
		image_rgb = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
		results = pose.process(image_rgb) # 姿态检测
		
		if results.pose_landmarks:
			print(f"✅ 在 {image_file} 中检测到人物姿势")
			
			# 绘制,过滤重绘
			draw_filtered_pose_landmarks(image, results, selectedPart)
		else:
			print(f"❌ 在 {image_file} 中未检测到人物")
		

		ty = f"{'.'}{typeArg}"
		
		_, buffer = cv2.imencode(ty, image)
		img_base64 = base64.b64encode(buffer).decode('utf-8')
		return jsonify({
            'type': typeArg,
			'str': img_base64,
        })

	print("99999999999999999",folder)
	return jsonify("展示成功!")

def draw_filtered_pose_landmarks(image, results, indices_to_keep):
    """
    只绘制指定索引的关键点及其相关的连接线
    """
    if not results.pose_landmarks:
        return

    # 获取原始的关键点列表
    pose_landmarks = results.pose_landmarks
    landmarks = pose_landmarks.landmark

    for i, landmark in enumerate(landmarks):
        if i not in indices_to_keep or landmark.visibility <= 0.5:  # 可选:增加可见性过滤
            landmark.visibility = 0.0
            landmark.presence = 0.0


    # 手动定义哪些连接线需要绘制(且两端都在保留点中)
    POSE_CONNECTIONS = mp_pose.POSE_CONNECTIONS  # 所有原始连接
    connections_to_draw = []

    for connection in POSE_CONNECTIONS:
        start_idx, end_idx = connection
        # 只有当两个端点都存在于保留集合中时才绘制
        if (start_idx in indices_to_keep and 
            end_idx in indices_to_keep and
            landmarks[start_idx].visibility > 0.5 and 
            landmarks[end_idx].visibility > 0.5):
            connections_to_draw.append(connection)

    # 使用 MediaPipe 绘图工具绘制(会自动跳过不可见点)
    mp_drawing.draw_landmarks(
        image,
        pose_landmarks,
        connections_to_draw,
        landmark_drawing_spec=mp_drawing.DrawingSpec(color=(0, 255, 0), thickness=2, circle_radius=2),
        connection_drawing_spec=mp_drawing.DrawingSpec(color=(255, 0, 0), thickness=2)
    )


@app.route('/image_proxy/<path:encoded_path>')
def image_proxy(encoded_path):
    try:
        img_path = unquote(encoded_path)
        img_path = Path(img_path)

        if not img_path.exists() or not img_path.is_file():
            return 'Image not found', 404

        # 自动判断 MIME 类型
        mimetype = 'image/jpeg'
        ext = img_path.suffix.lower()
        if ext in ['.png']: mimetype = 'image/png'
        elif ext in ['.gif']: mimetype = 'image/gif'
        elif ext in ['.webp']: mimetype = 'image/webp'
        elif ext in ['.bmp']: mimetype = 'image/bmp'

        return send_file(str(img_path), mimetype=mimetype)

    except Exception as e:
        print(f"Error serving image: {e}")
        return 'Error loading image', 500

if __name__ == '__main__':
    os.makedirs(UPLOAD_FOLDER, exist_ok=True)
    print("✅ 服务启动中...")
    print(f"📁 上传目录: {os.path.abspath(UPLOAD_FOLDER)}")
    print("🌐 访问 http://127.0.0.1:5000")
    app.run(debug=False, host='127.0.0.1', port=5000)

更多推荐