Python實現商品價格監控的方法

這篇文章將為大家詳細講解有關Python實現商品價格監控的方法，小編覺得挺實用的，因此分享給大家做個參考，希望大家閱讀完這篇文章后可以有所收獲。

一年一度的“雙十一”又要來了，很多人已經開始摩拳擦掌，畢竟幾天之后手還在不在就不好說了。

各種社交軟件也是跟著遭殃，整天就是“來幫我一起蓋樓”，各種字體繞過屏蔽，什么奇葩的腦洞也出來了：

不過也感謝這些電商平臺，讓多年未聯系的好友、加過但沒有對話的陌生人都找到了打破尷尬的話題。（讓場面更加尷尬）

月薪上萬的白領們為了2塊5毛錢的優惠券起早貪黑，也是堪稱人類迷惑行為大賞了……

問題是，你以為自己真的賺到了？

商品“明降暗升”的傳言早有耳聞：很多商品在雙十一之前早早地把價格調高，加上優惠之后也不過就是跟以前的原價相當。讓不知情的消費者在心理上感覺占了便宜。

這個傳言是不是真的，很好判斷，只要定期去訪問商品頁面，記錄價格就可以。不過一般人也沒閑工夫這么去做。于是，我們用 Python 做了一個可以定時監控商品的小工具，可以幫你監控想要關注的商品。

工具完成之后，我們隨機挑選了幾個商品作為測試，結果就有一個中招了……（真的是隨便選的）：

這款保暖背心產品，之前標價 39.9元，到11月之后卻突然調價為 49.9元，并標注上了“雙11狂歡價”，也就是原價……

商品價格監控

實現功能

·輸入天貓、蘇寧、京東、拼多多（網頁頁面 http://yangkeduo.com/）任一商品鏈接，不是口令。請復制選擇好商品配置的頁面鏈接，即返回相應商品價格，并保存到文件。商品頁面若有團購與單獨購買兩個價格，返回團購價格。

·使用 Windows 任務計劃或 Linux 定時任務，定時執行程序。獲取不同時段的商品價格信息。

·單獨運行畫圖程序，可根據定時任務獲取的數據，生成商品價格時間變化折線圖。

·程序監測的兩件商品截圖如下，具體文件在 pic 文件夾下 bnbx.html、kyy.html，推薦本地查看。

簡單的商品查看頁面 https://htmlpreview.github.io/?https://raw.githubusercontent.com/spiderbeg/price_monitor/master/search/search.html 。輸入查詢商品關鍵詞，選擇商城，即可查看相應商城商品列表。默認為蘇寧。效果圖如下。注意：點擊后請等待一段時間即可，請勿頻繁刷新。

運行環境

·python3.7

·Windows

·jupyter notebook

運行依賴包

·requests

·pyecharts

·beautifulsoup4

項目思路

部分問題回答

項目的大致思路流程：

·第一步：使用商品詳細頁鏈接獲取商品信息與商品價格，并保存獲取數據 時間、商品介紹，價格 到 csv 文件中；

·第二步：使用定時任務定時執行第一步完成的程序；

·第三步：讀取前兩步獲取到的時間、商品介紹、價格數據。使用 pyecharts 繪制繪制商品價格時間變化折線圖。

為什么不使用 pc 端來調試網頁，獲取價格信息？

因為在未登錄狀態天貓的詳細商品頁的信息是虛假的，同時從移動端網頁入手，可以降低調試難度。

谷歌瀏覽器如何開啟手機調試模式？

F12 進入開發者模式，然后鼠標點擊一下，具體見下圖,包括后文的查找價格接口信息。

實現代碼

test.py

測試商品鏈接是否能夠成功獲取到商品價格。

import timing
"""
1、調用 timing.py 中的 go 方法測試鏈接的可用性
2、調用 timing.py 中的 go, get_url() 方法測試 goods.csv 文件中鏈接的可用性
"""
# 鏈接測試
# urls = ['https://m.suning.com/product/0000000000/000000011210599174.html?utm_source=baidu&utm_midium=brand-wuxian 
&utm_content=&utm_campaign=title&safp=f73ee1cf.wapindex7.113464229882.4&safc=prd.1.rec_14-40_0_A_ab:A',
# 'https://m.suning.com/product/0070067092/000000000188392234.html?utm_source=baidu&utm_midium=brand-wuxian& 
utm_content=&utm_campaign=title&safp=f73ee1cf.wapindex7.113464229882.60&safc=prd.1.rec_5-5_1018C,1014C$
c3ae37eafeb814a098d120647449da6f_H_ab:A',
# 'https://m.suning.com/product/0000000000/000000000107426461.html?src=snsxpd_none_recssxcnxhq_1-3_p_0000000000 
_000000000107426461_rec_21-65_3_A&safp=f73ee1cf.71jyzx.112079032536.4&safc=prd.1.rec_21-65_3_A',
# 'https://m.suning.com/product/0000000000/10606656136.html?safp=f73ee1cf.phone2019.121927933306.2&safc=prd.0.0']
# 輸入文本的鏈接可用性測試
if __name__ == '__main__':
    urls = timing.get_url()
    for url in urls:
        try:
            timing.go(url) # 獲取返回信息 
        except BaseException as e:
            print(url,'\n',e)

timing.py

進行定時抓取任務時，運行的文件。

# encoding:utf8
import time
import os
import re
import csv
from shop.jd import JD # 自定義
from shop.tm import TM
from shop.sn import SN
from shop.pdd import PDD
from apscheduler.schedulers.blocking import BlockingScheduler

# import logging
# formats = "%(asctime)s %(name)s %(levelname)s function:%(funcName)s -> :%(message)s"
# logging.basicConfig(format=formats, datefmt='%m/%d/%Y %I:%M:%S %p') # ,handlers=[logging.FileHandler(log_path, 
'a+', 'utf-8')]
# LOGGER = logging.getLogger(__name__)
# LOGGER.setLevel(logging.INFO)
basePath = os.path.dirname(os.path.abspath(__file__)) # 當前文件夾
def get_date():
    """獲取日期"""
    timestamp = int(time.time())
    time_local = time.localtime(timestamp) # #時間戳 轉 時間數組
    dt = time.strftime("%Y-%m-%d %H:%M:%S",time_local) # #時間數組 轉 新的時間格式(2016-05-05 20:28:54)
    return dt
def get_url():
    """讀取商品鏈接
    返回：圖像名，商品名，商品鏈接 元組
    """
    urls = []
    with open(os.path.join(basePath, 'goods.csv'),'r',encoding='utf8') as f:
        f_csv = csv.reader(f)
        next(f_csv) # 返回標題,直接到內容
        for row in f_csv: # 內容
            if row:
                urls.append(row)
    return urls
def go(url):
    '''輸入：鏈接
    輸出：(時間，標題，商品價格), 文件路徑 元組
    統一價格輸出，以最低價格為標準，如有團購和單獨購買以單獨購買為準
    '''
    result = re.findall('://(.+?).com', url[2])
    if result:
        result = result[0]
        if 'yangkeduo' in result:
            pd = PDD(url[2])
            title,price = pd.main()
        elif 'suning' in result:
            sn = SN(url[2])
            title,price = sn.main()
        elif 'tmall' in result or 'taobao' in result:
            tm = TM(url[2]) # 605030977928：聯想筆記本 ； 603330883901 華為 mate30 pro ; 523962011119: 酸奶 
            title,price = tm.main()
        elif 'jd' in result:
            jd = JD(url[2]) # 測試 id：100009083152 商品：聯想 y9000x 筆記本電腦 2 熱水壺 or 薯條？
            title,price = jd.main()
        else:
            raise TypeError('請檢查輸入的網站鏈接')
        print('%s 標題 %s, 價格（多個價格以團購為準） %s. '%(result,title,price))
    else:
        raise TypeError('請檢查輸入是否為目標網站的商品詳細頁面鏈接')
    # 文件名
    replace_string = ['.',' ',r'/',r'\\']
    for rs in replace_string:
        url[1] = url[1].replace(rs,'_')
    path = os.path.join(os.path.join(basePath, 'data'), url[1]+'.csv')
    today = get_date() # 日期
    return (today, title, price),path
def addData(row, path):
    """數據寫入文件"""
    with open(path,'a+',encoding='utf8') as f:
        fieldnames = ['時間', '標題','價格']
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        if f.tell() == 0: # 如果內容為空則添加標題
            writer.writeheader()
        writer.writerow({'時間': row[0], '標題': row[1],'價格':row[2]})
def main():
    """運行程序"""
    urls = get_url()
    for url in urls:
        try:
            row,path = go(url) # 獲取返回信息 
            addData(row,path) # 寫入文件
        except BaseException as e:
            print('請求問題？報錯：%s'%e)
if __name__ == '__main__':
    print('時間',get_date())
    main()
    # scheduler = BlockingScheduler()
    # scheduler.add_job(go,'cron', args=[url],hour='8-23', minute= '5,35' , second='15')
    # # scheduler.add_job(main,'cron', args=[3088512],hour='8-23', minute= 5 , second='15')
    # print('Press Ctrl+{0} to exit'.format('Break' if os.name == 'nt' else 'C'))
    # try:
    #     scheduler.start()
    # except (KeyboardInterrupt, SystemExit):
    #     pass

draw.py

圖像文件生成在 pic 文件中。

# encoding: utf8
from pyecharts import options as opts
from pyecharts.charts import Page, Line
import os
import csv

basePath = os.path.dirname(os.path.abspath(__file__)) # 當前文件夾
def line(title,checktime,price) -> Line:
    """繪圖函數"""
    c = (
        Line()
        .add_xaxis(checktime)
        .add_yaxis(title, price, is_smooth=True)
        .set_global_opts(title_opts=opts.TitleOpts(title="商品價格"),
                yaxis_opts=opts.AxisOpts(name="元/臺"),
                xaxis_opts=opts.AxisOpts(name=title,
                axislabel_opts=opts.LabelOpts(formatter="{value}", font_size=12, rotate=30,) # x,y 軸標簽
                    )
                )
        )
    return c
def files():
    """
    輸出字典，每一個鍵值代表一張圖表
    """
    global basePath
    files = {}
    with open(os.path.join(basePath,'goods.csv'),'r',encoding='utf8') as f:
        f_csv = csv.reader(f)
        next(f_csv) # 標題
        for row in f_csv: # 內容
            if row:
                replace_string = ['.',' ',r'/',r'\\'] # 特殊字符處理
                for rs in replace_string:
                    row[1] = row[1].replace(rs,'_')
                files.setdefault(row[0],[]).append(row[1])
    return files
def draw(files):
    """繪制圖形文件"""
    datapath = os.path.join(basePath,'data')
    picpath = os.path.join(basePath,'pic')
    for k,i in files.items():
        page = Page()
        for n in i:
            try:
                with open(os.path.join(datapath, n +'.csv'),'r', encoding='utf8') as f:
                    f_csv = csv.DictReader(f)
                    price,checktime = [],[]
                    for row in f_csv:
                        checktime.append(row['時間'])
                        price.append(row['價格'])
                    title = n
                page.add(line(title,checktime,price)) # 24 發帖回帖變化圖、近3月變化圖、瀏覽、回復散點圖
            except:
                print('未制圖：',n)
        page.render(os.path.join(picpath, k +'.html'))
if __name__ == '__main__':
    draw(files())

關于Python實現商品價格監控的方法就分享到這里了，希望以上內容可以對大家有一定的幫助，可以學到更多知識。如果覺得文章不錯，可以把它分享出去讓更多的人看到。