You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
130 lines
3.7 KiB
130 lines
3.7 KiB
#!/usr/bin/env python3 |
|
"""Read-only extraction helper for the Jardins Mangueiral ranking.""" |
|
|
|
from __future__ import annotations |
|
|
|
import argparse |
|
import gzip |
|
import re |
|
import sqlite3 |
|
import sys |
|
from pathlib import Path |
|
|
|
from bs4 import BeautifulSoup |
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2] |
|
DB_PATH = ROOT / "dfimoveis_data" / "dfimoveis.sqlite3" |
|
DATA_ROOT = DB_PATH.parent |
|
SEARCH_FRAGMENT = "/jardins-mangueiral/" |
|
|
|
|
|
def clean(value: str | None) -> str: |
|
return re.sub(r"\s+", " ", value or "").strip() |
|
|
|
|
|
def sanitize_description(value: str) -> str: |
|
value = re.sub(r"\b[\w.+-]+@[\w.-]+\.[A-Za-z]{2,}\b", "[email]", value) |
|
value = re.sub( |
|
r"(?:\(\d{2}\)|\b\d{2}\b)\s*(?:\d[\s.-]*){5,11}\d", |
|
"[telefone]", |
|
value, |
|
) |
|
value = re.sub(r"\bCRECI\b.{0,30}", "", value, flags=re.IGNORECASE) |
|
return clean(value) |
|
|
|
|
|
def parse_html(relative_path: str) -> tuple[str, str, str]: |
|
with gzip.open(DATA_ROOT / relative_path, "rt", encoding="utf-8", errors="replace") as fh: |
|
soup = BeautifulSoup(fh.read(), "lxml") |
|
|
|
description_node = soup.select_one("div.assined-imv") |
|
description = "" |
|
if description_node: |
|
for node in description_node.select("span, a, button"): |
|
node.decompose() |
|
description = sanitize_description(description_node.get_text(" ", strip=True)) |
|
|
|
details: list[str] = [] |
|
for item in soup.select("ul.details-text li"): |
|
text = clean(item.get_text(" ", strip=True)) |
|
if text and text not in details: |
|
details.append(text) |
|
|
|
address_node = soup.select_one('[itemprop="address"]') |
|
address = clean(address_node.get_text(" ", strip=True) if address_node else "") |
|
return description, " | ".join(details), address |
|
|
|
|
|
def main() -> int: |
|
parser = argparse.ArgumentParser() |
|
parser.add_argument("--ids-only", action="store_true") |
|
args = parser.parse_args() |
|
|
|
connection = sqlite3.connect(f"file:{DB_PATH}?mode=ro", uri=True) |
|
connection.execute("PRAGMA query_only = ON") |
|
connection.execute("BEGIN") |
|
rows = connection.execute( |
|
""" |
|
SELECT |
|
l.listing_id, l.url, l.price_brl, l.condominium_brl, l.iptu_brl, |
|
l.area_m2, l.bedrooms, l.suites, l.parking_spaces, |
|
l.raw_html_path, COUNT(p.source_url) AS photo_count |
|
FROM listings AS l |
|
JOIN listing_searches AS s USING (listing_id) |
|
LEFT JOIN photos AS p ON p.listing_id=l.listing_id AND p.is_primary_gallery=1 |
|
WHERE instr(s.search_url, ?) > 0 |
|
AND l.inactive_at IS NULL |
|
GROUP BY l.listing_id |
|
ORDER BY l.price_brl, l.listing_id |
|
LIMIT 150 |
|
""", |
|
(SEARCH_FRAGMENT,), |
|
).fetchall() |
|
connection.close() |
|
|
|
if args.ids_only: |
|
for row in rows: |
|
print(row[0]) |
|
return 0 |
|
|
|
print( |
|
"listing_id\tprice_brl\tcondominium_brl\tiptu_brl\tarea_m2\tbedrooms\t" |
|
"suites\tparking_spaces\tphoto_count\taddress\tdetails\tdescription\turl" |
|
) |
|
for row in rows: |
|
( |
|
listing_id, |
|
url, |
|
price, |
|
condominium, |
|
iptu, |
|
area, |
|
bedrooms, |
|
suites, |
|
parking, |
|
raw_path, |
|
photo_count, |
|
) = row |
|
description, details, address = parse_html(raw_path) |
|
values = ( |
|
listing_id, |
|
price, |
|
condominium, |
|
iptu, |
|
area, |
|
bedrooms, |
|
suites, |
|
parking, |
|
photo_count, |
|
address, |
|
details, |
|
description, |
|
url, |
|
) |
|
print("\t".join("" if value is None else clean(str(value)) for value in values)) |
|
return 0 |
|
|
|
|
|
if __name__ == "__main__": |
|
sys.exit(main())
|
|
|