自分用・非汎用

Project description

domx

自分用・非汎用

インストール

uv add domx

※ patchright_page を使うとき：Google ChromeをPCにインストールしておく。
※ camoufox_page を使うとき：uv run camoufox fetch

実装機能

domx

WrappedPage
WrappedElement
WrappedElementGroup
ElementScan
WrappedFrame
WrappedShadowRoot
WrappedParser
WrappedNode
WrappedNodeGroup
NodeScan

domx.utils

parse_html(path: Path) -> LexborHTMLParser | None
meta_html(meta: Mapping[str, object | None]) -> str
from_here(file: str) -> Callable[[str], Path]
append_csv(path: Path, row: dict) -> None
write_csv(path: Path, rows: list[dict]) -> None
write_parquet(path: Path, rows: list[dict]) -> None
hash_name(key: str) -> str
write_text(path: Path, data: str) -> bool
write_bytes(path: Path, data: bytes) -> bool
save_log(path: Path, level: str = 'WARNING') -> None
process_map[T, R](worker: Callable[[T], R], items: Iterable[T], workers: int | None = None, *, chunksize: int | None = None) -> list[R | None]
glob_paths(dir_path: Path, pattern: str = '*.html') -> list[str]
counter(start: int = 1) -> Iterator[int]

domx.batch

Batch[T]
Batch.attach_patchright_page(*, browser_recreate_every: int | None = 1000, context_recreate_every: int | None = 200, browser_kwargs: dict | None = None, context_kwargs: dict | None = None) -> Iterator[Iterator[tuple[T, PatchrightPage]]]
Batch.attach_camoufox_page(*, browser_recreate_every: int | None = 1000, context_recreate_every: int | None = 200, browser_kwargs: dict | None = None, context_kwargs: dict | None = None) -> Iterator[Iterator[tuple[T, PlaywrightPage]]]
batch(items: list[T]) -> Batch[T]

domx.http

fetch_response(url: str | None, *, try_cnt: int = 3, wait_range: tuple[float, float] = (1, 3), sleep_after: tuple[float, float] | None = (0.5, 1), timeout: float = 30, impersonate: str = "chrome") -> Response | None
fetch_html(url: str | None, *, ...) -> str | None
fetch_bytes(url: str | None, *, ...) -> bytes | None

使用例

crawl.py

from urllib.parse import urlencode

from domx import wrap_page
from domx.batch import batch
from domx.utils import save_log, from_here, counter, write_csv

here = from_here(__file__)
save_log(here('log/crawling.log'))

prefecture_urls = ['https://home.katitas.jp/buyers_search/area/nagano']
n = len(prefecture_urls)
urls = []
with batch(prefecture_urls).attach_patchright_page(
    browser_kwargs={'channel': 'chrome', 'headless': False},
    context_kwargs={},
) as batch_pages:
    for i, (prefecture_url, page) in enumerate(batch_pages):
        p = wrap_page(page)
        print(f'prefecture_url {i}/{n - 1}')
        for page_num in counter():
            if not p.goto(f'{prefecture_url}?{urlencode({"page": page_num})}'):
                break
            if not (bukken_elems := p.ii('ul li div a[href^="https://home.katitas.jp"]:has(p)')):
                break
            urls.extend(bukken_elems.urls)
write_csv(here('csv/urls.csv'), [{'url': url} for url in urls])

scrape.py

from datetime import datetime, timezone
import time

from domx import wrap_page
from domx.batch import batch
from domx.utils import (
    save_log,
    append_csv,
    from_here,
    meta_html,
    hash_name,
    write_text,
    write_bytes,
)
import pandas as pd

here = from_here(__file__)
save_log(here('log/scraping.log'))

bukken_urls = pd.read_csv(here('csv/urls.csv'))['url'].tolist()
n = len(bukken_urls)
with batch(bukken_urls).attach_patchright_page(
    browser_kwargs={'channel': 'chrome', 'headless': False},
    context_kwargs={},
) as batch_pages:
    for url_index, (request_url, page) in enumerate(batch_pages):
        p = wrap_page(page)
        print(f'url_index {url_index}/{n - 1}')
        if not p.goto(request_url):
            append_csv(here('csv/failed.csv'), {
                'url_index': url_index,
                'request_url': request_url,
                'reason': 'goto',
            })
            continue
        html = meta_html({
            'domx:url_index': url_index,
            'domx:saved_at': datetime.now(timezone.utc),
            'domx:request_url': request_url,
            'domx:final_url': page.url,
        }) + page.content()
        if not write_text(here('html') / f'{hash_name(page.url)}.html', html):
            append_csv(here('csv/failed.csv'), {
                'url_index': url_index,
                'request_url': request_url,
                'reason': 'write_text',
            })

        page.screenshot(path=here(f'media/{url_index}-full-page.png'), full_page=True)

        elem_iframe = p.i('iframe[src^="https://home.katitas.jp"]')
        elem_iframe.scroll_into_view()
        time.sleep(3)
        elem_iframe.screenshot(here(f'media/{url_index}-gmap.png'), isolate=True)

        img_li_scan = p.ii('p.text-left').scan.m(r'画像をクリックすると拡大画像がご覧に').n('ul').ii('li').scan
        img_li = img_li_scan.m(r'外観') or img_li_scan.m(r'^(?!.*間取).*')
        img_url = img_li.i('a').url
        if (body := p.bytes_at(img_url)):
            write_bytes(here(f'media/{url_index}-img-desc.jpg'), body)

        main_img_url = p.i('img.w-full.object-contain').src
        if (body := p.bytes_at(main_img_url)):
            write_bytes(here(f'media/{url_index}-img-main.jpg'), body)

scrape_pairs.py（架空の2段パイプライン例）

import pandas as pd

from domx import wrap_page
from domx.batch import batch
from domx.utils import from_here

here = from_here(__file__)
items = pd.read_csv(here('csv/urls.csv'))['url'].tolist()

collected_urls = []

with batch(items).attach_patchright_page(
    browser_recreate_every=1000,
    context_recreate_every=200,
    browser_kwargs={'channel': 'chrome', 'headless': False},
    context_kwargs={'viewport': {'width': 1920, 'height': 1080}},
) as batch_pages:
    for request_url, page in batch_pages:
        p = wrap_page(page)
        if not p.goto(request_url):
            continue
        # collect urls...
        collected_urls.extend(p.ii('a[href]').urls)

with batch(collected_urls).attach_camoufox_page(
    browser_recreate_every=500,
    context_recreate_every=100,
    browser_kwargs={'headless': False, 'humanize': True},
    context_kwargs={'viewport': {'width': 1920, 'height': 1080}},
) as batch_pages:
    for request_url, page in batch_pages:
        p = wrap_page(page)
        if not p.goto(request_url):
            continue
        # scrape...

scrape_http.py

from datetime import datetime, timezone
from urllib.parse import urljoin

import pandas as pd
from selectolax.lexbor import LexborHTMLParser

from domx import wrap_parser
from domx.http import fetch_response, fetch_bytes
from domx.utils import (
    save_log,
    append_csv,
    from_here,
    meta_html,
    hash_name,
    write_text,
    write_bytes,
)

here = from_here(__file__)
save_log(here('log/scraping.log'))

bukken_urls = pd.read_csv(here('csv/urls.csv'))['url']
n = len(bukken_urls)
for url_index, request_url in bukken_urls.items():
    print(f'url_index {url_index}/{n - 1}')

    if not (res := fetch_response(request_url)):
        append_csv(here('csv/failed.csv'), {
            'url_index': url_index,
            'request_url': request_url,
            'reason': 'fetch_response',
        })
        continue

    html = meta_html({
        'domx:url_index': url_index,
        'domx:saved_at': datetime.now(timezone.utc),
        'domx:request_url': request_url,
        'domx:final_url': res.url,
    }) + res.text
    if not write_text(here('html') / f'{hash_name(res.url)}.html', html):
        append_csv(here('csv/failed.csv'), {
            'url_index': url_index,
            'request_url': request_url,
            'reason': 'write_text',
        })
        continue

    p = wrap_parser(LexborHTMLParser(res.text))
    img_li_scan = p.ii('p.text-left').scan.m(r'画像をクリックすると拡大画像がご覧に').n('ul').ii('li').scan
    img_li = img_li_scan.m(r'外観') or img_li_scan.m(r'^(?!.*間取).*')
    img_href = img_li.i('a').attr('href')
    if img_href and (body := fetch_bytes(urljoin(res.url, img_href))):
        write_bytes(here(f'media/{url_index}-img-desc.jpg'), body)

    main_img_src = p.i('img.w-full.object-contain').attr('src')
    if main_img_src and (body := fetch_bytes(urljoin(res.url, main_img_src))):
        write_bytes(here(f'media/{url_index}-img-main.jpg'), body)

extract.py

from pathlib import Path

from domx import wrap_parser
from domx.utils import from_here, glob_paths, parse_html, process_map, write_parquet

def main():
    here = from_here(__file__)
    html_paths = glob_paths(here('html'), '*.html')
    results = [r for r in process_map(extract, html_paths) if r]
    write_parquet(here('parquet/extract.parquet'), results)

def extract(file_path: str) -> dict | None:
    if not (parser := parse_html(Path(file_path))):
        return None
    p = wrap_parser(parser)
    dt_scan = p.ii('dt').scan
    dd_text = lambda pattern: dt_scan.m(pattern).n('dd').text
    return {
        'url_index': p.i('meta[name="domx:url_index"]').attr('content'),
        'saved_at': p.i('meta[name="domx:saved_at"]').attr('content'),
        'request_url': p.i('meta[name="domx:request_url"]').attr('content'),
        'final_url': p.i('meta[name="domx:final_url"]').attr('content'),
        'ファイル名': Path(file_path).name,

        '取り扱い店舗': p.ii('p').scan.m(r'取り扱い店舗').n('p').text,
        
        '価格': dd_text(r'価格'),
        '月々の支払い': dd_text(r'月々の支払い'),
        '間取': dd_text(r'間取'),
        '土地面積': dd_text(r'土地面積'),
        '建物面積': dd_text(r'建物面積'),
        
        '所在地': dd_text(r'所在地'),
        '交通': dd_text(r'交通'),
        '接道状況': dd_text(r'接道状況'),
        '私道面積': dd_text(r'私道面積'),
        'セットバック': dd_text(r'セットバック'),
        '建物構造': dd_text(r'建物構造'),
        '国土法提出': dd_text(r'国土法提出'),
        '駐車場': dd_text(r'駐車場'),
        '車庫区分': dd_text(r'車庫区分'),
        '都市計画': dd_text(r'都市計画'),
        '物件種別': dd_text(r'物件種別'),
        '建ぺい率 /容積率': dd_text(r'建ぺい率.*容積率'),
        '土地権利': dd_text(r'土地権利'),
        '地目': dd_text(r'地目'),
        '築年月': dd_text(r'築年月'),
        '取引態様': dd_text(r'取引態様'),
        '引渡日（入居予定日）': dd_text(r'引渡日.*入居予定日'),
        '用途地域': dd_text(r'用途地域'),
        '現況': dd_text(r'現況'),
        '設備・条件': dd_text(r'設備.*条件'),
        '備考': dd_text(r'備考'),
        '最寄りの学校': dd_text(r'最寄.*の学校'),
        '物件番号': dd_text(r'物件番号'),
        '情報更新日': dd_text(r'情報更新日'),
        '次回更新予定日': dd_text(r'次回更新予定日'),
        
        'スタッフからのコメント': p.ii('div').scan.m(r'スタッフからのコメント').n('div').text,
        '物件の魅力': p.ii('p').scan.m(r'物件の魅力').n('p').text,
        
        'img_desc': '\n'.join(p.ii('p.text-left').scan.m(r'画像をクリックすると拡大画像がご覧に').n('ul').ii('li').texts)
    }

if __name__ == '__main__':
    main()

clean.ipynb

import re

import pandas as pd

df_shikutyoson = pd.read_csv('./shikutyoson.csv')
cities = df_shikutyoson["市区町村"].dropna().sort_values(key=lambda x: x.str.len(), ascending=False)
shikutyoson_pattern = "|".join(cities.map(lambda x: re.escape(x)))

df_raw = pd.read_parquet('parquet/extract.parquet')
df_raw = df_raw.apply(lambda x: x.fillna('').str.normalize('NFKC').str.strip())

df = df_raw.sort_values('saved_at')[['url_index', 'saved_at', 'request_url', 'final_url']].copy()

df['事例種別'] = df_raw['物件種別'].str.contains(r'中古|土地').map({True: '中古売出'})
df['総額'] = (
    df_raw['価格']
    .str.extract(r'([,\d]+)\s*万円', expand=False)
    .replace(',', '', regex=True)
    .pipe(lambda s: pd.to_numeric(s, errors='coerce') * 10000)
)
df['土地面積'] = df_raw['土地面積'].str.extract(r'([\d\.]+)')
df['建物面積'] = df_raw['建物面積'].str.extract(r'([\d\.]+)')
df['建物種別'] = df_raw['物件種別'].map({'中古戸建': '戸建て', '中古マンション': 'マンション', '土地': '土地'})
df[['所在都道府県', '所在市', '所在字', '所在番地']] = df_raw['所在地'].str.extract(fr'^(京都府|.+?[都道府県])({shikutyoson_pattern})(\D*)(.*)')

s1 = (
    df_raw['築年月']
    .replace({r'元年': r'1年'}, regex=True)
    .str.extract(r'(\d+)年', expand=False)
    .pipe(lambda s: pd.to_numeric(s, errors='coerce'))
)
s2 = df_raw['築年月'].str[:2].map({'令和': 2018, '平成': 1988, '昭和': 1925, '大正': 1911, '明治': 1867})
df['建築年'] = s1 + s2

df['構造体'] = df_raw['建物構造'].str.extract(r'^(\S+)')
df['階層'] = df_raw['建物構造'].str.extract(r'(\d+)階')
df['リノベ内容'] = df_raw['備考'].str.extract(r'(?s)^(20\d{2}/.*?)\n\D')
df['間取'] = df_raw['間取']
df['成約年月'] = df_raw['現況'].map({'空': '販売中', '古家付': '販売中'})
df['私道負担'] = df_raw['私道面積']
df['接道'] = df_raw['接道状況']

s1 = df_raw['最寄りの学校'].str.extract(r'([^/\s【】・、(]+?小学校)', expand=False) 
s2 = df_raw['物件の魅力'].str.extract(r'([^/\s【】・、(]+?小学校)', expand=False)
s3 = df_raw['備考'].str.extract(r'([^/\s【】・、(]+?小学校)', expand=False)
s4 = df_raw['img_desc'].str.extract(r'([^/\s【】・、(]+?小学校)', expand=False) 
df['小学校'] = s1.fillna(s2).fillna(s3).fillna(s4)

s1 = df_raw['最寄りの学校'].str.extract(r'([^/\s【】・、(]+?中学校)', expand=False) 
s2 = df_raw['物件の魅力'].str.extract(r'([^/\s【】・、(]+?中学校)', expand=False)
s3 = df_raw['備考'].str.extract(r'([^/\s【】・、(]+?中学校)', expand=False)
s4 = df_raw['img_desc'].str.extract(r'([^/\s【】・、(]+?中学校)', expand=False) 
df['中学校'] = s1.fillna(s2).fillna(s3).fillna(s4)

df['周辺環境'] = df_raw['備考'].map(lambda x: '\n'.join(l for l in x.splitlines() if re.search(r'(?:\d分|\dm)$', l)))
df['都市計画'] = df_raw['都市計画']
df['用途地域'] = df_raw['用途地域']
df[['建ぺい率', '容積率']] = df_raw['建ぺい率 /容積率'].str.extract(r'(\d+%)\D*(\d+%)')
df['水道'] = df_raw['設備・条件'].str.extract(r'(公営水道|上水道)')
df['下水'] = df_raw['設備・条件'].str.extract(r'(本下水|個別浄化槽|汲取|下水道)')
df['ガス'] = df_raw['設備・条件'].str.extract(r'(個別LPG|集中LPG|都市ガス|プロパンガス|オール電化)')
df['契約態様'] = df_raw['取引態様']
df['問合せ先'] = df_raw['取り扱い店舗']
df['駐車場'] = df_raw['駐車場']
df['交通'] = df_raw['交通']
df['物件の特徴'] = df_raw['物件の魅力']
df['仕様'] = df_raw['設備・条件']

df['土地権利'] = df_raw['土地権利']
df['地目'] = df_raw['地目']
df['引渡日（入居予定日）'] = df_raw['引渡日（入居予定日）']
df['物件番号'] = df_raw['物件番号']
df['情報更新日'] = df_raw['情報更新日']

df.to_clipboard(index=False)

Project details

Release history Release notifications | RSS feed

0.1.18

Jun 11, 2026

0.1.17

Jun 10, 2026

0.1.16

Jun 9, 2026

0.1.15

Jun 9, 2026

This version

0.1.14

Jun 7, 2026

0.1.11

Jun 2, 2026

0.1.10

May 31, 2026

0.1.9

May 26, 2026

0.1.8

May 25, 2026

0.1.7

May 19, 2026

0.1.6

May 17, 2026

0.1.5

May 17, 2026

0.1.4

May 12, 2026

0.1.3

May 10, 2026

0.1.2

May 9, 2026

0.1.1

May 5, 2026

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

domx-0.1.14.tar.gz (20.6 kB view details)

Uploaded Jun 7, 2026 Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

The dropdown lists show the available interpreters, ABIs, and platforms. Enable javascript to be able to filter the list of wheel files.

domx-0.1.14-py3-none-any.whl (16.1 kB view details)

Uploaded Jun 7, 2026 Python 3

File details

Details for the file domx-0.1.14.tar.gz.

File metadata

Download URL: domx-0.1.14.tar.gz
Upload date: Jun 7, 2026
Size: 20.6 kB
Tags: Source
Uploaded using Trusted Publishing? No
Uploaded via: python-requests/2.34.2

File hashes

Hashes for domx-0.1.14.tar.gz
Algorithm	Hash digest
SHA256	`c8d2e48c25e0d48871fa0eb5bdc5774e1601b5051ba2824bd1a7fd5415a2e245`
MD5	`f415c084214eb5e88786d576643df322`
BLAKE2b-256	`0f332566b0f79161834f8bd5d59f2e8b5ea0d1cabbf00baba21ebfca983101ac`

See more details on using hashes here.

File details

Details for the file domx-0.1.14-py3-none-any.whl.

File metadata

Download URL: domx-0.1.14-py3-none-any.whl
Upload date: Jun 7, 2026
Size: 16.1 kB
Tags: Python 3
Uploaded using Trusted Publishing? No
Uploaded via: python-requests/2.34.2

File hashes

Hashes for domx-0.1.14-py3-none-any.whl
Algorithm	Hash digest
SHA256	`d60021df00d9e6c3e2f6c3b750c70021da52470b1b6871035f4dbd7634f43eb7`
MD5	`ebd3ea29f185517af1fec6ef0c69ce82`
BLAKE2b-256	`a57d1b950caa632438e6da662f1c1cb5aa5dca6a87389649b9055f141b22edf4`

See more details on using hashes here.

domx 0.1.14

Navigation

Verified details

Maintainers

Unverified details

Meta

Project description

domx

インストール

実装機能

domx

domx.utils

domx.batch

domx.http

使用例

crawl.py

scrape.py

scrape_pairs.py（架空の2段パイプライン例）

scrape_http.py

extract.py

clean.ipynb

Project details

Verified details

Maintainers

Unverified details

Meta

Release history Release notifications | RSS feed

Download files

Source Distribution

Built Distribution

File details

File metadata

File hashes

File details

File metadata

File hashes