· 7 years ago · Sep 09, 2018, 07:32 AM
1# -*- coding: utf-8 -*-
2# author: push.thanh@gmail.com
3
4import os
5import re
6import sys
7import time
8import smtplib
9import logging
10import requests
11import multiprocessing
12from settings import (ROOT,
13 LINK)
14from bs4 import BeautifulSoup
15from datetime import datetime
16from RedisQueue import RedisQueue
17from base import SeekScraperBase
18
19
20class SeekFullContentScraper(SeekScraperBase):
21 ''' A scraper to look for full description of jobs '''
22
23 def __init__(self):
24 super().__init__()
25 self.rqueue = RedisQueue('seek_jobs_queue')
26 self.exception_queue = RedisQueue('seek_exceptions_queuetest')
27 self.proxies = self.get_proxies()
28 self.headers = self.get_headers()
29
30 # self._init_log()
31
32 def _init_log(self):
33 logging.basicConfig(
34 filename="./logs/content_scrape_log_{}.log".format(
35 datetime.now().strftime("%Y-%m-%d %H.%M")),
36 format="%(levelname)s:%(message)s",
37 level=logging.DEBUG)
38
39 self.log = logging.getLogger(__name__)
40
41 def create_exception_table(self):
42 sql = """
43 CREATE EXTENSION IF NOT EXISTS "uuid-ossp";
44 CREATE TABLE IF NOT EXISTS jobs.seekjobsexception
45 (
46 id uuid PRIMARY KEY DEFAULT uuid_generate_v4(),
47 jobid VARCHAR NOT NULL,
48 scraped_at TIMESTAMP,
49 success BOOLEAN NOT NULL
50 );
51 """
52 with self.cursor() as cur:
53 cur.execute(sql)
54
55 def to_exception_tabel(self, jobid, time):
56 ''' Store jobid that jd not found'''
57 sql = """
58 INSERT INTO jobs.seekjobsexception (jobid, scraped_at, success)
59 VALUES ('{}', '{}', FALSE)
60 """.format(jobid, time)
61 with self.cursor() as cur:
62 cur.execute(sql)
63
64 def to_sql(self, jobid, content):
65 ''' Save jd to db'''
66 sql = """
67 UPDATE jobs.seekjobs
68 SET jd = '{}'
69 WHERE (info ->> 'jobid') = '{}';
70 """.format(content, jobid)
71
72 with self.cursor() as cur:
73 cur.execute(sql)
74
75 def jobid_scraped(self, jobid):
76 sql = """
77 SELECT jd FROM jobs.seekjobs
78 WHERE (info ->> 'jobid') = '{}'
79 AND jd IS NOT NULL;
80 """.format(jobid)
81 result = self.query_one(sql)
82 if result:
83 print("scraped: {}".format(jobid))
84 return True
85 return False
86
87 def is_job_expired(self, soup):
88 content = soup.find('div', attrs={'data-automation': 'expiredJobPage'})
89 if content:
90 return True
91 return False
92
93 def page_has_content(self, soup):
94 content = soup.find('div', class_='job-template__wrapper')
95 if content:
96 return True
97 return False
98
99 def scrape_job_content(self, jobid):
100
101 if self.jobid_scraped(jobid):
102 return
103 done = False
104 while not done:
105 try:
106 url = "".join(["https://www.seek.com.au/job/", jobid])
107 page = requests.get(url, headers=self.headers,
108 proxies=self.proxies)
109 done = False
110 status = page.status_code
111
112 if 300 <= status < 400:
113 self.reset_proxy_pool()
114 self.proxies = self.get_proxies()
115 self.headers = self.get_headers()
116
117 elif status == 200:
118 soup = BeautifulSoup(page.content.decode('utf-8',
119 'ignore'),
120 'html.parser')
121 if not self.is_job_expired(soup):
122 content = soup.find('div', class_="templatetext")
123 print("\n--\n{}".format(content))
124 if content:
125 s = content.findChildren().replace(r"'", r"''")
126 jd = re.sub(r'\n\s*\n', r'\n\n',
127 s.strip(), flags=re.M)
128
129 self.to_sql(jobid, jd)
130 print("saved: {}".format(jobid))
131 done = True
132 else:
133 self.to_exception_tabel(jobid, self.NOW)
134 done = True
135
136 except requests.exceptions.SSLError as s:
137 self.proxies = self.get_proxies()
138 self.log.debug("-SLLError: {} \n".format(s))
139
140 except requests.exceptions.ProxyError as p:
141 self.log.debug("-Proxy failed: {} \n".format(p))
142 self.proxies = self.get_proxies()
143
144 except requests.exceptions.RequestException as e:
145 time.sleep(180)
146 self.log.debug("-Request failed: {} \n".format(e))
147
148 except KeyboardInterrupt:
149 self.log.debug(
150 "-Keyboard Interrupted: wait putting jobid back to queue")
151 break
152
153 def scraper(self):
154 print("I'm a scraper")
155
156 self.connect_db()
157 info_scraper_finished = 0
158 while True:
159 jobid = self.rqueue.pop(timeout=10)
160 if jobid:
161 if jobid == "Done":
162 info_scraper_finished += 1
163 else:
164 self.scrape_job_content(jobid.decode("utf-8"))
165 else:
166 time.sleep(3)
167
168 if info_scraper_finished == len(LINK):
169 if self.rqueue.empty():
170 self.exception_queue.put("Done")
171 break
172
173 def run(self, slaves=10):
174 ''' create a pool of workers to scrape '''
175
176 workers = []
177 for i in range(slaves):
178 w = multiprocessing.Process(target=self.scraper)
179 w.start()
180 workers.append(w)
181 print("Start #", i)
182
183 for w in workers:
184 w.join()
185
186
187if __name__ == '__main__':
188 s = SeekFullContentScraper()
189 s.run()