314 lines
11 KiB
Python
314 lines
11 KiB
Python
"""
|
||
MySQL数据库管理器
|
||
使用SQLAlchemy ORM连接生产数据库（只读）
|
||
"""
|
||
import os
|
||
from datetime import datetime, timedelta
|
||
from typing import List, Dict, Optional, Tuple
|
||
from sqlalchemy import create_engine, Column, Integer, String, DateTime, Text
|
||
from sqlalchemy.ext.declarative import declarative_base
|
||
from sqlalchemy.orm import sessionmaker, Session
|
||
from sqlalchemy.pool import QueuePool
|
||
import sys
|
||
|
||
# 添加父目录到路径
|
||
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
from config import MYSQL_CONFIG, TABLE_MAPPING, IMAGE_URL_PREFIX
|
||
|
||
Base = declarative_base()
|
||
|
||
|
||
class DishCollection(Base):
|
||
"""菜品采集表模型"""
|
||
__tablename__ = TABLE_MAPPING['菜品']
|
||
|
||
id = Column(Integer, primary_key=True, autoincrement=True)
|
||
goods_id = Column(Integer, nullable=True, comment='物品ID')
|
||
goods_name = Column(String(255), nullable=True, comment='物品名称')
|
||
food_pic = Column(String(500), nullable=True, comment='图片URL')
|
||
equipment_code = Column(String(100), nullable=True, comment='设备编码')
|
||
create_time = Column(DateTime, nullable=True, comment='创建时间')
|
||
|
||
|
||
class InboundWeighing(Base):
|
||
"""食材入库秤表模型"""
|
||
__tablename__ = TABLE_MAPPING['食材']
|
||
|
||
id = Column(Integer, primary_key=True, autoincrement=True)
|
||
goods_id = Column(Integer, nullable=True, comment='物品ID')
|
||
goods_name = Column(String(255), nullable=True, comment='物品名称')
|
||
food_pic = Column(String(500), nullable=True, comment='图片URL')
|
||
equipment_code = Column(String(100), nullable=True, comment='设备编码')
|
||
create_time = Column(DateTime, nullable=True, comment='创建时间')
|
||
|
||
|
||
class ProcessedIngredient(Base):
|
||
"""半加工食材表模型（待替换表名）"""
|
||
__tablename__ = TABLE_MAPPING['半加工']
|
||
|
||
id = Column(Integer, primary_key=True, autoincrement=True)
|
||
goods_id = Column(Integer, nullable=True, comment='物品ID')
|
||
goods_name = Column(String(255), nullable=True, comment='物品名称')
|
||
food_pic = Column(String(500), nullable=True, comment='图片URL')
|
||
equipment_code = Column(String(100), nullable=True, comment='设备编码')
|
||
create_time = Column(DateTime, nullable=True, comment='创建时间')
|
||
|
||
|
||
class MySQLManager:
|
||
"""MySQL数据库管理器（只读）"""
|
||
|
||
def __init__(self):
|
||
"""初始化数据库连接"""
|
||
self.engine = None
|
||
self.Session = None
|
||
self._connect()
|
||
|
||
def _connect(self):
|
||
"""建立数据库连接"""
|
||
try:
|
||
# 构建连接字符串
|
||
connection_string = (
|
||
f"mysql+pymysql://{MYSQL_CONFIG['user']}:{MYSQL_CONFIG['password']}"
|
||
f"@{MYSQL_CONFIG['host']}:{MYSQL_CONFIG['port']}"
|
||
f"/{MYSQL_CONFIG['database']}?charset={MYSQL_CONFIG['charset']}"
|
||
)
|
||
|
||
# 创建引擎（使用连接池）
|
||
self.engine = create_engine(
|
||
connection_string,
|
||
poolclass=QueuePool,
|
||
pool_size=5,
|
||
max_overflow=10,
|
||
pool_recycle=3600,
|
||
echo=False # 设置为True可以看到SQL语句
|
||
)
|
||
|
||
# 创建Session工厂
|
||
self.Session = sessionmaker(bind=self.engine)
|
||
|
||
print("MySQL连接成功！")
|
||
|
||
except Exception as e:
|
||
print(f"MySQL连接失败: {e}")
|
||
raise
|
||
|
||
def get_model_by_datasource(self, datasource: str):
|
||
"""根据数据源获取对应的模型"""
|
||
model_mapping = {
|
||
'菜品': DishCollection,
|
||
'食材': InboundWeighing,
|
||
'半加工': ProcessedIngredient
|
||
}
|
||
return model_mapping.get(datasource)
|
||
|
||
def query_images(
|
||
self,
|
||
datasource: str,
|
||
start_date: Optional[datetime] = None,
|
||
end_date: Optional[datetime] = None,
|
||
goods_name: Optional[str] = None,
|
||
equipment_code: Optional[str] = None,
|
||
page: int = 1,
|
||
page_size: int = 20
|
||
) -> Tuple[List[Dict], int]:
|
||
"""
|
||
查询图片数据
|
||
|
||
Args:
|
||
datasource: 数据源（'菜品'/'食材'/'半加工'）
|
||
start_date: 开始日期
|
||
end_date: 结束日期
|
||
goods_name: 物品名称（模糊搜索）
|
||
equipment_code: 设备编码（精确匹配）
|
||
page: 页码（从1开始）
|
||
page_size: 每页数量
|
||
|
||
Returns:
|
||
(数据列表, 总数量)
|
||
"""
|
||
session: Session = self.Session()
|
||
try:
|
||
# 获取对应的模型
|
||
Model = self.get_model_by_datasource(datasource)
|
||
if Model is None:
|
||
return [], 0
|
||
|
||
# 构建查询
|
||
query = session.query(Model)
|
||
|
||
# 时间范围筛选
|
||
if start_date:
|
||
query = query.filter(Model.create_time >= start_date)
|
||
if end_date:
|
||
# 结束日期包含当天，所以需要加1天
|
||
end_date_inclusive = end_date + timedelta(days=1)
|
||
query = query.filter(Model.create_time < end_date_inclusive)
|
||
|
||
# 物品名称筛选（模糊搜索）
|
||
if goods_name:
|
||
query = query.filter(Model.goods_name.like(f'%{goods_name}%'))
|
||
|
||
# 设备编码筛选（精确匹配）
|
||
if equipment_code:
|
||
query = query.filter(Model.equipment_code == equipment_code)
|
||
|
||
# 只查询有图片的记录
|
||
query = query.filter(Model.food_pic.isnot(None))
|
||
query = query.filter(Model.food_pic != '')
|
||
|
||
# 获取总数
|
||
total_count = query.count()
|
||
|
||
# 按时间倒序排列
|
||
query = query.order_by(Model.create_time.desc())
|
||
|
||
# 分页
|
||
offset = (page - 1) * page_size
|
||
query = query.limit(page_size).offset(offset)
|
||
|
||
# 执行查询
|
||
results = query.all()
|
||
|
||
# 转换为字典列表
|
||
data_list = []
|
||
for item in results:
|
||
# 拼接完整图片URL
|
||
full_image_url = f"{IMAGE_URL_PREFIX}{item.food_pic}" if item.food_pic else ""
|
||
|
||
data_list.append({
|
||
'id': item.id,
|
||
'goods_id': item.goods_id,
|
||
'goods_name': item.goods_name or '未知',
|
||
'image_url': full_image_url,
|
||
'equipment_code': item.equipment_code or '',
|
||
'create_time': item.create_time.strftime('%Y-%m-%d %H:%M:%S') if item.create_time else '',
|
||
'datasource': datasource
|
||
})
|
||
|
||
return data_list, total_count
|
||
|
||
except Exception as e:
|
||
print(f"查询数据出错: {e}")
|
||
return [], 0
|
||
finally:
|
||
session.close()
|
||
|
||
def get_goods_names(self, datasource: str, limit: int = 100) -> List[str]:
|
||
"""
|
||
获取物品名称列表（用于搜索框自动补全）
|
||
|
||
Args:
|
||
datasource: 数据源
|
||
limit: 最多返回多少个
|
||
|
||
Returns:
|
||
物品名称列表
|
||
"""
|
||
session: Session = self.Session()
|
||
try:
|
||
Model = self.get_model_by_datasource(datasource)
|
||
if Model is None:
|
||
return []
|
||
|
||
# 查询不重复的物品名称
|
||
results = session.query(Model.goods_name)\
|
||
.filter(Model.goods_name.isnot(None))\
|
||
.filter(Model.goods_name != '')\
|
||
.distinct()\
|
||
.limit(limit)\
|
||
.all()
|
||
|
||
return [item[0] for item in results]
|
||
|
||
except Exception as e:
|
||
print(f"获取物品名称列表出错: {e}")
|
||
return []
|
||
finally:
|
||
session.close()
|
||
|
||
def get_statistics(
|
||
self,
|
||
datasource: str,
|
||
start_date: Optional[datetime] = None,
|
||
end_date: Optional[datetime] = None
|
||
) -> Dict:
|
||
"""
|
||
获取统计信息
|
||
|
||
Returns:
|
||
统计数据字典
|
||
"""
|
||
session: Session = self.Session()
|
||
try:
|
||
Model = self.get_model_by_datasource(datasource)
|
||
if Model is None:
|
||
return {}
|
||
|
||
# 基础查询
|
||
base_query = session.query(Model)\
|
||
.filter(Model.food_pic.isnot(None))\
|
||
.filter(Model.food_pic != '')
|
||
|
||
if start_date:
|
||
base_query = base_query.filter(Model.create_time >= start_date)
|
||
if end_date:
|
||
end_date_inclusive = end_date + timedelta(days=1)
|
||
base_query = base_query.filter(Model.create_time < end_date_inclusive)
|
||
|
||
# 总数
|
||
total_count = base_query.count()
|
||
|
||
# 今日数量
|
||
today_start = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
|
||
today_count = base_query.filter(Model.create_time >= today_start).count()
|
||
|
||
# 本周数量
|
||
week_start = today_start - timedelta(days=today_start.weekday())
|
||
week_count = base_query.filter(Model.create_time >= week_start).count()
|
||
|
||
# 本月数量
|
||
month_start = today_start.replace(day=1)
|
||
month_count = base_query.filter(Model.create_time >= month_start).count()
|
||
|
||
# TOP 10物品统计
|
||
from sqlalchemy import func
|
||
top_items = session.query(
|
||
Model.goods_name,
|
||
func.count(Model.id).label('count')
|
||
).filter(
|
||
Model.food_pic.isnot(None),
|
||
Model.food_pic != '',
|
||
Model.goods_name.isnot(None),
|
||
Model.goods_name != ''
|
||
)
|
||
|
||
if start_date:
|
||
top_items = top_items.filter(Model.create_time >= start_date)
|
||
if end_date:
|
||
top_items = top_items.filter(Model.create_time < end_date + timedelta(days=1))
|
||
|
||
top_items = top_items.group_by(Model.goods_name)\
|
||
.order_by(func.count(Model.id).desc())\
|
||
.limit(10)\
|
||
.all()
|
||
|
||
return {
|
||
'total_count': total_count,
|
||
'today_count': today_count,
|
||
'week_count': week_count,
|
||
'month_count': month_count,
|
||
'top_items': [{'name': item[0], 'count': item[1]} for item in top_items]
|
||
}
|
||
|
||
except Exception as e:
|
||
print(f"获取统计信息出错: {e}")
|
||
return {}
|
||
finally:
|
||
session.close()
|
||
|
||
def close(self):
|
||
"""关闭数据库连接"""
|
||
if self.engine:
|
||
self.engine.dispose()
|
||
print("MySQL连接已关闭")
|