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.
87 lines
2.8 KiB
87 lines
2.8 KiB
#!/usr/bin/env python3 |
|
"""Suggest duplicate Cruzeiro ads from visually similar local photos. |
|
|
|
Uses a simple difference hash as a screening signal. Results are candidates for |
|
manual review, never automatic proof that two ads describe the same property. |
|
""" |
|
|
|
from __future__ import annotations |
|
|
|
import sqlite3 |
|
from collections import defaultdict |
|
from pathlib import Path |
|
|
|
from PIL import Image, UnidentifiedImageError |
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2] |
|
DB_PATH = ROOT / "dfimoveis_data" / "dfimoveis.sqlite3" |
|
DATA_ROOT = DB_PATH.parent |
|
|
|
|
|
def difference_hash(path: Path) -> int | None: |
|
try: |
|
with Image.open(path) as image: |
|
pixels = list(image.convert("L").resize((9, 8)).get_flattened_data()) |
|
except (OSError, UnidentifiedImageError): |
|
return None |
|
value = 0 |
|
for row in range(8): |
|
offset = row * 9 |
|
for column in range(8): |
|
value = (value << 1) | (pixels[offset + column] > pixels[offset + column + 1]) |
|
return value |
|
|
|
|
|
def main() -> None: |
|
connection = sqlite3.connect(f"file:{DB_PATH}?mode=ro", uri=True) |
|
connection.execute("PRAGMA query_only = ON") |
|
connection.execute("BEGIN") |
|
rows = connection.execute( |
|
""" |
|
WITH cruzeiro AS ( |
|
SELECT s.listing_id |
|
FROM listing_searches AS s |
|
JOIN listings AS l USING (listing_id) |
|
WHERE instr(search_url, '/cruzeiro/novo/') > 0 |
|
AND l.inactive_at IS NULL |
|
), uncommon AS ( |
|
SELECT sha256 |
|
FROM photos |
|
WHERE sha256 IS NOT NULL AND is_primary_gallery=1 |
|
GROUP BY sha256 |
|
HAVING COUNT(DISTINCT listing_id) <= 5 |
|
) |
|
SELECT p.listing_id, p.local_path |
|
FROM photos AS p |
|
JOIN cruzeiro AS c USING (listing_id) |
|
JOIN uncommon AS u USING (sha256) |
|
WHERE p.local_path IS NOT NULL AND p.is_primary_gallery=1 |
|
ORDER BY p.listing_id, p.ordinal |
|
""" |
|
).fetchall() |
|
connection.close() |
|
|
|
hashes: dict[str, list[int]] = defaultdict(list) |
|
for listing_id, relative_path in rows: |
|
value = difference_hash(DATA_ROOT / relative_path) |
|
if value is not None: |
|
hashes[listing_id].append(value) |
|
|
|
ids = sorted(hashes) |
|
print("listing_a\tlisting_b\tvisually_similar_photos") |
|
for index, first_id in enumerate(ids): |
|
for second_id in ids[index + 1 :]: |
|
available = list(hashes[second_id]) |
|
matches = 0 |
|
for first_hash in hashes[first_id]: |
|
distances = [(first_hash ^ second_hash).bit_count() for second_hash in available] |
|
if distances and min(distances) <= 6: |
|
available.pop(distances.index(min(distances))) |
|
matches += 1 |
|
if matches >= 3: |
|
print(f"{first_id}\t{second_id}\t{matches}") |
|
|
|
|
|
if __name__ == "__main__": |
|
main()
|
|
|