utils/browser.py @@ -8,4 +8,4 @@ from_address = '"watchdog.net" <info@watchdog.net>' maildir_path = '/home/wathdog/Maildir' send_errors_to = 'bugs@watchdog.net' test_email = 'test@watchdog.net' #wyr test emails to go to this -production_test = 'wyr_prod@watchdog.net' #all production wyr msgs and their responses go here +production_test_email = 'wyr_prod@watchdog.net' #all production wyr msgs and their responses go here config.py @@ -3,30 +3,33 @@ Functions to send msgs to reps and deal with captchas """ import sys -from BeautifulSoup import BeautifulSoup -from ClientForm import ParseFile, ParseError, XHTMLCompatibleFormParser -from StringIO import StringIO import re, urllib2 from urlparse import urljoin import web from settings import db from wyrutils import * -from utils import captchasolver, messages, helpers +from utils import captchasolver, messages, helpers, browser +from config import production_test_email, from_address, test_email +from settings import production_mode __all__ = ["prepare", "send_msgs", "send", "writerep"] +test_mode = (not production_mode) + def prepare(pol): """ Checks if the `pol`'s contact url has captcha, and if so - takes care of - with cookies, return env with captcha url and form having captcha. + with cookies, return env with captcha url, cookies and form having captcha. """ env = {} url = getcontact(pol).get('contact') - response = urlopen(url) - captcha_src, form = get_img_and_form(response) + b = browser.Browser() + b.open(url) + captchas = b.find_nodes('img', attrs={'src': re.compile('.*[Cc]aptcha.*')}) + captcha_src, form = get_src_and_form(url, captchas) if captcha_src: - env['captcha_src'], env['form'] = captcha_src, form + env['captcha_src'], env['form'], env['cookies'] = captcha_src, form, b.get_state() return env def send(frm, to, subj, msg, user_details, source_id=None, env={}): @@ -71,6 +74,7 @@ def writerep(pol, i, env={}): handlers = dict(E=writerep_email, W=writerep_wyr, I=writerep_ima, Z=writerep_zipauth) try: handler = handlers[contacttype] + print handler.__name__, if contacttype == 'I': msg_sent = handler(pol, contact, i, env) else: @@ -80,17 +84,6 @@ def writerep(pol, i, env={}): msg_sent = False return msg_sent -def writerep_email(pol, pol_email, i): - name = '%s. %s %s' % (i.prefix, i.fname, i.lname) - from_addr = '%s <%s>' % (name, i.email) - - if production_mode: - to_addr = web.lstrips(pol_email, 'mailto:') - elif test_mode: - to_addr = test_email - web.sendmail(from_addr, to_addr, subject, msg) - return True - def writerep_wyr(pol, wyr_link, i): """Sends the msg along with the sender details from `i` through the WYR system. The WYR system has 3 forms typically (and a captcha form for few reps in between 1st and 2nd forms). @@ -98,55 +91,50 @@ def writerep_wyr(pol, wyr_link, i): Form 2 asks for sender's details such as prefix, name, city, address, email, phone etc Form 3 asks for the msg to send. """ + b = browser.Browser() + def wyr_step1(url): - forms, response = get_forms(url) - form = forms[0] + b.open(url) + form = get_form(b, not_signup_or_search) # state names are in form: "PRPuerto Rico" state_options = form.find_control_by_name('state').items state_l = [s.name for s in state_options if s.name[:2] == i.state] form.fill_all(state=state_l[0], zipcode=i.zip5, zip4=i.zip4) print 'step1 done', - request = form.click() - return request - - def get_challenge(soup): - labels = filter(lambda x: x.get('for') == 'HIP_response', soup.findAll('label')) - if labels: return labels[0].string + return form.click() + def get_challenge(): + labels = b.find_nodes('label', lambda x: x.get('for') == 'HIP_response') + if labels: return labels[0].string + def get_wyr_form2(request): - if not request: return - url, data = request.get_full_url(), request.get_data() - forms, response = get_forms(url, data) - soup = BeautifulSoup(response) - if len(forms) < 1: - if has_message(soup, "is shared by more than one"): raise ZipShared - elif has_message(soup, "not correct for the selected State"): raise ZipIncorrect - elif has_message(soup, "was not found in our database."): raise ZipNotFound - elif has_message(soup, "Use your web browser's <b>BACK</b> capability "): raise WyrError + b.open(request) + form = get_form(b, not_signup_or_search) + if not form: + if b.has_text("is shared by more than one"): raise ZipShared + elif b.has_text("not correct for the selected State"): raise ZipIncorrect + elif b.has_text("was not found in our database."): raise ZipNotFound + elif b.has_text("Use your web browser's <b>BACK</b> capability "): raise WyrError else: raise NoForm else: - challenge = get_challenge(soup) + challenge = get_challenge() if challenge: - form = forms[0] try: solution = captchasolver.solve(challenge) except Exception, detail: print >> sys.stderr, 'Exception in CaptchaSolve', detail - print 'Could not solve:"%s"' % challenge, + print >> sys.stderr, 'Could not solve:"%s"' % challenge, else: form.f['HIP_response'] = str(solution) request = form.click() form = get_wyr_form2(request) return form else: - return forms[0] + return form def wyr_step2(request): - if not request: return form = get_wyr_form2(request) - if not form: return - - if form.fill_name(i.prefix, i.fname, i.lname): + if form and form.fill_name(i.prefix, i.fname, i.lname): form.fill_address(i.addr1, i.addr2) form.fill_all(city=i.city, phone=i.phone, email=i.email) request = form.click() @@ -154,16 +142,11 @@ def writerep_wyr(pol, wyr_link, i): return request def wyr_step3(request): - if not request: return - forms, response = get_forms(request.get_full_url(), request.get_data()) - forms = filter(lambda f: f.has(type='textarea'), forms) - if forms: - form = forms[0] - if form.fill(i.full_msg, type='textarea'): - print 'step3 done', - return form.production_click() - else: - print >> sys.stderr, response + b.open(request) + form = get_form(b, lambda f: f.find_control_by_type('textarea')) + if form and form.fill(i.full_msg, type='textarea'): + print 'step3 done', + return submit_form(b, form) return wyr_step3(wyr_step2(wyr_step1(wyr_link))) @@ -173,25 +156,22 @@ def writerep_ima(pol, ima_link, i, env={}): and subject and msg (and with a captcha img for few reps/senators). If it has a captcha img, the form to fill captcha is taken from env. """ - try: - forms = ParseFile(StringIO(env.get('form', '')), ima_link, backwards_compat=False) - forms = [Form(f) for f in forms] - except: pass - - if not forms: - forms, response = get_forms(ima_link) - forms = filter(lambda f: f.has(type='textarea') , forms) + b = browser.Browser(env.get('cookies', [])) + b.url, b.page = ima_link, env.get('form') + f = get_form(b, lambda f: f.find_control_by_type('textarea')) + if not f: + b.open(ima_link) + f = get_form(b, lambda f: f.find_control_by_type('textarea')) - if forms: - f = forms[0] + if f: f.fill_name(i.prefix, i.fname, i.lname) f.fill_address(i.addr1, i.addr2) f.fill_phone(i.phone) f.fill(type='textarea', value=i.full_msg) captcha_val = i.get('captcha_%s' % pol, '') f.fill_all(city=i.city, state=i.state.upper(), zipcode=i.zip5, zip4=i.zip4, email=i.email,\ - issue=['GEN', 'OTH'], subject=i.subject, captcha=captcha_val, reply='yes') - return f.production_click() + issue=['GEN', 'OTH'], subject=i.subject, captcha=captcha_val, reply='yes') + return submit_form(b, f) else: print 'Error: No IMA form in', ima_link, @@ -201,7 +181,7 @@ def writerep_zipauth(pol, zipauth_link, i): Form 1 asks for zipcode and few user details Form 2 asks for the subject and msg to send and other sender's details. """ - def zipauth_step1(f): + def zipauth_step1(f): f.fill_name(i.prefix, i.fname, i.lname) f.fill_address(i.addr1, i.addr2) f.fill_phone(i.phone) @@ -211,13 +191,11 @@ def writerep_zipauth(pol, zipauth_link, i): print 'step1 done', return f.click() - def zipauth_step2(request): - if not request: return - headers = {'Cookie' : 'District=%s' % i.zip5} - forms, response = get_forms(request.get_full_url(), request.get_data(), headers) - forms = filter(lambda f: f.has(type='textarea'), forms) - if forms: - f = forms[0] + def zipauth_step2(request): + request.add_header('Cookie', 'District=%s' % i.zip5) #@@ done in ajax :( + response = b.open(request) + f = get_form(b, lambda f: f.find_control_by_type('textarea')) + if f: f.fill_name(i.prefix, i.fname, i.lname) f.fill_address(i.addr1, i.addr2) f.fill_phone(i.phone) @@ -225,65 +203,73 @@ def writerep_zipauth(pol, zipauth_link, i): f.fill_all(city=i.city, zipcode=i.zip5, zip4=i.zip4, state=i.state.upper(), email=i.email, issue=['GEN', 'OTH'], subject=i.subject, reply='yes') print 'step2 done', - return f.production_click() + return submit_form(b, f) else: - soup = BeautifulSoup(response) - if has_message(soup, 'zip code is split between more', 'p'): raise ZipShared - if has_message(soup, 'Access to the requested form is denied', ['p', 'font']): raise ZipIncorrect - if has_message(soup, 'you are outside', 'p'): raise ZipIncorrect + print 'no form with text area' + if b.has_text('zip code is split between more'): raise ZipShared + if b.has_text('Access to the requested form is denied'): raise ZipIncorrect + if b.has_text('you are outside'): raise ZipIncorrect - forms, response = get_forms(zipauth_link) - forms = filter(lambda f: f.has(name='zip'), forms) - if forms: - return zipauth_step2(zipauth_step1(forms[0])) - else: - if verbose: print 'Error: No zipauth form in', zipauth_link - return + b = browser.Browser() + b.open(zipauth_link) + form = get_form(b, lambda f: f.has(name='zip')) + if form: + return zipauth_step2(zipauth_step1(form)) + else: + print 'Error: No zipauth form in', zipauth_link -def get_forms(url, data=None, headers={}): - """Returns all the forms other than search and signup from the webpage with url `url`. - """ - def signup_or_search(form): - u = form.action - try: - form.find_control(type='textarea') - except: - return ('signup' in u and 'email' in u) or ('search' in u) or ('thomas.loc.gov' in u) - else: - return False +def writerep_email(pol, pol_email, i): + name = '%s. %s %s' % (i.prefix, i.fname, i.lname) + from_addr = '%s <%s>' % (name, i.email) - req = urllib2.Request(url, data, headers) - response = urlopen(req) or '' - forms = [] - if not response: return forms, response - response = response.read() - try: - forms = ParseFile(StringIO(response), url, backwards_compat=False) - except ParseError: - forms = ParseFile(StringIO(response), url, backwards_compat=False, form_parser_class=XHTMLCompatibleFormParser) - forms = [Form(f) for f in filter(lambda x: not signup_or_search(x), forms)] - return forms, response + if production_mode: + to_addr = web.lstrips(pol_email, 'mailto:') + elif test_mode: + to_addr = test_email + web.sendmail(from_addr, to_addr, i.subject, i.full_msg) + return True + +def submit_form(b, f): + """clicks the form `f` and opens the request in browser `b` and sends response.""" + if production_mode: + request = f.click() + response = b.open(request) + send_response(production_test_email, f.controls, response) + elif test_mode: + send_response(test_email, f.controls, response='') + return True -def has_message(soup, msg, tags='b'): - """Returns if the `tags` in `soup` have `msg` in them. +def send_response(to, form_controls, response): + """sends a mail to `to` to check if the form is submitted properly. """ - bs = soup.findAll(tags) - msg = msg.lower() - for b in bs: - errmsg = str(b.string).lower() - errmsg += ' '.join(str(c) for c in b.contents) - if (errmsg.find(msg) > -1): - return True - return False + form_values = "\n".join(["%s: %s" % (c.name, c.value) for c in form_controls]) + msg = 'Filled in the last form:\n\n' + form_values + if response: + msg += '\n\nResponse at' + response.geturl() + ':\n' + response.read() + else: + msg += '\n\n(Not Production click, no response)' + + subject = 'wyr mail' + web.sendmail(from_address, to, subject, msg) + +def not_signup_or_search(form): + has_textarea = form.find_control_by_type('textarea') + if has_textarea: + return True + else: + action = form.action + signup = 'signup' in action and 'email' in action + search = 'search' in action or 'thomas.loc.gov' in action + return not(search or signup) + +def get_form(browser, predicate=None): + """wrapper for browser.get_form method to return Form ojects instead of ClientForm objects""" + f = browser.get_form(lambda f: predicate is None or predicate(Form(f))) + if f: return Form(f) -def get_img_and_form(response): +def get_src_and_form(url, imgs): """Returns the source of captcha img(if any) and form containing it. """ - if not response: return None, None - url = response.geturl() - response = response.read() - soup = BeautifulSoup(response) - imgs = soup.findAll('img', attrs={'src': re.compile('.*[Cc]aptcha.*')}) try: img = imgs[0] img_src = img.get('src', '') @@ -293,4 +279,4 @@ def get_img_and_form(response): print >> sys.stderr, details return None, None else: - return captcha_src, str(form) + return captcha_src, form utils/writerep.py @@ -23,9 +23,9 @@ def captcha_box(pol, img_src): size='10', pre=pre, description='Validation') def render_captcha(c): - return "<tr><td><label for='%s'>Verification</label> %s</td>\ - <td style='valign:center'><label for='%s'>&nbsp;</label> %s </td></tr>" \ - % (c.name, c.pre, c.name, c.render()) + return """<tr><td colspan=3> + <label for='%s'>Verification</label> %s %s + </td></tr>""" % (c.name, c.pre, c.render()) def add_captcha(form, img_src, pol): c = captcha_box(pol, img_src) utils/wyrapp.py @@ -1,6 +1,6 @@ import web from settings import db -from writerep import writerep +from writerep import writerep, compose_msg from wyrutils import pol2dist def test(formtype=None): @@ -36,9 +36,10 @@ def test(formtype=None): dist = pol2dist(pol) zip5, zip4 = getzip(dist) print zip5, zip4, + msg = compose_msg(pol, 'testing...') u = web.Storage(zip5=zip5, zip4=zip4, prefix='Mr.', state=dist[:2], fname='test', lname ='Tester', addr1='111 av', addr2='addr extn', city='test city', - phone='0010010010', email='test@tryitout.net', subject='general', full_msg='testing...') + phone='0010010010', email='test@tryitout.net', subject='general', full_msg=msg) msg_sent = writerep(pol, u) print msg_sent and 'Success' or 'Failure' utils/wyrtest.py @@ -3,17 +3,11 @@ from zip2rep import zip2dist from settings import db import sys -import urllib2, cookielib from ClientForm import ControlNotFoundError, AmbiguityError -from urlparse import urljoin - -from config import production_test, from_address, test_email -from settings import production_mode __all__ = ['ZipShared', 'ZipIncorrect', 'ZipNotFound', 'NoForm', 'WyrError', #all exceptions 'numdists', 'getdist', 'getcontact', 'getpols', 'has_captcha', 'dist2pols', - 'Form', 'urlopen', 'require_captcha', - 'production_mode', 'test_mode', 'test_email'] + 'Form', 'require_captcha'] class ZipShared(Exception): pass class ZipIncorrect(Exception): pass @@ -21,8 +15,6 @@ class ZipNotFound(Exception): pass class WyrError(Exception): pass class NoForm(Exception): pass -test_mode = (not production_mode) - name_options = dict(prefix=['pre', 'salut', 'title'], lname=['lname', 'last'], fname=['fname', 'first', 'name'], @@ -83,14 +75,6 @@ def dist2pols(dist): except KeyError: return [] -def urlopen(url, data=None, cj=None): - cj = cj or cookielib.CookieJar() - opener = urllib2.build_opener(urllib2.HTTPCookieProcessor(cj)) - try: - return opener.open(url, data) - except Exception, details: - print url, details - def first(seq): """returns first True element""" if not seq: return False @@ -106,15 +90,6 @@ def require_captcha(i, pols=None): have_captcha = any(has_captcha(p) for p in pols) return (not captchas_filled) and have_captcha -def check_response(form_controls, response): - """sends a mail to check if the form is submitted properly. - """ - form_values = "\n".join(["%s: %s" % (c.name, c.value) for c in form_controls]) - msg = 'Filled in the form:\n\n' + form_values - if response: msg += '\n\nResponse at' + response.geturl() + ':\n' + response.read() - subject = 'wyr mail' - web.sendmail(from_address, production_test, subject, msg) - def matches(a, b): """`a` matches `b` if any name_options of `a` is a part of `b` in lower case >>> matches('name', 'required-fname') @@ -136,6 +111,7 @@ def matches(a, b): class Form(object): def __init__(self, f): self.f = f + self.action = self.f.action self.controls = filter(lambda c: not (c.readonly or c.type == 'hidden') and c.name, f.controls) def __repr__(self): @@ -147,15 +123,6 @@ class Form(object): def __getattr__(self, x): return getattr(self.f, x) - def production_click(self): - if production_mode: - request = self.f.click() - response = urlopen(request.get_full_url(), request.get_data()) - check_response(self.controls, response) - elif test_mode: - check_response(self.controls, response='') - return True - def click(self): try: return self.f.click() @@ -232,7 +199,7 @@ class Form(object): try: names = name_options[name] - except KeyError: + except KeyError: names = name and [name] c = None if type: c = self.find_control_by_type(type) utils/wyrutils.py b853369bead404cef99a5d170fb0fd6200840f48 Devi asldevi@gmail.com http://github.com/aaronsw/watchdog/commit/23d8499324e5d2466cedce3445ba037f072fd13e 23d8499324e5d2466cedce3445ba037f072fd13e 2008-11-28T04:53:59-08:00 2008-11-28T04:53:59-08:00 maintain state across form fillings of a politician, increase the success rate 357702b595b6725b237e440f9ef505e531d19f03 Devi asldevi@gmail.com