基于productid查询MySQL最低oldprice并在Scrapy管道实现Telegram通知

1. 数据库查询与Pipeline实现 #

直接在pipelines.py中实现数据库查询逻辑，按productid分组获取对应最低oldprice的完整记录，同时集成Telegram通知功能：

示例代码 #

import pymysql
import requests
from scrapy.exceptions import DropItem

class PriceNotificationPipeline:
    def open_spider(self, spider):
        # 初始化数据库连接
        self.conn = pymysql.connect(
            host='你的数据库地址',
            user='数据库用户名',
            password='数据库密码',
            db='数据库名',
            charset='utf8mb4'
        )
        self.cursor = self.conn.cursor()

    def close_spider(self, spider):
        # 关闭数据库连接
        self.cursor.close()
        self.conn.close()

    def get_min_price_records(self):
        # 查询每个productid对应的最低oldprice记录
        query = """
            SELECT ph.productid, ph.oldprice, ph.created_at
            FROM pricehistory ph
            INNER JOIN (
                SELECT productid, MIN(oldprice) AS min_price
                FROM pricehistory
                GROUP BY productid
            ) AS min_ph 
            ON ph.productid = min_ph.productid AND ph.oldprice = min_ph.min_price;
        """
        self.cursor.execute(query)
        return self.cursor.fetchall()

    def send_telegram_notification(self, records):
        # 构造Telegram通知消息并发送
        bot_token = '你的Telegram Bot Token'
        chat_id = '目标聊天ID'
        message = "商品最低价格记录：\n"
        
        for record in records:
            product_id, min_price, created_at = record
            message += f"商品ID: {product_id}\n最低价格: {min_price}\n记录时间: {created_at}\n\n"

        # 对消息内容编码，避免特殊字符导致请求失败
        encoded_message = requests.utils.quote(message)
        send_url = f'https://api.telegram.org/bot{bot_token}/sendMessage?chat_id={chat_id}&text={encoded_message}'
        
        response = requests.get(send_url)
        if response.status_code != 200:
            raise Exception(f"通知发送失败：{response.text}")

    def process_item(self, item, spider):
        # 先执行爬取数据入库逻辑（如果已实现）
        # ... 你的入库代码 ...

        # 获取最低价格记录并发送通知
        try:
            min_records = self.get_min_price_records()
            self.send_telegram_notification(min_records)
            spider.logger.info("最低价格通知已发送")
        except Exception as e:
            spider.logger.error(f"通知处理失败：{str(e)}")
        
        return item

2. 关键配置与排查点 #

• 激活Pipeline：在项目的settings.py中添加该Pipeline：

  ITEM_PIPELINES = {
      '你的项目名.pipelines.PriceNotificationPipeline': 300,
  }
  

• 数据库连接校验：确认数据库地址、账号密码正确，测试连接是否正常，避免因权限或网络问题导致查询失败。
• Telegram权限检查：确保Bot已加入目标聊天群，bot_token和chat_id准确（可通过https://api.telegram.org/bot<你的token>/getUpdates接口获取chat_id）。
• 查询逻辑优化：如果同一productid存在多条相同最低价格的记录，可修改SQL只取最早/最晚的一条：

  SELECT productid, oldprice, created_at
  FROM pricehistory
  WHERE (productid, oldprice) IN (
      SELECT productid, MIN(oldprice)
      FROM pricehistory
      GROUP BY productid
  )
  ORDER BY created_at DESC
  LIMIT 1;
  

3. 调试技巧 #

• 在get_min_price_records中打印查询结果，确认数据是否正确返回。
• 在send_telegram_notification中打印构造的消息内容，排查格式或特殊字符问题。
• 利用Scrapy的日志系统捕获异常，快速定位错误点。

内容的提问来源于stack exchange，提问作者cevreyolu