本文目录导读：
我来帮您编写一个库存数量监控脚本,这里提供几种常见场景的实现方案：
基于数据库的库存监控脚本
Python版本（使用MySQL）
import mysql.connector
from datetime import datetime
import smtplib
from email.mime.text import MIMEText
import logging
# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
class InventoryMonitor:
def __init__(self, db_config, alert_config):
self.db_config = db_config
self.alert_config = alert_config
def check_inventory(self, product_id=None):
"""检查库存数量"""
try:
conn = mysql.connector.connect(**self.db_config)
cursor = conn.cursor(dictionary=True)
if product_id:
query = "SELECT * FROM products WHERE id = %s"
cursor.execute(query, (product_id,))
else:
query = "SELECT * FROM products WHERE quantity < min_threshold"
cursor.execute(query)
results = cursor.fetchall()
return results
except Exception as e:
logging.error(f"数据库查询错误: {e}")
return None
finally:
if 'conn' in locals():
conn.close()
def send_alert(self, product):
"""发送库存预警"""
if self.alert_config['type'] == 'email':
self._send_email_alert(product)
elif self.alert_config['type'] == 'webhook':
self._send_webhook_alert(product)
def _send_email_alert(self, product):
"""发送邮件预警"""
try:
msg = MIMEText(f"""
库存预警通知
商品ID: {product['id']}
商品名称: {product['name']}
当前库存: {product['quantity']}
最低库存: {product['min_threshold']}
时间: {datetime.now()}
""")
msg['Subject'] = f"库存预警 - {product['name']}"
msg['From'] = self.alert_config['email_from']
msg['To'] = self.alert_config['email_to']
with smtplib.SMTP(self.alert_config['smtp_server'],
self.alert_config['smtp_port']) as server:
server.starttls()
server.login(self.alert_config['email_user'],
self.alert_config['email_password'])
server.send_message(msg)
logging.info(f"预警邮件已发送: {product['name']}")
except Exception as e:
logging.error(f"发送邮件失败: {e}")
def monitor_products(self, interval=3600):
"""定期监控库存"""
import schedule
import time
def job():
logging.info("开始库存检查...")
low_stock_products = self.check_inventory()
if low_stock_products:
for product in low_stock_products:
self.send_alert(product)
logging.warning(f"低库存预警: {product['name']} - {product['quantity']}")
else:
logging.info("所有商品库存正常")
# 定时执行
schedule.every(interval).seconds.do(job)
logging.info(f"库存监控已启动，每{interval}秒检查一次")
while True:
schedule.run_pending()
time.sleep(1)
# 使用示例
if __name__ == "__main__":
# 数据库配置
db_config = {
'host': 'localhost',
'user': 'your_user',
'password': 'your_password',
'database': 'inventory_db'
}
# 预警配置
alert_config = {
'type': 'email',
'email_from': 'monitor@example.com',
'email_to': 'admin@example.com',
'smtp_server': 'smtp.gmail.com',
'smtp_port': 587,
'email_user': 'your_email@gmail.com',
'email_password': 'your_app_password'
}
# 创建监控实例
monitor = InventoryMonitor(db_config, alert_config)
# 单次检查
results = monitor.check_inventory(101)
if results:
for product in results:
print(f"商品: {product['name']}, 库存: {product['quantity']}")
# 持续监控
monitor.monitor_products(interval=3600) # 每小时检查一次
基于Excel文件的库存监控
import pandas as pd
from datetime import datetime
import os
class ExcelInventoryMonitor:
def __init__(self, file_path):
self.file_path = file_path
self.last_modified = None
def load_inventory(self):
"""加载库存数据"""
try:
if not os.path.exists(self.file_path):
logging.error(f"文件不存在: {self.file_path}")
return None
df = pd.read_excel(self.file_path)
return df
except Exception as e:
logging.error(f"读取Excel失败: {e}")
return None
def check_thresholds(self, df):
"""检查库存阈值"""
alerts = []
for index, row in df.iterrows():
if row['quantity'] <= row['min_threshold']:
alerts.append({
'product': row['product_name'],
'sku': row['sku'],
'current': row['quantity'],
'threshold': row['min_threshold'],
'time': datetime.now()
})
return alerts
def generate_report(self, alerts):
"""生成库存报告"""
report = f"""
库存监控报告
时间: {datetime.now()}
文件: {self.file_path}
低库存商品:
"""
if alerts:
for alert in alerts:
report += f"""
商品: {alert['product']}
SKU: {alert['sku']}
当前库存: {alert['current']}
最低库存: {alert['threshold']}
"""
else:
report += "\n暂无低库存商品"
return report
def monitor_file_changes(self, check_interval=300):
"""监控文件变化"""
import time
while True:
try:
current_mtime = os.path.getmtime(self.file_path)
if self.last_modified is None:
self.last_modified = current_mtime
elif current_mtime != self.last_modified:
logging.info("检测到文件变化，重新检查库存...")
df = self.load_inventory()
if df is not None:
alerts = self.check_thresholds(df)
if alerts:
report = self.generate_report(alerts)
print(report)
# 可以添加发送通知的代码
self.last_modified = current_mtime
time.sleep(check_interval)
except KeyboardInterrupt:
logging.info("监控已停止")
break
except Exception as e:
logging.error(f"监控出错: {e}")
time.sleep(check_interval)
# 使用示例
monitor = ExcelInventoryMonitor("inventory.xlsx")
monitor.monitor_file_changes(300) # 每5分钟检查一次
基于API的库存监控
import requests
import json
from datetime import datetime
import hashlib
class APIInventoryMonitor:
def __init__(self, api_config):
self.base_url = api_config['base_url']
self.api_key = api_config['api_key']
self.api_secret = api_config['api_secret']
def get_inventory(self, product_id=None):
"""从API获取库存数据"""
headers = {
'Authorization': f'Bearer {self.api_key}',
'Content-Type': 'application/json'
}
try:
if product_id:
url = f"{self.base_url}/products/{product_id}"
else:
url = f"{self.base_url}/products/inventory"
response = requests.get(url, headers=headers, timeout=10)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
logging.error(f"API请求失败: {e}")
return None
def update_inventory(self, product_id, quantity_change):
"""更新库存"""
headers = {
'Authorization': f'Bearer {self.api_key}',
'Content-Type': 'application/json'
}
payload = {
'product_id': product_id,
'quantity_change': quantity_change,
'reason': '库存调整',
'timestamp': datetime.now().isoformat()
}
try:
url = f"{self.base_url}/products/{product_id}/inventory"
response = requests.post(url, json=payload, headers=headers)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
logging.error(f"库存更新失败: {e}")
return None
def check_low_stock(self, threshold=10):
"""检查低库存商品"""
inventory = self.get_inventory()
if not inventory:
return []
low_stock_items = []
for item in inventory:
if item['quantity'] <= threshold:
low_stock_items.append(item)
return low_stock_items
# 使用示例
api_config = {
'base_url': 'https://api.example.com/v1',
'api_key': 'your_api_key',
'api_secret': 'your_api_secret'
}
monitor = APIInventoryMonitor(api_config)
low_stock = monitor.check_low_stock(threshold=5)
print(f"低库存商品数: {len(low_stock)}")
Docker容器化部署
# Dockerfile FROM python:3.9-slim WORKDIR /app # 安装依赖 COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # 复制脚本 COPY inventory_monitor.py . # 设置定时任务（可选） RUN apt-get update && apt-get install -y cron COPY crontab /etc/cron.d/inventory-monitor RUN chmod 0644 /etc/cron.d/inventory-monitor # 运行监控脚本 CMD ["python", "inventory_monitor.py"]
# docker-compose.yml
version: '3.8'
services:
inventory-monitor:
build: .
environment:
- DB_HOST=mysql
- DB_USER=root
- DB_PASSWORD=rootpassword
- DB_NAME=inventory
volumes:
- ./logs:/app/logs
restart: unless-stopped
depends_on:
- mysql
mysql:
image: mysql:8.0
environment:
- MYSQL_ROOT_PASSWORD=rootpassword
- MYSQL_DATABASE=inventory
volumes:
- ./mysql-data:/var/lib/mysql
监控脚本的高级功能
class AdvancedInventoryMonitor:
def __init__(self):
self.alert_channels = [] # 多渠道通知
self.history_buffer = [] # 历史记录
self.prediction_model = None # 预测模型
def add_alert_channel(self, channel_type, config):
"""添加通知渠道"""
channels = {
'email': EmailAlert,
'slack': SlackAlert,
'sms': SMSAlert,
'wechat': WeChatAlert
}
if channel_type in channels:
alert_obj = channels[channel_type](config)
self.alert_channels.append(alert_obj)
def predict_stockout(self, product_id, days=7):
"""预测缺货时间"""
# 使用历史数据预测
history = self.get_product_history(product_id)
if not history:
return None
# 简单移动平均预测
consumption_rates = []
for i in range(1, len(history)):
rate = history[i]['consumption'] / history[i]['days']
consumption_rates.append(rate)
avg_daily_consumption = sum(consumption_rates) / len(consumption_rates)
current_stock = history[-1]['quantity']
days_until_stockout = current_stock / avg_daily_consumption
return {
'product_id': product_id,
'current_stock': current_stock,
'daily_consumption': avg_daily_consumption,
'days_until_stockout': days_until_stockout,
'stockout_date': datetime.now() + timedelta(days=days_until_stockout)
}
def auto_reorder(self, product_id, quantity):
"""自动补货"""
# 连接到供应商API
# 生成采购订单
order = {
'product_id': product_id,
'quantity': quantity,
'supplier': self.get_preferred_supplier(product_id),
'order_date': datetime.now(),
'expected_delivery': datetime.now() + timedelta(days=3)
}
# 发送订单
if self.place_order(order):
logging.info(f"已自动补货: 商品{product_id}，数量{quantity}")
return order
return None
# 使用示例
monitor = AdvancedInventoryMonitor()
monitor.add_alert_channel('email', {
'recipient': 'admin@example.com'
})
monitor.add_alert_channel('slack', {
'webhook_url': 'https://hooks.slack.com/services/xxx'
})
# 预测缺货
prediction = monitor.predict_stockout(101)
if prediction and prediction['days_until_stockout'] < 3:
monitor.auto_reorder(101, 100)
使用建议
-
安装依赖：
pip install pandas pymongo mysql-connector-python schedule requests
-
配置文件（config.json）：
{ "database": { "type": "mysql", "host": "localhost", "port": 3306, "database": "inventory" }, "monitoring": { "check_interval": 3600, "low_stock_threshold": 10, "alert_channels": ["email", "slack"] }, "alerts": { "email": { "enabled": true, "recipients": ["admin@example.com"] } } } -
运行方式：
# 直接运行 python inventory_monitor.py
Docker运行
docker-compose up -d
设置定时任务（Linux）
crontab -e
添加：0 /2 /usr/bin/python3 /path/to/inventory_monitor.py
这个脚本可以根据您的具体需求进行定制，支持多种数据源、通知方式和部署方案。
