Skip to content

Commit

Permalink
Added loop to repeat update in python instead of OS. Added random-nes…
Browse files Browse the repository at this point in the history 
…s to the loop sleep time to make it seem more natural. Added multiple user-agents, update uses a random one each time. Added back off when given a captcha, eases back in on subsequent successes.
  • Loading branch information
@GusWard
GusWard committed Apr 14, 2017
1 parent c1e6ec2 commit eadd7ce
Show file tree
Hide file tree
Showing 3 changed files with 189 additions and 170 deletions.
260 changes: 147 additions & 113 deletions DSAChecker.py
View file
Expand Up @@ -5,10 +5,12 @@
For finding cancellations quickly and easily!
"""
import urllib.request, urllib.parse, urllib.error, urllib.request, urllib.error, urllib.parse, http.cookiejar, time, sys, os
from datetime import timedelta
from datetime import datetime
from bs4 import BeautifulSoup

import http.cookiejar
import time
import random

from DSACheckerClasses import Page

##################################################################
Expand All @@ -35,7 +37,7 @@

# Enter your gmail account details here so that the script can send emails
emailUsername = 'example@gmail.com'
emailPassword = 'mypassword' # the password to your "example@gmail.com" account
emailPassword = 'mypassword' # the password to your "example@gmail.com" account

# Change this (at your own risk) if you don't use gmail (e.g. to hotmail/yahoo/etc smtp servers
emailSMTPserver = 'smtp.gmail.com'
Expand All @@ -61,131 +63,163 @@

cookieJar = http.cookiejar.CookieJar()

def isBeforeMyTest(dt):
if dt <= myTestDate:
return True
else:
return False

def sendEmail(datetimeList):
# i should probably point out i pinched this from stackoverflow or something
SMTPserver = emailSMTPserver
sender = emailFrom
destination = emailAddresses

USERNAME = emailUsername
PASSWORD = emailPassword

# typical values for text_subtype are plain, html, xml
text_subtype = 'plain'

content = "Available DSA test slots at your selected test centre:\n\n"

for dt in datetimeList:
content += "* %s\n" % dt.strftime('%A %d %b %Y at %H:%M')

content += "\nChecked at [%s]\n\n" % time.strftime('%d-%m-%Y @ %H:%M')

subject = emailSubject

import sys
import os
import re

from smtplib import SMTP as SMTP # this invokes the secure SMTP protocol (port 465, uses SSL)
# from smtplib import SMTP # use this for standard SMTP protocol (port 25, no encryption)
from email.mime.text import MIMEText

try:
msg = MIMEText(content, text_subtype)
msg['Subject']= subject
msg['From'] = sender # some SMTP servers will do this automatically, not all

conn = SMTP(SMTPserver, 587)
conn.set_debuglevel(False)
conn.ehlo()
conn.starttls() # Use TLS

conn.login(USERNAME, PASSWORD)
try:
conn.sendmail(sender, destination, msg.as_string())
finally:
conn.close()

except Exception as exc:
sys.exit( "mail failed; %s" % str(exc) ) # give a error message

def isBeforeMyTest(dt):
if dt <= myTestDate:
return True
else:
return False

def performUpdate():

# this should point at the DSA login page
launchPage = 'https://driverpracticaltest.direct.gov.uk/login'
def sendEmail(datetimeList):
# i should probably point out i pinched this from stackoverflow or something
SMTPserver = emailSMTPserver
sender = emailFrom
destination = emailAddresses

print('[%s]' % (time.strftime('%Y-%m-%d @ %H:%M'),))
print('---> Starting update...')
USERNAME = emailUsername
PASSWORD = emailPassword

launcher = Page(launchPage, cookieJar)
launcher.connect()
launcher.fields['username'] = licenceNumber
launcher.fields['password'] = theoryNumber

# check to see if captcha
captcha = launcher.html.find('div', id='recaptcha-check')
if captcha:
print('Captcha was present, retry later')
# TODO: implement something to solve these or prompt you for them
return
print('')
# typical values for text_subtype are plain, html, xml
text_subtype = 'plain'

time.sleep(pauseTime)
content = "Available DSA test slots at your selected test centre:\n\n"

launcher.connect()
if captcha:
print(launcher.html.find("Enter details below to access your booking"))
for dt in datetimeList:
content += "* %s\n" % dt.strftime('%A %d %b %Y at %H:%M')

dateChangeURL = launcher.html.find(id="date-time-change").get('href')
# example URL: href="/manage?execution=e1s1&amp;csrftoken=hIRXetGR5YAOdERH7aTLi14fHfOqnOgt&amp;_eventId=editTestDateTime"
# i am probably screwing up the POST bit on the forms
dateChangeURL = 'https://driverpracticaltest.direct.gov.uk' + dateChangeURL
content += "\nChecked at [%s]\n\n" % time.strftime('%d-%m-%Y @ %H:%M')

subject = emailSubject

slotPickingPage = Page(dateChangeURL, cookieJar)
slotPickingPage.fields = launcher.fields
import sys

slotPickingPage.connect()
from smtplib import SMTP as SMTP # this invokes the secure SMTP protocol (port 465, uses SSL)
# from smtplib import SMTP # use this for standard SMTP protocol (port 25, no encryption)
from email.mime.text import MIMEText

e1s2URL = slotPickingPage.html.form.get('action')
e1s2URL = 'https://driverpracticaltest.direct.gov.uk' + e1s2URL
datePickerPage = Page(e1s2URL, cookieJar)
try:
msg = MIMEText(content, text_subtype)
msg['Subject'] = subject
msg['From'] = sender # some SMTP servers will do this automatically, not all

datePickerPage.fields['testChoice'] = 'ASAP'
datePickerPage.fields['drivingLicenceSubmit'] = 'Continue'
datePickerPage.fields['csrftoken'] = dateChangeURL.split('=')[3]

datePickerPage.connect()
conn = SMTP(SMTPserver, 587)
conn.set_debuglevel(False)
conn.ehlo()
conn.starttls() # Use TLS

# earliest available date
conn.login(USERNAME, PASSWORD)
try:
conn.sendmail(sender, destination, msg.as_string())
finally:
conn.close()

availableDates = []
except Exception as exc:
sys.exit("mail failed; %s" % str(exc)) # give a error message

for slot in datePickerPage.html.find_all(class_='SlotPicker-slot'):
try:
availableDates.append(datetime.strptime(slot['data-datetime-label'].strip(), '%A %d %B %Y %I:%M%p'))
except Exception as ex:
print("".join(traceback.format_exception(etype=type(ex),value=ex,tb=ex.__traceback__)))
print ('---> Available slots:')

soonerDates = []

for dt in availableDates:
if isBeforeMyTest(dt):
print ('-----> [CANCELLATION] %s' % (dt.strftime('%A %d %b %Y at %H:%M'),))
soonerDates.append(dt)
else:
print ('-----> %s' % (dt.strftime('%A %d %b %Y at %H:%M'),))
if len(soonerDates):
sendEmail(soonerDates)
soonerDates = []
baseWaitTime = 600
userAgents = [
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/56.0.2924.87 Safari/537.36',
'Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/56.0.2924.87 Safari/537.36',
'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/56.0.2924.87 Safari/537.36',
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_3) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/56.0.2924.87 Safari/537.36',
'Mozilla/5.0 (Windows NT 10.0; WOW64; rv:52.0) Gecko/20100101 Firefox/52.0',
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_3) AppleWebKit/602.4.8 (KHTML, like Gecko) Version/10.0.3 Safari/602.4.8',
'Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/56.0.2924.87 Safari/537.36'
]

performUpdate()
print('')

def performUpdate():
global baseWaitTime
global userAgents
global soonerDates

# this should point at the DSA login page
launchPage = 'https://driverpracticaltest.direct.gov.uk/login'

print('[%s]' % (time.strftime('%Y-%m-%d @ %H:%M'),))
print('---> Starting update...')

# use a random agent for each run through
agent = userAgents[random.randint(0, len(userAgents) - 1)]
print("---> Using agent " + agent)

launcher = Page(launchPage, cookieJar)
launcher.connect(agent)
launcher.fields['username'] = licenceNumber
launcher.fields['password'] = theoryNumber

# check to see if captcha
captcha = launcher.html.find('div', id='recaptcha-check')
if captcha:
# server is suspicious, back off a bit!
baseWaitTime *= 2
print('Captcha was present, increased baseline wait time to ' + str(baseWaitTime/60)) + ' minutes'
# TODO: implement something to solve these or prompt you for them
return
print('')

time.sleep(pauseTime)

launcher.connect(agent)
if captcha:
print(launcher.html.find("Enter details below to access your booking"))

dateChangeURL = launcher.html.find(id="date-time-change").get('href')
# example URL: href="/manage?execution=e1s1&amp;csrftoken=hIRXetGR5YAOdERH7aTLi14fHfOqnOgt&amp;_eventId=editTestDateTime"
# i am probably screwing up the POST bit on the forms
dateChangeURL = 'https://driverpracticaltest.direct.gov.uk' + dateChangeURL

slotPickingPage = Page(dateChangeURL, cookieJar)
slotPickingPage.fields = launcher.fields

slotPickingPage.connect(agent)

e1s2URL = slotPickingPage.html.form.get('action')
e1s2URL = 'https://driverpracticaltest.direct.gov.uk' + e1s2URL
datePickerPage = Page(e1s2URL, cookieJar)

datePickerPage.fields['testChoice'] = 'ASAP'
datePickerPage.fields['drivingLicenceSubmit'] = 'Continue'
datePickerPage.fields['csrftoken'] = dateChangeURL.split('=')[3]

datePickerPage.connect(agent)

# earliest available date

availableDates = []

for slot in datePickerPage.html.find_all(class_='SlotPicker-slot'):
try:
availableDates.append(datetime.strptime(slot['data-datetime-label'].strip(), '%A %d %B %Y %I:%M%p'))
except Exception as ex:
print("".join(traceback.format_exception(etype=type(ex), value=ex, tb=ex.__traceback__)))
print ('---> Available slots:')

newSoonerDates = []
for dt in availableDates:
# only show / send new appointments
if isBeforeMyTest(dt) and (dt not in soonerDates):
print ('-----> [CANCELLATION] %s' % (dt.strftime('%A %d %b %Y at %H:%M'),))
soonerDates.append(dt)
newSoonerDates.append(dt)
else:
print ('-----> %s' % (dt.strftime('%A %d %b %Y at %H:%M'),))

if len(newSoonerDates):
print('---> Sending to ' + ', '.join(emailAddresses))
sendEmail(newSoonerDates)

if baseWaitTime > 300:
# decrease the baseline wait time as this was a success
baseWaitTime = int(baseWaitTime / 2)

while True:
print('***************************************')
performUpdate()
# wait for baseline + random time so its less robotic
sleepTime = baseWaitTime + random.randint(60, 300)
print('---> Waiting for ' + str(sleepTime / 60) + ' minutes...')
time.sleep(int(sleepTime))
91 changes: 41 additions & 50 deletions DSACheckerClasses.py
View file
@@ -1,53 +1,44 @@
import urllib.request, urllib.parse, urllib.error, urllib.request, urllib.error, urllib.parse, http.cookiejar
import urllib.error
import urllib.error
import urllib.parse
import urllib.parse
import urllib.request
import urllib.request
from bs4 import BeautifulSoup
from datetime import datetime
import re

class Page:
fields = {}
url = None
connection = None
html = None # BeautifulSoup object
cookieJar = None
opener = None
response = None

def __init__(self, url, cj):
self.url = url
self.cookieJar = cj

def connect(self):
print("---> Connecting to %s" % (self.url,))

self.opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(self.cookieJar))
self.opener.addheaders.append(('User-agent', 'Mozilla/4.0'))

if self.fields:
data = urllib.parse.urlencode(self.fields)
binary_data = data.encode('ascii')
self.response = self.opener.open(self.url, binary_data)
print("-----> Sending data:")
for c in list(self.fields.keys()):
print("-------> %s = %s" % (c, self.fields[c][:20]))
else:
self.response = self.opener.open(self.url)

self.html = BeautifulSoup(self.response.read(), "html.parser")

# save the pages for diagnostic info
# save = open(re.sub(r'\W+', '', self.html.title.string) + '.html', 'w')
# save.write(str(self.html))
# save.close()













class Page:
fields = {}
url = None
connection = None
html = None # BeautifulSoup object
cookieJar = None
opener = None
response = None

def __init__(self, url, cj):
self.url = url
self.cookieJar = cj

def connect(self, agent):
print("---> Connecting to %s" % (self.url,))

self.opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(self.cookieJar))
self.opener.addheaders.append(('User-agent', agent))

if self.fields:
data = urllib.parse.urlencode(self.fields)
binary_data = data.encode('ascii')
self.response = self.opener.open(self.url, binary_data)
print("-----> Sending data:")
for c in list(self.fields.keys()):
print("-------> %s = %s" % (c, self.fields[c][:20]))
else:
self.response = self.opener.open(self.url)

self.html = BeautifulSoup(self.response.read(), "html.parser")

# save the pages for diagnostic info
# save = open(re.sub(r'\W+', '', self.html.title.string) + '.html', 'w')
# save.write(str(self.html))
# save.close()

0 comments on commit eadd7ce

Please sign in to comment.