database.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
数据库模型和操作
用于存储用户水电费查询历史记录
"""
import os
from datetime import datetime, date
from sqlalchemy import create_engine, Column, Integer, String, Float, Date, DateTime, Text, Boolean
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from pathlib import Path

# 数据库文件路径
DATABASE_PATH = "/home/cxy/log/cardquery_data.db"
DATABASE_URL = f"sqlite:///{DATABASE_PATH}"

# 创建基类
Base = declarative_base()

class QueryRecord(Base):
    """查询记录模型"""
    __tablename__ = 'query_records'
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    email = Column(String(255), nullable=False, index=True)
    query_date = Column(Date, nullable=False, index=True)
    query_time = Column(DateTime, nullable=False, default=datetime.now)
    
    # 房间信息
    building_code = Column(String(50))
    building_name = Column(String(100))
    floor_code = Column(String(50))
    floor_name = Column(String(100))
    room_code = Column(String(50))
    room_name = Column(String(100))
    
    # 电费信息
    electricity_balance = Column(Float)
    electricity_unit = Column(String(20), default='度')
    electricity_query_success = Column(Boolean, default=False)
    electricity_error_message = Column(Text)
    
    # 水费信息
    water_balance = Column(Float)
    water_unit = Column(String(20), default='元')
    water_query_success = Column(Boolean, default=False)
    water_error_message = Column(Text)
    
    # 创建时间
    created_at = Column(DateTime, default=datetime.now)
    
    def __repr__(self):
        return f"<QueryRecord(email='{self.email}', date='{self.query_date}', " \
               f"elec={self.electricity_balance}, water={self.water_balance})>"

class UserConfig(Base):
    """用户配置模型"""
    __tablename__ = 'user_configs'
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    email = Column(String(255), nullable=False, unique=True, index=True)
    
    # API配置
    card_user_id = Column(String(255))
    room_user_id = Column(String(255))
    device_id = Column(String(255))
    area_id_elec = Column(String(50))
    area_id_water = Column(String(50))
    
    # APNs推送配置
    device_token = Column(String(255))  # iOS设备推送token
    enable_push = Column(Boolean, default=True)  # 是否启用推送
    
    # 查询配置
    enable_auto_query = Column(Boolean, default=True)
    last_query_date = Column(Date)
    
    # 创建和更新时间
    created_at = Column(DateTime, default=datetime.now)
    updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now)
    
    def __repr__(self):
        return f"<UserConfig(email='{self.email}', auto_query={self.enable_auto_query})>"

# 创建数据库引擎
engine = create_engine(DATABASE_URL, echo=False)

# 创建会话工厂
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

def init_database():
    """初始化数据库"""
    try:
        Base.metadata.create_all(bind=engine)
        print(f"✅ 数据库初始化成功: {DATABASE_PATH}")
        return True
    except Exception as e:
        print(f"❌ 数据库初始化失败: {e}")
        return False

def get_db_session() -> Session:
    """获取数据库会话"""
    return SessionLocal()

def save_query_record(email: str, room_info: dict, electricity_result: dict, water_result: dict, query_date=None):
    """保存查询记录"""
    if query_date is None:
        query_date = date.today()
    
    session = get_db_session()
    try:
        # 检查今天是否已经有记录
        existing_record = session.query(QueryRecord).filter(
            QueryRecord.email == email,
            QueryRecord.query_date == query_date
        ).first()
        
        if existing_record:
            # 更新现有记录
            record = existing_record
            record.query_time = datetime.now()
        else:
            # 创建新记录
            record = QueryRecord(
                email=email,
                query_date=query_date
            )
            session.add(record)
        
        # 更新房间信息
        record.building_code = room_info.get('building_code', '')
        record.building_name = room_info.get('building_name', '')
        record.floor_code = room_info.get('floor_code', '')
        record.floor_name = room_info.get('floor_name', '')
        record.room_code = room_info.get('room_code', '')
        record.room_name = room_info.get('room_name', '')
        
        # 更新电费信息
        record.electricity_query_success = electricity_result.get('success', False)
        record.electricity_balance = electricity_result.get('balance')
        record.electricity_unit = electricity_result.get('unit', '度')
        record.electricity_error_message = electricity_result.get('error_message', '')
        
        # 更新水费信息
        record.water_query_success = water_result.get('success', False)
        record.water_balance = water_result.get('balance')
        record.water_unit = water_result.get('unit', '元')
        record.water_error_message = water_result.get('error_message', '')
        
        session.commit()
        session.refresh(record)  # 刷新对象以获取数据库生成的ID
        
        record_id = record.id
        print(f"💾 保存查询记录: {email} - 电费:{record.electricity_balance}{record.electricity_unit}, "
              f"水费:{record.water_balance}{record.water_unit}")
        
        return record_id
        
    except Exception as e:
        session.rollback()
        print(f"❌ 保存查询记录失败 ({email}): {e}")
        return None
    finally:
        session.close()

def get_user_query_history(email: str, limit: int = 30):
    """获取用户查询历史"""
    session = get_db_session()
    try:
        records = session.query(QueryRecord).filter(
            QueryRecord.email == email
        ).order_by(QueryRecord.query_date.desc()).limit(limit).all()
        
        return [
            {
                'date': record.query_date.isoformat(),
                'electricity_balance': record.electricity_balance,
                'electricity_unit': record.electricity_unit,
                'electricity_success': record.electricity_query_success,
                'water_balance': record.water_balance,
                'water_unit': record.water_unit,
                'water_success': record.water_query_success,
                'room_name': record.room_name,
                'query_time': record.query_time.isoformat() if record.query_time is not None else None
            }
            for record in records
        ]
    except Exception as e:
        print(f"❌ 获取查询历史失败 ({email}): {e}")
        return []
    finally:
        session.close()

def get_yesterday_record(email: str):
    """获取用户昨天的查询记录"""
    from datetime import timedelta
    
    yesterday = date.today() - timedelta(days=1)
    session = get_db_session()
    try:
        record = session.query(QueryRecord).filter(
            QueryRecord.email == email,
            QueryRecord.query_date == yesterday
        ).first()
        
        if record:
            return {
                'electricity_balance': record.electricity_balance,
                'water_balance': record.water_balance,
                'electricity_success': record.electricity_query_success,
                'water_success': record.water_query_success,
                'date': record.query_date.isoformat()
            }
        return None
    except Exception as e:
        print(f"❌ 获取昨日记录失败 ({email}): {e}")
        return None
    finally:
        session.close()

def calculate_usage_difference(email: str, today_electricity: float, today_water: float):
    """计算用户今日与昨日的使用量差值"""
    yesterday_record = get_yesterday_record(email)
    
    if not yesterday_record:
        return {
            'has_yesterday_data': False,
            'electricity_usage': None,
            'water_usage': None,
            'message': '无昨日数据'
        }
    
    # 计算用电量（昨天余额 - 今天余额 = 使用量）
    electricity_usage = None
    if (yesterday_record['electricity_success'] and 
        yesterday_record['electricity_balance'] is not None and 
        today_electricity is not None):
        electricity_usage = round(yesterday_record['electricity_balance'] - today_electricity, 2)
    
    # 计算用水量（昨天余额 - 今天余额 = 使用量）
    water_usage = None
    if (yesterday_record['water_success'] and 
        yesterday_record['water_balance'] is not None and 
        today_water is not None):
        water_usage = round(yesterday_record['water_balance'] - today_water, 2)
    
    return {
        'has_yesterday_data': True,
        'yesterday_electricity': yesterday_record['electricity_balance'],
        'yesterday_water': yesterday_record['water_balance'],
        'today_electricity': today_electricity,
        'today_water': today_water,
        'electricity_usage': electricity_usage,
        'water_usage': water_usage,
        'yesterday_date': yesterday_record['date']
    }

def get_query_statistics():
    """获取查询统计信息"""
    session = get_db_session()
    try:
        total_records = session.query(QueryRecord).count()
        total_users = session.query(QueryRecord.email).distinct().count()
        today_records = session.query(QueryRecord).filter(
            QueryRecord.query_date == date.today()
        ).count()
        
        return {
            'total_records': total_records,
            'total_users': total_users,
            'today_records': today_records,
            'database_path': DATABASE_PATH
        }
    except Exception as e:
        print(f"❌ 获取统计信息失败: {e}")
        return {}
    finally:
        session.close()

def save_user_device_token(email: str, device_token: str):
    """保存用户设备推送token"""
    session = get_db_session()
    try:
        # 查找或创建用户配置
        user_config = session.query(UserConfig).filter(
            UserConfig.email == email
        ).first()
        
        if not user_config:
            user_config = UserConfig(email=email)
            session.add(user_config)
        
        # 更新device token
        user_config.device_token = device_token
        user_config.enable_push = True
        user_config.updated_at = datetime.now()
        
        session.commit()
        print(f"💾 保存设备token: {email}")
        return True
        
    except Exception as e:
        session.rollback()
        print(f"❌ 保存设备token失败 ({email}): {e}")
        return False
    finally:
        session.close()

def get_user_device_token(email: str):
    """获取用户设备推送token"""
    session = get_db_session()
    try:
        user_config = session.query(UserConfig).filter(
            UserConfig.email == email
        ).first()
        
        if user_config and user_config.enable_push is not None and user_config.enable_push:
            device_token = user_config.device_token
            return device_token
        return None
        
    except Exception as e:
        print(f"❌ 获取设备token失败 ({email}): {e}")
        return None
    finally:
        session.close()

def cleanup_old_records(days: int = 90):
    """清理旧记录"""
    from datetime import timedelta
    
    cutoff_date = date.today() - timedelta(days=days)
    session = get_db_session()
    
    try:
        deleted_count = session.query(QueryRecord).filter(
            QueryRecord.query_date < cutoff_date
        ).delete()
        
        session.commit()
        print(f"🗑️ 清理了 {deleted_count} 条超过 {days} 天的旧记录")
        return deleted_count
        
    except Exception as e:
        session.rollback()
        print(f"❌ 清理旧记录失败: {e}")
        return 0
    finally:
        session.close()

if __name__ == "__main__":
    # 测试数据库功能
    print("🔧 测试数据库功能...")
    
    if init_database():
        print("📊 数据库统计信息:")
        stats = get_query_statistics()
        for key, value in stats.items():
            print(f"  - {key}: {value}")
    else:
        print("❌ 数据库测试失败")