ArchiveBox/archivebox/archive_methods.py

655 lines
22 KiB
Python

import os
from functools import wraps
from collections import defaultdict
from datetime import datetime
from index import (
parse_json_link_index,
write_link_index,
patch_index_title_hack,
)
from config import (
CURL_BINARY,
GIT_BINARY,
WGET_BINARY,
YOUTUBEDL_BINARY,
CHROME_BINARY,
FETCH_FAVICON,
FETCH_TITLE,
FETCH_WGET,
FETCH_WGET_REQUISITES,
FETCH_PDF,
FETCH_SCREENSHOT,
FETCH_DOM,
FETCH_WARC,
FETCH_GIT,
FETCH_MEDIA,
RESOLUTION,
CHECK_SSL_VALIDITY,
SUBMIT_ARCHIVE_DOT_ORG,
COOKIES_FILE,
WGET_USER_AGENT,
CHROME_USER_DATA_DIR,
CHROME_HEADLESS,
CHROME_SANDBOX,
TIMEOUT,
MEDIA_TIMEOUT,
ANSI,
ARCHIVE_DIR,
GIT_DOMAINS,
GIT_SHA,
)
from util import (
domain,
without_query,
without_fragment,
fetch_page_title,
progress,
chmod_file,
pretty_path,
print_error_hints,
check_link_structure,
wget_output_path,
run, PIPE, DEVNULL,
)
_RESULTS_TOTALS = { # globals are bad, mmkay
'skipped': 0,
'succeded': 0,
'failed': 0,
}
def load_link_index(link_dir, link):
"""check for an existing link archive in the given directory,
and load+merge it into the given link dict
"""
is_new = not os.path.exists(link_dir)
if is_new:
os.makedirs(link_dir)
else:
link = {
**parse_json_link_index(link_dir),
**link,
}
check_link_structure(link)
print_link_status_line(link_dir, link, is_new)
return link
def archive_link(link_dir, link, overwrite=True):
"""download the DOM, PDF, and a screenshot into a folder named after the link's timestamp"""
ARCHIVE_METHODS = (
(FETCH_TITLE, fetch_title),
(FETCH_FAVICON, fetch_favicon),
(FETCH_WGET, fetch_wget),
(FETCH_PDF, fetch_pdf),
(FETCH_SCREENSHOT, fetch_screenshot),
(FETCH_DOM, fetch_dom),
(FETCH_GIT, fetch_git),
(FETCH_MEDIA, fetch_media),
(SUBMIT_ARCHIVE_DOT_ORG, archive_dot_org),
)
active_methods = [method for toggle, method in ARCHIVE_METHODS if toggle]
try:
link = load_link_index(link_dir, link)
for archive_method in active_methods:
archive_method(link_dir, link, overwrite=overwrite)
write_link_index(link_dir, link)
except Exception as err:
print(' ! Failed to archive link: {}: {}'.format(err.__class__.__name__, err))
return link
def print_link_status_line(link_dir, link, is_new):
print('[{symbol_color}{symbol}{reset}] [{now}] "{title}"\n {blue}{url}{reset}'.format(
symbol='+' if is_new else '*',
symbol_color=ANSI['green' if is_new else 'black'],
now=datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
**{**link, 'title': link['title'] or link['url']},
**ANSI,
))
print(' > {}{}'.format(pretty_path(link_dir), ' (new)' if is_new else ''))
# if link['type']:
# print(' i {}'.format(link['type']))
def attach_result_to_link(method):
"""
Instead of returning a result={output:'...', status:'success'} object,
attach that result to the links's history & latest fields, then return
the updated link object.
"""
def decorator(fetch_func):
@wraps(fetch_func)
def timed_fetch_func(link_dir, link, overwrite=False, **kwargs):
# initialize methods and history json field on link
link['latest'] = link.get('latest') or {}
link['latest'][method] = link['latest'].get(method) or None
link['history'] = link.get('history') or {}
link['history'][method] = link['history'].get(method) or []
start_ts = datetime.now().timestamp()
# if a valid method output is already present, dont run the fetch function
if link['latest'][method] and not overwrite:
print('{}'.format(method))
result = None
else:
print(' > {}'.format(method))
result = fetch_func(link_dir, link, **kwargs)
end_ts = datetime.now().timestamp()
duration = str(end_ts * 1000 - start_ts * 1000).split('.')[0]
# append a history item recording fail/success
history_entry = {
'timestamp': str(start_ts).split('.')[0],
}
if result is None:
history_entry['status'] = 'skipped'
elif isinstance(result.get('output'), Exception):
history_entry['status'] = 'failed'
history_entry['duration'] = duration
history_entry.update(result or {})
link['history'][method].append(history_entry)
else:
history_entry['status'] = 'succeded'
history_entry['duration'] = duration
history_entry.update(result or {})
link['history'][method].append(history_entry)
link['latest'][method] = result['output']
_RESULTS_TOTALS[history_entry['status']] += 1
return link
return timed_fetch_func
return decorator
@attach_result_to_link('wget')
def fetch_wget(link_dir, link, requisites=FETCH_WGET_REQUISITES, warc=FETCH_WARC, timeout=TIMEOUT):
"""download full site using wget"""
domain_dir = os.path.join(link_dir, domain(link['url']))
existing_file = wget_output_path(link)
if os.path.exists(domain_dir) and existing_file:
return {'output': existing_file, 'status': 'skipped'}
if warc:
warc_dir = os.path.join(link_dir, 'warc')
os.makedirs(warc_dir, exist_ok=True)
warc_path = os.path.join('warc', str(int(datetime.now().timestamp())))
# WGET CLI Docs: https://www.gnu.org/software/wget/manual/wget.html
CMD = [
WGET_BINARY,
# '--server-response', # print headers for better error parsing
'--no-verbose',
'--adjust-extension',
'--convert-links',
'--force-directories',
'--backup-converted',
'--span-hosts',
'--no-parent',
'-e', 'robots=off',
'--restrict-file-names=unix',
'--timeout={}'.format(timeout),
*(() if warc else ('--timestamping',)),
*(('--warc-file={}'.format(warc_path),) if warc else ()),
*(('--page-requisites',) if FETCH_WGET_REQUISITES else ()),
*(('--user-agent={}'.format(WGET_USER_AGENT),) if WGET_USER_AGENT else ()),
*(('--load-cookies', COOKIES_FILE) if COOKIES_FILE else ()),
*((() if CHECK_SSL_VALIDITY else ('--no-check-certificate', '--no-hsts'))),
link['url'],
]
end = progress(timeout, prefix=' ')
try:
result = run(CMD, stdout=PIPE, stderr=PIPE, cwd=link_dir, timeout=timeout)
end()
output = wget_output_path(link, look_in=domain_dir)
output_tail = [' ' + line for line in (result.stdout + result.stderr).decode().rsplit('\n', 3)[-3:] if line.strip()]
# parse out number of files downloaded from "Downloaded: 76 files, 4.0M in 1.6s (2.52 MB/s)"
files_downloaded = (
int(output_tail[-1].strip().split(' ', 2)[1] or 0)
if 'Downloaded:' in output_tail[-1]
else 0
)
# Check for common failure cases
if result.returncode > 0 and files_downloaded < 1:
print(' Got wget response code {}:'.format(result.returncode))
print('\n'.join(output_tail))
if b'403: Forbidden' in result.stderr:
raise Exception('403 Forbidden (try changing WGET_USER_AGENT)')
if b'404: Not Found' in result.stderr:
raise Exception('404 Not Found')
if b'ERROR 500: Internal Server Error' in result.stderr:
raise Exception('500 Internal Server Error')
raise Exception('Got an error from the server')
except Exception as e:
end()
output = e
print_error_hints(cmd=CMD, pwd=link_dir, err=e)
return {
'cmd': CMD,
'output': output,
}
@attach_result_to_link('pdf')
def fetch_pdf(link_dir, link, timeout=TIMEOUT, user_data_dir=CHROME_USER_DATA_DIR):
"""print PDF of site to file using chrome --headless"""
if link['type'] in ('PDF', 'image'):
return {'output': wget_output_path(link)}
if os.path.exists(os.path.join(link_dir, 'output.pdf')):
return {'output': 'output.pdf', 'status': 'skipped'}
CMD = [
*chrome_headless(user_data_dir=user_data_dir),
'--print-to-pdf',
'--hide-scrollbars',
'--timeout={}'.format((timeout) * 1000),
*(() if CHECK_SSL_VALIDITY else ('--disable-web-security', '--ignore-certificate-errors')),
link['url']
]
end = progress(timeout, prefix=' ')
try:
result = run(CMD, stdout=PIPE, stderr=PIPE, cwd=link_dir, timeout=timeout)
end()
if result.returncode:
print(' ', (result.stderr or result.stdout).decode())
raise Exception('Failed to print PDF')
chmod_file('output.pdf', cwd=link_dir)
output = 'output.pdf'
except Exception as e:
end()
output = e
print_error_hints(cmd=CMD, pwd=link_dir, err=e)
return {
'cmd': CMD,
'output': output,
}
@attach_result_to_link('screenshot')
def fetch_screenshot(link_dir, link, timeout=TIMEOUT, user_data_dir=CHROME_USER_DATA_DIR, resolution=RESOLUTION):
"""take screenshot of site using chrome --headless"""
if link['type'] in ('PDF', 'image'):
return {'output': wget_output_path(link)}
if os.path.exists(os.path.join(link_dir, 'screenshot.png')):
return {'output': 'screenshot.png', 'status': 'skipped'}
CMD = [
*chrome_headless(user_data_dir=user_data_dir),
'--screenshot',
'--window-size={}'.format(resolution),
'--hide-scrollbars',
'--timeout={}'.format((timeout) * 1000),
*(() if CHECK_SSL_VALIDITY else ('--disable-web-security', '--ignore-certificate-errors')),
# '--full-page', # TODO: make this actually work using ./bin/screenshot fullPage: true
link['url'],
]
end = progress(timeout, prefix=' ')
try:
result = run(CMD, stdout=PIPE, stderr=PIPE, cwd=link_dir, timeout=timeout)
end()
if result.returncode:
print(' ', (result.stderr or result.stdout).decode())
raise Exception('Failed to take screenshot')
chmod_file('screenshot.png', cwd=link_dir)
output = 'screenshot.png'
except Exception as e:
end()
output = e
print_error_hints(cmd=CMD, pwd=link_dir, err=e)
return {
'cmd': CMD,
'output': output,
}
@attach_result_to_link('dom')
def fetch_dom(link_dir, link, timeout=TIMEOUT, user_data_dir=CHROME_USER_DATA_DIR):
"""print HTML of site to file using chrome --dump-html"""
if link['type'] in ('PDF', 'image'):
return {'output': wget_output_path(link)}
output_path = os.path.join(link_dir, 'output.html')
if os.path.exists(output_path):
return {'output': 'output.html', 'status': 'skipped'}
CMD = [
*chrome_headless(user_data_dir=user_data_dir),
'--dump-dom',
'--timeout={}'.format((timeout) * 1000),
link['url']
]
end = progress(timeout, prefix=' ')
try:
with open(output_path, 'w+') as f:
result = run(CMD, stdout=f, stderr=PIPE, cwd=link_dir, timeout=timeout)
end()
if result.returncode:
print(' ', (result.stderr).decode())
raise Exception('Failed to fetch DOM')
chmod_file('output.html', cwd=link_dir)
output = 'output.html'
except Exception as e:
end()
output = e
print_error_hints(cmd=CMD, pwd=link_dir, err=e)
return {
'cmd': CMD,
'output': output,
}
def parse_archive_dot_org_response(response):
# Parse archive.org response headers
headers = defaultdict(list)
# lowercase all the header names and store in dict
for header in response.splitlines():
if b':' not in header or not header.strip():
continue
name, val = header.decode().split(':', 1)
headers[name.lower().strip()].append(val.strip())
# Get successful archive url in "content-location" header or any errors
content_location = headers['content-location']
errors = headers['x-archive-wayback-runtime-error']
return content_location, errors
@attach_result_to_link('archive_org')
def archive_dot_org(link_dir, link, timeout=TIMEOUT):
"""submit site to archive.org for archiving via their service, save returned archive url"""
output = 'archive.org.txt'
archive_org_url = None
path = os.path.join(link_dir, output)
if os.path.exists(path):
archive_org_url = open(path, 'r').read().strip()
return {'output': archive_org_url, 'status': 'skipped'}
submit_url = 'https://web.archive.org/save/{}'.format(link['url'])
CMD = [
CURL_BINARY,
'--location',
'--head',
'--user-agent', 'ArchiveBox/{} (+https://github.com/pirate/ArchiveBox/)'.format(GIT_SHA), # be nice to the Archive.org people and show them where all this ArchiveBox traffic is coming from
'--max-time', str(timeout),
*(() if CHECK_SSL_VALIDITY else ('--insecure',)),
submit_url,
]
end = progress(timeout, prefix=' ')
try:
result = run(CMD, stdout=PIPE, stderr=DEVNULL, cwd=link_dir, timeout=timeout)
end()
content_location, errors = parse_archive_dot_org_response(result.stdout)
if content_location:
archive_org_url = 'https://web.archive.org{}'.format(content_location[0])
elif len(errors) == 1 and 'RobotAccessControlException' in errors[0]:
archive_org_url = None
# raise Exception('Archive.org denied by {}/robots.txt'.format(domain(link['url'])))
elif errors:
raise Exception(', '.join(errors))
else:
raise Exception('Failed to find "content-location" URL header in Archive.org response.')
except Exception as e:
end()
output = e
print_error_hints(cmd=CMD, pwd=link_dir, err=e)
if not isinstance(output, Exception):
# instead of writing None when archive.org rejects the url write the
# url to resubmit it to archive.org. This is so when the user visits
# the URL in person, it will attempt to re-archive it, and it'll show the
# nicer error message explaining why the url was rejected if it fails.
archive_org_url = archive_org_url or submit_url
with open(os.path.join(link_dir, output), 'w', encoding='utf-8') as f:
f.write(archive_org_url)
chmod_file('archive.org.txt', cwd=link_dir)
output = archive_org_url
return {
'cmd': CMD,
'output': output,
}
@attach_result_to_link('favicon')
def fetch_favicon(link_dir, link, timeout=TIMEOUT):
"""download site favicon from google's favicon api"""
output = 'favicon.ico'
if os.path.exists(os.path.join(link_dir, output)):
return {'output': output, 'status': 'skipped'}
CMD = [
CURL_BINARY,
'--max-time', str(timeout),
'--location',
'--output', output,
*(() if CHECK_SSL_VALIDITY else ('--insecure',)),
'https://www.google.com/s2/favicons?domain={}'.format(domain(link['url'])),
]
end = progress(timeout, prefix=' ')
try:
run(CMD, stdout=PIPE, stderr=PIPE, cwd=link_dir, timeout=timeout)
end()
chmod_file('favicon.ico', cwd=link_dir)
output = 'favicon.ico'
except Exception as e:
end()
output = e
print_error_hints(cmd=CMD, pwd=link_dir, err=e)
return {
'cmd': CMD,
'output': output,
}
@attach_result_to_link('title')
def fetch_title(link_dir, link, timeout=TIMEOUT):
"""try to guess the page's title from its content"""
# if link already has valid title, skip it
if link['title'] and not link['title'].lower().startswith('http'):
return {'output': link['title'], 'status': 'skipped'}
end = progress(timeout, prefix=' ')
try:
title = fetch_page_title(link['url'], timeout=timeout, progress=False)
end()
output = title
except Exception as e:
end()
print(' {}Failed: {} {}{}'.format(ANSI['red'], e.__class__.__name__, e, ANSI['reset']))
output = e
# titles should show up in the global index immediatley for better UX,
# do a hacky immediate replacement to add them in as we're archiving
# TODO: figure out how to do this without gnarly string replacement
if title:
link['title'] = title
patch_index_title_hack(link['url'], title)
return {
'cmd': 'fetch_page_title("{}")'.format(link['url']),
'output': output,
}
@attach_result_to_link('media')
def fetch_media(link_dir, link, timeout=MEDIA_TIMEOUT, overwrite=False):
"""Download playlists or individual video, audio, and subtitles using youtube-dl"""
# import ipdb; ipdb.set_trace()
output = os.path.join(link_dir, 'media')
already_done = os.path.exists(output) # and os.listdir(output)
if already_done and not overwrite:
return {'output': 'media', 'status': 'skipped'}
os.makedirs(output, exist_ok=True)
CMD = [
YOUTUBEDL_BINARY,
'--write-description',
'--write-info-json',
'--write-annotations',
'--yes-playlist',
'--write-thumbnail',
'--no-call-home',
'--no-check-certificate',
'--user-agent',
'--all-subs',
'--extract-audio',
'--keep-video',
'--ignore-errors',
'--geo-bypass',
'--audio-format', 'mp3',
'--audio-quality', '320K',
'--embed-thumbnail',
'--add-metadata',
*(() if CHECK_SSL_VALIDITY else ('--no-check-certificate',)),
link['url'],
]
end = progress(timeout, prefix=' ')
try:
result = run(CMD, stdout=PIPE, stderr=PIPE, cwd=output, timeout=timeout + 1)
chmod_file('media', cwd=link_dir)
output = 'media'
end()
if result.returncode:
if (b'ERROR: Unsupported URL' in result.stderr
or b'HTTP Error 404' in result.stderr
or b'HTTP Error 403' in result.stderr
or b'URL could be a direct video link' in result.stderr
or b'Unable to extract container ID' in result.stderr):
# These happen too frequently on non-media pages to warrant printing to console
pass
else:
print(' got youtubedl response code {}:'.format(result.returncode))
print(result.stderr)
raise Exception('Failed to download media')
except Exception as e:
end()
output = e
print_error_hints(cmd=CMD, pwd=link_dir, err=e)
return {
'cmd': CMD,
'output': output,
}
@attach_result_to_link('git')
def fetch_git(link_dir, link, timeout=TIMEOUT):
"""download full site using git"""
url_is_clonable = (
domain(link['url']) in GIT_DOMAINS
or link['url'].endswith('.git')
or link['type'] == 'git'
)
if not url_is_clonable:
return {'output': None, 'status': 'skipped'}
git_dir = os.path.join(link_dir, 'git')
if os.path.exists(git_dir):
return {'output': 'git', 'status': 'skipped'}
os.makedirs(git_dir, exist_ok=True)
output = 'git'
CMD = [
GIT_BINARY,
'clone',
'--mirror',
'--recursive',
*(() if CHECK_SSL_VALIDITY else ('-c', 'http.sslVerify=false')),
without_query(without_fragment(link['url'])),
]
end = progress(timeout, prefix=' ')
try:
result = run(CMD, stdout=PIPE, stderr=PIPE, cwd=git_dir, timeout=timeout + 1)
end()
if result.returncode == 128:
# ignore failed re-download when the folder already exists
pass
elif result.returncode > 0:
print(' got git response code {}:'.format(result.returncode))
raise Exception('Failed git download')
except Exception as e:
end()
output = e
print_error_hints(cmd=CMD, pwd=link_dir, err=e)
return {
'cmd': CMD,
'output': output,
}
def chrome_headless(binary=CHROME_BINARY, user_data_dir=CHROME_USER_DATA_DIR, headless=CHROME_HEADLESS, sandbox=CHROME_SANDBOX):
global USER_DATA_DIR
user_data_dir = user_data_dir or USER_DATA_DIR
cmd_args = [binary]
if headless:
cmd_args += ('--headless',)
if not sandbox:
# dont use GPU or sandbox when running inside docker container
cmd_args += ('--no-sandbox', '--disable-gpu')
# Find chrome user data directory
default_profile_paths = (
'~/.config/chromium',
'~/.config/google-chrome',
'~/.config/google-chrome-beta',
'~/.config/google-chrome-unstable',
'~/Library/Application Support/Chromium',
'~/Library/Application Support/Google/Chrome',
'~/Library/Application Support/Google/Chrome Canary',
'~/AppData/Local/Chromium/User Data',
'~/AppData/Local/Google/Chrome/User Data',
'~/AppData/Local/Google/Chrome SxS/User Data',
)
if user_data_dir:
cmd_args.append('--user-data-dir={}'.format(user_data_dir))
else:
for path in default_profile_paths:
full_path = os.path.expanduser(path)
if os.path.exists(full_path):
USER_DATA_DIR = full_path
cmd_args.append('--user-data-dir={}'.format(full_path))
break
return cmd_args
USER_DATA_DIR = CHROME_USER_DATA_DIR