#!/usr/bin/env python
##########################################################################
#
# Copyright (c) 2005 Imaginary Landscape LLC and Contributors.
#
# Permission is hereby granted, free of charge, to any person obtaining
# a copy of this software and associated documentation files (the
# "Software"), to deal in the Software without restriction, including
# without limitation the rights to use, copy, modify, merge, publish,
# distribute, sublicense, and/or sell copies of the Software, and to
# permit persons to whom the Software is furnished to do so, subject to
# the following conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
# LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
# OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
# WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
##########################################################################
"""
zopen.py
This script allows you to avoid several portions of the Zope web
management interface, instead interacting with Zope through a
command-line interface (and External Editor).
This script implements multiple commands. It should be installed as
simply 'z', like:
$ ln -s zopen.py ~/bin/z
Which creates the z. Alternately, you can link it multiple times,
like:
$ ln -s zopen.py ~/bin/zls
$ ls -s zopen.py ~/bin/zopen
And so on. Do 'z help' to get a list of commands.
Each command supports the -h option, which displays specific usage.
(Each is implemented in a top-level class by a similar name)
Everything is done with screen scraping (using standard modules), and
no software needs to be installed on the Zope server except for
External Editor.
"""
############################################################
## Imports
############################################################
try:
import optparse
except ImportError:
import optik as optparse
try:
import textwrap
except ImportError:
import simple_textwrap as textwrap
True, False = 1==1, 1==0
import os
import sys
import re
import urllib
import time
import urllib2
import base64
import stat
import gzip
try:
from cStringIO import StringIO
except ImportError:
from StringIO import StringIO
import urlparse
import difflib
import types
import getpass
import fnmatch
__version__ = '0.1'
############################################################
## Command infrastructure
############################################################
class Command:
"""
This is an abstract command. You initialize it with sys.argv,
then activate it with .run().
Subclasses set various class variables, and override run().
"""
# Set by subclasses:
names = None
"The filename of the command"
parser = None
"The optparse.OptionParser object"
max_args = 0
"Maximum number of positional arguments: -1 for no maximum"
min_args = 0
"Minimum number of positional arguments"
help = None
"The help text"
def __init__(self, args):
options, args = self.parse_args(args)
if len(args) > self.max_args and self.max_args != -1:
self.usage_error("Too many arguments (only %i allowed)"
% self.max_args)
if len(args) < self.min_args:
self.usage_error("Too few arguments (%i required)"
% self.min_args)
self.args = args
self.options = options
self.conf = PathConfig()
self.verbose = self.options.verbose
def parse_args(self, args):
try:
return self.parser.parse_args(args)
except SystemExit:
print
print format_docstring(self.help)
raise
def usage_error(self, msg):
print msg
print self.parser.usage
print
print format_docstring(self.help)
sys.exit(1)
def notify(self, msg):
sys.stdout.write(msg)
sys.stdout.flush()
def temp_filename(self):
return '/tmp/zopeedit-%s-%s.html' % (time.time(), os.environ['USER'])
def parse_dir(self, page):
lines = page.split('\n')
result = []
last = ''
while 1:
if not lines:
break
if not re.search(r'\s*$', lines[0], re.I):
lines.pop(0)
continue
lines.pop(0)
while not lines[0].strip():
lines.pop(0)
match = re.search(r'^\s*([a-z\.\_][^ \n\r]*)', lines[0], re.I)
if match:
if match.group(1).endswith(last):
last = ''
result.append((match.group(1), last))
lines.pop(0)
last = ' (Unknown)'
continue
match = re.search(r'([a-zA-Z_]+)\.gif" alt="([^"]+)"', lines[0])
if match:
if icon_types.has_key(match.group(1)):
last = icon_types[match.group(1)]
else:
last = ' (%s)' % match.group(2)
lines.pop(0)
return result
def get_url(self, url):
if self.verbose:
print "Getting %s" % self.conf.expand_path(url)
try:
config = self.conf.config(url)
if self.verbose:
print 'Config:'
self.print_config(config)
url = self.conf.expand_path(url)
headers = {'Authorization':
'Basic ' + base64.encodestring('%s:%s' % (config['user'], config['password']))}
req = urllib2.Request(url, headers=headers)
try:
f = urllib2.urlopen(req)
except Exception, e:
print url
raise
output = f.read()
f.close()
return output
except Exception, e:
#print "URL:", url
#print 'Config:'
#self.print_config(config)
raise
def print_config(self, config):
items = config.items()
items.sort()
for n, v in items:
if n == 'password':
v = 'xxxxx'
print ' %s: %s' % (n, v)
def external_edit(self, url):
url = self.conf.expand_path(url)
try:
output = self.get_url('%s/externalEdit_/%s'
% (url_dir(url),
url_base(url)))
return output
except urllib2.HTTPError, e:
if e.code in (404, 500):
print '%i File not found: %s' % (e.code, url)
sys.exit(1)
else:
raise
def format_exception(self, error):
"""
Reformats the HTML errors Zope produces as text.
"""
pos = error.find('')
if pos != -1:
error = error[pos+len(''):]
pos = error.find('')
if pos != -1:
error = error[:pos]
error = re.sub(r'\n', '', error)
error = re.sub(r'', '', error)
return error
class _CommandRegistry:
"""
This keeps track of all the Commands, and dispatches based on how
the script was called. Subclasses of Command are automatically
registered with this registry at the end of the module.
There is only a single instance of this class, and that instance
is `CommandRegistry`.
"""
def __init__(self):
self.commands = {}
self.classes = []
def add_class(self, cls):
for name in cls.names:
self.commands[name] = cls
self.classes.append(cls)
self.classes.sort(lambda a, b: cmp(a.names[0], b.names[0]))
def run(self, args=None):
if args is None:
args = sys.argv
name = os.path.basename(args[0])
if name.endswith('.py'):
name = name[:-3]
if name == 'z':
args = args[1:]
if not args:
name = 'help'
else:
name = args[0]
sys.argv[0] = 'z %s' % name
self.run_command(name, args[1:])
def run_command(self, name, args):
if name == '-h':
name = 'help'
try:
cmdClass = self.commands[name]
except KeyError:
raise
print "No such command: %s" % name
self.commands['help']([]).run()
sys.exit(1)
try:
cmd = cmdClass(args)
cmd.run()
except BadArgs, e:
print str(e)
print format_docstring(self.commands[name].help)
sys.exit(1)
CommandRegistry = _CommandRegistry()
class BadArgs(Exception):
"""
Exception to throw when the command-line syntax is incorrect;
there are a variety of restrictions that can't be natively
expressed with optparse, so you can throw this exception if you
encounter such an error.
"""
def __init__(self, *args):
self.args = args
def __str__(self):
if self.args:
return ' '.join(self.args)
else:
return 'Bad Arguments'
############################################################
## Configuration
############################################################
class PathConfig:
"""
This represents the hierarchical path-based configuration of the
system. Basically, you can store a configuration parameter for
any path, and it applies to all subpaths.
"""
def __init__(self):
self.read_config()
self.read_pwd()
def config(self, path):
"""
Return a dictionary of configuration parameters for the given
path.
"""
possible = []
trim_path = path
while 1:
if self.path_config.has_key(trim_path):
possible.append((trim_path, self.path_config[trim_path]))
if not trim_path:
break
pieces = trim_path.split('/')
trim_path = '/'.join(pieces[:-1])
possible.sort(lambda a, b: cmp(len(a[0]), len(b[0])))
config = {'user': os.environ['USER'],
'password': os.environ.get('ZOPEPASSWORD', '')}
for path, path_config in possible:
config.update(path_config)
return config
def set_config(self, path, values):
"""
Set a configuration parameters for a specific path. values
is a dictionary.
"""
config = self.path_config.setdefault(path, {})
config.update(values)
self.write_config()
def expand_path(self, path):
"""
Dereference bookmarks in a path, and join it to the current
location if the URL is relative.
"""
match = re.search(r'^([a-zA-Z0-1_\-]+):(.*)$', path)
if match and match.group(1) in ('http', 'https'):
# absolute path...
return path
elif match and match.group(1) not in ('http', 'https'):
bookmark = self.find_bookmark(match.group(1))
if bookmark is None:
raise BadArgs, "No bookmark by the name %r found" % match.group(1)
return (bookmark + '/' + match.group(2))
else:
cur = self.pwd
if not cur.endswith('/'):
cur += '/'
return urlparse.urljoin(cur, path)
def find_bookmark(self, name):
for path, config in self.path_config.items():
if config.get('bookmark') == name:
return path
return None
def all_bookmarks(self):
"""
Returns a dictionary of bookmark_name: url.
"""
bookmarks = {}
for path, config in self.path_config.items():
if config.get('bookmark'):
bookmarks[config['bookmark']] = path
return bookmarks
def read_config(self):
filename = self.config_filename('zopen.conf', ifempty='#empty\n')
self.path_config = {}
current_path = ''
file = open(filename)
for line in file.readlines():
line = line.strip()
if not line or line.startswith('#'):
continue
name, value = line.split('=', 1)
name = name.strip()
value = value.strip()
if name == 'path':
current_path = value
while current_path.endswith('/'):
current_path = current_path[:-1]
if name == 'password':
value = decrypt(value)
local_config = self.path_config.setdefault(current_path, {})
local_config[name] = value
file.close()
def write_config(self):
filename = self.config_filename('zopen.conf')
file = open(filename, 'w')
for path, config in self.path_config.items():
file.write('path=%s\n' % path)
for name, value in config.items():
if name == 'path':
continue
if name == 'password':
value = encrypt(value)
assert value.find('\n') == -1, "Newlines are not allowed in configuration values: %s=%r in %s" % (name, value, path)
file.write('%s=%s\n' % (name, value))
file.close()
def config_filename(self, name, ifempty=None):
filename = os.path.join(self.config_dir(), name)
if not os.path.exists(filename) and ifempty is not None:
f = open(filename, 'w')
f.write(ifempty)
f.close()
return filename
def config_dir(self):
base = os.path.join(os.environ['HOME'], '.zopen')
if not os.path.exists(base):
os.mkdir(base)
return base
def read_pwd(self):
filename = self.config_filename('pwd.txt', ifempty='')
if not os.path.exists(filename):
self.pwd = ''
else:
file = open(filename)
self.pwd = file.read().strip()
file.close()
def set_pwd(self, path):
path = self.expand_path(path)
filename = self.config_filename('pwd.txt')
file = open(filename, 'w')
file.write(path)
file.close()
self.pwd = path
def save_pwd(self):
filename = self.config_filename('last_pwd.txt')
file = open(filename, 'w')
file.write(self.pwd)
file.close()
def last_pwd(self):
filename = self.config_filename('last_pwd.txt')
if not os.path.exists(filename):
return None
file = open(filename)
c = file.read().strip()
file.close()
return c
############################################################
## Commands
############################################################
def new_parser(usage):
parser = optparse.OptionParser(usage=usage,
version='%%prog %s' % __version__)
parser.add_option('-v', '--verbose',
help="Verbose mode",
dest='verbose',
action='store_true')
return parser
class cd_Command(Command):
names = ['cd', 'zcd']
usage = 'usage: %prog [options] PATH'
parser = new_parser(usage=usage)
max_args = 1
min_args = 1
help = """
Change Zope directory/context. Use a PATH of - to return to your
last location.
"""
def run(self):
new_path = self.args[0]
if new_path == '-':
new_path = self.conf.last_pwd()
if new_path is None:
raise BadArgs, "No previous directory"
self.conf.save_pwd()
self.conf.set_pwd(new_path)
class pwd_Command(Command):
names = ['pwd', 'zpwd']
usage = 'usage: %prog [-h]'
parser = new_parser(usage=usage)
help = """
Print the present Zope directory/context
"""
def run(self):
self.conf.read_pwd()
if not self.conf.pwd:
print >> sys.stderr, "No working directory"
else:
print self.conf.pwd
class open_Command(Command):
names = ['open', 'zopen']
usage = 'usage: %prog [-pstd] location'
max_args = 1
min_args = 1
help = """
Opens the location. The options create the resource before opening
it.
"""
parser = new_parser(usage=usage)
parser.add_option('-p', '--python',
help='Create Python Script',
dest='create',
action='store_const',
const='python')
parser.add_option('-s', '--sql',
help='Create Z SQL Method',
dest='create',
action='store_const',
const='sql')
parser.add_option('-t', '--template',
help='Create Page Template',
dest='create',
action='store_const',
const='template')
parser.add_option('-f', '--folder',
help='Create Folder (does not "open" folder)',
dest='create',
action='store_const',
const='folder')
parser.add_option('--dtml',
help="Create DTML Document",
dest='create',
action='store_const',
const='dtml')
parser.add_option('--method',
help="Create DTML Method",
dest='create',
action='store_const',
const='method')
parser.add_option('--title',
help="Set new object title",
dest='title',
action='store',
metavar='TITLE')
def run(self):
location = self.conf.expand_path(self.args[0])
if self.options.create is not None:
can_edit = self.create(self.options.create,
self.options.title,
location)
if not can_edit:
return
filename = self.temp_filename()
output = self.external_edit(location)
file = open(filename, 'w')
file.write(output)
file.close()
os.system('zopeedit.py %s &' % filename)
_type_data = {
'python':
{'title': 'Python Script',
'url': '%(addProduct)s/PythonScripts/manage_addPythonScript?'
'id=%(id)s'},
'sql':
{'title': 'Z SQL Method',
'url': '%(addProduct)s/ZSQLMethods/manage_addZSQLMethod?'
'title=%(title)s&connection_id=%(db)s&arguments=&'
'template=&id=%(id)s',
'required_options': ['db']},
'template':
{'title': 'Page Template',
'url': '%(addProduct)s/PageTemplates/manage_addPageTemplate'
'?submit=%%20Add%%20&id=%(id)s'},
'folder':
{'title': 'Folder',
'url': '%(addProduct)s/OFSP/manage_addFolder?'
'submit=%%20Add%%20&id=%(id)s&title=%(title)s',
'can_edit': False},
'dtml':
{'title': 'DTML Document',
'url': '%(addProduct)s/OFSP/addDTMLDocument?'
'submit=%%20Add%%20&id=%(id)s&title=%(title)s'},
'method':
{'title': 'DTML Method',
'url': '%(addProduct)s/OFSP/addDTMLMethod?'
'submit=%%20Add%%20&id=%(id)s&title=%(title)s'},
}
def create(self, create_type, title, url):
data = self._type_data[create_type]
self.notify('Creating %s...' % data['title'])
args = {
'url_dir': url_dir(url),
'addProduct': url_dir(url) + '/manage_addProduct',
'id': urllib.quote(url_base(url)),
'title': urllib.quote(title or ''),
}
config = self.conf.config(url)
for option in data.get('required_options', []):
if not config.get(option):
raise BadArgs, "No %s defined for %s" % (option, url)
args[option] = urllib.quote(config[option])
req = data['url'] % args
try:
output = self.get_url(req)
except urllib2.HTTPError, e:
if e.code == 404:
print '404 Folder does not exist: %s' % url_dir(url)
sys.exit(1)
elif e.code == 400:
print '400 Object already exists: %s' % url
sys.exit(1)
else:
raise
self.notify('done.\n')
return data.get('can_edit', True)
class ls_Command(Command):
names = ['ls', 'zls', 'dir']
usage = 'usage: %prog [URL]'
max_args = 1
min_args = 0
help = """
Lists the files/object in URL (or pwd). URL can be a simple pattern.
"""
parser = new_parser(usage=usage)
parser.add_option('-e', '--no-extensions',
dest='no_extensions',
action='store_true',
default=False,
help="Do not display artificial extensions")
def run(self):
if self.args:
location = self.args[0]
else:
location = self.conf.pwd
if '*' in location:
pattern = location.split('/')[-1]
location = '/'.join(location.split('/')[:-1])
else:
pattern = None
if not location:
location = self.conf.pwd
try:
output = self.get_url(location + '/manage_main')
except urllib2.HTTPError, e:
if e.code == 404:
print '404 Folder does not exist: %s' % url_dir(location)
sys.exit(1)
raise
for name, ext in self.parse_dir(output):
if self.options.no_extensions:
full = name
else:
full = name + ext
if pattern and not fnmatch.fnmatchcase(full, pattern):
continue
print full
class grep_Command(Command):
names = ['grep']
usage = 'usage: %prog [-n] [-u URL] TEXT'
max_args = 1
min_args = 1
help = """
Finds objects that contain TEXT, looking in DIR or the current
location (if -d not provided).
"""
parser = new_parser(usage=usage)
parser.add_option('-u', '--url',
help="Search in URL",
metavar="URL",
action="store",
dest="url",
default=None)
def run(self):
text = self.args[0]
if self.options.url is not None:
url = self.options.url
else:
url = self.conf.pwd
output = self.get_url(
url + '/manage_findResult?searchtype=simple&obj_metatypes:list=all&search_sub:int=1&btn_submit=Find&batch_size=1000&obj_searchterm=%s'
% urllib.quote(text))
for name, ext in parse_dir(output):
print name + ext
class find_Command(Command):
names = ['find', 'zfind']
usage = 'usage: %prog [-u URL] [-i] TEXT'
max_args = 1
min_args = 1
help = """
Finds objects whose name contains TEXT, looking in URL or the
current location (if -u not provided).
"""
parser = new_parser(usage=usage)
parser.add_option('-u', '--url',
help="Search in URL",
metavar="URL",
action="store",
dest="url",
default=None)
parser.add_option('-i', '--ignore-case',
help="Ignore case",
action="store_true",
dest="ignore_case")
def run(self):
text = self.args[0]
if self.options.url is not None:
url = self.options.url
else:
url = self.conf.pwd
if self.options.ignore_case:
query = 'id.lower().find(%r)!=-1' % text.lower()
else:
query = 'id.find(%r)!=-1' % text
output = self.get_url(
url + '/manage_findResult?searchtype=simple&obj_metatypes:list=all&search_sub:int=1&btn_submit=Find&batch_size=1000&obj_expr=%s'
% urllib.quote(query))
for name, ext in parse_dir(output):
print name + ext
class sqlrefresh_Command(Command):
names = ['sqlrefresh', 'zsqlrefresh', 'zrefresh', 'refresh']
usage = 'usage: %prog URL [URL...]'
max_args = -1
min_args = 1
help = """
Refreshes SQL brains (the class that is mixed into Z SQL rows) for
the given URL. (Changes to files for SQL brains are not otherwise
detected.)
"""
parser = new_parser(usage=usage)
def run(self):
for arg in self.args:
sys.stdout.write('Refreshing ' + arg + '...')
sys.stdout.flush()
self.refresh_sql(arg)
sys.stdout.write('done.\n')
def refresh_sql(self, url):
output = self.get_url(url + '/manage_advancedForm')
try:
output = output[re.search(r'', output).start()]
args = []
while 1:
match = re.search(r']*value="([^"]*)">', output)
if not match:
break
# @@: I should un-html-quote it too...
args.append('%s=%s' %
(match.group(1), urllib.quote(match.group(2))))
output = output[match.end():]
output = self.get_url(
url
+ '/manage_advanced?submit=Save+Changes&'
+ '&'.join(args))
class cp_Command(Command):
names = ['cp', 'zcp', 'copy']
usage = 'usage: %prog SOURCE DEST'
max_args = 2
min_args = 2
help = """
Copies SOURCE to DEST, exporting SOURCE then importing it to DEST.
You must set the import_path option for the DEST.
"""
parser = new_parser(usage=usage)
parser.add_option('-e', '--external',
help="Copy a single file with ExternalEditor (destination must already exist)",
action="store_true",
dest="external")
def run(self):
source, dest = self.args
if self.options.external:
return self.run_external(source, dest)
config = self.conf.config(dest)
if not config.has_key('import_path'):
raise BadArgs, "You must provide an import_path option for the destination (%s, %s)" % (dest, config)
import_path = config['import_path']
source = self.conf.expand_path(source)
# @@: broken on Windows:
url_dir = os.path.dirname(source)
url_id = os.path.basename(source)
sys.stdout.write('Downloading...')
sys.stdout.flush()
output = self.get_url(
url_dir
+ '/manage_exportObject?id=%s&download:int=1&submit=Export'
% url_id)
filename = self.temp_filename()
base_filename = os.path.basename(filename)
dest_filename = os.path.join(import_path, base_filename)
f = open(filename, 'wb')
f.write(output)
f.close()
sys.stdout.write('done.\n')
sys.stdout.write('Uploading...')
sys.stdout.flush()
assert import_path.find("'") == -1, "You cannot have ' in an import_path (%r)" % import_path
result = os.system("scp '%s' '%s'" % (filename, dest_filename))
if result:
print "\nscp failed with code %i" % result
os.unlink(filename)
sys.exit(1)
sys.stdout.write('done.')
sys.stdout.write('Importing...')
sys.stdout.flush()
output = self.get_url(
dest + '/manage_importObject?file=%s&set_owner:int=1&submit=Import'
% filename)
if dest_filename.find(':') != -1:
dest_host, dest_path = dest_filename.split(':', 1)
os.system("ssh %s 'rm %s'"
% (dest_host, dest_path))
else:
os.unlink(dest_filename)
sys.stdout.write('done.\n')
def run_external(self, source, dest):
source_text = self.external_edit(source)
dest_text = self.external_edit(source)
# @@: Ack, I'm not sure what to do next; some sort of PUT.
raise BadArgs, "external editor copying is not yet supported"
all_options = [
('user', 'Zope username'),
('password', 'Zope password'),
('db', 'Connection name for new Z SQL Methods'),
('import_path', 'Path to the import/ directory; may use SSH syntax'),
]
class setopt_Command(Command):
names = ['setopt', 'zsetopt', 'set']
usage = 'usage: %prog [-u URL] [-g] [NAME=VALUE...]'
max_args = -1
min_args = 0
help = """\
Sets options for the current directory. Currently available options
are:
%s
If no URL is given, the current location is used. Without any options,
this will simply display the current options.
""" % '\n '.join([('%%-%is: %%s'
% (max([len(a) for (a, b) in all_options])))
% op_desc for op_desc in all_options])
parser = new_parser(usage=usage)
parser.add_option('-u', '--url',
help="Set for URL",
metavar="URL",
action="store",
dest="url",
default=None)
parser.add_option('-g', '--global',
help="Set a global option",
action="store_true",
dest="global_url")
def run(self):
if self.options.url is not None:
url = self.options.url
elif self.options.global_url:
url = ''
else:
url = self.conf.pwd
opts = {}
all_options_dict = dict(all_options)
for arg in self.args:
if not '=' in arg:
raise BadArgs, "Not in format NAME=VALUE: %r" % arg
name, value = arg.split('=', 1)
if not all_options_dict.has_key(name):
raise BadArgs, "Not a known option name: %r (not one of: %s)" % (name, ', '.join(all_options_dict.keys()))
opts[name] = value
self.conf.set_config(url, opts)
print "Current config for %s:" % url
self.print_config(self.conf.config(url))
class bookmark_Command(Command):
names = ['bookmark', 'zbookmark', 'book']
usage = 'usage: %prog [-l] [-u URL] [BOOKMARK]'
max_args = 1
min_args = 0
help = """
Display and set bookmarks. Bookmarks are used like 'bookmark:'
in a URL.
"""
parser = new_parser(usage=usage)
parser.add_option('-l', '--list',
help="Display all bookmarks",
action="store_true",
dest="list")
parser.add_option('-u', '--url',
help="Set bookmark for this URL (instead of current location",
metavar="URL",
action="store",
dest="url",
default=None)
def run(self):
if self.options.list:
if self.args:
raise BadArgs, "You cannot both provide a BOOKMARK and the list option"
elif not self.args:
raise BadArgs, "You must provide a BOOKMARK"
if self.options.list:
self.run_list()
else:
if self.options.url is not None:
url = self.options.url
else:
url = self.conf.pwd
url = self.conf.expand_path(url)
bookmark = self.args[0].replace(':', '')
self.add_bookmark(url, bookmark)
def run_list(self):
bookmarks = self.conf.all_bookmarks().items()
bookmarks.sort()
if not bookmarks:
print "No bookmarks"
for name, url in bookmarks:
print "%-12s: %s" % (name, url)
def add_bookmark(self, url, name):
self.conf.set_config(
url,
{'bookmark': name})
class diff_Command(Command):
names = ['diff', 'zdiff']
usage = 'usage: %prog OBJ1 OBJ2'
max_args = 2
min_args = 2
help = """
Show the differences between OBJ1 and OBJ2.
"""
parser = new_parser(usage=usage)
parser.add_option('-u', '--unified',
help="Display in Unified Diff format",
action="store_true",
dest="unified")
parser.add_option('-n', '--ndiff',
help="Display in ndiff format",
action="store_true",
dest="ndiff")
_ignore_lines = [
re.compile(r'url:.*\n'),
re.compile(r'auth:.*\n'),
re.compile(r'lock-token:.*\n'),
]
def run(self):
source, dest = self.args
source_text = self.strip_ignore(self.external_edit(source))
dest_text = self.strip_ignore(self.external_edit(dest))
if self.options.unified:
func = difflib.unified_diff
elif self.options.ndiff:
func = self.ndiff
else:
func = difflib.context_diff
diff = func(
source_text.splitlines(), dest_text.splitlines(),
source, dest)
for line in diff:
if line.endswith('\n'):
line = line[:-1]
print line
def strip_ignore(self, text):
for _ignore_re in self._ignore_lines:
text = _ignore_re.sub('', text)
return text
def ndiff(self, source, dest, source_filename, dest_filename):
return difflib.ndiff(source, dest)
class help_Command(Command):
names = ['help']
usage = 'usage: %prog [COMMAND]'
max_args = 1
min_args = 0
help = """
Gives help on this program, or for a specific command.
"""
parser = new_parser(usage=usage)
def run(self):
if self.args:
self.print_help(self.args[0])
return
print "Commands:"
for cls in CommandRegistry.classes:
if cls == self.__class__:
continue
name = cls.names[0]
usage = cls.usage.replace('%prog', name)
if usage.startswith('usage:'):
usage = usage[len('usage:'):].strip()
print 'z', usage
print indent(format_docstring(cls.help), 4)
print
print "Use z help COMMAND for more help, or z COMMAND -h"
def print_help(self, name):
command = CommandRegistry.commands[name]
sys.argv[0] = 'z %s' % name
command(['-h'])
class password_Command(Command):
names = ['passwd', 'zpasswd', 'password']
usage = 'usage: %prog [-u URL] [-g]'
max_args = 0
min_args = 0
help = """
Change the password
"""
parser = new_parser(usage=usage)
parser.add_option('-u', '--url',
help="Set password for given URL",
action="store",
dest="url",
default=None)
parser.add_option('-g', '--global',
help="Set global/default password",
action="store_true",
dest="global_url")
def run(self):
if self.options.url is not None:
url = self.options.url
elif self.options.global_url:
url = ''
else:
url = self.conf.pwd
password = getpass.getpass()
self.conf.set_config(url, {'password': password})
class cat_Command(Command):
names = ['cat', 'zcat', 'print']
usage = 'usage: %prog URL'
max_args = 1
min_args = 1
help = """
Print the resource.
"""
parser = new_parser(usage=usage)
def run(self):
url = self.args[0]
text = self.external_edit(url)
# Get rid of the ExternalEditor headers:
lines = text.splitlines()
while lines[0]:
lines.pop(0)
lines.pop(0)
print '\n'.join(lines)
class runsql_Command(Command):
names = ['runsql', 'zrunsql', 'sqlrun']
usage = 'usage: %prog [-u URL] [SQLSTATEMENTS...]'
max_args = -1
min_args = 1
help = """
Execute the SQL with the database connection that is configured
for the current location (or URL).
"""
parser = new_parser(usage=usage)
parser.add_option('-u', '--url',
metavar='URL',
dest='url',
action='store',
default=None)
def run(self):
url = self.options.url or self.conf.pwd
config = self.conf.config(url)
if not config.has_key('db'):
raise BadArgs, "The URL %s has no database configured (used z setopt db=dbname)" % url
db = config['db']
query = ' '.join(self.args)
try:
result = self.get_url(
'%s/%s/manage_test?query=%s&submit=Submit%%20Query'
% (url, db, urllib.quote(query)))
except urllib2.HTTPError, e:
error = e.read()
print self.format_exception(error)
else:
self.print_table(result, sys.stdout)
def text_repr(self, text):
if text == 'None':
return 'NULL'
elif '\n' in text:
return repr(text)[1:-1]
else:
return text
def print_table(self, page, out):
page = self.tag_contents(page, '')
if not page:
out.write('No results\n')
return
trs = self.tag_contents(page[0], '', '
')
headers = self.tag_contents(trs[0], '', ' | ')
body = []
longest = {}
for row in trs[1:]:
body.append({})
for header, value in zip(
headers, self.tag_contents(row, '', ' | ')):
body[-1][header] = value
if len(self.text_repr(value)) > longest.get(header, 0):
longest[header] = len(self.text_repr(value))
if len(header) > longest.get(header, 0):
longest[header] = len(header)
if sum(longest.values()) > 60:
vertical = True
else:
vertical = False
if not vertical:
for header in headers:
out.write(header)
out.write(' '*(longest[header]-len(header)+1))
out.write('\n')
for header in headers:
out.write('='*len(header))
out.write(' '*(longest[header]-len(header)+1))
out.write('\n')
for row in body:
for header in headers:
out.write(self.text_repr(row[header]))
out.write(' '*(longest[header]-len(self.text_repr(row[header]))+1))
out.write('\n')
else:
longest_header = max(map(len, headers))
for row in body:
for header in headers:
out.write(header + ':')
out.write(' '*(longest_header-len(header)+1))
out.write(self.text_repr(row[header]))
out.write('\n')
out.write('\n')
def tag_contents(self, page, start, end):
results = []
while 1:
pos = page.find(start)
if pos == -1:
break
page = page[pos+len(start):]
pos = page.find(end)
if pos != -1:
text = page[:pos]
page = page[pos:]
else:
text = page
page = ''
results.append(text.strip())
return results
############################################################
## Utility functions
############################################################
def encrypt(password):
"""
Not really encryption, because there's no key. Just a bunch of
obfuscation, so someone can't casually read your password off the
disk.
"""
password = ' '*6 + password + ' '*6
password = base64.encodestring(password)
out = StringIO()
f = gzip.GzipFile('x', 'wb', 9, out)
f.write(password)
f.close()
password = out.getvalue()
password = password[10:] + password[:10]
password = password.encode('base64').replace('\n', '')
return password
def decrypt(password):
"""
Fake encrypt/decryption; goes with encrypt().
"""
password = password.decode('base64')
password = password[-10:] + password[:-10]
filein = StringIO(password)
f = gzip.GzipFile('x', 'rb', 9, filein)
password = f.read()
f.close()
password = base64.decodestring(password)
assert password.endswith(' '*6) and password.startswith(' '*6)
password = password[6:-6]
return password
icon_types = {
'pyscript': '.py',
'sqlmethod': '.sql',
'Folder_icon': '/',
'dtmlmethod': '.dtml',
'dtmldoc': '.dtml',
'File_icon': '.txt',
'zpt': '',
}
def parse_dir(page):
"""
Parses the output of the Zope management screen (manage_main), and
returns a list of (object_name, object_type).
"""
lines = page.split('\n')
result = []
last = ''
while 1:
if not lines:
break
if not re.search(r'\s*$', lines[0], re.I):
lines.pop(0)
continue
lines.pop(0)
while not lines[0].strip():
lines.pop(0)
match = re.search(r'^\s*([a-z\.\_][^ \n\r]*)', lines[0], re.I)
if match:
if match.group(1).endswith(last):
last = ''
result.append((match.group(1), last))
lines.pop(0)
last = ' (Unknown)'
continue
match = re.search(r'([a-zA-Z_]+)\.gif" alt="([^"]+)"', lines[0])
if match:
if icon_types.has_key(match.group(1)):
last = icon_types[match.group(1)]
else:
last = ' (%s)' % match.group(2)
lines.pop(0)
return result
def format_docstring(s):
if s.startswith('\n'):
s = s[1:]
lines = s.split('\n')
indent = 0
for indent in range(len(lines[0])):
if lines[0][indent] != ' ':
break
newlines = []
for line in lines:
if line.startswith(' '*indent):
line = line[indent:]
newlines.append(line)
return '\n'.join(newlines)
def indent(s, indent):
return '\n'.join([' '*indent+line for line in s.splitlines()])
def url_base(url):
return url.split('/')[-1]
def url_dir(url):
return '/'.join(url.split('/')[:-1])
############################################################
## Main runner
############################################################
# Registery all the classes:
for name, value in globals().items():
if ((isinstance(value, type)
or type(value) is types.ClassType)
and issubclass(value, Command)
and value != Command):
CommandRegistry.add_class(value)
if __name__ == '__main__':
CommandRegistry.run()