external/iiab
1
0
Fork 0
mirror of https://github.com/iiab/iiab.git synced 2025-03-09 15:40:17 +00:00
Code Issues Releases Wiki Activity
iiab/roles/activity-server/files/xs_activities/__init__.py
George Hunt 845632d0ac initial checkin -- May 27, 2017
2017-05-27 11:09:50 -07:00

535 lines
18 KiB
Python
Raw Blame History

# Copyright (C) 2008 One Laptop Per Child Association, Inc.
# Licensed under the terms of the GNU GPL v2 or later; see COPYING for details.
#
# written by Douglas Bagnall <douglas@paradise.net.nz>
"""Functions for processing XO activities, either for indexing and
presentaton to the laptops, or for diagnostics.
"""
import os, sys, shutil
import zipfile
import re
from cStringIO import StringIO
#import traceback
import syslog
from ConfigParser import SafeConfigParser
# we no longer really have a default in that it is set in the conf file
# we assume that we have a lang_template for the default language
TEMPLATE_DIR = '/library/xs-activity-server/lang_templates'
DEFAULT_LANG = 'en'
# how many versions before the latest are worth having around.
KEEP_OLD_VERSIONS = 3
#print to stderr, rathe than syslog?
USE_STDERR=False
REQUIRED_TAGS = ('bundle_id', 'activity_version', 'host_version', 'name', 'license')
OPTIONAL_TAGS = ('show_launcher', 'exec', 'mime_types', 'icon')
#XXX need either icon or show_launcher=no
def log(msg, level=syslog.LOG_NOTICE):
if USE_STDERR:
print >> sys.stderr, msg
else:
syslog.openlog( 'xs-activity-server', 0, syslog.LOG_USER )
syslog.syslog(level, msg)
syslog.closelog()
class BundleError(Exception):
pass
class Bundle(object):
def __init__(self, bundle):
self.linfo = {}
self.zf = zipfile.ZipFile(bundle)
# The activity path will be 'Something.activity/activity/activity.info'
for p in self.zf.namelist():
if p.endswith(self.INFO_PATH):
self.raw_data = read_info_file(self.zf, p, self.INFO_SECTION)
# the file name itself is needed for the URL
self.url = os.path.basename(bundle)
self.mtime = os.stat(bundle).st_mtime
self.name = self.raw_data.get('name')
self.license = self.raw_data.get('license', None)
# child ctor should now call
# _set_bundle_id
# _set_version
# _set_description
def _set_bundle_id(self, id):
if id is None:
raise BundleError("bad bundle: No bundle ID")
self.bundle_id = id
if self.name is None:
self.name = id
def _set_version(self, version):
self.version = version
def _set_description(self, description):
self.description = description
def __cmp__(self, other):
"""Alphabetical sort (locale dependant of course)"""
if self.bundle_id == other.bundle_id:
return cmp(self.version, other.version)
return cmp(self.name, other.name)
def set_older_versions(self, versions):
"""Versions should be a list of (version number, version tuples)"""
self.older_versions = ', '.join('<a href="%s">%s</a>' % (v.url, v.version) for v in versions)
def to_html(self, locale, template=None):
"""Fill in the template with data approriate for the locale."""
if template is None:
template = read_template('activity', locale)
d = {'older_versions': self.older_versions,
'bundle_id': self.bundle_id,
'activity_version': self.version,
'bundle_url': self.url,
'name': self.name,
'description': self.description,
}
d.update(self.linfo.get(locale, {}))
if d['older_versions']:
d['show_older_versions'] = 'inline'
else:
d['show_older_versions'] = 'none'
return template % d
def get_name(self, locale=None):
return self.name
class Content(Bundle):
INFO_PATH = "library/library.info"
INFO_SECTION = "Library"
def __init__(self, bundle):
super(Content, self).__init__(bundle)
d = self.raw_data
# bundle_id is often missing; service name is used instead.
self._set_bundle_id(d.get('global_name', None))
self._set_version(d.get('library_version', 1))
self._set_description(d.get('long_name', ''))
def debug(self, force_recheck=False):
# FIXME: implement debug checking for content bundles
return {}
class Activity(Bundle):
INFO_PATH = "activity/activity.info"
INFO_SECTION = "Activity"
#Activities appear to be looser than RFC3066, using e.g. _ in place of -.
linfo_re = re.compile(r'/locale/([A-Za-z]+[\w-]*)/activity.linfo$')
def __init__(self, bundle):
"""Takes a zipped .xo bundle name, returns a dictionary of its
activity info. Can raise a variety of exceptions, all of
which should indicate the bundle is invalid."""
super(Activity, self).__init__(bundle)
# The locale info will be Something.activity/locale/xx_XX/activity.linfo
for p in self.zf.namelist():
linfo = self.linfo_re.search(p)
if linfo:
lang = canonicalise(linfo.group(1))
self.linfo[lang] = read_info_file(self.zf, p, self.INFO_SECTION)
# Unfortunately the dictionary lacks some information, and
# stores other bits in inconsistent ways.
d = self.raw_data
# bundle_id is often missing; service name is used instead.
self._set_bundle_id(d.get('bundle_id', d.get('service_name')))
self._set_version(d.get('activity_version', 1))
self._set_description(d.get('description', ''))
def debug(self, force_recheck=False):
"""Make a copy of the raw data with added bits so we can work
out what is going on. This is useful for diagnosing problems
with odd activities and composing tut-tut-ing emails to their
authors.
Not used in production."""
if hasattr(self, '_debug_data') and not force_recheck:
return self._debug_data
d = self.raw_data.copy()
correct_forms = {
'name': str.upper,
'activity_version': str.isdigit,
'host_version': str.isdigit,
'bundle_id': re.compile(r'^[\w.]+$').match,
'service_name': re.compile(r'^[\w.]+$').match,
'icon': re.compile(r'^[\S]+$').match,
'exec': str.upper,
'mime_types': re.compile(r'^([\w.+-]+/[\w.+-]+;?)*$').match,
'update_url': re.compile(r'^http://([\w-]+\.?)+(:\d+)?(/[\w~%.-]+)*$').match,
#'update_url': re.compile(r'^$').match,
'show_launcher': re.compile(r'^(yes)|(no)$').match,
'class': re.compile(r'^(\w+.?)+$').match,
'license': str.upper,
#'license': re.compile(r'^GPLv[23]\+?$').match,
}
for k, v in d.items():
if k in correct_forms:
f = correct_forms.get(k, len)
if not f(v):
d['BAD ' + k] = v
rcount = 0
for k in REQUIRED_TAGS:
if k not in d:
d['LACKS %s' % k] = True
rcount += 1
d['MISSING KEYS'] = rcount
for t in OPTIONAL_TAGS:
if t not in d:
d['NO ' + t] = True
if not 'icon' in d and d.get('show_launcher') != 'no':
d['NO icon AND show_launcher'] = True
self._debug_data = d
return d
def get_name(self, locale):
"""Return the best guess at a name for the locale."""
for loc in locale_search_path(locale):
if loc in self.linfo and 'name' in self.linfo[loc]:
return self.linfo[loc]['name']
return super(Activity, self).get_name()
def check_all_bundles(directory, show_all_bundles=False):
"""A verbose debug function."""
all_bundles = []
unique_bundles = {}
counts = {}
# watch for these tags and print out the lists
bad_contents = {}
all_linfo = {}
unique_linfo = {}
linfo_keys = {}
log('Checking all activities in %s\n' % directory)
for f in os.listdir(directory):
if not f.endswith('.xo') and not f.endswith('.xol'):
continue
#log(f)
try:
if f.endswith('.xo'):
bundle = Activity(os.path.join(directory, f))
else:
bundle = Content(os.path.join(directory, f))
except Exception, e:
log("IRREDEEMABLE bundle %-25s (Error: %s)" % (f, e), syslog.LOG_WARNING)
#Clump together bundles of the same ID
x = unique_bundles.setdefault(bundle.bundle_id, [])
x.append(bundle)
all_bundles.append(bundle)
if not show_all_bundles:
#only show the newest one of each set.
bundles = []
for versions in unique_bundles.values():
versions.sort()
bundles.append(versions[-1])
else:
bundles = all_bundles
licenses = {}
for bundle in bundles:
bid = bundle.bundle_id
for k, v in bundle.debug().iteritems():
counts[k] = counts.get(k, 0) + 1
if k.startswith('BAD '):
bc = bad_contents.setdefault(k, {})
bc[bid] = v
for k, v in bundle.linfo.iteritems():
linfo_l = all_linfo.setdefault(k, [])
linfo_l.append(bundle)
for x in v:
linfo_keys[x] = linfo_keys.get(x, 0) + 1
if v['name'] != bundle.name:
linfo_l = unique_linfo.setdefault(k, [])
linfo_l.append(bundle)
if bundle.license:
lic = licenses.setdefault(bundle.license, [])
lic.append(bundle.bundle_id)
citems = counts.items()
rare_keys = [k for k, v in citems if v < 10]
lack_counts = dict((k, v) for k, v in citems if k.startswith('LACKS '))
bad_counts = dict((k, v) for k, v in citems if k.startswith('BAD '))
no_counts = dict((k, v) for k, v in citems if k.startswith('NO '))
tag_counts = dict((k, v) for k, v in citems if k not in lack_counts and
k not in bad_counts and k not in no_counts and k != 'MISSING KEYS')
# flag whether the tag is needed, ok, or not
tag_quality = dict((k, '*') for k in REQUIRED_TAGS)
tag_quality.update((k, '+') for k in OPTIONAL_TAGS)
linfo_counts = dict((k, len(v)) for k, v in all_linfo.iteritems())
linfo_uniq_counts = dict((k, len(v)) for k, v in unique_linfo.iteritems())
log('\nFound: %s bundles\n %s unique bundles' % (len(all_bundles), len(unique_bundles)))
for d, name, d2 in [(tag_counts, '\nattribute counts:', tag_quality),
(lack_counts, '\nmissing required keys:', {}),
(no_counts, '\nunused optional keys:', {}),
(bad_counts, '\nill-formed values:', {}),
(linfo_counts, '\nlinfo counts: total localised', linfo_uniq_counts),
(linfo_keys, '\nlinfo keys:', {})]:
log(name)
counts_reversed = [(v, k) for (k, v) in d.iteritems()]
counts_reversed.sort()
counts_reversed.reverse()
for (k, v) in counts_reversed:
log("%-25s %4s %4s" % (v, k, d2.get(v, '')))
log("\nRare keys:")
for k in rare_keys:
if k.startswith('BAD '):
continue
log(k)
for b in bundles:
v = b.debug().get(k)
if v:
log(' %-25s %s' % (b.bundle_id, v))
log("\nInteresting contents:")
for k, v in bad_contents.iteritems():
log(k)
for x in v.iteritems():
log(' %s: %s' % x)
log("\nInteresting linfo:")
for k in ('pseudo',):
log(k)
for a in all_linfo[k]:
if a in unique_linfo.get(k, []):
log(' * %s (%s vs. %s)' % (a.bundle_id, a.name, a.linfo[k]['name']))
else:
log(' %s (%s)' % (a.bundle_id, a.name))
log("\nLicenses:")
for lic, v in licenses.iteritems():
log("%-20s %s" %(repr(lic), len(v)))
log("\nRare licenses:")
for lic, v in licenses.iteritems():
if len(v) < 3:
log(' %s' % lic)
for x in v:
log(" %s" %(x))
log("\nAlmost valid activities:")
for b in bundles:
d = b.debug()
if d['MISSING KEYS'] == 1:
missing = ', '.join(x for x in d if x.startswith('LACKS'))
bad_values = ', '. join(x for x in d if x.startswith('BAD'))
log("%-20s %s %s" %(b.name, missing, bad_values))
log("\nValid activities (maybe):")
for b in bundles:
d = b.debug()
bid = b.bundle_id
if (d['MISSING KEYS'] == 0 and
bid not in bad_contents['BAD mime_types']):
log("%-20s - %s" %(b.name, bid))
#log(a.raw_data)
def read_info_file(zipfile, path, section):
"""Return a dictionary matching the contents of the config file at
path in zipfile"""
cp = SafeConfigParser()
info = StringIO(zipfile.read(path))
cp.readfp(info)
return dict(cp.items(section))
def canonicalise(lang):
"""Make all equivalent language strings the same.
>>> canonicalise('Zh-cN')
zh-CN
>>> canonicalise('zh_CN')
zh-CN
"""
lang = lang.replace('_', '-').upper()
bits = lang.split('-', 1)
bits[0] = bits[0].lower()
return '-'.join(bits)
def locale_search_path(locale):
"""Find a series of sensible locales to try, including
DEFAULT_LANG. For example 'zh-CN' would become ('zh-CN', 'zh',
'DEFAULT_LANG')."""
#XXX might be better to be storing locale as tuple
if '-' in locale:
return (locale, locale.split('-')[0], DEFAULT_LANG)
return (locale, DEFAULT_LANG)
def read_metadata(bundle_dir):
"""Attempt to read data in a metadata file. Raises expected
exceptions if the metadata file isn't readable. The file should
look something like this:
[org.laptop.Pippy]
description = Succinct description of this activity.
[org.laptop.Develop]
description = Succinct description of this activity.
web_icon = develop.png
"""
md_files = [os.path.join(bundle_dir, x)
for x in os.listdir(bundle_dir) if x.endswith('.info')]
cp = SafeConfigParser()
cp.read(md_files)
metadata = {}
for section in cp.sections():
metadata[section] = dict(x for x in cp.items(section))
return metadata
def htmlise_bundles(bundle_dir, dest_html):
"""Makes a nice html manifest for the bundles in a directory. The
manifest only shows the newest version of each bundle.
"""
#so, we collect up a dictionary of lists, then sort each list on
#the version number to find the newest.
bundles = [os.path.join(bundle_dir, x)
for x in os.listdir(bundle_dir) if x.endswith('.xo') or x.endswith('.xol')]
try:
metadata = read_metadata(bundle_dir)
except Exception, e:
log("had trouble reading metadata: %s" % e)
metadata = {}
all_bundles = {}
for filename in bundles:
try:
if filename.endswith('.xo'):
bundle = Activity(filename)
else:
bundle = Content(filename)
x = all_bundles.setdefault(bundle.bundle_id, [])
x.append((bundle.mtime, bundle))
except Exception, e:
log("Couldn't find good activity/library info in %s (Error: %s)" % (filename, e))
newest = []
# create an index for each language that has a template
# but track any locales in bundles in case we do not have templates for them
locales = [os.path.join(o) for o in os.listdir(TEMPLATE_DIR) if os.path.isdir(os.path.join(TEMPLATE_DIR,o))]
locales_found = set ()
for versions in all_bundles.values():
versions = [x[1] for x in sorted(versions)]
# end of list is the newest; beginning of list might need deleting
latest = versions.pop()
locales_found.update(latest.linfo)
newest.append(latest)
goners = versions[:-KEEP_OLD_VERSIONS]
keepers = versions[-KEEP_OLD_VERSIONS:]
for v in goners:
fn = os.path.join(bundle_dir, v.url)
os.remove(fn)
latest.set_older_versions(keepers)
if latest.bundle_id in metadata:
# we have extra metadata with which to fill out details
# presumably this is mainly human-oriented description.
d = metadata[latest.bundle_id]
for k in ('description', 'name'):
if k in d:
setattr(latest, k, d[k])
log('found locales: %s' % locales)
# assume locales is not empty as we have at least the default language
for locale in locales:
try:
make_html(newest, locale, '%s.%s' % (dest_html, locale))
except Exception, e:
log("Couldn't make page for %s (Error: %s)" % (locale, e), syslog.LOG_WARNING)
# make_varfile(locales, dest_html)- have switched to multiviews, so var not needed
def make_varfile(locales, dest_html):
f = open(dest_html + '.var', 'w')
index = os.path.basename(dest_html)
f.write('URI: %s\n\n' % index)
for locale in locales:
f.write('URI: %s.%s\n' % (index, locale))
f.write('Content-type: text/html; charset=utf-8\n')
f.write('Content-language: %s\n\n' % locale)
# now the default, slightly higher qs
f.write('URI: %s.DEFAULT\n' % index)
f.write('Content-type: text/html; charset=utf-8\n')
f.write('Content-language: en\n\n')
f.close()
def read_template(name, locale):
"""Try to read the locale's template, falling back to the
default."""
#also try containing locales, eg 'zh' for 'zh-CN'
for x in locale_search_path(locale):
try:
f = open(os.path.join(TEMPLATE_DIR, x, name))
break
except (OSError, IOError), e:
#log(str(e))
continue
s = f.read()
f.close()
return s
def make_html(bundles, locale, filename):
"""Write a microformated index for the activities in the appropriate language,
and save it to filename."""
page_tmpl = read_template('page', locale)
act_tmpl = read_template('activity', locale)
#bundles.sort() won't cut it.
schwartzian = [ (x.get_name(locale), x.to_html(locale, act_tmpl)) for x in bundles ]
schwartzian.sort()
s = page_tmpl % {'activities': '\n'.join(x[1] for x in schwartzian)}
if os.path.exists(filename):
shutil.move(filename, filename + '~')
f = open(filename, 'w')
f.write(s)
f.close()
Reference in a new issue View git blame Copy permalink