-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathloader.py
More file actions
109 lines (91 loc) · 3.29 KB
/
Copy pathloader.py
File metadata and controls
109 lines (91 loc) · 3.29 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
import logging
import shutil
from os.path import splitext
from pathlib import Path
from urllib.parse import urljoin, urlparse
import requests
from bs4 import BeautifulSoup
from progress.bar import Bar
from page_loader import logging as lo
logger = logging.getLogger(__name__)
# @log_func
def download(url, destination=Path.cwd(), externals=False) -> list:
logger.info("Starting download")
file_name = get_filename_by_url(url)
resources_path = (Path(destination) / (file_name + "_files")).resolve()
if not resources_path.parent.exists():
logger.error("Could not find the path specified: %s" % destination)
raise lo.PathAccessError("Directory not found.")
resources_path.mkdir(exist_ok=True)
try:
page_contents = load_resources(
url, resources_path, externals=externals)
except lo.ConnectionError as e:
logger.debug("Removing resources directory. %s" % resources_path)
shutil.rmtree(resources_path)
raise e
page_path = Path(destination).resolve() / (file_name + '.html')
page_path.write_text(page_contents, 'utf-8')
return page_path, resources_path
@lo.log_func
def get_filename_by_url(url) -> str:
parsed = urlparse(url)
hostname = parsed.hostname if parsed.hostname else ''
file_name = "{}{}".format(hostname, parsed.path)\
.strip('/')\
.replace('/', '-')\
.replace('.', '-')
return file_name
tag_map = {
'img': 'src',
'script': 'src',
'link': 'href',
}
# TODO: img tag has two type of sources `src` and `data-src`
# find way to handle it.
@lo.log_func
def load_resources(url, local_dir, externals=False) -> str:
soup = get_parsed_html(url)
parsed_url = urlparse(url)
for tag, attr in tag_map.items():
for elem in soup.find_all(tag):
src = elem.get(attr)
if not src:
continue
elif not urlparse(src).netloc:
src = urljoin(url, src.strip('/'))
elif not externals and parsed_url.hostname not in src:
logger.debug("Skipping side resource {}".format(src))
continue
elem[attr] = save_resource(src, local_dir)
return str(soup)
def save_resource(src, destination) -> str:
bar = Bar(src, max=2)
base, ext = splitext(src)
file_path = Path(destination) / (get_filename_by_url(base) + ext)
try:
response = requests.get(src)
bar.next()
file_path.write_bytes(response.content)
bar.next()
except OSError as e:
# Skip unsuccessfull download
bar.finish()
logger.warning(
"Unable to save resource to destination path {}".format(file_path))
logger.debug(e, exc_info=True)
return src
bar.finish()
logger.debug("Resource {} was saved to {}".format(src, file_path))
res = (Path(destination.name) / file_path.name).as_posix()
logger.debug("Chaned link to local address {}".format(res))
return res
def get_parsed_html(url):
try:
soup = BeautifulSoup(requests.get(url).text, features='html.parser')
except requests.exceptions.RequestException as e:
logger.error(
"Failed to establish connection with: {}".format(url))
logger.debug(e, exc_info=True)
# raise lo.ConnectionError() from e
return soup