※ 開発中ですが、以下のロゴのリンクから、ご利用可能です。

先日の記事では、不動産の物件情報を収集しました。
不動産物件情報の掲載ページには、販売する不動産会社の名称も記載されていますが、
フランチャイズチェーン等の場合、宅建業者の免許番号と社名が一致しないケースがあるようです。

そこで、国交省が提供する「宅地建物取引業者検索システム」で不動産会社情報を収集することにしました。

download_and_save_master() で、このシステムにある不動産会社情報を一括抽出 & DB登録することもできますし、
find_licence_def(licence_no) で、免許番号を検索し、該当する不動産会社情報を抽出 & DB登録することもできます。

ただし?、「宅地建物取引業者検索システム」は、元々、サーバのリソースが潤沢でない為が、動作が固まります。
また、なぜか、BeautifulSoup では動作しなかった為、seleniumを使用しています。

db table

CREATE TABLE IF NOT EXISTS real_estate_shop (
government              varchar(16),
licence                 varchar(16),
shop                    varchar(64),
primary key(government,licence)
);
COMMENT ON COLUMN real_estate_shop.government
     IS '例; 国土交通大臣、東京都';

python class

#!python
# -*- coding: utf-8 -*-

from selenium.webdriver.common.by  import By
from selenium.webdriver.support.ui import Select
from util.db import Db

import appbase
import os
import jaconv        # pip install jaconv
import re
import time
import unicodedata   # 標準module

# 国土交通省の宅地建物取引業者検索システム用
url_base = "https://etsuran.mlit.go.jp/TAKKEN/takkenKensaku.do"
api_args_tmpl = \
    "CMD=search&sv_rdoSelect=1&sv_rdoSelectJoken=1&sv_rdoSelectSort=1&"+\
    "sv_kenCode=&sv_choice=1&sv_sortValue=1&sv_pageListNo1=1&sv_pageListNo2=1&"+\
    "sv_comNameKanaOnly=&sv_comNameKanjiOnly=&sv_licenseNoKbn=&"+\
    "sv_licenseNoFrom=%s&sv_licenseNoTo=%s&sv_licenseNo=&sv_dispCount=50&"+\
    "sv_dispPage=1&resultCount=5&pageCount=1&dispPage=1&caller=TK&rdoSelect=1&"+\
    "comNameKanaOnly=&comNameKanjiOnly=&rdoSelectJoken=1&licenseNoKbn=&"+\
    "licenseNoFrom=%s&licenseNoTo=%s&choice=1&kenCode=&sortValue=1&"+\
    "rdoSelectSort=1&dispCount=50&pageListNo1=1&pageListNo2=1"

insert_cols = ["government","licence","shop"]

logger = appbase.AppBase().get_logger()

class MlitRealEstateShopService(appbase.AppBase):

    def __init__(self):
        pass

    # 宅地建物取引業者検索システムを 免許番号で検索し
    # 不動産会社の名称をDBへ登録します。
    # なぜか、BeautifulSoup では動作しなかった為、seleniumを使用.
    # しかも、宅地建物取引業者検索システムは、
    # 元々、不安定? リソース不足? の為、動作が固まります。
    def find_licence_def(self,licence_no):

        licence_no = licence_no.replace("第","").replace("号","")
        
        api_args = api_args_tmpl % (licence_no,licence_no,licence_no,licence_no)
        req_url = url_base + "?" + api_args
        browser = self.get_browser()
        time.sleep(2)
        browser.get(req_url)
        shops = self.parse_found_shops_pages(browser)

        util_db = Db()
        util_db.bulk_upsert("real_estate_shop",
                            ['government','licence'],
                            ['government','licence','shop'],
                            ['shop'],
                            shops )
        return shops

    def download_and_save_master(self):
        logger.info("start")

        pref_no     = 1
        max_pref_no = 47 # 47都道府県

        while pref_no <= max_pref_no:
            req_url = url_base + "?dispCount=50&choice=1&kenCode=%02d" %(pref_no)
            logger.info(req_url)

            browser = self.get_browser()
            time.sleep(2)
            browser.get(req_url)
            time.sleep(2)
            search_btn = self.find_search_btn( browser )
            search_btn.click()
            
            # parseした不動産会社情報のdb保存
            shops = self.parse_found_shops_pages(browser)
                
            self.save_tbl_rows(shops)

            browser.close()
            pref_no += 1

    def parse_found_shops_pages(self, browser):
        shops_hash = {}
        i = 0
        while(i < 1000 ):  # 1000は 最終pageを判定できない場合に備えたもの
            body = browser.find_element(by=By.CSS_SELECTOR,value="body")
            body_text = body.text.replace("\n"," ")
            # 「接続拒否」の場合は、一旦、BACK
            if "The requested URL was rejected" in body_text:
                tmp_msg = "The requested URL was rejected... and bak"
                logger.warning(tmp_msg)
                a_elms = browser.find_elements(by=By.CSS_SELECTOR, value="a")
                a_elms[0].click() # 「Go Back」link
            
            shops_hash_tmp = self.parse_shops( browser )
            shops_hash.update( shops_hash_tmp )

            select_elms = browser.find_elements(by=By.CSS_SELECTOR,
                                                value="#pageListNo1")
            if len(select_elms) == 0:
                tmp_msg = "fail find_elements() for " + browser.current_url
                logger.error( tmp_msg )
                i += 1
                continue
                
            page_no = 0
            try:
                page_no = \
                    Select(select_elms[0]).first_selected_option.text.split("/")
            except Exception as e:
                logger.error(e)
                logger.error("fail find page no... retry")
                browser.back()
                continue
                
            if i % 10 == 0:
                logger.info("%s/%s %s" %
                            (page_no[0],page_no[1],browser.current_url) )
            
            if page_no[0] == page_no[1]:  #最終pageに達したら、終了
                break

            next_btn = self.find_next_btn( browser )
            next_btn.click()
            time.sleep(3)
            i += 1

        ret_datas = []
        for pref_licence_str,shop in shops_hash.items():
            pref_licence = pref_licence_str.split("\t")
            ret_datas.append({"government": pref_licence[0],
                              "licence"   : pref_licence[1],
                              "shop"      : shop} )
        return ret_datas


    def conv_shop_name(self, shop):
        replace_strs = ["株式会社","有限会社","合資会社","合同会社",'合名会社',
                        '独立行政法人','特定非営利活動法人','社会福祉法人',
                        '一般社団法人',"一般財団法人","公益財団法人"]
        for replace_str in replace_strs:
            shop = shop.replace(replace_str,"")

        shop = shop.strip().strip("　")

        # 英数字とカナを半角化
        shop = unicodedata.normalize("NFKC", shop)
        shop = jaconv.z2h(shop, kana=True, ascii=False, digit=False)
        
        return shop

    def parse_shops( self, browser ):
        tr_elms = browser.find_elements(by=By.CSS_SELECTOR,
                                        value="table.re_disp tr")
        
        if len(tr_elms) == 0:
            logger.error("fail parse table.re_disp tr %s" % (browser.current_url))
            return {}
        
        tr_elms.pop(0) # 先頭行はヘッダの為、削除

        re_compile = re.compile("[\(（].+[\)）]")

        shops_tmp = {}
        for tr_elm in tr_elms:
            cols_str = tr_elm.text
            cols = tr_elm.text.split(" ")

            government = "";
            licence    = "";
            shop       = "";

            if len(cols) == 7:
                government = re_compile.sub('',cols[1])
                government = government.replace("各地方整備局等","国土交通大臣")
                
                licence    = re_compile.sub('',cols[2])
                shop = cols[3]
            elif len(cols) == 6:    #信託銀行の場合
                government = "-"
                licence    = re_compile.sub('',cols[1])
                shop       = cols[2]
            else:
                continue

            

            shop = self.conv_shop_name(shop)

            shop_key = government +"\t"+ licence
            shops_tmp[shop_key] = shop

        return shops_tmp

    def find_search_btn( self, browser ):
        img_elms = browser.find_elements(by=By.CSS_SELECTOR, value="img")
        for img_elm in img_elms:
            img_src = img_elm.get_attribute("src")
            if "btn_search_off.png" in img_src:
                return img_elm
        return None

    def find_next_btn( self, browser ):
        img_elms = browser.find_elements(by=By.CSS_SELECTOR, value="img")
        for img_elm in img_elms:
            img_src = img_elm.get_attribute("src")
            if "result_move_r.jpg" in img_src:
                return img_elm
        return None
            
    def get_def_by_licence(self,government,licence):

        sql = """
select * from real_estate_shop
where government=%s and  licence=%s
"""
        db_conn = self.db_connect()
        with self.db_cursor(db_conn) as db_cur:
            try:
                db_cur.execute(sql,(government,licence))
            except Exception as e:
                logger.error(e)
                logger.error(sql)
                return {}

            ret_rows = db_cur.fetchall()
            if len(ret_rows):
                return dict(ret_rows[0])

            logger.warning("Not found %s %s" % (government,licence))
            return None
        
    def del_tbl_rows(self):
        logger.info("start")
        util_db = Db()
        util_db.del_tbl_rows("real_estate_shop")

    def save_tbl_rows(self, rows):
        logger.info("start")
        util_db = Db()
        util_db.save_tbl_rows("real_estate_shop",insert_cols,rows )