SRL Archive - Python/Perl Help and Tutorials

Newegg Price Monitor.


noidea

2010-11-25 16:43:32

*Linux only*
Requires espeak installed.

This will monitor the items on newegg you put in the array and check every X seconds.

forgot to update :S
# To change this template, choose Tools | Templates
# and open the template in the editor.

__author__="lucas"
__date__ ="$Nov 24, 2010 11:52:39 PM$"

import os
import urllib2
import threading
import time
import math

class NeweggCheck(threading.Thread):
def __init__(self, url, timecheck=30):
threading.Thread.__init__(self)
self.url, self.timecheck = url, timecheck
self.exit = False

def stop(self):
self.exit = True

def run(self):
print 'thread started', self.url
price = 0
while self.exit == False:
try:
request = urllib2.build_opener()
request.addheaders = [('User-agent', 'Mozilla/5.0')]
line = request.open(self.url).read()
except IOError, e:
print 'There was an exception'
print e
continue
line_catch = 'Biz.Product.DetailPage.switchCrossSellItem(this,\''
# check for the price
if line.find(line_catch, 0, len(line)-1) != -1: # Searches for the phrase from within the line
line_catch_posA = line.find(line_catch)+len(line_catch)+1
# Gets the price of the item from the line
price_string = line[line_catch_posA:line[line.find(line_catch)+len(line_catch)+1:len(line)-1].find('\',\'')+ line_catch_posA]
price_float = float(price_string)
print '$' + price_string, self.url
if price != price_float:
os.popen("espeak --stdout 'the. prices. have. updated. . . It. is. now."+price_string+"' | aplay")
# os.popen("espeak 'the. prices. have. updated. . . It. is. now. "+ price_string+"'")
print 'Price of', self.url, 'has changed'
price = price_float
with open('newegg.txt', 'a') as f:
f.write(price_string+' '+self.url+'\n')
else:
print 'POSSIBLE PRICE CHANGE, CANNOT FIND PRICE', self.url
os.popen("espeak --stdout 'possible price. change' | aplay") #("espeak 'possible price. change'")
wait = self.timecheck if self.timecheck < 1 else 0.5
for t in range(int(math.ceil(self.timecheck))*2):
time.sleep(wait)
if self.exit: break


if __name__ == "__main__":
urls = ['http://www.newegg.com/Product/Product.aspx?Item=N82E16856119018', #case, mobo, psu
'http://www.newegg.com/Product/Product.aspx?Item=N82E16814121397', #gpu
'http://www.newegg.com/Product/Product.aspx?Item=20-144-259', #ram
'http://www.newegg.com/Product/Product.aspx?Item=N82E16820148235', #ram
'http://www.newegg.com/Product/Product.aspx?Item=N82E16819116264', #cpu
'http://www.newegg.com/Product/Product.aspx?Item=N82E16827136181'] #bluraty burner

time_between = 30 # amount in seconds to wait before checking the site again.
thread_list = []
for url in urls:
thread_list.append(NeweggCheck(url, timecheck=time_between))
thread_list[len(thread_list)-1].start()
try:
while True:
time.sleep(0.5)
except KeyboardInterrupt:
print 'Interrupt called'
for single_thread in thread_list:
single_thread.stop()

Thats an example. time_between is the amount in seconds to wait before checking the webpage again.

It logs the price changed in newegg.txt (which it will create).
It will espeak when the prices change or it cannot detect a price. All the pages I've tried so far have worked, but I ran into an issue earlier with one and fixed it.
Tell me if you find one that doesnt work.

Update: Fixed url exceptions, threading, etc.
yet to add: getting the items name. Should be piss easy.

Nava2

2010-11-25 17:10:10

I will definitely try this out, it looks great!

Just what I need right now, especially with all the sales going on.

noidea

2010-11-25 17:57:33

None of the things I've been watching have changed yet, so please try it and see if it works :3

Nava2

2010-11-26 03:06:16

So, I'm currently using this.

Advice:
- Create a queue for terminal/speech output, thus staggering the outputs as time goes on.

- Make the Output more informative, since seeing a large url with an ID is kinda lame.
– I'm sure that there is some HTML on the page which can be parsed for the name of the item. Maybe store that in the thread object on start?

So.. I implemented the Speech queue.. I really should be in bed though.. haha. :(

I also changed os.popen() to subprocess, since popen is depreciated. I moved some things around, and fixed some formatting issues. Use tab, boy. Set tab to 4 spaces in your editor haha.

# -*- coding: utf-8 -*-
# To change this template, choose Tools | Templates
# and open the template in the editor.

__author__="lucas"
__date__ ="$Nov 24, 2010 11:52:39 PM$"

import os
import urllib
import threading
import time
import Queue
import subprocess


class NeweggCheck(threading.Thread):
def __init__(self, url, timecheck=30):
threading.Thread.__init__(self)
self.url, self.timecheck = url, timecheck

def run(self):
print 'thread started', self.url
price = 0
while True:
lines = urllib.urlopen(self.url).readlines()
# The line to search for in the website's code
# line_catch = 'onclick="javascript:Biz.Product.DetailPage.switchCrossSellItem(this,\''
line_catch = 'Biz.Product.DetailPage.switchCrossSellItem(this,\''
# Check each line for that line above
found = False
for line in lines:
if line.find(line_catch, 0, len(line)-1) != -1: # Searches for the phrase from within the line
print line[line.find("'$")+1: line.find('\',')], self.url# Gets the price of the item from the line
price_string = line[line.find("'$")+1: line.find('\',')]
price_float = float(price_string[1:])
if price != price_float:
# add the speech into the speech queue
speechThread.add_speech("'the. prices. have. updated. . . It. is. now. %s'" % price_string)

print 'Price of', self.url, 'has changed'
price = price_float
f = open('newegg.txt', 'a')
f.write(price_string+' '+self.url+'\n')
f.close()
found = True
break
if not found:
print 'POSSIBLE PRICE CHANGE, CANNOT FIND PRICE', self.url
speechThread.add_speech("espeak 'possible price. change'")
time.sleep(self.timecheck)



class SpeechQueue(threading.Thread):
'''
Thread to make speech un-garbled.
'''

def __init__(self,min_wait=0.016):
threading.Thread.__init__(self)

self.queue = Queue.Queue()
if min_wait < 0.016:
self.min_wait = 0.016
else:
self.min_wait = min_wait

def add_speech(self, speech):
'''
Adds the speech into the queue to be spoken later.
'''
self.queue.put(speech)

def run(self):
print('started speech thread ..')

while True:
if not self.queue.empty():
words = self.queue.get()
# create subprocess to speak.
subprocess.call(['espeak', "'%s'"%words])
time.sleep(self.min_wait)


if __name__ == "__main__":
urls = ['http://www.newegg.com/Product/Product.aspx?Item=N82E16822136533',
'http://www.newegg.com/Product/Product.aspx?Item=N82E16820148357',
'http://www.newegg.com/Product/Product.aspx?Item=N82E16812123282',
'http://www.newegg.com/Product/Product.aspx?Item=N82E16814500115',
'http://www.newegg.com/Product/Product.aspx?Item=N82E16820231310',
'http://www.newegg.com/Product/Product.aspx?Item=N82E16823201033',
'http://www.newegg.com/Product/Product.aspx?Item=N82E16826125108',
'http://www.newegg.com/Product/Product.aspx?Item=N82E16827135204']
time_between = 20 * 60 # amount in seconds to wait before checking the site again.
thread_list = []

# start the speech thread
speechThread = SpeechQueue()
speechThread.start()

#start the price check threads
for url in urls:
thread_list.append(NeweggCheck(url, timecheck=time_between))
thread_list[len(thread_list)-1].start()
[/highlight]

Nava2

2010-11-26 11:59:08

Just a note:

POSSIBLE PRICE CHANGE, CANNOT FIND PRICE http://www.newegg.com/Product/Product.aspx?Item=N82E16814500115
$11.99 http://www.newegg.com/Product/Product.aspx?Item=N82E16823201033
$7.99 http://www.newegg.com/Product/Product.aspx?Item=N82E16826125108
$134.99 http://www.newegg.com/Product/Product.aspx?Item=N82E16820148357
POSSIBLE PRICE CHANGE, CANNOT FIND PRICE http://www.newegg.com/Product/Product.aspx?Item=N82E16822136533
$17.99 http://www.newegg.com/Product/Product.aspx?Item=N82E16827135204
$4.29 http://www.newegg.com/Product/Product.aspx?Item=N82E16812123282
POSSIBLE PRICE CHANGE, CANNOT FIND PRICE http://www.newegg.com/Product/Product.aspx?Item=N82E16820231310
$114.99 http://www.newegg.com/Product/Product.aspx?Item=N82E16814500115
$11.99 http://www.newegg.com/Product/Product.aspx?Item=N82E16823201033
$7.99 http://www.newegg.com/Product/Product.aspx?Item=N82E16826125108
$134.99 http://www.newegg.com/Product/Product.aspx?Item=N82E16820148357
POSSIBLE PRICE CHANGE, CANNOT FIND PRICE http://www.newegg.com/Product/Product.aspx?Item=N82E16822136533
$4.29 http://www.newegg.com/Product/Product.aspx?Item=N82E16812123282
$17.99 http://www.newegg.com/Product/Product.aspx?Item=N82E16827135204
POSSIBLE PRICE CHANGE, CANNOT FIND PRICE http://www.newegg.com/Product/Product.aspx?Item=N82E16820231310

noidea

2010-11-26 16:06:23

From: Nava2
Just a note:


weird, they seem to work for me on both newegg.com and newegg.ca.
$89.99 http://www.newegg.com/Product/Product.aspx?Item=N82E16822136533
$56.99 http://www.newegg.com/Product/Product.aspx?Item=N82E16820231310
$114.99 http://www.newegg.com/Product/Product.aspx?Item=N82E16814500115

$58.49 http://www.newegg.ca/Product/Product.aspx?Item=N82E16820231310
$99.99 http://www.newegg.ca/Product/Product.aspx?Item=N82E16822136533
$116.14 http://www.newegg.ca/Product/Product.aspx?Item=N82E16814500115



From: Nava2
Advice:
- Create a queue for terminal/speech output, thus staggering the outputs as time goes on.

I dont really see the point of staggering them. Sure it could be an option, but I want to see info asap. I can see issues like if all of the prices change at the same time and it tries to espeak them all.

From: Nava2
- Make the Output more informative, since seeing a large url with an ID is kinda lame.
– I'm sure that there is some HTML on the page which can be parsed for the name of the item. Maybe store that in the thread object on start?
\

Will do. I should have added that before.

From: Nava2

I also changed os.popen() to subprocess, since popen is depreciated. I moved some things around, and fixed some formatting issues. Use tab, boy. Set tab to 4 spaces in your editor haha.


Sorry I've been using python 2.65 :S
I've been using tab. I guess copying it from netbeans to paste.villavu.com to forums changed it.

Thanks for your suggestions!

Nava2

2010-11-26 16:12:22

So, I have been out of coding for too long, thus why I'm spending so much time on this..

I am thinking of adding a way to kill this, because doing `ps aux | grep newegg.py' then killing the ID is annoying.

Also, I've noticed that exceptions are thrown by urllib.. these should probably be handled, if not just passed off. The exception I get is ``IOError: [Errno socket error] [Errno 110] Connection timed out'' which kills the thread that was running, forcing the application to be restarted.

noidea

2010-11-26 16:14:48

From: Nava2
I am thinking of adding a way to kill this, because doing `ps aux | grep newegg.py' then killing the ID is annoying.

Completely forgot about that, sorry :S

From: Nava2
Also, I've noticed that exceptions are thrown by urllib.. these should probably be handled, if not just passed off. The exception I get is ``IOError: [Errno socket error] [Errno 110] Connection timed out'' which kills the thread that was running, forcing the application to be restarted.


I havent encountered that yet, but I'll fix it.

edit: Ok, working on stopping the threads atm because that seems to be the biggest issue. Just fixed the urllib exception.

Nava2

2010-11-26 16:54:24

From: noidea
Completely forgot about that, sorry :S



I havent encountered that yet, but I'll fix it.

edit: Ok, working on stopping the threads atm because that seems to be the biggest issue. Just fixed the urllib exception.


# -*- coding: utf-8 -*-
# To change this template, choose Tools | Templates
# and open the template in the editor.

__author__="lucas"
__date__ ="$Nov 24, 2010 11:52:39 PM$"

import os
import urllib
import threading
import time
import Queue
import subprocess
import sys

shutdown_event = threading.Event()

class NeweggCheck(threading.Thread):
def __init__(self, url, timecheck=30):
threading.Thread.__init__(self)
self.url, self.timecheck = url, timecheck

def run(self):
print 'thread started', self.url
price = 0
while not shutdown_event.is_set():

#read the page into a string list
try:
lines = urllib.urlopen(self.url).readlines()
except (KeyboardInterrupt, SysExit):
pass
except:
time.sleep(self.timecheck)
continue

# The line to search for in the website's code
# line_catch = 'onclick="javascript:Biz.Product.DetailPage.switchCrossSellItem(this,\''
line_catch = 'Biz.Product.DetailPage.switchCrossSellItem(this,\''
# Check each line for that line above
found = False
for line in lines:
if line.find(line_catch, 0, len(line)-1) != -1: # Searches for the phrase from within the line
print line[line.find("'$")+1: line.find('\',')], self.url# Gets the price of the item from the line
price_string = line[line.find("'$")+1: line.find('\',')]
price_float = float(price_string[1:])
if price != price_float:
# add the speech into the speech queue
speechThread.add_speech("'the. prices. have. updated. . . It. is. now. %s'" % price_string)

print 'Price of', self.url, 'has changed'
price = price_float
f = open('newegg.txt', 'a')
f.write(price_string+' '+self.url+'\n')
f.close()
found = True
break
if not found:
print 'POSSIBLE PRICE CHANGE, CANNOT FIND PRICE', self.url
speechThread.add_speech("espeak 'possible price. change'")

wait_time = time.time() + self.timecheck
while time.time() < wait_time and not shutdown_event.is_set():
time.sleep(0.016)



class SpeechQueue(threading.Thread):
'''
Thread to make speech un-garbled.
'''

def __init__(self,min_wait=0.016):
threading.Thread.__init__(self)

self.queue = Queue.Queue()
if min_wait < 0.016:
self.min_wait = 0.016
else:
self.min_wait = min_wait

def add_speech(self, speech):
'''
Adds the speech into the queue to be spoken later.
'''
self.queue.put(speech)

def run(self):
print('started speech thread ..')

while not shutdown_event.is_set():
if not self.queue.empty():
words = self.queue.get()
# create subprocess to speak.
subprocess.call(['espeak', "'%s'"%words])
time.sleep(self.min_wait)


if __name__ == "__main__":
urls = ['http://www.newegg.com/Product/Product.aspx?Item=N82E16822136533',
'http://www.newegg.com/Product/Product.aspx?Item=N82E16820148357',
'http://www.newegg.com/Product/Product.aspx?Item=N82E16812123282',
'http://www.newegg.com/Product/Product.aspx?Item=N82E16814500115',
'http://www.newegg.com/Product/Product.aspx?Item=N82E16820231310',
'http://www.newegg.com/Product/Product.aspx?Item=N82E16823201033',
'http://www.newegg.com/Product/Product.aspx?Item=N82E16826125108',
'http://www.newegg.com/Product/Product.aspx?Item=N82E16827135204']
time_between = 20 * 60 # amount in seconds to wait before checking the site again.
thread_list = []

# start the speech thread
speechThread = SpeechQueue()
speechThread.start()

#start the price check threads
for url in urls:
thread_list.append(NeweggCheck(url, timecheck=time_between))
thread_list[len(thread_list)-1].start()

try:
while True:
time.sleep(0.016)
except (KeyboardInterrupt, SystemExit):
shutdown_event.set()

while speechThread.is_alive():
time.sleep(0.016)
print("speech thread killed.")

for th in thread_list:
while th.is_alive():
time.sleep(0.016)
print("killed all price threads.")

print("Application Terminated successfully")

sys.exit(0)
[/highlight]

Threads are all killed gracefully and the application exits nice! :)

E: More thread-safe version:
[highlight=python]# -*- coding: utf-8 -*-
# To change this template, choose Tools | Templates
# and open the template in the editor.

__author__="lucas"
__date__ ="$Nov 24, 2010 11:52:39 PM$"

import os
import urllib
import threading
import time
import Queue
import subprocess
import sys

shutdown_event = threading.Event()

class NeweggCheck(threading.Thread):
def __init__(self, url, timecheck=30):
threading.Thread.__init__(self)
self.url, self.timecheck = url, timecheck

def sleep(self, sleep_time):
'''
Sleeps the thread, but also checks for the shutdown signal. sleep_time is in
seconds.
'''
wait_time = time.time() + (sleep_time * 1000)
while time.time() < wait_time and not shutdown_event.is_set():
time.sleep(0.016)


def run(self):
print 'thread started', self.url
price = 0
while not shutdown_event.is_set():

#read the page into a string list
try:
lines = urllib.urlopen(self.url).readlines()
except (KeyboardInterrupt, SysExit):
pass
except:
self.sleep(self.timecheck)
continue

# The line to search for in the website's code
# line_catch = 'onclick="javascript:Biz.Product.DetailPage.switchCrossSellItem(this,\''
line_catch = 'Biz.Product.DetailPage.switchCrossSellItem(this,\''
# Check each line for that line above
found = False
for line in lines:
if line.find(line_catch, 0, len(line)-1) != -1: # Searches for the phrase from within the line
print line[line.find("'$")+1: line.find('\',')], self.url# Gets the price of the item from the line
price_string = line[line.find("'$")+1: line.find('\',')]
price_float = float(price_string[1:])
if price != price_float:
# add the speech into the speech queue
speechThread.add_speech("'the. prices. have. updated. . . It. is. now. %s'" % price_string)

print 'Price of', self.url, 'has changed'
price = price_float
f = open('newegg.txt', 'a')
f.write(price_string+' '+self.url+'\n')
f.close()
found = True
break
if not found:
print 'POSSIBLE PRICE CHANGE, CANNOT FIND PRICE', self.url
speechThread.add_speech("espeak 'possible price. change'")

self.sleep(self.timecheck)


class SpeechQueue(threading.Thread):
'''
Thread to make speech un-garbled.
'''

def __init__(self,min_wait=0.016):
threading.Thread.__init__(self)

self.queue = Queue.Queue()
if min_wait < 0.016:
self.min_wait = 0.016
else:
self.min_wait = min_wait

def add_speech(self, speech):
'''
Adds the speech into the queue to be spoken later.
'''
self.queue.put(speech)

def run(self):
print('started speech thread ..')

while not shutdown_event.is_set():
if not self.queue.empty():
words = self.queue.get()
# create subprocess to speak.
subprocess.call(['espeak', "'%s'"%words])
time.sleep(self.min_wait)


if __name__ == "__main__":
urls = ['http://www.newegg.com/Product/Product.aspx?Item=N82E16822136533',
'http://www.newegg.com/Product/Product.aspx?Item=N82E16820148357',
'http://www.newegg.com/Product/Product.aspx?Item=N82E16812123282',
'http://www.newegg.com/Product/Product.aspx?Item=N82E16814500115',
'http://www.newegg.com/Product/Product.aspx?Item=N82E16820231310',
'http://www.newegg.com/Product/Product.aspx?Item=N82E16823201033',
'http://www.newegg.com/Product/Product.aspx?Item=N82E16826125108',
'http://www.newegg.com/Product/Product.aspx?Item=N82E16827135204']
time_between = 20 * 60 # amount in seconds to wait before checking the site again.
thread_list = []

# start the speech thread
speechThread = SpeechQueue()
speechThread.start()

#start the price check threads
for url in urls:
thread_list.append(NeweggCheck(url, timecheck=time_between))
thread_list[len(thread_list)-1].start()

try:
while True:
time.sleep(0.016)
except (KeyboardInterrupt, SystemExit):
shutdown_event.set()

while speechThread.is_alive():
time.sleep(0.016)
print("speech thread killed.")

for th in thread_list:
while th.is_alive():
time.sleep(0.016)
print("killed all price threads.")

print("Application Terminated successfully")

sys.exit(0)
[/highlight]

E2: Fixed a small math issue.. :)

noidea

2010-12-04 13:23:57

I think I fixed the issue of not being able to find the item. I think the newegg site might have been giving the script a fake result.
So I have it present the firefox user agent, and it hasn't failed yet.

Thanks nava for help and suggestions on the irc.

Edit: nvm.. after a day of it working successfully, all of a sudden 3 of my items stopped working..
edit2: ffffff its been working for like 3 days straight now after that issue.