improve_gps_extraction.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
改进GPS信息提取的脚本
优化GPS数据解析和显示
"""

import re

def improve_gps_extraction():
    """改进GPS提取逻辑的脚本"""
    
    script_content = '''#!/bin/bash
# -*- coding: utf-8 -*-
# 改进GPS信息提取脚本

set -e

# 颜色定义
RED='\\033[0;31m'
GREEN='\\033[0;32m'
YELLOW='\\033[1;33m'
BLUE='\\033[0;34m'
NC='\\033[0m'

log_info() {
    echo -e "${BLUE}[INFO]${NC} $1"
}

log_success() {
    echo -e "${GREEN}[SUCCESS]${NC} $1"
}

log_warning() {
    echo -e "${YELLOW}[WARNING]${NC} $1"
}

log_error() {
    echo -e "${RED}[ERROR]${NC} $1"
}

APP_DIR="/home/ubuntu/image-extractor"

# 改进extract_info.py的GPS处理
improve_gps_handling() {
    log_info "改进GPS信息提取处理..."
    
    cat > "$APP_DIR/extract_info.py" << 'EOFF'
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
图片信息提取核心模块
用于提取图片的EXIF信息，包括拍摄时间、GPS位置等
"""

import os
import logging
from datetime import datetime
from PIL import Image
from PIL.ExifTags import TAGS, GPSTAGS
import json
import re

def make_json_serializable(obj):
    """
    将对象转换为JSON可序列化的格式
    
    Args:
        obj: 需要序列化的对象
        
    Returns:
        JSON可序列化的对象
    """
    if obj is None:
        return None
    elif isinstance(obj, (str, int, float, bool)):
        return obj
    elif isinstance(obj, dict):
        return {key: make_json_serializable(value) for key, value in obj.items()}
    elif isinstance(obj, (list, tuple)):
        return [make_json_serializable(item) for item in obj]
    elif hasattr(obj, '__iter__') and not isinstance(obj, str):
        return [make_json_serializable(item) for item in obj]
    else:
        # 处理PIL的特殊数据类型
        try:
            # 尝试转换为字符串
            return str(obj)
        except:
            # 如果无法转换，返回类型信息
            return f"Unserializable({type(obj).__name__})"

def is_valid_gps_data(gps_data):
    """
    检查GPS数据是否有效
    
    Args:
        gps_data: GPS数据字典
        
    Returns:
        bool: GPS数据是否有效
    """
    if not gps_data or not isinstance(gps_data, dict):
        return False
    
    # 检查是否有有效的经纬度
    has_valid_coords = False
    
    # 检查纬度
    if 'GPSLatitude' in gps_data and 'GPSLatitudeRef' in gps_data:
        lat = gps_data['GPSLatitude']
        if isinstance(lat, (list, tuple)) and len(lat) >= 3:
            try:
                # 检查是否为有效数字（不是nan或0）
                if (float(lat[0]) > 0 and float(lat[1]) > 0 and 
                    not str(lat[0]).lower() in ['nan', 'inf'] and
                    not str(lat[1]).lower() in ['nan', 'inf']):
                    has_valid_coords = True
            except (ValueError, TypeError):
                pass
    
    # 检查经度
    if 'GPSLongitude' in gps_data and 'GPSLongitudeRef' in gps_data:
        lon = gps_data['GPSLongitude']
        if isinstance(lon, (list, tuple)) and len(lon) >= 3:
            try:
                # 检查是否为有效数字（不是nan或0）
                if (float(lon[0]) > 0 and float(lon[1]) > 0 and 
                    not str(lon[0]).lower() in ['nan', 'inf'] and
                    not str(lon[1]).lower() in ['nan', 'inf']):
                    has_valid_coords = True
            except (ValueError, TypeError):
                pass
    
    return has_valid_coords

def clean_gps_data(gps_data):
    """
    清理GPS数据，移除无效信息
    
    Args:
        gps_data: 原始GPS数据
        
    Returns:
        dict: 清理后的GPS数据
    """
    if not gps_data:
        return {}
    
    cleaned = {}
    
    for key, value in gps_data.items():
        # 跳过空字节字符串
        if isinstance(value, str):
            if '\\x00' in repr(value) or value.strip() == '':
                continue
        
        # 跳过nan值
        if isinstance(value, (str, float)):
            if str(value).lower() == 'nan':
                continue
        
        # 跳过无效的时间戳
        if key == 'GPSTimeStamp' and isinstance(value, (list, tuple)):
            try:
                if (len(value) >= 3 and 
                    (str(value[0]).lower() == 'nan' or 
                     str(value[1]).lower() == 'nan' or 
                     str(value[2]).lower() == 'nan')):
                    continue
            except:
                continue
        
        cleaned[key] = value
    
    return cleaned

class ImageExtractor:
    """图片信息提取器"""
    
    def __init__(self):
        """初始化提取器"""
        self.logger = logging.getLogger(__name__)
        
        # 支持的图片格式
        self.supported_formats = {
            '.jpg', '.jpeg', '.png', '.tiff', '.tif', 
            '.bmp', '.gif', '.webp', '.raw', '.cr2', 
            '.nef', '.arw', '.dng'
        }
    
    def is_supported_format(self, file_path):
        """检查文件格式是否支持"""
        _, ext = os.path.splitext(file_path.lower())
        return ext in self.supported_formats
    
    def extract_exif_data(self, image_path):
        """
        提取图片的EXIF数据
        
        Args:
            image_path (str): 图片文件路径
            
        Returns:
            dict: 包含所有EXIF数据的字典
        """
        try:
            # 打开图片
            image = Image.open(image_path)
            
            # 获取EXIF数据
            exif_data = image._getexif()
            
            if not exif_data:
                self.logger.warning(f"图片 {image_path} 不包含EXIF数据")
                return {}
            
            # 解析EXIF标签
            exif_dict = {}
            for tag_id, value in exif_data.items():
                tag = TAGS.get(tag_id, tag_id)
                
                # 特殊处理GPS信息
                if tag == "GPSInfo":
                    gps_data = self._parse_gps_info(value)
                    exif_dict[tag] = gps_data
                else:
                    # 尝试转换bytes为字符串
                    if isinstance(value, bytes):
                        try:
                            value = value.decode('utf-8')
                        except UnicodeDecodeError:
                            try:
                                value = value.decode('latin-1')
                            except UnicodeDecodeError:
                                value = str(value)
                    
                    exif_dict[tag] = value
            
            return exif_dict
            
        except Exception as e:
            self.logger.error(f"提取EXIF数据失败: {str(e)}")
            return {}
    
    def _parse_gps_info(self, gps_info):
        """
        解析GPS信息
        
        Args:
            gps_info: GPS原始数据
            
        Returns:
            dict: 解析后的GPS信息
        """
        gps_data = {}
        
        try:
            for tag_id, value in gps_info.items():
                tag = GPSTAGS.get(tag_id, tag_id)
                gps_data[tag] = value
            
            # 清理GPS数据
            gps_data = clean_gps_data(gps_data)
            
            # 只有当数据有效时才进行转换
            if is_valid_gps_data(gps_data):
                # 转换GPS坐标为十进制格式
                if 'GPSLatitude' in gps_data and 'GPSLatitudeRef' in gps_data:
                    gps_data['latitude_decimal'] = self._convert_to_decimal(
                        gps_data['GPSLatitude'], 
                        gps_data['GPSLatitudeRef']
                    )
                
                if 'GPSLongitude' in gps_data and 'GPSLongitudeRef' in gps_data:
                    gps_data['longitude_decimal'] = self._convert_to_decimal(
                        gps_data['GPSLongitude'], 
                        gps_data['GPSLongitudeRef']
                    )
            else:
                # 标记GPS数据无效
                gps_data['gps_status'] = 'invalid_or_missing'
            
            return gps_data
            
        except Exception as e:
            self.logger.error(f"解析GPS信息失败: {str(e)}")
            return {'gps_status': 'parse_error', 'error': str(e)}
    
    def _convert_to_decimal(self, gps_coord, ref):
        """
        将GPS坐标转换为十进制格式
        
        Args:
            gps_coord: GPS坐标 (度, 分, 秒)
            ref: 参考方向 (N/S, E/W)
            
        Returns:
            float: 十进制坐标
        """
        try:
            degrees = float(gps_coord[0])
            minutes = float(gps_coord[1])
            seconds = float(gps_coord[2])
            
            decimal = degrees + (minutes / 60.0) + (seconds / 3600.0)
            
            if ref in ['S', 'W']:
                decimal = -decimal
            
            return decimal
            
        except (TypeError, ValueError, IndexError) as e:
            self.logger.error(f"坐标转换失败: {str(e)}")
            return None
    
    def get_capture_time(self, exif_data):
        """
        获取拍摄时间
        
        Args:
            exif_data (dict): EXIF数据字典
            
        Returns:
            str: 格式化的拍摄时间，如果无法获取则返回None
        """
        time_fields = [
            'DateTimeOriginal',  # 原始拍摄时间
            'DateTime',          # 修改时间
            'DateTimeDigitized'  # 数字化时间
        ]
        
        for field in time_fields:
            if field in exif_data:
                try:
                    time_str = exif_data[field]
                    # 解析时间格式
                    if isinstance(time_str, str):
                        # 尝试不同的时间格式
                        time_formats = [
                            '%Y:%m:%d %H:%M:%S',
                            '%Y-%m-%d %H:%M:%S',
                            '%Y/%m/%d %H:%M:%S'
                        ]
                        
                        for fmt in time_formats:
                            try:
                                dt = datetime.strptime(time_str, fmt)
                                return dt.strftime('%Y-%m-%d %H:%M:%S')
                            except ValueError:
                                continue
                
                except Exception as e:
                    self.logger.warning(f"解析拍摄时间失败: {str(e)}")
                    continue
        
        return None
    
    def get_gps_info(self, exif_data):
        """
        获取GPS信息
        
        Args:
            exif_data (dict): EXIF数据字典
            
        Returns:
            dict: GPS信息字典
        """
        gps_info = {}
        
        if 'GPSInfo' not in exif_data:
            return {
                'status': 'no_gps_data',
                'message': '图片不包含GPS信息'
            }
        
        gps_data = exif_data['GPSInfo']
        
        try:
            # 检查GPS数据状态
            if 'gps_status' in gps_data:
                if gps_data['gps_status'] == 'invalid_or_missing':
                    return {
                        'status': 'invalid_gps_data',
                        'message': 'GPS数据无效或已被隐私处理',
                        'raw_data_available': len(gps_data) > 1
                    }
                elif gps_data['gps_status'] == 'parse_error':
                    return {
                        'status': 'parse_error',
                        'message': 'GPS数据解析失败',
                        'error': gps_data.get('error', 'Unknown error')
                    }
            
            # 获取经纬度
            if 'latitude_decimal' in gps_data:
                gps_info['latitude'] = gps_data['latitude_decimal']
                gps_info['latitude_ref'] = gps_data.get('GPSLatitudeRef', 'N')
                gps_info['has_coordinates'] = True
            else:
                gps_info['has_coordinates'] = False
            
            if 'longitude_decimal' in gps_data:
                gps_info['longitude'] = gps_data['longitude_decimal']
                gps_info['longitude_ref'] = gps_data.get('GPSLongitudeRef', 'E')
            
            # 获取海拔高度
            if 'GPSAltitude' in gps_data:
                try:
                    altitude = float(gps_data['GPSAltitude'])
                    if str(altitude).lower() != 'nan':
                        gps_info['altitude'] = altitude
                        gps_info['altitude_ref'] = gps_data.get('GPSAltitudeRef', 0)
                        gps_info['has_altitude'] = True
                    else:
                        gps_info['has_altitude'] = False
                except (ValueError, TypeError):
                    gps_info['has_altitude'] = False
            
            # GPS时间戳
            if 'GPSTimeStamp' in gps_data and 'GPSDateStamp' in gps_data:
                try:
                    time_stamp = gps_data['GPSTimeStamp']
                    date_stamp = gps_data['GPSDateStamp']
                    
                    if isinstance(time_stamp, (list, tuple)) and len(time_stamp) >= 3:
                        hour = int(time_stamp[0])
                        minute = int(time_stamp[1])
                        second = int(time_stamp[2])
                        
                        gps_time = f"{date_stamp} {hour:02d}:{minute:02d}:{second:02d}"
                        gps_info['gps_time'] = gps_time
                        gps_info['has_timestamp'] = True
                    else:
                        gps_info['has_timestamp'] = False
                except Exception as e:
                    gps_info['has_timestamp'] = False
            
            # 尝试获取地理位置描述
            if gps_info.get('has_coordinates') and 'latitude' in gps_info and 'longitude' in gps_info:
                location = self._get_location_description(
                    gps_info['latitude'], 
                    gps_info['longitude']
                )
                if location:
                    gps_info['location'] = location
                    gps_info['status'] = 'location_found'
                else:
                    gps_info['status'] = 'coordinates_no_location'
            else:
                gps_info['status'] = 'no_valid_coordinates'
            
            # 添加数据质量信息
            gps_info['data_quality'] = self._assess_gps_quality(gps_data)
            
        except Exception as e:
            self.logger.error(f"处理GPS信息失败: {str(e)}")
            return {
                'status': 'processing_error',
                'message': 'GPS信息处理失败',
                'error': str(e)
            }
        
        return gps_info
    
    def _assess_gps_quality(self, gps_data):
        """
        评估GPS数据质量
        
        Args:
            gps_data: GPS数据
            
        Returns:
            str: 数据质量等级
        """
        quality_indicators = 0
        
        # 检查基本坐标
        if 'latitude_decimal' in gps_data and 'longitude_decimal' in gps_data:
            quality_indicators += 2
        
        # 检查海拔
        if 'GPSAltitude' in gps_data and str(gps_data['GPSAltitude']).lower() != 'nan':
            quality_indicators += 1
        
        # 检查时间戳
        if 'GPSTimeStamp' in gps_data and 'GPSDateStamp' in gps_data:
            quality_indicators += 1
        
        # 检查其他GPS标签
        gps_tags = ['GPSDOP', 'GPSSpeed', 'GPSTrack', 'GPSSatellites']
        for tag in gps_tags:
            if tag in gps_data:
                quality_indicators += 0.5
        
        if quality_indicators >= 4:
            return 'excellent'
        elif quality_indicators >= 3:
            return 'good'
        elif quality_indicators >= 2:
            return 'fair'
        elif quality_indicators >= 1:
            return 'poor'
        else:
            return 'very_poor'
    
    def _get_location_description(self, latitude, longitude):
        """
        根据经纬度获取地理位置描述
        
        Args:
            latitude (float): 纬度
            longitude (float): 经度
            
        Returns:
            str: 位置描述
        """
        try:
            # 中国主要城市的大致坐标范围
            locations = {
                '北京': (39.9, 116.4),
                '上海': (31.2, 121.5),
                '广州': (23.1, 113.3),
                '深圳': (22.5, 114.1),
                '杭州': (30.3, 120.2),
                '成都': (30.7, 104.1),
                '武汉': (30.6, 114.3),
                '西安': (34.3, 108.9),
                '天津': (39.1, 117.2),
                '南京': (32.1, 118.8),
                '重庆': (29.6, 106.5),
                '青岛': (36.1, 120.4),
                '大连': (38.9, 121.6),
                '厦门': (24.5, 118.1),
                '苏州': (31.3, 120.6),
            }
            
            # 找到最近的城市
            min_distance = float('inf')
            nearest_city = None
            
            for city, (lat, lon) in locations.items():
                distance = ((latitude - lat) ** 2 + (longitude - lon) ** 2) ** 0.5
                if distance < min_distance:
                    min_distance = distance
                    nearest_city = city
            
            # 如果距离在2度范围内，返回城市名
            if min_distance < 2.0:
                return nearest_city
            
            return f"位置未知 ({latitude:.4f}, {longitude:.4f})"
            
        except Exception as e:
            self.logger.error(f"获取位置描述失败: {str(e)}")
            return None
    
    def get_camera_info(self, exif_data):
        """
        获取相机信息
        
        Args:
            exif_data (dict): EXIF数据字典
            
        Returns:
            dict: 相机信息字典
        """
        camera_info = {}
        
        # 相机制造商
        if 'Make' in exif_data:
            make = str(exif_data['Make']).strip()
            if make and '\\x00' not in repr(make):
                camera_info['make'] = make
        
        # 相机型号
        if 'Model' in exif_data:
            model = str(exif_data['Model']).strip()
            if model and '\\x00' not in repr(model):
                camera_info['model'] = model
        
        # 从LensModel中提取信息
        if 'LensModel' in exif_data:
            lens_model = str(exif_data['LensModel']).strip()
            if lens_model and '\\x00' not in repr(lens_model):
                camera_info['lens_model'] = lens_model
                # 如果没有相机制造商，尝试从镜头型号中提取
                if 'make' not in camera_info:
                    if 'apple' in lens_model.lower():
                        camera_info['make'] = 'Apple'
                    elif 'samsung' in lens_model.lower():
                        camera_info['make'] = 'Samsung'
                    elif 'oppo' in lens_model.lower():
                        camera_info['make'] = 'OPPO'
                    elif 'vivo' in lens_model.lower():
                        camera_info['make'] = 'vivo'
                    elif 'xiaomi' in lens_model.lower():
                        camera_info['make'] = 'Xiaomi'
        
        # 软件/固件
        if 'Software' in exif_data:
            software = str(exif_data['Software']).strip()
            if software and '\\x00' not in repr(software):
                camera_info['software'] = software
            else:
                # 尝试从LensModel推断设备类型
                if 'LensModel' in exif_data:
                    lens = str(exif_data['LensModel'])
                    if 'iphone' in lens.lower():
                        camera_info['software'] = 'iOS'
                    elif 'android' in lens.lower() or any(brand in lens.lower() for brand in ['oppo', 'vivo', 'xiaomi', 'huawei', 'samsung']):
                        camera_info['software'] = 'Android'
        
        # ISO感光度
        if 'ISOSpeedRatings' in exif_data:
            camera_info['iso'] = exif_data['ISOSpeedRatings']
        
        # 光圈值
        if 'FNumber' in exif_data:
            try:
                f_number = exif_data['FNumber']
                if isinstance(f_number, (list, tuple)):
                    f_number = f_number[0] / f_number[1] if len(f_number) == 2 else f_number[0]
                camera_info['f_number'] = float(f_number)
            except (ValueError, TypeError, IndexError):
                pass
        
        # 曝光时间
        if 'ExposureTime' in exif_data:
            try:
                exposure_time = exif_data['ExposureTime']
                if isinstance(exposure_time, (list, tuple)):
                    exposure_time = exposure_time[0] / exposure_time[1] if len(exposure_time) == 2 else exposure_time[0]
                camera_info['exposure_time'] = float(exposure_time)
            except (ValueError, TypeError, IndexError):
                pass
        
        # 焦距
        if 'FocalLength' in exif_data:
            try:
                focal_length = exif_data['FocalLength']
                if isinstance(focal_length, (list, tuple)):
                    focal_length = focal_length[0] / focal_length[1] if len(focal_length) == 2 else focal_length[0]
                camera_info['focal_length'] = float(focal_length)
            except (ValueError, TypeError, IndexError):
                pass
        
        return camera_info
    
    def get_image_dimensions(self, image_path):
        """
        获取图片尺寸信息
        
        Args:
            image_path (str): 图片文件路径
            
        Returns:
            dict: 包含宽度、高度的字典
        """
        try:
            with Image.open(image_path) as image:
                return {
                    'width': image.width,
                    'height': image.height,
                    'format': image.format,
                    'mode': image.mode
                }
        except Exception as e:
            self.logger.error(f"获取图片尺寸失败: {str(e)}")
            return {}
    
    def extract_all_info(self, image_path):
        """
        提取图片的所有信息
        
        Args:
            image_path (str): 图片文件路径
            
        Returns:
            dict: 包含所有信息的字典
        """
        if not os.path.exists(image_path):
            raise FileNotFoundError(f"图片文件不存在: {image_path}")
        
        if not self.is_supported_format(image_path):
            raise ValueError(f"不支持的文件格式: {image_path}")
        
        # 获取文件基本信息
        file_stat = os.stat(image_path)
        
        result = {
            'filename': os.path.basename(image_path),
            'file_size': file_stat.st_size,
            'capture_time': None,
            'gps_info': {},
            'camera_info': {},
            'exif_data': {},
            'dimensions': {}
        }
        
        try:
            # 提取EXIF数据
            exif_data = self.extract_exif_data(image_path)
            result['exif_data'] = exif_data
            
            # 提取拍摄时间
            result['capture_time'] = self.get_capture_time(exif_data)
            
            # 提取GPS信息
            result['gps_info'] = self.get_gps_info(exif_data)
            
            # 提取相机信息
            result['camera_info'] = self.get_camera_info(exif_data)
            
            # 获取图片尺寸
            result['dimensions'] = self.get_image_dimensions(image_path)
            
        except Exception as e:
            self.logger.error(f"提取图片信息失败: {str(e)}")
            raise
        
        return result


def main():
    """测试函数"""
    extractor = ImageExtractor()
    
    # 设置日志
    logging.basicConfig(level=logging.INFO)
    
    # 测试图片路径（需要替换为实际路径）
    test_image = "test.jpg"
    
    if os.path.exists(test_image):
        try:
            info = extractor.extract_all_info(test_image)
            print(json.dumps(info, indent=2, ensure_ascii=False))
        except Exception as e:
            print(f"错误: {str(e)}")
    else:
        print(f"测试图片不存在: {test_image}")


if __name__ == "__main__":
    main()
EOFF
    
    if [ $? -eq 0 ]; then
        log_success "GPS信息提取功能改进完成"
    else
        log_error "GPS信息提取功能改进失败"
        return 1
    fi
}

# 重启服务
restart_service() {
    log_info "重启服务..."
    
    cd "$APP_DIR"
    
    # 停止现有服务
    pkill -f "python.*app.py" 2>/dev/null || true
    sleep 3
    
    # 激活虚拟环境
    source "$APP_DIR/image_extractor_env/bin/activate"
    
    # 启动服务
    nohup python app.py > logs/app.log 2>&1 &
    local pid=$!
    
    echo $pid > app.pid
    
    log_success "服务启动完成，PID: $pid"
    
    # 等待启动
    sleep 5
    
    # 测试健康检查
    response=$(curl -s http://localhost:5000/health 2>/dev/null || echo "")
    
    if [[ -n "$response" ]]; then
        log_success "服务健康检查通过"
    else
        log_warning "服务健康检查失败"
    fi
}

# 主函数
main() {
    echo
    echo "========================================"
    echo "改进GPS信息提取功能"
    echo "========================================"
    echo
    
    # 进入项目目录
    cd "$APP_DIR"
    
    # 1. 改进GPS处理
    improve_gps_handling
    
    echo
    
    # 2. 重启服务
    restart_service
    
    echo
    echo "========================================"
    log_success "GPS信息提取功能改进完成！"
    echo
    echo "改进内容："
    echo "1. 增加GPS数据有效性检查"
    echo "2. 清理无效的GPS数据"
    echo "3. 提供更详细的GPS状态信息"
    echo "4. 改进相机信息提取"
    echo "5. 更好的错误处理和状态报告"
    echo
    echo "现在重新测试您的图片："
    echo "curl -X POST -F \"image=@/path/to/image.jpg\" http://60.205.165.40:5000/extract"
    echo
    echo "GPS信息现在会显示更详细的状态："
    echo "- 如果没有GPS：显示 'no_gps_data'"
    echo "- 如果GPS无效：显示 'invalid_gps_data'"
    echo "- 如果有GPS：显示坐标和位置信息"
    echo "========================================"
}

# 执行主函数
main "$@"
'''

    # 保存脚本
    with open('/tmp/improve_gps_extraction.sh', 'w', encoding='utf-8') as f:
        f.write(script_content)
    
    print("✓ 已创建GPS改进脚本")
    return '/tmp/improve_gps_extraction.sh'

# 创建脚本
script_path = improve_gps_extraction()
print(f"\n📁 脚本已保存到: {script_path}")
print("\n🚀 在服务器上执行以下命令：")
print(f"cat {script_path} > improve_gps_extraction.sh")
print("chmod +x improve_gps_extraction.sh")
print("./improve_gps_extraction.sh")