ansible/roles/badges/backend/files/cron/award-oldschool-badges
2017-10-07 21:14:41 +00:00

208 lines
6.6 KiB
Python

#!/usr/bin/env python
import __main__
# This is going to require sqlalchemy 0.8 sooner than later.
__main__.__requires__ = __requires__ = ["tahrir-api", "sqlalchemy>=0.7"];
import pkg_resources
pkg_resources.require(__requires__)
import itertools
import string
import time
import urllib
import socket
from hashlib import md5
import getpass
import pprint
from tahrir_api.dbapi import TahrirDatabase
import transaction
_fas_cache = {}
import logging
log = logging.getLogger()
logging.basicConfig(level=logging.INFO)
import fedora.client.fas2
import fedmsg
import fedmsg.config
fm_config = fedmsg.config.load_config()
fm_config['cert_prefix'] = 'fedbadges'
fm_config['name'] = 'relay_inbound'
fm_config['active'] = True
fedmsg.init(**fm_config)
import fedbadges.utils
def user_in_group(user, group_name):
# First, bail out if they're not in the group at all
if not any([g.name == group_name for g in user.memberships]):
return False
# Find the group_id of the group we're looking for..
group_id = None
for g in user.memberships:
if g.name == group_name:
group_id = g.id
break
if not group_id:
return False
# For that group_id, find the relevant role
relevant_role = None
for role in user.roles:
if role.group_id == group_id:
relevant_role = role
break
if not relevant_role:
return False
# They must be actually 'approved' in that group for this to count
return relevant_role.role_status == 'approved'
def get_all_fas_users(creds):
fasclient = fedora.client.fas2.AccountSystem(
username=creds['username'],
password=creds['password'],
)
timeout = socket.getdefaulttimeout()
socket.setdefaulttimeout(600)
try:
log.info("Downloading FAS cache")
for a, b in itertools.product(string.lowercase, string.lowercase):
term = a + b + '*'
log.info(" Querying FAS for %r" % term)
request = fasclient.send_request('/user/list',
req_params={'search': term},
auth=True)
log.info(" Found %r matching users." % len(request['people']))
for person in request['people']:
yield person
finally:
socket.setdefaulttimeout(timeout)
def get_fas_groupings(fas_credentials, lookup, **config):
results = {}
packager_id, ambassadors_id = None, None
sponsor_types = ['sponsor', 'administrator']
mega_list = get_all_fas_users(fas_credentials)
for user in mega_list:
# This is the main check.
for group_name, badge_id in lookup.iteritems():
if user_in_group(user, group_name):
results[group_name] = results.get(group_name, []) + [user]
# Beyond the main check, here is a special check that makes sure they
# are a sponsor in the packager group.
if not packager_id:
for group in user.memberships:
if group.name == 'packager':
packager_id = group.id
if not ambassadors_id:
for group in user.memberships:
if group.name == 'ambassadors':
ambassadors_id = group.id
for role in user.roles:
if role.group_id == packager_id:
if role.role_type in sponsor_types and role.role_status == 'approved':
results['sponsors'] = results.get('sponsors', []) + [user]
if role.group_id == ambassadors_id:
if role.role_type in sponsor_types and role.role_status == 'approved':
results['ambassadors_sponsors'] = results.get('ambassadors_sponsors', []) + [user]
return results
def main():
d = {}
print "fascache.db code is commented out -- querying fas."
# A mapping of fas groups to badge ids
mapping = {
'3d-printing-sig': '3d-printing-sig-member',
'cla_done': 'involvement',
'ambassadors': 'ambassador',
'ambassadors_sponsors': 'ambassadors-sponsor',
'commops': 'commops-superstar',
'council': 'council-member',
'designteam': 'pixel-ninja',
'dotnet-team': 'dotnet-sig-member',
'fedora-hams': 'amateur-radio-sig-member',
'fi-apprentice': 'no-longer-a-ronin',
'gitfedora-web': 'rock-the-web!',
'git-fedora-electronic-lab': 'fel-member',
'irc-support-operators':
'your-call-may-be-monitored-for-quality-assurance',
'lxqt-sig': 'lxqt-sig-member',
'modularity-wg': 'modularity-wg-member',
'provenpackager': 'proven-packager',
'proventesters': 'proven-tester',
'robotics-sig': 'domo-arigato',
'security-team': 'security-team',
'sponsors': 'packager-sponsor',
'summer-coding': 'summer-coder',
'sysadmin-main': 'trust-me,-i-know-what-i-am-doing',
'sysadmin-badges': 'inside-job',
'videos': 'videographer',
}
# First, some validation that the badge ids actually exist.
for fas_group, badge_id in mapping.items():
badge = tahrir.get_badge(badge_id=badge_id)
if not badge:
raise ValueError("%r is not a valid badge id" % badge_id)
# Then, do a long query against FAS for our candidates.
results = get_fas_groupings(fas_credentials=fm_config['fas_credentials'],
lookup=mapping)
for fas_group, members in results.iteritems():
print fas_group, "had", len(members), "members for", mapping[fas_group]
for fas_group, members in results.iteritems():
print "processing", fas_group
print "======================"
badge = tahrir.get_badge(badge_id=mapping[fas_group])
hit_em_up(badge, members)
def hit_em_up(badge, group):
for fas_user in group:
email = fas_user.username + "@fedoraproject.org"
user = tahrir.get_person(email)
if not user:
continue
if tahrir.assertion_exists(badge.id, email):
print email, "already has", badge.id, "skipping."
continue
time.sleep(1)
print "awarding", badge.id, "to", email
try:
transaction.begin()
tahrir.add_assertion(badge.id, email, None)
transaction.commit()
except Exception as e:
transaction.abort()
print "Failure:", e
if __name__ == '__main__':
uri = fm_config['badges_global']['database_uri']
tahrir = TahrirDatabase(
uri,
notification_callback=fedbadges.utils.notification_callback,
)
main()