391 lines
9.5 KiB
Python
391 lines
9.5 KiB
Python
"""
|
|
Celery tasks for djeuscan
|
|
"""
|
|
|
|
from datetime import datetime
|
|
|
|
from celery.task import task, group
|
|
|
|
#import portage
|
|
|
|
from django.conf import settings
|
|
from django.template.loader import render_to_string
|
|
from django.core.mail import send_mail
|
|
from django.db.models import Q
|
|
|
|
from euscan.version import gentoo_unstable
|
|
|
|
from djeuscan.models import Package, RefreshPackageQuery, VersionLog
|
|
from djeuscan.processing import scan, misc
|
|
|
|
from euscan_accounts.helpers import get_account_versionlogs, get_user_fav_infos
|
|
from euscan_accounts.models import UserProfile
|
|
|
|
|
|
class TaskFailedException(Exception):
|
|
"""
|
|
Exception for failed tasks
|
|
"""
|
|
pass
|
|
|
|
|
|
def group_one(task, seq, *args, **kwargs):
|
|
"""
|
|
Create a group of tasks, each task handle one element of seq
|
|
"""
|
|
tasks = []
|
|
|
|
if "attr_name" in kwargs:
|
|
attr_name = kwargs['attr_name']
|
|
del kwargs["attr_name"]
|
|
else:
|
|
attr_name = None
|
|
|
|
for elem in seq:
|
|
if attr_name:
|
|
kwargs[attr_name] = elem
|
|
tasks.append(task.subtask(args=list(args), kwargs=dict(kwargs),
|
|
immutable=True))
|
|
else:
|
|
tasks.append(task.subtask(args=[elem] + list(args),
|
|
kwargs=dict(kwargs), immutable=True))
|
|
|
|
return group(tasks)
|
|
|
|
|
|
def group_chunks(task, seq, n, *args, **kwargs):
|
|
"""
|
|
Creates a group of tasks, each subtask has <n> elements to handle
|
|
"""
|
|
tasks = []
|
|
for i in xrange(0, len(seq), n):
|
|
tasks.append(
|
|
task.subtask(args=[seq[i:i + n]] + list(args), kwargs=kwargs,
|
|
immutable=True)
|
|
)
|
|
return group(tasks)
|
|
|
|
|
|
@task
|
|
def regen_rrds():
|
|
"""
|
|
Regenerate RRDs
|
|
"""
|
|
misc.regen_rrds()
|
|
return True
|
|
|
|
|
|
@task
|
|
def update_counters(fast=False):
|
|
"""
|
|
Updates counters
|
|
"""
|
|
logger = update_counters.get_logger()
|
|
logger.info("Updating counters (fast=%s)...", fast)
|
|
misc.update_counters(fast=fast)
|
|
logger.info("Done")
|
|
return True
|
|
|
|
|
|
@task
|
|
def scan_metadata(packages=[], category=None, populate=False):
|
|
"""
|
|
Scans metadata for the given set of packages
|
|
"""
|
|
logger = scan_metadata.get_logger()
|
|
|
|
if packages:
|
|
logger.info("Starting metadata scan for %d packages...",
|
|
len(packages))
|
|
elif category:
|
|
logger.info("Starting metadata scan for %s...",
|
|
category)
|
|
else:
|
|
logger.info("Starting metadata scan...")
|
|
|
|
scan.scan_metadata(
|
|
packages=packages,
|
|
category=category,
|
|
logger=logger,
|
|
populate=populate,
|
|
)
|
|
return True
|
|
|
|
|
|
@task
|
|
def scan_portage(packages=None, category=None,
|
|
no_log=False, purge_packages=False,
|
|
purge_versions=False):
|
|
"""
|
|
Scans portage for the given set of packages
|
|
"""
|
|
logger = scan_portage.get_logger()
|
|
|
|
if packages:
|
|
logger.info("Starting portage scan for %d packages...",
|
|
len(packages))
|
|
elif category:
|
|
logger.info("Starting portage scan for %s...",
|
|
category)
|
|
else:
|
|
logger.info("Starting portage scan...")
|
|
|
|
return scan.scan_portage(
|
|
packages=packages,
|
|
category=category,
|
|
no_log=no_log,
|
|
purge_packages=purge_packages,
|
|
purge_versions=purge_versions,
|
|
logger=logger,
|
|
)
|
|
|
|
|
|
@task
|
|
def scan_upstream(packages=[], purge_versions=False):
|
|
"""
|
|
Scans upstream for the given set of packages
|
|
"""
|
|
logger = scan_upstream.get_logger()
|
|
|
|
if packages:
|
|
logger.info("Starting upstream scan subtask for %d packages...",
|
|
len(packages))
|
|
else:
|
|
logger.info("Starting upstream scan...")
|
|
|
|
scan.scan_upstream(
|
|
packages=packages,
|
|
purge_versions=purge_versions,
|
|
logger=logger,
|
|
)
|
|
return True
|
|
|
|
|
|
@task
|
|
def update_portage_trees():
|
|
"""
|
|
Update portage tree
|
|
"""
|
|
logger = update_portage_trees.get_logger()
|
|
misc.update_portage_trees(logger=logger)
|
|
return True
|
|
|
|
|
|
@task
|
|
def update_portage(packages=None):
|
|
#categories = portage.settings.categories
|
|
|
|
# Workaround for celery bug when chaining groups
|
|
update_portage_trees()
|
|
updated_packages = scan_portage(
|
|
packages=None,
|
|
purge_packages=True,
|
|
purge_versions=True,
|
|
)
|
|
scan_metadata(packages=None, populate=True)
|
|
|
|
if not updated_packages:
|
|
# Simple/Fast path
|
|
update_counters(fast=False)
|
|
return
|
|
|
|
(
|
|
group_chunks(scan_upstream, updated_packages,
|
|
settings.TASKS_UPSTREAM_GROUPS,
|
|
purge_versions=True) |
|
|
update_counters.si(fast=False)
|
|
)()
|
|
|
|
""" Currently broken
|
|
update_portage_trees()
|
|
scan_metadata(packages=None, populate=True)
|
|
(
|
|
group_one(scan_portage, categories,
|
|
attr_name="category", purge_packages=True,
|
|
purge_versions=True) |
|
|
group_one(scan_metadata, categories,
|
|
attr_name="category") |
|
|
update_counters.si(fast=True)
|
|
)()
|
|
"""
|
|
return True
|
|
|
|
|
|
@task
|
|
def update_upstream():
|
|
if settings.TASKS_UPSTREAM_GROUPS >= 1:
|
|
packages = Package.objects.all().order_by('pk') # .order_by('?') ?
|
|
|
|
scan_upstream_sub = group_chunks(scan_upstream, packages,
|
|
settings.TASKS_UPSTREAM_GROUPS,
|
|
purge_versions=True)
|
|
else:
|
|
scan_upstream_sub = scan_upstream.si(purge_versions=True)
|
|
|
|
(
|
|
scan_upstream_sub |
|
|
update_counters.si(fast=False) |
|
|
send_update_email.si()
|
|
)()
|
|
return True
|
|
|
|
|
|
@task
|
|
def scan_package(package):
|
|
logger = scan_package.get_logger()
|
|
logger.info("Scanning package %s", package)
|
|
scan_portage([package], purge_packages=True, purge_versions=True)
|
|
scan_metadata([package])
|
|
scan_upstream([package], purge_versions=True)
|
|
return True
|
|
|
|
|
|
@task(rate_limit="1/m")
|
|
def scan_package_user(package):
|
|
scan_package(package)
|
|
return True
|
|
|
|
|
|
@task(rate_limit="1/m")
|
|
def consume_refresh_queue(locked=False):
|
|
"""
|
|
Satisfies user requests for package refreshing, runs every minute
|
|
"""
|
|
logger = consume_refresh_queue.get_logger()
|
|
logger.info('Consuming package refresh request queue...')
|
|
|
|
try:
|
|
query = RefreshPackageQuery.objects.all().order_by('-priority')[0]
|
|
pkg = query.package
|
|
query.delete()
|
|
scan_package_user.delay(pkg)
|
|
logger.info('Selected: %s' % pkg)
|
|
except IndexError:
|
|
return
|
|
|
|
if RefreshPackageQuery.objects.count():
|
|
logger.info('Restarting myself in 60s')
|
|
consume_refresh_queue.apply_async(
|
|
kwargs={'locked': True}, countdown=60
|
|
)
|
|
return True
|
|
|
|
|
|
@task(max_retries=10, default_retry_delay=10 * 60)
|
|
def send_user_email(address, subject, text):
|
|
try:
|
|
send_mail(
|
|
subject, text, settings.DEFAULT_FROM_EMAIL, [address],
|
|
fail_silently=False
|
|
)
|
|
except Exception, exc:
|
|
raise send_user_email.retry(exc=exc)
|
|
return True
|
|
|
|
|
|
@task
|
|
def process_emails(profiles, only_if_vlogs=False):
|
|
for profile in profiles:
|
|
now = datetime.now()
|
|
user = profile.user
|
|
|
|
vlogs = get_account_versionlogs(profile)
|
|
vlogs = vlogs.filter(
|
|
datetime__gt=profile.last_email,
|
|
overlay="", # only upstream versions
|
|
action=VersionLog.VERSION_ADDED, # only adds
|
|
)
|
|
if profile.email_ignore_pre:
|
|
vlogs = vlogs.exclude(vtype__in=gentoo_unstable)
|
|
if profile.email_ignore_pre_if_stable:
|
|
vlogs = vlogs.exclude(
|
|
~Q(package__last_version_gentoo__vtype__in=gentoo_unstable),
|
|
vtype__in=gentoo_unstable
|
|
)
|
|
|
|
if only_if_vlogs and not vlogs.count():
|
|
continue
|
|
|
|
vlogs = vlogs.order_by("-datetime")
|
|
|
|
infos = get_user_fav_infos(user)
|
|
infos["user"] = user
|
|
infos["vlogs"] = vlogs
|
|
|
|
mail_text = render_to_string(
|
|
"euscan/accounts/euscan_email.txt",
|
|
infos
|
|
)
|
|
|
|
send_user_email.delay(
|
|
user.email, "euscan updates - %s" % str(now.date()), mail_text
|
|
)
|
|
|
|
profile.last_email = now
|
|
profile.save(force_update=True)
|
|
return True
|
|
|
|
|
|
@task
|
|
def send_update_email():
|
|
profiles = UserProfile.objects.filter(
|
|
email_every=UserProfile.EMAIL_SCAN,
|
|
email_activated=True
|
|
)
|
|
group_chunks(
|
|
process_emails,
|
|
profiles,
|
|
settings.TASKS_EMAIL_GROUPS,
|
|
only_if_vlogs=True
|
|
)()
|
|
return True
|
|
|
|
|
|
@task
|
|
def send_weekly_email():
|
|
profiles = UserProfile.objects.filter(
|
|
email_every=UserProfile.EMAIL_WEEKLY,
|
|
email_activated=True
|
|
)
|
|
group_chunks(process_emails, profiles, settings.TASKS_EMAIL_GROUPS)()
|
|
return True
|
|
|
|
|
|
@task
|
|
def send_monthly_email():
|
|
profiles = UserProfile.objects.filter(
|
|
email_every=UserProfile.EMAIL_MONTHLY,
|
|
email_activated=True
|
|
)
|
|
group_chunks(process_emails, profiles, settings.TASKS_EMAIL_GROUPS)()
|
|
return True
|
|
|
|
|
|
@task
|
|
def collect_stabilization_candidates():
|
|
"""
|
|
This task collects the packages that are stabilization candidates
|
|
"""
|
|
logger = collect_stabilization_candidates.get_logger()
|
|
misc.stabilization_candidates(
|
|
days_to_stabilization=settings.DAYS_TO_STABILIZATION,
|
|
logger=logger
|
|
)
|
|
|
|
|
|
admin_tasks = [
|
|
regen_rrds,
|
|
update_counters,
|
|
scan_metadata,
|
|
scan_portage,
|
|
scan_upstream,
|
|
update_portage_trees,
|
|
update_portage,
|
|
update_upstream,
|
|
scan_package,
|
|
send_update_email,
|
|
send_weekly_email,
|
|
send_monthly_email,
|
|
collect_stabilization_candidates,
|
|
]
|