$ python -m SimpleHTTPServer
import optparse
import socket, select, threading
class Pipe(threading.Thread):
time_out = 5
buffer_size = 4096
max_idle = 5
def __init__(self, server_host, server_port, dclient_connxn):
super(Pipe, self).__init__()
self.server_host = server_host
self.server_port = server_port
self.server_addr = (server_host, server_port)
self.client_connxn = client_connxn
def connect_to_server(self):
self.server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
print(self.server_addr)
self.server_sock.connect(self.server_addr)
def log_request_response(self, req, resp):
print('request')
print(req)
print('response')
print(resp)
def run(self):
self.connect_to_server()
to_read = [self.client_connxn, self.server_sock]
to_write = []
to_watch_for_exception = to_read
request = ''
response = ''
seconds_idle = 0
while True:
(ready_to_read, ready_to_write, experienced_exception) = select.select(to_read, to_write, to_watch_for_exception, 1)
if (experienced_exception):
self.log_request_response(request, response)
self.client_connxn.close()
self.server_sock.close()
break
if (ready_to_read):
seconds_idle = 0
for waitable in ready_to_read:
data = waitable.recv(buffer_size)
if (waitable == self.client_connxn):
out = self.server_sock
request += data
else:
out = self.client_connxn
log_to = response
response += data
out.send(data)
else:
seconds_idle += 1
if (seconds_idle == max_idle):
self.log_request_response(request, response)
self.client_connxn.close()
self.server_sock.close()
break
class PortListener():'''
'''
def __init__(self, listen_on_port, pipe_to_host, pipe_to_port, max_connxns=1):
self.listen_on_port = listen_on_port
self.pipe_to_host = pipe_to_host
self.pipe_to_port = pipe_to_port
def listen():
listen_to_addr = ('127.0.0.1', self.listen_to_port)
listen_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
listen_socket.bind(listen_to_addr)
listen_socket.listen(max_connxns)
while True:
# pause until client requests connection
(client_connxn, client_addr) = listen_socket.accept()
pipe = Pipe(self.pipe_to_host, self.pipe_to_port, client_connxn)
pipe.start()
def setup_info_logger(log_file_path):
logging.basicConfig(filename=log_file_path, level=logging.INFO)
return logging.getLogger()
def parse_command_line():'''
parse command line argument to populate dictionary
return dictionary with keys = client_port, server_host, server_port, log_file_path
returns None if mandatory option missing
'''
parser = optparse.OptionParser()
parser.add_option('--clientport', dest='client_port', help='port client to this port on localhost')
parser.add_option('--serverhost', dest='server_host', help='server host, default = 127.0.0.1')
parser.add_option('--serverport', dest='server_port', help='server port')
parser.add_option('--logfilepath', dest='log_file_path', help='log file path')
(options, args) = parser.parse_args()
if (options.client_port == None):
print('--clientport is mandatory')
return None
if (options.server_port == None):
print('--serverport is mandatory')
return None
d = {}
if (options.log_file != None): d['log_file'] = options.log_file
else: d['log_file'] = None
d['client_port'] = options.client_port
if (options.server_host != None): d['server_host'] = options.server_host
else: d['server_host'] = '127.0.0.1'
d['server_port'] = options.server_port
return d
def main():
opt = parse_command_line()
logger = None
if (opt['log_file_path'] != None):
logger = setup_info_logger(opt['log_file_path'])
listener = PortListener(opt['client_port'], opt['server_host'], opt['server_port']]
if (__name__ == '__main__'):
main()
#!/bin/sh -
"exec" "python" "-O" "$0" "$@"
__doc__ = """Tiny HTTP Proxy.
This module implements GET, HEAD, POST, PUT and DELETE methods
on BaseHTTPServer, and behaves as an HTTP proxy. The CONNECT
method is also implemented experimentally, but has not been
tested yet.
Any help will be greatly appreciated. SUZUKI Hisao
"""
__version__ = "0.2.1"
import BaseHTTPServer, select, socket, SocketServer, urlparse
class ProxyHandler (BaseHTTPServer.BaseHTTPRequestHandler):
__base = BaseHTTPServer.BaseHTTPRequestHandler
__base_handle = __base.handle
server_version = "TinyHTTPProxy/" + __version__
rbufsize = 0 # self.rfile Be unbuffered
def handle(self):
(ip, port) = self.client_address
if hasattr(self, 'allowed_clients') and ip not in self.allowed_clients:
self.raw_requestline = self.rfile.readline()
if self.parse_request(): self.send_error(403)
else:
self.__base_handle()
def _connect_to(self, netloc, soc):
i = netloc.find(':')
if i >= 0:
host_port = netloc[:i], int(netloc[i+1:])
else:
host_port = netloc, 80
print "\t" "connect to %s:%d" % host_port
try: soc.connect(host_port)
except socket.error, arg:
try: msg = arg[1]
except: msg = arg
self.send_error(404, msg)
return 0
return 1
def do_CONNECT(self):
soc = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
if self._connect_to(self.path, soc):
self.log_request(200)
self.wfile.write(self.protocol_version +
" 200 Connection established\r\n")
self.wfile.write("Proxy-agent: %s\r\n" % self.version_string())
self.wfile.write("\r\n")
self._read_write(soc, 300)
finally:
print "\t" "bye"
soc.close()
self.connection.close()
def do_GET(self):
(scm, netloc, path, params, query, fragment) = urlparse.urlparse(
self.path, 'http')
if scm != 'http' or fragment or not netloc:
self.send_error(400, "bad url %s" % self.path)
return
soc = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
if self._connect_to(netloc, soc):
self.log_request()
soc.send("%s %s %s\r\n" % (
self.command,
urlparse.urlunparse(('', '', path, params, query, '')),
self.request_version))
self.headers['Connection'] = 'close'
del self.headers['Proxy-Connection']
for key_val in self.headers.items():
soc.send("%s: %s\r\n" % key_val)
soc.send("\r\n")
self._read_write(soc)
finally:
print "\t" "bye"
soc.close()
self.connection.close()
def _read_write(self, soc, max_idling=20):
iw = [self.connection, soc]
ow = []
count = 0
while 1:
count += 1
(ins, _, exs) = select.select(iw, ow, iw, 3)
if exs: break
if ins:
for i in ins:
if i is soc:
out = self.connection
else:
out = soc
data = i.recv(8192)
if data:
out.send(data)
count = 0
else:
print "\t" "idle", count
if count == max_idling: break
do_HEAD = do_GET
do_POST = do_GET
do_PUT = do_GET
do_DELETE=do_GET
class ThreadingHTTPServer (SocketServer.ThreadingMixIn,
BaseHTTPServer.HTTPServer): pass
if __name__ == '__main__':
from sys import argv
if argv[1:] and argv[1] in ('-h', '--help'):
print argv[0], "[port [allowed_client_name ...]]"
else:
if argv[2:]:
allowed = []
for name in argv[2:]:
client = socket.gethostbyname(name)
allowed.append(client)
print "Accept: %s (%s)" % (client, name)
ProxyHandler.allowed_clients = allowed
del argv[2:]
else:
print "Any clients will be served..."
BaseHTTPServer.test(ProxyHandler, ThreadingHTTPServer)
# --- --- --- --- --- --- --- --- --- --- --- ---
from optparse import OptionParser
# pyGTK --- --- --- --- --- --- --- --- ---
import pygtk
pygtk.require('2.0')
import gtk
# ElementTree - --- --- --- --- --- --- ---
import xml.etree.ElementTree as ET
# --- --- --- --- --- --- --- --- --- --- --- ---
import sys
xml_2_ns = {
'http://www.s3.org/XML/1998/namespace' : 'xml',
}
ns_2_xml = {
'xml' : 'http://www.s3.org/XML/1998/namespace',
}
def extend_tree(treestore, element, parent_node, ns_list):
'''
'''
#import pdb; pdb.set_trace()
node_label = '%s' % strip_ns_from_str(element.tag, ns_list)
if element.text != None:
element_text = element.text.strip()
if (len(element_text) > 0):
element_text = strip_ns_from_str(element_text, ns_list)
#discard = treestore.append(node, [element_text])
node_label = 'e %s : %s' % (node_label, element_text)
node = treestore.append(parent_node, [node_label])
# add attribute name/val pairs to node
for name, value in element.items():
name = strip_ns_from_str(name, ns_list)
value = strip_ns_from_str(value, ns_list)
label = 'a %s = %s' % (name, value)
treestore.append(node, [label])
# recurse over children
for child_element in element.getchildren():
child_node = extend_tree(treestore, child_element, node, ns_list)
return node
def etree_to_gtk_treestore(tree):
'''
construct gtk.TreeStore(str) from ElementTree
'''
treestore = gtk.TreeStore(str)
ns_list = ns_list_from_tree(tree)
element = tree.getroot()
node_label = '%s' % strip_ns_from_str(element.tag, ns_list)
if element.text != None:
element_text = element.text.strip()
if (len(element_text) > 0):
element_text = strip_ns_from_str(element_text, ns_list)
node_label = 'e %s : %s' % (node_label, element_text)
node = treestore.append(None, [node_label])
for child_e in element.getchildren():
extend_tree(treestore, child_e, node, ns_list)
# add attribute name/val pairs to node
for name, value in element.items():
name = strip_ns_from_str(name, ns_list)
value = strip_ns_from_str(value, ns_list)
label = '%s = %s' % (name, value)
treestore.append(None, [label])
# if ns is global
for ns in ns_list:
label = 'xmlns : %s' % ns
treestore.append(None, [label])
return treestore
def ns_from_string(s):
left = s.find('{')
right = s.find('}')
ns = s[left+1:right]
return ns
def process_potential_ns_string(s, ns_list):
if s == None:
return
if (s.find('{') != -1) and (s.find('}') != -1):
ns = ns_from_string(s)
if (ns not in ns_list):
ns_list.append(ns)
def ns_list_from_element(el, ns_list):
# tag
process_potential_ns_string(el.tag, ns_list)
# text
process_potential_ns_string(el.text, ns_list)
# attributes
for name, value in el.items():
process_potential_ns_string(name, ns_list)
process_potential_ns_string(value, ns_list)
for child_el in el.getchildren():
ns_list_from_element(child_el, ns_list)
return ns_list
def ns_list_from_tree(etree):
'''
'''
ns_list = []
root_e = etree.getroot()
# tag
process_potential_ns_string(root_e.tag, ns_list)
# text
process_potential_ns_string(root_e.text, ns_list)
# attributes
for name, value in root_e.items():
process_potential_ns_string(name, ns_list)
process_potential_ns_string(value, ns_list)
# child elements
for child_el in root_e.getchildren():
ns_list_from_element(child_el, ns_list)
return ns_list
def strip_ns_from_str(s, ns_list):
stripped = str(s)
for ns in ns_list:
token = '{%s}' % ns
loc = stripped.find(token)
if (loc != -1):
stripped = stripped[:loc] + stripped[loc + len(token):len(stripped)]
return stripped
class XMLTreeView:
'''
'''
def delete_event(widget, event, data=None):
'''
not an instance method, cuz gtk will not call as such
'''
gtk.main_quit()
return False
def __init__(self, path, title='pyxtree - python XML tree viewer', xsize=900, ysize = 500):
'''
'''
etree = ET.parse(path)
self.treestore = etree_to_gtk_treestore(etree)
self.window = gtk.Dialog()
self.window.connect("destroy", self.delete_event)
self.window.set_border_width(0)
self.swin_tree = gtk.ScrolledWindow()
self.swin_tree.set_border_width(10)
self.swin_tree.set_policy(gtk.POLICY_AUTOMATIC, gtk.POLICY_ALWAYS)
self.window.vbox.pack_start(self.swin_tree, True, True, 0)
self.swin_text = gtk.ScrolledWindow()
self.swin_text.set_border_width(10)
self.swin_text.set_policy(gtk.POLICY_AUTOMATIC, gtk.POLICY_ALWAYS)
self.window.vbox.pack_start(self.swin_text, True, True, 0)
self.window.set_title(title)
self.window.set_size_request(xsize, ysize)
self.window.connect("delete_event", self.delete_event)
# create the TreeView using treestore
self.treeview = gtk.TreeView(self.treestore)
self.tvcolumn = gtk.TreeViewColumn(path)
self.treeview.append_column(self.tvcolumn)
self.cell = gtk.CellRendererText()
# add the cell to the tvcolumn and allow it to expand
self.tvcolumn.pack_start(self.cell, True)
# set the cell "text" attribute to column 0 - retrieve text
# from that column in treestore
self.tvcolumn.add_attribute(self.cell, 'text', 0)
# make it searchable
self.treeview.set_search_column(0)
# Allow sorting on the column
#self.tvcolumn.set_sort_column_id(0)
# Allow drag and drop reordering of rows
#self.treeview.set_reorderable(True)
self.swin_tree.add_with_viewport(self.treeview)
self.swin_tree.show()
xfile = open(path, 'r')
xtext = xfile.read()
xfile.close()
label = gtk.Label(xtext)
# ns_list = ns_list_from_tree(etree)
# ns_str = ''
# for ns in ns_list:
# ns_str = ns_str + str(ns) + '\n'
# label = gtk.Label(ns_str)
label.set_alignment(xalign=0, yalign=0.5)
self.swin_text.add_with_viewport(label)
self.swin_text.show()
self.window.show_all()
def main(path = None):
'''
'''
if (path == None):
filechooser = gtk.FileChooserDialog(
title=None,
action=gtk.FILE_CHOOSER_ACTION_OPEN,
buttons=(gtk.STOCK_CANCEL,gtk.RESPONSE_CANCEL,gtk.STOCK_OPEN,gtk.RESPONSE_OK))
filter_stdxmlfiles = gtk.FileFilter()
filter_stdxmlfiles.set_name("Std XML Files")
filter_stdxmlfiles.add_pattern("*.xml")
filter_stdxmlfiles.add_pattern("*.wsdl")
filechooser.add_filter(filter_stdxmlfiles)
filter_allfiles = gtk.FileFilter()
filter_allfiles.set_name("All files")
filter_allfiles.add_pattern("*")
filechooser.add_filter(filter_allfiles)
response = filechooser.run()
if response == gtk.RESPONSE_OK:
path = filechooser.get_filename()
elif response == gtk.RESPONSE_CANCEL:
pass
#log failure
filechooser.destroy()
if (path != None):
tree_view = XMLTreeView(path)
gtk.main()
return
def src_file_from_cmd_args(args):
'''
'''
akapy = [
'pyxtree',
'pyxtree.py',
'python',
'python.exe',
'python2.5',
'python25.exe',
]
cleaned = []
for arg in args:
washed = arg.strip().lower()
if (washed not in akapy) and (washed.find('python') == -1):
cleaned.append(washed)
if len(cleaned) == 0:
cleaned = [None]
return cleaned
def configure_logging(LOG_FILENAME='pyxtree.log'):
import logging
logging.basicConfig(filename=LOG_FILENAME,level=logging.INFO)
log = logging.getLogger()
return log
def setup_command_line_options():
'''
using optparse
define command line arguments
return target filename, None if none specified
'''
parser = OptionParser()
parser.add_option('-f', dest='filename', help='target xml file to view')
(options, args) = parser.parse_args()
return(options.filename)
if __name__ == "__main__":
print('\nPyXTree - david.barkhuizen@gmail.com')
print('help: pyxtree -h\n')
xmlfilepath = setup_command_line_options()
log = configure_logging()
log.info('target file specified at command line = %s' % xmlfilepath)
main(path=xmlfilepath)
#!/usr/bin/env python
import sys
import traceback
def throws():
raise RuntimeError('error from throws')
def nested():
try:
throws()
except Exception, original_error:
try:
raise
finally:
try:
cleanup()
except:
pass # ignore errors in cleanup
def cleanup():
raise RuntimeError('error from cleanup')
def main():
try:
nested()
return 0
except Exception, err:
traceback.print_exc()
return 1
if __name__ == '__main__':
sys.exit(main())
# docpy.py - david.barkhuizn@gmail.com
# 1. module methods - signature, followed by docstring
# 2. module fields ?
# 3. module classes
# class name, parent class, __init__ method parameters & docstring
# instance fields (get from self.? refs in __init__ method)
# ---------------------------------------------------------------
# ISSUES
# need to strip out classes before extracting method info
# can't handle multi-line method signatures
# ---------------------------------------------------------------
import sys
# ---------------------------------------------------------------
single_marker = '\'\'\''
double_marker = '\"\"\"'
# ---------------------------------------------------------------
class MethodInfo:
def __init__(self, method_name, params, docstrings):
'''
method_name - string
params - list of param name strings
'''
self.method_name = method_name
self.params = params
self.docstrings = docstrings
# ---------------------------------------------------------------
def load_text_file(file_path):
'''
return list of lines in text file @ 'file_path'
'''
try:
text_file = open(file_path, 'r')
text_block = text_file.read()
lines = text_block.split('\n')
text_file.close()
return lines
except Exception, err:
print('error during reading of text file %s' % file_path)
print(err)
return []
def src_file_path_from_cmdargs():
valid_args = []
for arg in sys.argv:
if arg.lower() not in ['docpy.py', 'python']:
valid_args.append(arg.lower())
return 'docpy.py'
if len(valid_args) != 1:
return ''
else:
return valid_args[0]
def get_sig_line_idxs(lines):
'''
lines = list of lines in python source file (in sequence)
'''
token = 'def '
indices = []
for i in range(len(lines)):
if lines[i].strip()[:4] == token:
indices.append(i)
return indices
def parse_method_sig(sig):
# e.g.' def parse_methodsig(sig):#comment '
sig = sig.strip()
# e.g.'def parse_methodsig(sig):#comment'
sig = sig[4:len(sig)]
# e.g.'parse_methodsig(sig):#comment'
i = sig.find('#')
if i != -1:
sig = sig[:i-1]
# e.g.'parse_methodsig(sig):
left = sig.find('(')
right = sig.rfind(')')
meth_name = sig[:left]
params = []
if right - left != 1:
params = sig[left+1:right]
params = params.replace(',', ' ')
got_double = True
while got_double == True:
got_double = (params.find(' ') != -1)
if got_double:
params = params.replace(' ', ' ')
params = params.split(' ')
return MethodInfo(meth_name, params, [])
def extract_single_line_docstring(docstring):
'''example of single line method docstring'''
lsingle = docstring.find(single_marker)
ldouble = docstring.find(double_marker)
marker = ''
left = -1
if lsingle != -1:
marker = single_marker
left = lsingle
elif ldouble != -1:
marker = double_marker
left = ldouble
else:
return ''
right = docstring.rfind(marker)
if (right != -1) and (right != left):
return docstring[left+3:right]
else:
return ''
def strip_leading_triplequotes(line):
left = line.find(single_marker)
if left == -1:
left = line.find(double_marker)
if left == -1:
return ''
return line[left + 3:]
def strip_trailing_triplequotes(line):
right = line.rfind(double_marker)
if right == -1:
right = line.rfind(double_marker)
if right == -1:
return ''
return line[:right]
def extract_docstrings(lines, meth_sig_idx, next_meth_sig_idx):
# determine line indices of starting and ending triple quotes
start = -1
end = -1
for i in range(meth_sig_idx + 1, next_meth_sig_idx):
line = lines[i]
sidx = line.find(single_marker)
if sidx == -1:
sidx = line.find(double_marker)
if sidx != -1:
if start == -1:
start = i
else:
end = i
break
if start == -1:
return []
elif end == -1:
single_line = extract_single_line_docstring(lines[start])
if single_line != '':
return [ single_line ]
else:
return []
else:
docstrings = []
for i in range(start, end + 1):
no_whitespace = lines[i].strip()
if i == start:
stripped = strip_leading_triplequotes(no_whitespace)
if stripped != '':
docstrings.append(stripped)
elif i == end:
stripped = strip_trailing_triplequotes(no_whitespace)
if stripped != '':
docstrings.append(stripped)
else:
docstrings.append(no_whitespace)
return docstrings
def get_method_info(lines, meth_sig_idx, next_meth_sig_idx):
method_info = parse_method_sig(lines[meth_sig_idx])
method_info.docstrings = extract_docstrings(lines, meth_sig_idx, next_meth_sig_idx)
return method_info
def mock_method(one, two, three, four):
print('vokol')
def display_module_method_info(meth_info):
print(meth_info.method_name)
for p in meth_info.params:
print(' ' + p)
for docstring in meth_info.docstrings:
print(docstring)
def main():
text_file_path = src_file_path_from_cmdargs()
lines = load_text_file(text_file_path)
sig_line_idxs = get_sig_line_idxs(lines)
module_methods_info = []
for i in range(len(sig_line_idxs)):
idx = sig_line_idxs[i]
if i < len(sig_line_idxs) - 1:
next_idx = sig_line_idxs[i + 1]
else:
next_idx = len(sig_line_idxs)
module_methods_info.append( get_method_info(lines, idx, next_idx) )
for meth_info in module_methods_info:
display_module_method_info(meth_info)
if __name__ == '__main__':
main()
import httplib
httplib.HTTPConnection.debuglevel = 1
import urllib
def enc_quote(ticker, fromY, fromM, fromD, toY, toM, toD):
quote = dict()
quote['s'] = ticker
quote['a'] = fromM
quote['b'] = fromD
quote['c'] = fromY
quote['d'] = toD
quote['e'] = toM
quote['f'] = toY
quote['g'] = "d"
return urllib.urlencode(quote)
url_stem = 'http://ichart.yahoo.com/table.csv?'
quote_tokens = enc_quote('BP', '1900', '01', '01', '2010', '07', '18')
url = url_stem + "&ignore=.csv" + quote_tokens
f = urllib.urlopen(url)
body = f.read()
f.close()
try:
f = open('BP.csv', 'w')
f.write(body)
f.close()
except Exception, err:
print('ERROR: %s\n' % str(err))