如何编写库存数量监控脚本

wen 实用脚本 2026-06-28 237

本文目录导读：

如何编写库存数量监控脚本

Docker运行
设置定时任务（Linux）
添加：0 /2 /usr/bin/python3 /path/to/inventory_monitor.py

我来帮您编写一个库存数量监控脚本,这里提供几种常见场景的实现方案：

基于数据库的库存监控脚本

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


这个脚本可以根据您的具体需求进行定制，支持多种数据源、通知方式和部署方案。

本文地址： https://www.bmjup.com/post/11229.html