Version 20260629.01. Initial.
This commit is contained in:
arkiver
374
pipeline.py
Normal file
374
pipeline.py
Normal file
@@ -0,0 +1,374 @@
|
||||
# encoding=utf8
|
||||
from distutils.version import StrictVersion
|
||||
import datetime
|
||||
import hashlib
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import socket
|
||||
import sys
|
||||
import time
|
||||
|
||||
if sys.version_info[0] < 3:
|
||||
from urllib import unquote
|
||||
else:
|
||||
from urllib.parse import unquote
|
||||
|
||||
import seesaw
|
||||
from seesaw.config import realize, NumberConfigValue
|
||||
from seesaw.externalprocess import WgetDownload
|
||||
from seesaw.item import ItemInterpolation, ItemValue
|
||||
from seesaw.pipeline import Pipeline
|
||||
from seesaw.project import Project
|
||||
from seesaw.task import SimpleTask, LimitConcurrent
|
||||
from seesaw.tracker import GetItemFromTracker, PrepareStatsForTracker, \
|
||||
UploadWithTracker, SendDoneToTracker
|
||||
from seesaw.util import find_executable
|
||||
|
||||
if StrictVersion(seesaw.__version__) < StrictVersion('0.8.5'):
|
||||
raise Exception('This pipeline needs seesaw version 0.8.5 or higher.')
|
||||
|
||||
|
||||
###########################################################################
|
||||
# Find a useful Wget+Lua executable.
|
||||
#
|
||||
# WGET_AT will be set to the first path that
|
||||
# 1. does not crash with --version, and
|
||||
# 2. prints the required version string
|
||||
|
||||
class HigherVersion:
|
||||
def __init__(self, expression, min_version):
|
||||
self._expression = re.compile(expression)
|
||||
self._min_version = min_version
|
||||
|
||||
def search(self, text):
|
||||
for result in self._expression.findall(text):
|
||||
if result >= self._min_version:
|
||||
print('Found version {}.'.format(result))
|
||||
return True
|
||||
|
||||
|
||||
WGET_AT = find_executable(
|
||||
'Wget+AT',
|
||||
HigherVersion(
|
||||
r'(GNU Wget 1\.[0-9]{2}\.[0-9]{1}-at\.[0-9]{8}\.[0-9]{2})[^0-9a-zA-Z\.-_]',
|
||||
'GNU Wget 1.21.3-at.20241119.01'
|
||||
),
|
||||
[
|
||||
'./wget-at',
|
||||
'/home/warrior/data/wget-at-nss'
|
||||
]
|
||||
)
|
||||
|
||||
if not WGET_AT:
|
||||
raise Exception('No usable Wget+At found.')
|
||||
|
||||
|
||||
###########################################################################
|
||||
# The version number of this pipeline definition.
|
||||
#
|
||||
# Update this each time you make a non-cosmetic change.
|
||||
# It will be added to the WARC files and reported to the tracker.
|
||||
VERSION = '20260629.01'
|
||||
TRACKER_ID = 'blice'
|
||||
TRACKER_HOST = 'legacy-api.arpa.li'
|
||||
MULTI_ITEM_SIZE = 100
|
||||
|
||||
|
||||
###########################################################################
|
||||
# This section defines project-specific tasks.
|
||||
#
|
||||
# Simple tasks (tasks that do not need any concurrency) are based on the
|
||||
# SimpleTask class and have a process(item) method that is called for
|
||||
# each item.
|
||||
class CheckIP(SimpleTask):
|
||||
def __init__(self):
|
||||
SimpleTask.__init__(self, 'CheckIP')
|
||||
self._counter = 0
|
||||
|
||||
def process(self, item):
|
||||
# NEW for 2014! Check if we are behind firewall/proxy
|
||||
|
||||
if self._counter <= 0:
|
||||
item.log_output('Checking IP address.')
|
||||
ip_set = set()
|
||||
|
||||
ip_set.add(socket.gethostbyname('twitter.com'))
|
||||
#ip_set.add(socket.gethostbyname('facebook.com'))
|
||||
ip_set.add(socket.gethostbyname('youtube.com'))
|
||||
ip_set.add(socket.gethostbyname('microsoft.com'))
|
||||
ip_set.add(socket.gethostbyname('icanhas.cheezburger.com'))
|
||||
ip_set.add(socket.gethostbyname('archiveteam.org'))
|
||||
|
||||
if len(ip_set) != 5:
|
||||
item.log_output('Got IP addresses: {0}'.format(ip_set))
|
||||
item.log_output(
|
||||
'Are you behind a firewall/proxy? That is a big no-no!')
|
||||
raise Exception(
|
||||
'Are you behind a firewall/proxy? That is a big no-no!')
|
||||
|
||||
# Check only occasionally
|
||||
if self._counter <= 0:
|
||||
self._counter = 10
|
||||
else:
|
||||
self._counter -= 1
|
||||
|
||||
|
||||
class PrepareDirectories(SimpleTask):
|
||||
def __init__(self, warc_prefix):
|
||||
SimpleTask.__init__(self, 'PrepareDirectories')
|
||||
self.warc_prefix = warc_prefix
|
||||
|
||||
def process(self, item):
|
||||
item_name = item['item_name']
|
||||
item_name_hash = hashlib.sha1(item_name.encode('utf8')).hexdigest()
|
||||
escaped_item_name = item_name_hash
|
||||
dirname = '/'.join((item['data_dir'], escaped_item_name))
|
||||
|
||||
if os.path.isdir(dirname):
|
||||
shutil.rmtree(dirname)
|
||||
|
||||
os.makedirs(dirname)
|
||||
|
||||
item['item_dir'] = dirname
|
||||
item['warc_file_base'] = '-'.join([
|
||||
self.warc_prefix,
|
||||
item_name_hash,
|
||||
time.strftime('%Y%m%d-%H%M%S')
|
||||
])
|
||||
|
||||
open('%(item_dir)s/%(warc_file_base)s.warc.gz' % item, 'w').close()
|
||||
open('%(item_dir)s/%(warc_file_base)s_data.txt' % item, 'w').close()
|
||||
|
||||
|
||||
class MoveFiles(SimpleTask):
|
||||
def __init__(self):
|
||||
SimpleTask.__init__(self, 'MoveFiles')
|
||||
|
||||
def process(self, item):
|
||||
os.rename('%(item_dir)s/%(warc_file_base)s.warc.gz' % item,
|
||||
'%(data_dir)s/%(warc_file_base)s.warc.gz' % item)
|
||||
os.rename('%(item_dir)s/%(warc_file_base)s_data.txt' % item,
|
||||
'%(data_dir)s/%(warc_file_base)s_data.txt' % item)
|
||||
|
||||
shutil.rmtree('%(item_dir)s' % item)
|
||||
|
||||
|
||||
def normalize_string(s):
|
||||
while True:
|
||||
temp = unquote(s).strip().lower()
|
||||
if temp == s:
|
||||
break
|
||||
s = temp
|
||||
return s
|
||||
|
||||
|
||||
class SetBadUrls(SimpleTask):
|
||||
def __init__(self):
|
||||
SimpleTask.__init__(self, 'SetBadUrls')
|
||||
|
||||
def process(self, item):
|
||||
item['item_name_original'] = item['item_name']
|
||||
items = item['item_name'].split('\0')
|
||||
items_lower = [normalize_string(s) for s in items]
|
||||
with open('%(item_dir)s/%(warc_file_base)s_bad-items.txt' % item, 'r') as f:
|
||||
for s in {
|
||||
normalize_string(s) for s in f
|
||||
}:
|
||||
index = items_lower.index(s)
|
||||
item.log_output('Item {} is aborted.'.format(s))
|
||||
items.pop(index)
|
||||
items_lower.pop(index)
|
||||
item['item_name'] = '\0'.join(items)
|
||||
|
||||
|
||||
class MaybeUploadWithTracker(UploadWithTracker):
|
||||
def enqueue(self, item):
|
||||
if len(item['item_name']) == 0:
|
||||
item.log_output('Skipping UploadWithTracker.')
|
||||
return self.complete_item(item)
|
||||
return super(UploadWithTracker, self).enqueue(item)
|
||||
|
||||
|
||||
class MaybeSendDoneToTracker(SendDoneToTracker):
|
||||
def enqueue(self, item):
|
||||
if len(item['item_name']) == 0:
|
||||
return self.complete_item(item)
|
||||
return super(MaybeSendDoneToTracker, self).enqueue(item)
|
||||
|
||||
|
||||
def get_hash(filename):
|
||||
with open(filename, 'rb') as in_file:
|
||||
return hashlib.sha1(in_file.read()).hexdigest()
|
||||
|
||||
CWD = os.getcwd()
|
||||
PIPELINE_SHA1 = get_hash(os.path.join(CWD, 'pipeline.py'))
|
||||
LUA_SHA1 = get_hash(os.path.join(CWD, 'blice.lua'))
|
||||
|
||||
def stats_id_function(item):
|
||||
d = {
|
||||
'pipeline_hash': PIPELINE_SHA1,
|
||||
'lua_hash': LUA_SHA1,
|
||||
'python_version': sys.version,
|
||||
}
|
||||
|
||||
return d
|
||||
|
||||
|
||||
class WgetArgs(object):
|
||||
def realize(self, item):
|
||||
wget_args = [
|
||||
WGET_AT,
|
||||
'-nv',
|
||||
'--no-cookies',
|
||||
'--host-lookups', 'dns',
|
||||
'--hosts-file', '/dev/null',
|
||||
'--resolvconf-file', '/dev/null',
|
||||
'--dns-servers', '9.9.9.10,149.112.112.10,2620:fe::10,2620:fe::fe:10',
|
||||
'--reject-reserved-subnets',
|
||||
#'--prefer-family', ('IPv4' if 'PREFER_IPV4' in os.environ else 'IPv6'),
|
||||
'--content-on-error',
|
||||
'--lua-script', 'blice.lua',
|
||||
'-o', ItemInterpolation('%(item_dir)s/wget.log'),
|
||||
'--output-document', ItemInterpolation('%(item_dir)s/wget.tmp'),
|
||||
'--truncate-output',
|
||||
'-e', 'robots=off',
|
||||
'--recursive', '--level=inf',
|
||||
'--no-parent',
|
||||
'--page-requisites',
|
||||
'--timeout', '30',
|
||||
'--connect-timeout', '1',
|
||||
'--tries', 'inf',
|
||||
'--domains', ','.join([
|
||||
'blice.co.kr',
|
||||
'www.blice.co.kr',
|
||||
'cds.blice.co.kr'
|
||||
]),
|
||||
'--span-hosts',
|
||||
'--waitretry', '30',
|
||||
'--warc-file', ItemInterpolation('%(item_dir)s/%(warc_file_base)s'),
|
||||
'--warc-header', 'operator: Archive Team',
|
||||
'--warc-header', 'x-wget-at-project-version: ' + VERSION,
|
||||
'--warc-header', 'x-wget-at-project-name: ' + TRACKER_ID,
|
||||
'--warc-dedup-url-agnostic',
|
||||
'--impersonate', 'firefox148-h1',
|
||||
'--header', 'Accept-Encoding: identity'
|
||||
]
|
||||
|
||||
if '--concurrent' in sys.argv:
|
||||
concurrency = int(sys.argv[sys.argv.index('--concurrent')+1])
|
||||
else:
|
||||
concurrency = os.getenv('CONCURRENT_ITEMS')
|
||||
if concurrency is None:
|
||||
concurrency = 2
|
||||
item['concurrency'] = str(concurrency)
|
||||
|
||||
for item_name in item['item_name'].split('\0'):
|
||||
item_type, item_value = item_name.split(':', 1)
|
||||
wget_args.extend(['--warc-header', 'x-wget-at-project-item-name: '+item_name])
|
||||
if item_type == 'novel':
|
||||
wget_args.extend(['--warc-header', 'blice-novel: '+item_value])
|
||||
wget_args.append('https://blice.co.kr/web/detail.kt?novelId={}'.format(item_value))
|
||||
elif item_type == 'times':
|
||||
wget_args.extend(['--warc-header', 'blice-times: '+item_value])
|
||||
wget_args.append('https://blice.co.kr/web/viewer.kt?timesId={}'.format(item_value))
|
||||
elif item_type == 'event':
|
||||
wget_args.extend(['--warc-header', 'blice-event: '+item_value])
|
||||
wget_args.append('https://blice.co.kr/event/eventPage.kt?eventSeq={}'.format(item_value))
|
||||
elif item_type == 'document':
|
||||
wget_args.extend(['--warc-header', 'blice-document: '+item_value])
|
||||
wget_args.append('https://blice.co.kr/web/documents/detail.kt?documentSeq={}'.format(item_value))
|
||||
elif item_type == 'notice':
|
||||
wget_args.extend(['--warc-header', 'blice-notice: '+item_value])
|
||||
wget_args.append('https://blice.co.kr/web/support/noticeDetail.kt?noticeSeq={}'.format(item_value))
|
||||
elif item_type == 'user':
|
||||
wget_args.extend(['--warc-header', 'blice-user: '+item_value])
|
||||
wget_args.append('https://blice.co.kr/mw/retrieveSpecificUserProfile.kt?onlyPenYn=N&userType=USER&userNicknameSeq={}'.format(item_value))
|
||||
elif item_type == 'asset':
|
||||
url = 'https://' + item_value
|
||||
wget_args.extend(['--warc-header', 'blice-asset: '+url])
|
||||
wget_args.append(url)
|
||||
else:
|
||||
raise Exception('Unknown item')
|
||||
|
||||
item['item_name_newline'] = item['item_name'].replace('\0', '\n')
|
||||
|
||||
if 'bind_address' in globals():
|
||||
wget_args.extend(['--bind-address', globals()['bind_address']])
|
||||
print('')
|
||||
print('*** Wget will bind address at {0} ***'.format(
|
||||
globals()['bind_address']))
|
||||
print('')
|
||||
|
||||
return realize(wget_args, item)
|
||||
|
||||
|
||||
###########################################################################
|
||||
# Initialize the project.
|
||||
#
|
||||
# This will be shown in the warrior management panel. The logo should not
|
||||
# be too big. The deadline is optional.
|
||||
project = Project(
|
||||
title=TRACKER_ID,
|
||||
project_html='''
|
||||
<img class="project-logo" alt="Project logo" src="https://wiki.archiveteam.org/images/4/4b/Blice-icon.png" height="50px" title=""/>
|
||||
<h2>Blice <span class="links"><a href="https://blice.co.kr/">Website</a> · <a href="https://tracker.archiveteam.org/blice/">Leaderboard</a> · <a href="https://wiki.archiveteam.org/index.php/Blice">Wiki</a></span></h2>
|
||||
<p>Archiving Blice.</p>
|
||||
''',
|
||||
utc_deadline=datetime.datetime(2026, 6, 30, 15, 0, 0)
|
||||
)
|
||||
|
||||
|
||||
pipeline = Pipeline(
|
||||
CheckIP(),
|
||||
GetItemFromTracker('https://{}/{}/multi={}/'
|
||||
.format(TRACKER_HOST, TRACKER_ID, MULTI_ITEM_SIZE),
|
||||
downloader, VERSION),
|
||||
PrepareDirectories(warc_prefix=TRACKER_ID),
|
||||
WgetDownload(
|
||||
WgetArgs(),
|
||||
max_tries=1,
|
||||
accept_on_exit_code=[0, 4, 8],
|
||||
env={
|
||||
'item_dir': ItemValue('item_dir'),
|
||||
'item_names': ItemValue('item_name_newline'),
|
||||
'warc_file_base': ItemValue('warc_file_base'),
|
||||
'concurrency': ItemValue('concurrency')
|
||||
}
|
||||
),
|
||||
SetBadUrls(),
|
||||
PrepareStatsForTracker(
|
||||
defaults={'downloader': downloader, 'version': VERSION},
|
||||
file_groups={
|
||||
'data': [
|
||||
ItemInterpolation('%(item_dir)s/%(warc_file_base)s.warc.gz')
|
||||
]
|
||||
},
|
||||
id_function=stats_id_function,
|
||||
),
|
||||
MoveFiles(),
|
||||
LimitConcurrent(NumberConfigValue(min=1, max=20, default='20',
|
||||
name='shared:rsync_threads', title='Rsync threads',
|
||||
description='The maximum number of concurrent uploads.'),
|
||||
MaybeUploadWithTracker(
|
||||
'https://%s/%s' % (TRACKER_HOST, TRACKER_ID),
|
||||
downloader=downloader,
|
||||
version=VERSION,
|
||||
files=[
|
||||
ItemInterpolation('%(data_dir)s/%(warc_file_base)s.warc.gz'),
|
||||
ItemInterpolation('%(data_dir)s/%(warc_file_base)s_data.txt')
|
||||
],
|
||||
rsync_target_source_path=ItemInterpolation('%(data_dir)s/'),
|
||||
rsync_extra_args=[
|
||||
'--recursive',
|
||||
'--min-size', '1',
|
||||
'--no-compress',
|
||||
'--compress-level', '0'
|
||||
]
|
||||
),
|
||||
),
|
||||
MaybeSendDoneToTracker(
|
||||
tracker_url='https://%s/%s' % (TRACKER_HOST, TRACKER_ID),
|
||||
stats=ItemValue('stats')
|
||||
)
|
||||
)
|
||||
Reference in New Issue
Block a user