Newer
Older
#!/usr/bin/python3
# coding: utf-8
# aurbot - Archlinux User Repository Builder Bot
#
# Started, October 30th 2011
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
from argparse import ArgumentParser
from configparser import ConfigParser
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from logging import StreamHandler, getLogger, Formatter, DEBUG, INFO
from os import chdir, environ, getcwd, mkdir
from os.path import exists, join
from subprocess import Popen, check_call, DEVNULL, PIPE
from systemd.daemon import notify
from tarfile import open as tar
from tempfile import TemporaryDirectory
AUR_URL = 'https://aur.archlinux.org'
USER_AGENT = "aurbot"
XDG_DIRECTORY = "aurbot"
ERR_USAGE = 1
ERR_FATAL = 2
ERR_ABORT = 3
ERR_UNKNOWN = 4
class ABFormatter(Formatter):
'''
Customer logging formater
'''
def __init__(self, fmt="[%(levelname)s] %(msg)s"):
Formatter.__init__(self, fmt)
def format(self, record):
format_orig = self._style._fmt
if record.levelno == INFO and getLogger(record.name).getEffectiveLevel() != DEBUG:
self._style._fmt = "%(msg)s"
result = Formatter.format(self, record)
self._style._fmt = format_orig
return result
class AURPackage(dict):
'''
Abstract AUR package action
'''
def __init__(self, name, timeout=None):
self.name = name
debug("getting %s aur infos" % self.name)
url = "%s/rpc.php?type=info&arg=%s" % (AUR_URL, name)
url_req = Request(url, headers={"User-Agent": USER_AGENT})
url_fd = urlopen(url_req, timeout=timeout)
d = jloads(url_fd.read().decode("utf-8"))
if d["version"] != 1:
raise Exception("Unknown AUR Backend version: %s" % d["version"])
if len(d["results"]) == 0:
raise Exception("No such package: %s" % name)
self._info = d["results"]
def __getattr__(self, name):
for k, v in self._info.items():
if name == k.lower():
return v
raise AttributeError()
def __repr__(self):
def extract(self, path):
'''
Extract aur source tarball inside a directory path
'''
fo = urlopen('%s/%s' % (AUR_URL, self.urlpath))
tarball = tar(mode='r|*', fileobj=fo)
tarball.extractall(path)
fo.close()
class LocalPackage(dict):
'''Local package data abstraction'''
def __init__(self, name):
self.name = name
self.path = join(save_data_path(XDG_DIRECTORY), name)
debug("local path is: %s" % self.path)
if not exists(self.path):
mkdir(self.path)
@property
def logdir(self):
logdir = join(self.path, "log")
if not exists(logdir):
mkdir(logdir)
return logdir
return int(open(join(self.path, X), "r").read())
except Exception as exp:
debug("Failed to read %s time: %s" % (X, exp))
return 0
def setlastX(self, X, value):
try:
open(join(self.path, X), "w").write("%d\n" % value)
except Exception as exp:
error("Failed to save %s time: %s" % (X, exp))
# store the moment where the build was done locally
lastbuild = property(lambda x: LocalPackage.getlastX(x, "lastbuild"),
lambda x, y: LocalPackage.setlastX(x, "lastbuild", y))
# store the aur lastmodified value of the last sucessful build
lastsuccess = property(lambda x: LocalPackage.getlastX(x, "lastsuccess"),
lambda x, y: LocalPackage.setlastX(x, "lastsuccess", y))
# store the aur lastmodified value of the last failed build
lastfailed = property(lambda x: LocalPackage.getlastX(x, "lastfailed"),
lambda x, y: LocalPackage.setlastX(x, "lastfailed", y))
# store the last time we check the aur
lastchecked = property(lambda x: LocalPackage.getlastX(x, "lastchecked"),
lambda x, y: LocalPackage.setlastX(x, "lastchecked", y))
def send_report(config, localpkg, aurpkg, status, logfile):
'''Send build notification'''
# generate message
msg = MIMEMultipart()
"successful" if status else "failure", localpkg.name, aurpkg.version)
msg["From"] = "Aurbot"
msg["To"] = config["notify"]
msg["Date"] = formatdate(localtime=True)
# attach logfile
with open(logfile, "r") as fd:
mt = MIMEText(fd.read())
msg.attach(mt)
# send message
proc = Popen(["sendmail", "-i", "-t"], stdin=PIPE, close_fds=True)
proc.stdin.write(msg.as_bytes())
proc.stdin.close()
proc.wait()
def build(config, localpkg, aurpkg):
'''
Build and commit a package
Notify if succeeded
'''
# register the build
localpkg.lastbuild = time()
# log files
fp = join(localpkg.logdir, strftime("build-%Y-%m-%d-%H-%M-%S.log", localtime(time())))
debug("Build log file path: %s" % fp)
with open(fp, "w") as fd:
try:
cwd = getcwd()
chdir("%s/%s" % (build_dir.name, aurpkg.name))
# build
info("Starting build command")
debug(config["build_cmd"])
fd.write("Build command: %s\n" % config["build_cmd"])
start_time = time()
try:
check_call(config["build_cmd"], stdin=DEVNULL, stdout=fd,
stderr=fd, shell=True, close_fds=True)
except Exception as exp:
error("Build command failure: %s" % exp)
raise
end_time = time()
info("Build duration: %.2fs" % (end_time - start_time))
fd.write("Build duration: %.2fs\n" % (end_time - start_time))
# commit
if "commit_cmd" in config:
info("Starting commit command")
debug(config["commit_cmd"])
fd.write("Commit command: %s\n" % config["commit_cmd"])
start_time = time()
try:
check_call(config["commit_cmd"], stdin=DEVNULL, stdout=fd,
stderr=fd, shell=True, close_fds=True)
except Exception as exp:
error("Commit command failure: %s" % exp)
raise
end_time = time()
info("Commit duration: %.2fs" % (end_time - start_time))
fd.write("Commit duration: %.2fs\n" % (end_time - start_time))
status = True
except Exception as exp:
status = False
# we have to register after chdir in the original directory
if status:
localpkg.lastsuccess = aurpkg.lastmodified
else:
localpkg.lastfailed = aurpkg.lastmodified
# notify
if "notify" in config:
send_report(config, localpkg, aurpkg, status, fp)
# parse package list
packages = ConfigParser()
packages.read(config_path)
for name, config in packages.items():
if name == "DEFAULT":
continue
local = LocalPackage(name)
check_interval = config.getint("check_interval", DEFAULT_CHECK_INTERVAL)
debug("Check interval is %ss" % check_interval)
check_delta = int(local.lastchecked - time() + check_interval)
debug("Check delta is %ss" % check_delta)
if check_delta > 0:
# next check is in the future
next_checks.add(check_delta)
info("Nothing to check. Next is planned in %ss" % check_delta)
continue
next_checks.add(check_interval)
# get remote data
aur = AURPackage(name, config.getint("timeout"))
except Exception as exp:
error("Unable to get AUR package info: %s" % exp)
continue
maintainer = config.get("maintainer")
if maintainer != aur.maintainer:
debug("Local last success lastmodified: %s" % local.lastbuild)
debug("Local last failed lastmodified: %s" % local.lastfailed)
debug("Local last build time: %s" % local.lastbuild)
if local.lastsuccess >= aur.lastmodified :
if "force" in config:
if config["force"].isdigit() is False:
warning("Invalid force value, ignore it")
continue
# if lastbuild not exists, it will be equal to 0
# too small to be > to time() even with big force time
now = int(time())
force = int(config["force"])
debug("Force is: %ss" % force)
force_delta = local.lastbuild - now + force
debug("Force Delta is: %ss" % force_delta)
if force_delta < 0:
info("Forced update")
build(config, local, aur)
else:
else:
info("Up to date, nothing to do.")
elif local.lastfailed >= aur.lastmodified:
info("Last build has failed. We skip.")
else:
# sleep until next check
# len(next_checks) is 0 when there is no package configured
timeout = min(next_checks) if len(next_checks) > 0 else DEFAULT_CHECK_INTERVAL
debug("waiting for %ds" % timeout)
sleep(timeout)
def parse_argv():
'''Parse command line arguments'''
# load parser
parser = ArgumentParser()
parser.add_argument("-c", "--config", help="packages config file path")
parser.add_argument("-d", "--debug", action="store_true", help="debug mode")
parser.epilog = "You could set $XDG_DATA_HOME to change the path of the local package cache."
# parse it!
args = parser.parse_args()
# set global debug mode
if args.debug:
getLogger().setLevel(DEBUG)
# set default paths
if args.config is None:
args.config = join(save_config_path(XDG_DIRECTORY), "packages.conf")
return args
def main():
'''Program entry point'''
try:
# set logger config
hdlr = StreamHandler()
hdlr.setFormatter(ABFormatter())
getLogger().addHandler(hdlr)
getLogger().setLevel(DEBUG if "AURBOT_DEBUG" in environ else INFO)
# use sighup to unblock sleep syscall
signal(SIGHUP, lambda signum, frame: info("SIGHUP received"))
# tell to systemd we are ready
notify("READY=1\n")
# while 42
except KeyboardInterrupt:
exit(ERR_ABORT)
except Exception as exp:
error(exp)
if getLogger().getEffectiveLevel() != DEBUG:
error("Unknown error. Please report it with --debug.")
else:
raise
exit(ERR_UNKNOWN)
if __name__ == '__main__':