Python－スクレイピングでフリーランス案件を探そう(ランサーズ編)

# ドメイン
la_domain = "https://www.lancers.jp/"

# デフォルトのクロール間隔
defalt_crawl_delay = 1

# ユーザーエージェント
user_agent = "Crawler/1.0.0"

# 除外するサプライヤー(ユーザー名で可)
la_except_users = []

import os
import urllib.robotparser


# robot.txtをパースする
def parse_robots(domain):
    robots_url = os.path.join(domain, 'robots.txt')
    rp = urllib.robotparser.RobotFileParser()
    rp.set_url(robots_url)
    rp.read()
    return rp

# 記述ルール
# 1. 検索したい語句はシングルクオーテーション(')か、ダブルクオーテーション(")で囲む
# 2. アンド検索をする場合は各語句をプラス記号(+)でつなぐ
#
# 記載例
# search_word = ["スクレイピング", "wordpress+SEO"]
search_words = ["スクレイピング", 'Python']

import re
import time

from bs4 import BeautifulSoup as bs
import requests

from config import config
from functions import parse_robots


# ランサーズのスクレイピング実行プログラム
def lancers(search_words):
    print("Start scraping for Lancers...")

    # 1) robots.txtの確認
    domain = config.cw_domain
    robot_parser = parse_robots(domain)
    crawl_delay = robot_parser.crawl_delay("*")
    if not crawl_delay:
        crawl_delay = config.defalt_crawl_delay

    # ヘッダー設定
    result = [{"service": "検索サービス", "word": "検索ワード", "id": "ID", "job_title": "案件名", "url": "URL",
               "job_type": "報酬体型", 'fee': '報酬', 'remaining_period': "残期間"}]
    # 2) 検索ワードごとにクローリングとスクレイピング
    for word in search_words:
        print("Start searching for '{}' in Lancers...".format(word))

        # クロール間隔の調整
        time.sleep(crawl_delay)

        # クローリングの実行
        html = la_crawling(word, robot_parser)

        # クロールできなかった場合(html=はNone)
        if not html:
            result.append({"service": "CrowWorks", "word": word, "id": "",
                           "job_title": '「{}」に該当する案件はありませんでした。'.format(word), "url": "", "job_type": "",
                           "fee": "", "remaining_period": ""})
            continue
        # スクレイピング
        result += la_scraping(html, word)
    return result


# クローリング
def la_crawling(word, robot_parser):
    headers = {'User-Agent': config.user_agent, "Accept-Language": "ja,en-US;q=0.9,en;q=0.8"}
    search_url = "https://www.lancers.jp/work/search?keyword={}&open=1&sort=started&work_rank%5B%5D=0&" \
                 "work_rank%5B%5D=1&work_rank%5B%5D=2&work_rank%5B%5D=3".format(word)

    # 1)robots.txtのチェック
    if not robot_parser.can_fetch('*', search_url):  # falseの場合
        return None
    res = requests.get(search_url, headers=headers)

    # 2)ステータスコードチェック
    if str(res.status_code) != "200":
        return None

    return res.text


# スクレイピング
def la_scraping(html, word):
    result = []
    soup = bs(html, 'lxml')

    # 1) 検索結果が存在しない場合
    if soup.find_all('div', class_="p-search__empty"):
        result.append({"service": "Lancers", "word": word, "id": "",
                       "job_title": '「{}」に該当する案件はありませんでした。'.format(word), "url": "", "job_type": "",
                       "fee": "", "remaining_period": ""})
        return result

    # 2) 各要素の検索
    num = 0
    for elem in soup.select(".c-media-list.c-media-list--forClient > div"):

        # 2-1) ユーザーチェック -> 該当する場合飛ばす
        user = user_info(elem.select('.c-avatar__note'))
        if user in config.la_except_users:
            print("Out of Supplier : {}".format(user))
            continue

        # 2-2) 該当する案件情報の取得
        num += 1
        job_title = get_title(elem)
        job_url = get_job_url(elem)
        job_type = elem.find(class_="c-badge__text").text
        fee = ''.join(elem.find(class_="c-media__job-price").text.split())
        remaining_period = get_period(elem)

        result.append({"service": "Lancers", "word": word, "id": num, "job_title": job_title, "url": job_url,
                       "job_type": job_type, "fee": fee, "remaining_period": remaining_period})

        print('id {}'.format(num))
    if not result:
        result.append({"service": "Lancers", "word": word, "id": "",
                       "job_title": '「{}」に該当する案件はありませんでした。'.format(word), "url": "", "job_type": "",
                       "fee": "", "remaining_period": ""})
    return result


# ユーザー情報を抽出する関数
def user_info(tag):
    if not tag:
        user = None
    else:
        user = tag[0].text
    return user


# ジョブタイトルを求める関数
def get_title(elem):
    tags = elem.find_all(class_="c-media__job-tags")
    if tags:
        tags_text = tags[0].text
        title_elem = elem.find_all(class_="c-media__title-inner")[0].text.split()
        title_list = [i for i in title_elem if i not in tags_text]
    else:
        title_list = elem.find_all(class_="c-media__title-inner")[0].text.split()
    job_title = ' '.join(title_list)
    return job_title


# ジョブURLを抽出する関数
def get_job_url(elem):
    url = elem.find(class_="c-media__title")['href']
    if re.search(r'https://', url):
        job_url = url
    else:
        job_url = config.la_domain + url
    return job_url


# 残期間を抽出する関数
def get_period(elem):
    period_tag = elem.find(class_="c-media__job-time__remaining")
    if period_tag:
        remaining_period = period_tag.text
    else:
        remaining_period = None
    return remaining_period


if __name__ == "__main__":
    import csv
    import os
    from search_words import search_words

    s = time.time()
    result = lancers(search_words)
    with open(os.path.join(os.environ['HOME'], 'Desktop/results.csv'), 'w') as csvfile:
        fieldnames = list(result[0].keys())
        writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
        writer.writerows(result)
    print("Total time: {}".format(time.time() - s))