mirror of
https://github.com/timberjoegithub/GoogleScrape.git
synced 2026-07-22 00:19:48 +00:00
timberjoegithub
13ae301667
- Remove dead imports/duplication from Google2xls.py and social.py - Centralize shared logic in review_workflow_utils.py - Add focused posting-flow tests in tests/test_social.py (all pass) - Update .gitignore for secrets and local artifacts - Trim requirements.txt to direct deps - Improve CI: compile all scripts, run focused tests - Prep repo for clean commit (no secrets, no local artifacts)
121 lines
4.2 KiB
Python
121 lines
4.2 KiB
Python
import re
|
|
import time
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
|
|
from dateutil.relativedelta import relativedelta
|
|
|
|
|
|
GOOGLE_REVIEW_SCROLL_XPATH = (
|
|
'//*[@id="QA0Szd"]/div/div/div[1]/div[2]/div/div[1]/div/div/div[5]/div[2]'
|
|
)
|
|
GOOGLE_REVIEW_SCROLL_SCRIPT = (
|
|
'document.getElementsByClassName("dS8AEf")[0].scrollTop = '
|
|
'document.getElementsByClassName("dS8AEf")[0].scrollHeight'
|
|
)
|
|
|
|
|
|
def build_google_chrome_options(
|
|
headless=False,
|
|
log_level=None,
|
|
ignore_certificate_errors=False,
|
|
ignore_certificate_errors_spki_list=False,
|
|
ignore_ssl_errors=False,
|
|
remote_debugging_pipe=False,
|
|
):
|
|
from selenium import webdriver
|
|
|
|
options = webdriver.ChromeOptions()
|
|
if log_level is not None:
|
|
options.add_argument(f"--log-level={log_level}")
|
|
if ignore_certificate_errors:
|
|
options.add_argument("--ignore-certificate-error")
|
|
if ignore_certificate_errors_spki_list:
|
|
options.add_argument("--ignore-certificate-errors-spki-list")
|
|
if ignore_ssl_errors:
|
|
options.add_argument("--ignore-ssl-errors")
|
|
if headless:
|
|
options.add_argument("--headless")
|
|
options.add_argument("--lang=en-US")
|
|
options.add_argument("--disable-blink-features=AutomationControlled")
|
|
if remote_debugging_pipe:
|
|
options.add_argument("--remote-debugging-pipe")
|
|
options.add_experimental_option("excludeSwitches", ["enable-automation"])
|
|
options.add_experimental_option("useAutomationExtension", False)
|
|
return options
|
|
|
|
|
|
def set_stealth_driver(driver):
|
|
driver.execute_script(
|
|
"Object.defineProperty(navigator, 'webdriver', {get: () => undefined})"
|
|
)
|
|
|
|
|
|
def count_google_review_pages(driver):
|
|
result = driver.find_element("class name", 'Qha3nb').text
|
|
result = result.replace(',', '')
|
|
result = result.split(' ')
|
|
result = result[0].split('\n')
|
|
return int(result[0]) // 10 + 1
|
|
|
|
|
|
def scroll_google_reviews(driver, review_count, label='google_scroll...', delay_seconds=3):
|
|
print(label, end="")
|
|
time.sleep(delay_seconds)
|
|
scrollable_div = driver.find_element("xpath", GOOGLE_REVIEW_SCROLL_XPATH)
|
|
google_scroller = None
|
|
for _ in range(review_count):
|
|
try:
|
|
google_scroller = driver.execute_script(
|
|
GOOGLE_REVIEW_SCROLL_SCRIPT,
|
|
scrollable_div,
|
|
)
|
|
time.sleep(delay_seconds)
|
|
print('.', end="")
|
|
except Exception as error:
|
|
print(f"Error while {label.strip('.')}: {error}")
|
|
break
|
|
print('')
|
|
return google_scroller
|
|
|
|
|
|
def sanitize_review_name(name):
|
|
return re.sub(r'[^a-zA-Z0-9]', '', name)
|
|
|
|
|
|
def normalize_google_media_url(style_text):
|
|
return re.sub(r'=\S*-p-k-no', '=-no', (re.findall(r"['\"](.*?)['\"]", style_text))[0])
|
|
|
|
|
|
def build_review_media_path(review_name, visit_date, filename, base_dir="./Output/Pics"):
|
|
review_dir = Path(base_dir) / sanitize_review_name(review_name) / visit_date
|
|
review_dir.mkdir(parents=True, exist_ok=True)
|
|
return str(review_dir / filename)
|
|
|
|
|
|
def normalize_wordpress_date(date_text, now=None):
|
|
reference_time = now or datetime.now()
|
|
normalized = str(date_text).strip()
|
|
lowered = normalized.lower()
|
|
|
|
if lowered == "a day ago":
|
|
publish_at = reference_time - timedelta(days=1)
|
|
elif "day" in lowered:
|
|
publish_at = reference_time - timedelta(days=int(re.sub(r'[^0-9]', '', lowered) or 0))
|
|
elif lowered == "a week ago":
|
|
publish_at = reference_time - relativedelta(weeks=1)
|
|
elif "week" in lowered:
|
|
publish_at = reference_time - relativedelta(weeks=int(re.sub(r'[^0-9]', '', lowered) or 0))
|
|
elif lowered == "a month ago":
|
|
publish_at = reference_time - relativedelta(months=1)
|
|
elif "month" in lowered:
|
|
publish_at = reference_time - relativedelta(months=int(re.sub(r'[^0-9]', '', lowered) or 0))
|
|
elif lowered == "a year ago":
|
|
publish_at = reference_time - relativedelta(years=1)
|
|
elif "year" in lowered:
|
|
publish_at = reference_time - relativedelta(years=int(re.sub(r'[^0-9]', '', lowered) or 0))
|
|
else:
|
|
compact_date = normalized.replace(" ", "")
|
|
publish_at = datetime.strptime(compact_date[:3] + "/" + compact_date[3:] + "/01", '%b/%Y/%d')
|
|
|
|
return publish_at.strftime('%Y-%m-%dT22:00:00') |