Permalink
Cannot retrieve contributors at this time
Name already in use
A tag already exists with the provided branch name. Many Git commands accept both tag and branch names, so creating this branch may cause unexpected behavior. Are you sure you want to create this branch?
lsoft_api/lsoft_api.py
Go to fileThis commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
300 lines (266 sloc)
8.89 KB
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin/python3 | |
# Daniel Trinkle | |
# Purdue University | |
# 1 May 2024 | |
import configparser | |
from html.parser import HTMLParser | |
import os | |
import re | |
import sys | |
try: | |
from twill.commands import formvalue, timeout | |
from twill import browser, set_output | |
except: | |
print('''Requires twill module: | |
$ pip3 install --user twill | |
to install.''') | |
sys.exit(1) | |
# URL templates for LSoft operations | |
lsoft_urlfmt = { | |
'get': '%s/scripts/wa.exe?LCMD=REVIEW+%s%0.0s+NOH+MSG+ALL&p=0&X=%s&Y=%s', # server list ignored token account | |
'add': '%s/scripts/wa.exe?LCMD=QUIET+ADD+%s+%s+*&p=0&X=%s&Y=%s', # server list member token account | |
'del': '%s/scripts/wa.exe?LCMD=QUIET+DELETE+%s+%s&p=0&X=%s&Y=%s' # server list member token account | |
} | |
class LSOFTHTMLParser(HTMLParser): | |
''' | |
Parse returned LSoft HTML | |
Collect non-empty lines from <table class="status"> element | |
Provide data_clear() to reset, and data_get() method to get it | |
''' | |
def __init__(self): | |
HTMLParser.__init__(self) | |
self.status = False | |
self.data = [] | |
def handle_endtag(self, tag): | |
if tag == 'table': | |
self.status = False | |
def handle_starttag(self, tag, attrs): | |
if tag == 'table': | |
self.status = False | |
for a in attrs: | |
if a[0] == 'class' and a[1] == 'status': | |
self.status = True | |
def handle_data(self,data): | |
if self.status: | |
data = data.strip() | |
if data: | |
self.data.append(data) | |
def data_clear(self): | |
self.data = [] | |
def data_get(self): | |
return self.data | |
class DevNull(object): | |
''' | |
Divert twill output | |
''' | |
def write(self, str): | |
pass | |
def flush(self): | |
pass | |
def lsoft_login(cred=None): | |
''' | |
Use twill package to log in to LSoft web interface | |
Capture token from response URL and save in cred dict | |
''' | |
timeout(30) # LSoft is sometimes slow | |
# go to login dialog | |
try: | |
set_output(DevNull()) | |
browser.go('%s/scripts/wa.exe?LOGON=INDEX' % cred['serv']) | |
set_output() | |
except Exception as e: | |
set_output() | |
return 'FAIL', e | |
if browser.code != 200: | |
return 'FAIL', 'failed to get login page: %s' % browser.code | |
# enter login, passwd, click "Log In" button | |
try: | |
set_output(DevNull()) | |
formvalue('1', 'email', cred['acct']) # "email" is the id of the email field | |
formvalue('1', 'pw', cred['pass']) # "pw" is the id of the password field | |
browser.submit('e') # "e" is the name of the "Log In" button (from browser.show_forms()) | |
set_output() | |
except Exception as e: | |
set_output() | |
return 'FAIL', e | |
if browser.code != 200: | |
return 'FAIL', 'login failed: %s' % browser.code | |
if re.search(r'invalid password', browser.html, re.IGNORECASE): | |
return 'FAIL', 'login failed: invalid password' | |
# we get back a <head> element that contains a META element with a "refresh" URL (portion after server) | |
token = None | |
m = re.search(r'URL=([^"]+)">', browser.html) | |
if m: | |
url = cred['serv'] + m.group(1) | |
# visit the "refresh" URL | |
try: | |
set_output(DevNull()) | |
browser.go(url) | |
set_output() | |
except Exception as e: | |
set_output() | |
return 'FAIL', e | |
if browser.code == 200: | |
# if that succeeds, extract the token from the current URL | |
m = re.search(r'.*X=(\w+).*', browser.url) | |
if m: | |
token = m.group(1) | |
else: | |
return 'FAIL', 'did not get refresh token: %s' % browser.url | |
else: | |
return 'FAIL', 'login failed: %s' % browser.code | |
else: | |
return 'FAIL', 'unknown failure mode' | |
# save token, return OK | |
if ('token' not in cred) or (token != cred['token']): | |
cred['token'] = token | |
return 'OK', 'Logged in' | |
def lsoft_transact(cred=None, op=None, list=None, member=None): | |
''' | |
Process a transaction with LSoft via twill | |
Collect and update token from response URL | |
Return is a tuple (status, data) where status is 'OK' or | |
'FAIL', data is text output from the transaction (OK) or error | |
message (FAIL) | |
''' | |
if op not in lsoft_urlfmt: | |
return 'FAIL', 'ERROR: Unknown op: %s' % op | |
if ('token' not in cred) or not cred['token']: | |
# try to log in | |
(status, mesg) = lsoft_login(cred) | |
if status != 'OK': | |
return status, mesg | |
# divert output, issue operation, restore output | |
try: | |
set_output(DevNull()) | |
browser.go(lsoft_urlfmt[op] % (cred['serv'], list, member, cred['token'], cred['acct'])) | |
set_output() | |
except Exception as e: | |
set_output() | |
return 'FAIL', e | |
if browser.code != 200: | |
return 'FAIL', 'op failed: %s' % browser.code | |
# update token (so we don't time out connection) | |
m = re.search(r'.*X=(\w+).*', browser.url) | |
if m: | |
token = m.group(1) | |
if token != cred['token']: | |
cred['token'] = token | |
else: | |
print('ERROR: Cant find token in response URL') | |
# parse HTML response | |
if ('parse' not in cred) or not cred['parse']: | |
cred['parse'] = LSOFTHTMLParser() | |
cred['parse'].data_clear() | |
cred['parse'].feed(browser.html) | |
ldata = cred['parse'].data_get() | |
if re.search(r'list is unknown to', ' '.join(ldata)): | |
return 'FAIL', 'list is unknown' | |
elif re.search(r'ou are not the owner of the', ' '.join(ldata)): | |
return 'FAIL', 'not owner' | |
return 'OK', ldata | |
def addLSoftMember(name=None, addr=None, cred=None): | |
''' | |
Add a member to LSoft list | |
Return OK if member added, or already on the list | |
''' | |
(status, data) = lsoft_transact(cred=cred,op='add',list=name,member=addr) | |
if status == 'OK': | |
if re.search(r'has been added to the|is already subscribed to the', ' '.join(data)): | |
return status, data | |
else: | |
return 'FAIL', data | |
else: | |
return status, data | |
def delLSoftMember(name=None, addr=None, cred=None): | |
''' | |
Delete a member from LSoft list | |
Return OK if the member removed, or not on the list | |
''' | |
(status, data) = lsoft_transact(cred=cred,op='del',list=name,member=addr) | |
if status == 'OK': | |
if re.search(r'has been removed from the|is not subscribed to the', ' '.join(data)): | |
return status, data | |
else: | |
return 'FAIL', data | |
else: | |
return status, data | |
def getLSoftMembers(name=None, cred=None): | |
''' | |
Return an array of member addresses for LSoft list | |
''' | |
(status, data) = lsoft_transact(cred=cred,op='get',list=name,member=None) | |
if status != 'OK': | |
print('ERROR: %s' % data) | |
return status, data | |
members = [] | |
for l in data: | |
if not re.search(r'@', l): | |
continue | |
l = l.split()[0] | |
members.append(l.lower()) | |
return 'OK', members | |
def run_examples(short=False): | |
# get server, account info from ~/.lsoft.cfg: | |
# | |
# [lsoft] | |
# server = https://lists.purdue.edu | |
# account = lsoftuser@purdue.edu | |
# password = somethingveryeasytoguess:-) | |
# | |
config = configparser.ConfigParser() | |
config.read(os.path.join(os.environ['HOME'], 'etc/lsoft.cfg')) | |
# set up LSoft connection dict | |
cred = { | |
'serv': config.get('lsoft','server'), | |
'acct': config.get('lsoft','account'), | |
'pass': config.get('lsoft','password') | |
} | |
mylist = 'dept-listname' | |
myaddr = 'nobody@purdue.edu' | |
(status, members) = getLSoftMembers(name=mylist,cred=cred) | |
print(members) | |
if status != 'OK': | |
print('ERROR') | |
elif len(members): | |
print('Members of %s' % mylist) | |
for m in members: | |
print(m) | |
else: | |
print('%s is empty' % mylist) | |
if short: | |
return | |
(status, data) = addLSoftMember(name=mylist,addr=myaddr,cred=cred) | |
print(data) | |
if status == 'OK': | |
print('%s added to %s' % (myaddr, mylist)) | |
else: | |
print('%s not added' % myaddr) | |
(status, members) = getLSoftMembers(name=mylist,cred=cred) | |
print(members) | |
if status != 'OK': | |
print('ERROR') | |
elif len(members): | |
print('Members of %s' % mylist) | |
for m in members: | |
print(m) | |
else: | |
print('%s is empty' % mylist) | |
(status, data) = delLSoftMember(name=mylist,addr=myaddr,cred=cred) | |
print(data) | |
if status == 'OK': | |
print('%s removed from %s' % (myaddr, mylist)) | |
else: | |
print('%s not removed' % myaddr) | |
(status, members) = getLSoftMembers(name=mylist,cred=cred) | |
print(members) | |
if status != 'OK': | |
print('ERROR') | |
elif len(members): | |
print('Members of %s' % mylist) | |
for m in members: | |
print(m) | |
else: | |
print('%s is empty' % mylist) | |
if __name__ == '__main__': | |
run_examples(short=False) |