Python 制作查询商品历史价格的小工具

白榆

发布时间：2026-10-15| 5人浏览过

原创

本文介绍如何利用 Python 编写一个监控京东商品历史价格的小工具。通过抓取商品价格并存入 SQLite 数据库，结合 schedule 模块实现每日定时任务，最后使用 pyecharts 绘制折线图直观展示价格波动，帮助用户在购物时识别“明降暗升”，获取真正优惠。

Python 监控商品历史价格思路

双十一临近，各类砍价、盖楼及挖现金活动将在未来一个月内占据朋友圈和微信群。作为多次参与双十一的用户，虽然操作繁琐，但最终收益往往微乎其微，例如仅获得 2 毛 5 的优惠。这种耗时且未必带来真实优惠的活动，背后隐藏着电商“明降暗升”的潜规则。要打破这一规则，可以通过编写 Python 脚本，创建一个定时监控商品价格的小工具。

思路

第一步，抓取商品价格并存入 Python 内置的 SQLite 数据库。
第二步，设置每日定时任务抓取商品价格。
第三步，利用 pyecharts 模块绘制价格折线图，使低价信息一目了然。

抓取京东商品价格

抓取京东价格

在商品详情页打开 F12 开发者工具，定位包含 p.3 的链接，并在旁边的 preview 面板中查看当前商品价格。


def get_jd_price(skuId):

  sku_detail_url = 'http://item.jd.com/{}.html'
  sku_price_url = 'https://p.3.cn/prices/get?type=1&skuid=J_{}'

  r = requests.get(sku_detail_url.format(skuId)).content

  soup = BeautifulSoup(r, 'html.parser', from_encoding='utf-8')
  sku_name_div = soup.find('div', class_="sku-name")

  if not sku_name_div:
    print('您输入的商品ID有误！')
    return
  else:
    sku_name = sku_name_div.text.strip()

  r = requests.get(sku_price_url.format(skuId))
  price = json.loads(r.text)[0]['p']

  data = {
    'sku_id': skuId,
    'sku_name': sku_name,
    'price': price
  }
  return data

数据存储与插入

将抓取到的价格存入 SQLite 数据库，建议使用 PyCharm 的 Database 功能来创建和管理 sqlite 数据库。

最终将获取的数据插入到数据库中。


# 新增
def insert(data):
  conn = sqlite3.connect('price.db')
  c = conn.cursor()
  sql = 'INSERT INTO price (sku_id,sku_name,price) VALUES ("{}", "{}", "{}")'.format(data.get("sku_id"), data.get("sku_name"), data.get('price') )
  c.execute(sql)
  conn.commit()
  conn.close()

# 查询
def select(sku_id):
  conn = sqlite3.connect('price.db')
  c = conn.cursor()
  sql = 'select sku_id, sku_name, price, time from price where sku_id = "{}" order by time asc'.format(sku_id)
  cursor = c.execute(sql)

  datas = []
  for row in cursor:
    data = {
      'sku_id': row[0],
      'sku_name': row[1],
      'price': row[2],
      'time': row[3]
    }
    datas.append(data)
  conn.close()

  return datas

示例结果

设置定时抓取任务

计划任务

使用轻量级的 schedule 模块，设定每天早上 10 点执行抓取京东价格的步骤。

安装 schedule 模块。


pip install schedule


def run_price_job(skuId):

  # 使用不占主线程的方式启动 计划任务
  def run_continuously(interval=1):
    cease_continuous_run = threading.Event()

    class ScheduleThread(threading.Thread):
      @classmethod
      def run(cls):
        while not cease_continuous_run.is_set():
          schedule.run_pending()
          time.sleep(interval)

    continuous_thread = ScheduleThread()
    continuous_thread.start()
    return cease_continuous_run
  
  # 每天10点运行，get_jd_price：任务方法，skuId：任务方法的参数
  schedule.every().day.at("10:00").do(get_jd_price, skuId=skuId)
  run_continuously()

可视化历史价格

查看历史价格

使用 pytharts 模块绘制折线图，直观地查看每日价格差异。


datas = select(skuId)

def line(datas):
  x_data = []
  y_data = []
  for data in datas:
    x_data.append(data.get('time'))
    y_data.append(data.get('price'))

  (
    Line()
    .add_xaxis(x_data)
    .add_yaxis(datas[0].get('sku_name'), y_data, is_connect_nones=True)
    .render("商品历史价格.html")
  )