<commit_before>import atexit
import datetime
import hashlib
import os
import re
import psutil
import socket
import string
import shutil
import redis
import time
import json

from os import environ as env
from urlparse import urlparse
from seesaw.project import *
from seesaw.item import *
from seesaw.task import *
from seesaw.pipeline import *
from seesaw.externalprocess import *

from seesaw.util import find_executable

VERSION = "20140112.01"
USER_AGENT = "ArchiveTeam ArchiveBot/%s" % VERSION
EXPIRE_TIME = 60 * 60 * 48  # 48 hours between archive requests
WGET_LUA = find_executable('Wget+Lua', "GNU Wget 1.14.0-archivebot1",
        [ './wget-lua' ])

if not WGET_LUA:
    raise Exception("No usable Wget+Lua found.")

if 'RSYNC_URL' not in env:
    raise Exception('RSYNC_URL not set.')

if 'REDIS_URL' not in env:
    raise Exception('REDIS_URL not set.')

if 'LOG_CHANNEL' not in env:
    raise Exception('LOG_CHANNEL not set.')

RSYNC_URL = env['RSYNC_URL']
REDIS_URL = env['REDIS_URL']
LOG_CHANNEL = env['LOG_CHANNEL']

# ------------------------------------------------------------------------------
# REDIS CONNECTION
# ------------------------------------------------------------------------------

redis_url = urlparse(REDIS_URL)
redis_db = int(redis_url.path[1:])
r = redis.StrictRedis(host=redis_url.hostname, port=redis_url.port, db=redis_db)

# ------------------------------------------------------------------------------
# SYSTEM MONITORING
# ------------------------------------------------------------------------------

hostname = socket.gethostname()
fqdn = socket.getfqdn()
pid = os.getpid()

pipeline_id_input = "%s:%s:%s" % (hostname, fqdn, pid)
m = hashlib.md5()
m.update(pipeline_id_input)
pipeline_id = 'pipeline:%s' % m.hexdigest()

def do_report(pipeline, redis):
    process_report = {
        'id': pipeline_id,
        'hostname': hostname,
        'fqdn': fqdn,
        'pid': pid,
        'version': VERSION,
        'mem_usage': psutil.virtual_memory().percent,
        'disk_usage': psutil.disk_usage(pipeline.data_dir).percent
    }

    redis.sadd('pipelines', pipeline_id)
    redis.hmset(pipeline_id, process_report)

def unregister_pipeline():
    r.delete(pipeline_id)
    r.srem('pipelines', pipeline_id)<commit_msg>Add a pipeline to the active set only after its data has been written.<commit_after>import atexit
import datetime
import hashlib
import os
import re
import psutil
import socket
import string
import shutil
import redis
import time
import json

from os import environ as env
from urlparse import urlparse
from seesaw.project import *
from seesaw.item import *
from seesaw.task import *
from seesaw.pipeline import *
from seesaw.externalprocess import *

from seesaw.util import find_executable

VERSION = "20140112.01"
USER_AGENT = "ArchiveTeam ArchiveBot/%s" % VERSION
EXPIRE_TIME = 60 * 60 * 48  # 48 hours between archive requests
WGET_LUA = find_executable('Wget+Lua', "GNU Wget 1.14.0-archivebot1",
        [ './wget-lua' ])

if not WGET_LUA:
    raise Exception("No usable Wget+Lua found.")

if 'RSYNC_URL' not in env:
    raise Exception('RSYNC_URL not set.')

if 'REDIS_URL' not in env:
    raise Exception('REDIS_URL not set.')

if 'LOG_CHANNEL' not in env:
    raise Exception('LOG_CHANNEL not set.')

RSYNC_URL = env['RSYNC_URL']
REDIS_URL = env['REDIS_URL']
LOG_CHANNEL = env['LOG_CHANNEL']

# ------------------------------------------------------------------------------
# REDIS CONNECTION
# ------------------------------------------------------------------------------

redis_url = urlparse(REDIS_URL)
redis_db = int(redis_url.path[1:])
r = redis.StrictRedis(host=redis_url.hostname, port=redis_url.port, db=redis_db)

# ------------------------------------------------------------------------------
# SYSTEM MONITORING
# ------------------------------------------------------------------------------

hostname = socket.gethostname()
fqdn = socket.getfqdn()
pid = os.getpid()

pipeline_id_input = "%s:%s:%s" % (hostname, fqdn, pid)
m = hashlib.md5()
m.update(pipeline_id_input)
pipeline_id = 'pipeline:%s' % m.hexdigest()

def do_report(pipeline, redis):
    process_report = {
        'id': pipeline_id,
        'hostname': hostname,
        'fqdn': fqdn,
        'pid': pid,
        'version': VERSION,
        'mem_usage': psutil.virtual_memory().percent,
        'disk_usage': psutil.disk_usage(pipeline.data_dir).percent
    }

    redis.hmset(pipeline_id, process_report)
    redis.sadd('pipelines', pipeline_id)

def unregister_pipeline():
    r.delete(pipeline_id)
    r.srem('pipelines', pipeline_id)
{'fraction_non_alphanumeric': 0.23512747875354106, 'fraction_numerical': 0.010895619960775768, 'mean_word_length': 2.3284989122552573, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6571813', 'n_tokens_mistral': 1460, 'n_tokens_neox': 1452, 'n_words': 354}