586 lines
14 KiB
Python
586 lines
14 KiB
Python
import re
|
|
import time
|
|
from datetime import datetime, timedelta
|
|
|
|
import pandas as pd
|
|
from dateutil.relativedelta import relativedelta
|
|
from playwright.sync_api import sync_playwright
|
|
|
|
|
|
# ============================================================
|
|
# USTAWIENIA
|
|
# ============================================================
|
|
|
|
URL = "https://maps.app.goo.gl/USRCto35jhoVtmj4A"
|
|
|
|
HEADLESS = False
|
|
|
|
# Ile kolejnych scrolli bez nowych opinii oznacza koniec
|
|
MAX_EMPTY_SCROLLS = 8
|
|
|
|
# Czas między scrollami
|
|
SCROLL_DELAY = 1.2
|
|
|
|
|
|
# ============================================================
|
|
# PARSOWANIE DATY GOOGLE
|
|
# ============================================================
|
|
|
|
def parse_google_date(text: str):
|
|
if not text:
|
|
return None
|
|
|
|
text = text.strip().lower()
|
|
now = datetime.now()
|
|
|
|
if text == "dzisiaj":
|
|
return now.date()
|
|
|
|
if text == "wczoraj":
|
|
return (now - timedelta(days=1)).date()
|
|
|
|
patterns = [
|
|
(r"(\d+)\s+dni?\s+temu", lambda n: now - timedelta(days=n)),
|
|
(r"(\d+)\s+dzień\s+temu", lambda n: now - timedelta(days=n)),
|
|
(r"(\d+)\s+tygodni?\s+temu", lambda n: now - timedelta(weeks=n)),
|
|
(
|
|
r"(\d+)\s+miesi(?:ąc|ące|ęcy)\s+temu",
|
|
lambda n: now - relativedelta(months=n)
|
|
),
|
|
(
|
|
r"(\d+)\s+lat(?:a)?\s+temu",
|
|
lambda n: now - relativedelta(years=n)
|
|
),
|
|
]
|
|
|
|
for pattern, func in patterns:
|
|
match = re.search(pattern, text)
|
|
|
|
if match:
|
|
return func(int(match.group(1))).date()
|
|
|
|
if "tydzień temu" in text:
|
|
return (now - timedelta(weeks=1)).date()
|
|
|
|
if "miesiąc temu" in text:
|
|
return (now - relativedelta(months=1)).date()
|
|
|
|
if "rok temu" in text:
|
|
return (now - relativedelta(years=1)).date()
|
|
|
|
return None
|
|
|
|
|
|
# ============================================================
|
|
# COOKIES
|
|
# ============================================================
|
|
|
|
def accept_cookies(page):
|
|
labels = [
|
|
"Zaakceptuj wszystko",
|
|
"Accept all",
|
|
"Odrzuć wszystko",
|
|
"Reject all",
|
|
]
|
|
|
|
for label in labels:
|
|
try:
|
|
button = page.get_by_role(
|
|
"button",
|
|
name=re.compile(label, re.I)
|
|
)
|
|
|
|
if button.count() and button.first.is_visible():
|
|
button.first.click(timeout=3000)
|
|
time.sleep(2)
|
|
return
|
|
except:
|
|
pass
|
|
|
|
|
|
# ============================================================
|
|
# OTWIERANIE PANELU OPINII
|
|
# ============================================================
|
|
|
|
def open_reviews_panel(page):
|
|
print("Szukam właściwego przycisku z opiniami...")
|
|
|
|
buttons = page.locator("button")
|
|
|
|
for i in range(buttons.count()):
|
|
try:
|
|
button = buttons.nth(i)
|
|
|
|
if not button.is_visible():
|
|
continue
|
|
|
|
aria = button.get_attribute("aria-label") or ""
|
|
|
|
try:
|
|
text = button.inner_text(timeout=500)
|
|
except:
|
|
text = ""
|
|
|
|
label = f"{aria} {text}".strip()
|
|
|
|
# Musi zawierać liczbę oraz opini/recenz/review
|
|
if (
|
|
re.search(r"\d+", label)
|
|
and re.search(r"(opini|recenz|reviews?)", label, re.I)
|
|
):
|
|
print("Kandydat:", repr(label[:200]))
|
|
|
|
try:
|
|
button.scroll_into_view_if_needed()
|
|
time.sleep(0.5)
|
|
button.click(timeout=3000)
|
|
|
|
print("Kliknięto:", repr(label[:200]))
|
|
|
|
time.sleep(3)
|
|
|
|
cards = page.locator('div[data-review-id]')
|
|
|
|
if cards.count() > 5:
|
|
print("Panel opinii otwarty.")
|
|
return True
|
|
|
|
except Exception as e:
|
|
print("Nie udało się kliknąć:", e)
|
|
|
|
except:
|
|
continue
|
|
|
|
# Fallback dla role=button
|
|
print("Próbuję alternatywnego selektora...")
|
|
|
|
candidates = page.locator(
|
|
'[role="button"][aria-label*="opini"], '
|
|
'[role="button"][aria-label*="recenz"], '
|
|
'[role="button"][aria-label*="review"]'
|
|
)
|
|
|
|
for i in range(candidates.count()):
|
|
try:
|
|
el = candidates.nth(i)
|
|
|
|
if not el.is_visible():
|
|
continue
|
|
|
|
aria = el.get_attribute("aria-label") or ""
|
|
|
|
if not re.search(r"\d+", aria):
|
|
continue
|
|
|
|
print("Klikam:", repr(aria))
|
|
|
|
el.scroll_into_view_if_needed()
|
|
el.click(timeout=3000)
|
|
|
|
time.sleep(3)
|
|
|
|
cards = page.locator('div[data-review-id]')
|
|
|
|
if cards.count() > 5:
|
|
print("Panel opinii otwarty.")
|
|
return True
|
|
|
|
except:
|
|
continue
|
|
|
|
return False
|
|
|
|
|
|
# ============================================================
|
|
# ZNAJDOWANIE SCROLLOWANEGO PANELU
|
|
# ============================================================
|
|
|
|
def get_scrollable_parent(card):
|
|
try:
|
|
return card.evaluate_handle("""
|
|
el => {
|
|
let p = el.parentElement;
|
|
|
|
while (p) {
|
|
const style = getComputedStyle(p);
|
|
|
|
if (
|
|
(style.overflowY === 'auto' ||
|
|
style.overflowY === 'scroll') &&
|
|
p.scrollHeight > p.clientHeight
|
|
) {
|
|
return p;
|
|
}
|
|
|
|
p = p.parentElement;
|
|
}
|
|
|
|
return null;
|
|
}
|
|
""")
|
|
except:
|
|
return None
|
|
|
|
|
|
# ============================================================
|
|
# OCENA GWIAZDKOWA
|
|
# ============================================================
|
|
|
|
def extract_rating(card):
|
|
selectors = [
|
|
'[role="img"][aria-label*="gwiazd"]',
|
|
'[aria-label*="gwiazd"]',
|
|
'[role="img"][aria-label*="star"]',
|
|
'[aria-label*="star"]',
|
|
'span.kvMYJc[aria-label]',
|
|
]
|
|
|
|
for selector in selectors:
|
|
try:
|
|
el = card.locator(selector).first
|
|
|
|
if not el.count():
|
|
continue
|
|
|
|
label = el.get_attribute("aria-label") or ""
|
|
|
|
match = re.search(r"([1-5](?:[.,]\d+)?)", label)
|
|
|
|
if match:
|
|
value = match.group(1).replace(",", ".")
|
|
return int(float(value))
|
|
|
|
except:
|
|
pass
|
|
|
|
return None
|
|
|
|
|
|
# ============================================================
|
|
# DATA OPINII
|
|
# ============================================================
|
|
|
|
def extract_date_text(card):
|
|
selectors = [
|
|
".rsqaWe",
|
|
"span.rsqaWe",
|
|
]
|
|
|
|
for selector in selectors:
|
|
try:
|
|
el = card.locator(selector).first
|
|
|
|
if el.count():
|
|
text = el.inner_text().strip()
|
|
|
|
if text:
|
|
return text
|
|
|
|
except:
|
|
pass
|
|
|
|
return None
|
|
|
|
|
|
# ============================================================
|
|
# ŁADOWANIE WSZYSTKICH OPINII
|
|
# ============================================================
|
|
|
|
def count_unique_reviews(cards):
|
|
"""Policz opinie, a nie zduplikowane elementy DOM Google Maps."""
|
|
try:
|
|
return cards.evaluate_all("""
|
|
elements => new Set(
|
|
elements
|
|
.map(el => el.getAttribute('data-review-id'))
|
|
.filter(Boolean)
|
|
).size
|
|
""")
|
|
except:
|
|
return 0
|
|
|
|
|
|
def load_all_reviews(page):
|
|
previous_count = 0
|
|
empty_scrolls = 0
|
|
|
|
print("\nŁadowanie opinii...")
|
|
|
|
while empty_scrolls < MAX_EMPTY_SCROLLS:
|
|
cards = page.locator('div[data-review-id]')
|
|
count = count_unique_reviews(cards)
|
|
|
|
print(
|
|
f"\rZaładowane opinie: {count}",
|
|
end="",
|
|
flush=True
|
|
)
|
|
|
|
if count > previous_count:
|
|
previous_count = count
|
|
empty_scrolls = 0
|
|
else:
|
|
empty_scrolls += 1
|
|
|
|
if count == 0:
|
|
page.mouse.wheel(0, 5000)
|
|
time.sleep(SCROLL_DELAY)
|
|
continue
|
|
|
|
last_card = cards.last
|
|
|
|
scrollable = get_scrollable_parent(last_card)
|
|
|
|
if scrollable:
|
|
try:
|
|
scrollable.evaluate("""
|
|
el => {
|
|
el.scrollTop = el.scrollHeight;
|
|
}
|
|
""")
|
|
except:
|
|
page.mouse.wheel(0, 5000)
|
|
else:
|
|
try:
|
|
last_card.scroll_into_view_if_needed()
|
|
except:
|
|
page.mouse.wheel(0, 5000)
|
|
|
|
time.sleep(SCROLL_DELAY)
|
|
|
|
print()
|
|
|
|
|
|
# ============================================================
|
|
# ZBIERANIE DANYCH
|
|
# ============================================================
|
|
|
|
def extract_reviews(page):
|
|
results = {}
|
|
|
|
cards = page.locator('div[data-review-id]')
|
|
|
|
unique_count = count_unique_reviews(cards)
|
|
print(f"\nZbieram dane z {unique_count} unikalnych opinii...")
|
|
|
|
for i in range(cards.count()):
|
|
card = cards.nth(i)
|
|
|
|
try:
|
|
review_id = card.get_attribute("data-review-id")
|
|
|
|
if not review_id:
|
|
continue
|
|
|
|
# Deduplikacja
|
|
if review_id in results:
|
|
continue
|
|
|
|
rating = extract_rating(card)
|
|
|
|
if rating is None:
|
|
continue
|
|
|
|
date_raw = extract_date_text(card)
|
|
date_approx = parse_google_date(date_raw)
|
|
|
|
results[review_id] = {
|
|
"review_id": review_id,
|
|
"rating": rating,
|
|
"date_raw": date_raw,
|
|
"date_approx": date_approx,
|
|
}
|
|
|
|
except Exception as e:
|
|
print(f"Błąd przy opinii #{i}: {e}")
|
|
|
|
return pd.DataFrame(results.values())
|
|
|
|
|
|
# ============================================================
|
|
# PODSUMOWANIE
|
|
# ============================================================
|
|
|
|
def print_summary(df):
|
|
print("\n==============================")
|
|
print("ŁĄCZNIE")
|
|
print("==============================")
|
|
print(len(df))
|
|
|
|
print("\n==============================")
|
|
print("ROZKŁAD GWIAZDEK")
|
|
print("==============================")
|
|
|
|
distribution = (
|
|
df["rating"]
|
|
.value_counts()
|
|
.reindex([5, 4, 3, 2, 1], fill_value=0)
|
|
)
|
|
|
|
total = len(df)
|
|
|
|
for stars, count in distribution.items():
|
|
percent = (count / total * 100) if total else 0
|
|
|
|
print(
|
|
f"{stars}★: "
|
|
f"{count:>4} "
|
|
f"({percent:>5.1f}%)"
|
|
)
|
|
|
|
print("\n==============================")
|
|
print("ŚREDNIA")
|
|
print("==============================")
|
|
|
|
if len(df):
|
|
print(round(df["rating"].mean(), 3))
|
|
else:
|
|
print("Brak danych")
|
|
|
|
|
|
# ============================================================
|
|
# FILTROWANIE PO OKRESIE
|
|
# ============================================================
|
|
|
|
def print_period_summary(df, date_from=None, date_to=None):
|
|
temp = df.copy()
|
|
|
|
temp["date_approx"] = pd.to_datetime(
|
|
temp["date_approx"],
|
|
errors="coerce"
|
|
)
|
|
|
|
if date_from:
|
|
temp = temp[
|
|
temp["date_approx"] >= pd.to_datetime(date_from)
|
|
]
|
|
|
|
if date_to:
|
|
temp = temp[
|
|
temp["date_approx"] <= pd.to_datetime(date_to)
|
|
]
|
|
|
|
print("\n==============================")
|
|
print(f"OKRES: {date_from} -> {date_to}")
|
|
print("==============================")
|
|
|
|
if temp.empty:
|
|
print("Brak opinii w tym okresie.")
|
|
return
|
|
|
|
distribution = (
|
|
temp["rating"]
|
|
.value_counts()
|
|
.reindex([5, 4, 3, 2, 1], fill_value=0)
|
|
)
|
|
|
|
print(f"Liczba opinii: {len(temp)}")
|
|
|
|
for stars, count in distribution.items():
|
|
print(f"{stars}★: {count}")
|
|
|
|
print("Średnia:", round(temp["rating"].mean(), 3))
|
|
|
|
|
|
# ============================================================
|
|
# MAIN
|
|
# ============================================================
|
|
|
|
def main():
|
|
with sync_playwright() as p:
|
|
browser = p.chromium.launch(
|
|
headless=HEADLESS
|
|
)
|
|
|
|
context = browser.new_context(
|
|
locale="pl-PL",
|
|
viewport={
|
|
"width": 1400,
|
|
"height": 900,
|
|
},
|
|
)
|
|
|
|
page = context.new_page()
|
|
|
|
print("Otwieram Google Maps...")
|
|
|
|
page.goto(
|
|
URL,
|
|
wait_until="domcontentloaded",
|
|
timeout=60_000
|
|
)
|
|
|
|
time.sleep(3)
|
|
|
|
accept_cookies(page)
|
|
|
|
opened = open_reviews_panel(page)
|
|
|
|
if not opened:
|
|
print(
|
|
"\nNie udało się automatycznie otworzyć panelu opinii."
|
|
)
|
|
print(
|
|
"Kliknij ręcznie liczbę opinii / „Więcej opinii”."
|
|
)
|
|
|
|
input(
|
|
"Gdy panel będzie otwarty, naciśnij ENTER..."
|
|
)
|
|
|
|
time.sleep(2)
|
|
|
|
cards = page.locator('div[data-review-id]')
|
|
|
|
print(
|
|
"Opinie widoczne po otwarciu panelu:",
|
|
count_unique_reviews(cards)
|
|
)
|
|
|
|
if count_unique_reviews(cards) < 5:
|
|
print(
|
|
"\nUWAGA: panel opinii prawdopodobnie "
|
|
"nie został poprawnie otwarty."
|
|
)
|
|
|
|
input(
|
|
"Otwórz ręcznie wszystkie opinie "
|
|
"i naciśnij ENTER..."
|
|
)
|
|
|
|
load_all_reviews(page)
|
|
|
|
df = extract_reviews(page)
|
|
|
|
browser.close()
|
|
|
|
if df.empty:
|
|
print("\nNie znaleziono żadnych opinii.")
|
|
return
|
|
|
|
df.to_csv(
|
|
"google_reviews_ratings.csv",
|
|
index=False,
|
|
encoding="utf-8-sig"
|
|
)
|
|
|
|
print_summary(df)
|
|
|
|
print("\nZapisano:")
|
|
print("google_reviews_ratings.csv")
|
|
|
|
# ========================================================
|
|
# PRZYKŁADOWE FILTROWANIE
|
|
# Odkomentuj jeśli chcesz
|
|
# ========================================================
|
|
|
|
# print_period_summary(
|
|
# df,
|
|
# "2025-01-01",
|
|
# "2025-12-31"
|
|
# )
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|