Merge branch '3.0' of https://github.com/timberjoegithub/GoogleScrape into 3.0

# Conflicts:
#	social.py

.github/workflows/pylint.yml

@@ -1,23 +1,26 @@
-name: Pylint
+name: Validate
 
 on: [push]
 
 jobs:
-  build:
+  validate:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: ["3.8", "3.9", "3.10"]
+        python-version: ["3.11"]
     steps:
     - uses: actions/checkout@v3
     - name: Set up Python ${{ matrix.python-version }}
       uses: actions/setup-python@v3
       with:
         python-version: ${{ matrix.python-version }}
-    - name: Install dependencies
+    - name: Install validation dependencies
       run: |
         python -m pip install --upgrade pip
-        pip install pylint
+        pip install python-dateutil selenium
-    - name: Analysing the code with pylint
+    - name: Compile core scripts
       run: |
-        pylint $(git ls-files '*.py')
+        python -m py_compile Google2xls.py WP2xls.py social.py text.py review_workflow_utils.py
+    - name: Run focused unit tests
+      run: |
+        python -m unittest discover -s tests

.gitignore

@@ -3,19 +3,7 @@ __pycache__/
 out.xlsx
 Output/
 env.py
-requirements.txt
-test.py
-ThreadsPost.py
-TikTokPost.py
-WP2xls.py
-Xpost.py
-Instapost.py
-InstaBot.py
-Google2xls.py
-google.py
 geckodriver.log
-Facepost.py
 config/
-env.py
-social.py
 .vscode/
+debug.log

Google2xls.py

@@ -0,0 +1,152 @@
+import time
+import os
+from selenium import webdriver
+from selenium.webdriver.chrome.webdriver import WebDriver
+from selenium.webdriver.chrome.service import Service
+from selenium.webdriver.common.by import By
+import re
+import urllib3
+from urllib.request import urlretrieve
+from openpyxl import Workbook
+import pandas as pd
+from env import URL, DriverLocation
+from datetime import datetime
+
+from review_workflow_utils import (
+    build_google_chrome_options,
+    build_review_media_path,
+    count_google_review_pages,
+    normalize_google_media_url,
+    sanitize_review_name,
+    scroll_google_reviews,
+    set_stealth_driver,
+)
+
+today = datetime.today().strftime('%Y-%m-%d')
+
+# Rotate User Agent would be helpful
+
+def get_data(driver):
+    """
+    this function get main text, score, name
+    """
+    print('get data...')
+    # Click on more botton on each text reviews
+    more_elemets = driver.find_elements(By.CSS_SELECTOR, '.w8nwRe.kyuRq')
+    for list_more_element in more_elemets:
+        list_more_element.click()
+    # Find Pictures that have the expansion indicator to see the rest of the pictures under them and click it to expose them all
+    more_pics = driver.find_elements(By.CLASS_NAME, 'Tya61d') 
+    for list_more_pics in more_pics:
+        if 'showMorePhotos' in  list_more_pics.get_attribute("jsaction") :
+            print('Found extra pics')
+            list_more_pics.click()
+    elements = driver.find_elements(By.CLASS_NAME, 'jftiEf')
+    lst_data = []
+    for data in elements:
+        name = data.find_element(By.CSS_SELECTOR, 'div.d4r55.YJxk2d').text 
+        try: address = data.find_element(By.CSS_SELECTOR, 'div.RfnDt.xJVozb').text
+        except: address = 'Unknonwn'
+        print ('Name of location: ',name, '   Address:',address)
+        try: visitdate = data.find_element(By.CSS_SELECTOR, 'span.rsqaWe').text
+        except: visitdate = "Unknown"
+        print('Visited: ',visitdate)  
+        try: text = data.find_element(By.CSS_SELECTOR, 'div.MyEned').text 
+        except: text = '' 
+        try: score = data.find_element(By.CSS_SELECTOR, 'span.kvMYJc').get_attribute("aria-label")  #find_element(By.CSS_SELECTOR,'aria-label').text #)  ##QA0Szd > div > div > div.w6VYqd > div:nth-child(2) > div > div.e07Vkf.kA9KIf > div > div > div.m6QErb.DxyBCb.kA9KIf.dS8AEf > div.m6QErb > div:nth-child(3) > div:nth-child(2) > div > div:nth-child(4) > div.DU9Pgb > span.kvMYJc
+        except: score = "Unknown"
+        more_specific_pics = data.find_elements(By.CLASS_NAME, 'Tya61d') 
+        pics= [] 
+        pics2 = []
+        # check to see if folder for pictures and videos already exists, if not, create it
+        cleanname = sanitize_review_name(name)
+        if not os.path.exists('./Output/Pics/'+cleanname):
+            os.makedirs('./Output/Pics/'+cleanname)
+        # Walk through all the pictures and videos for a given review
+        for lmpics in more_specific_pics:  
+            # Grab URL from style definiton (long multivalue string), and remove the -p-k so that it is full size 
+            urlmedia = normalize_google_media_url(lmpics.get_attribute("style"))
+            print ('URL : ',urlmedia)
+            pics.append(urlmedia)
+#            time.sleep(2)
+#            photoindex = str(lmpics.get_attribute("data-photo-index"))
+            # Grab the name of the file and remove all spaces and special charecters to name the folder
+            filename = re.sub( r'[^a-zA-Z0-9]','', str(lmpics.get_attribute("aria-label")))
+#            filename = re.sub( r'[^a-zA-Z0-9]','', filename) 
+            
+            if lmpics == more_specific_pics[0]:
+                lmpics.click()
+                time.sleep(2)
+                #iframe = driver.find_element(By.TAG_NAME, "iframe")
+                tempdate = str((driver.find_element(By.CLASS_NAME,'mqX5ad')).text).rsplit("-",1)
+                visitdate = re.sub( r'[^a-zA-Z0-9]','',tempdate[1])
+                print ('Visited: ',visitdate)
+                # driver.switch_to.default_content()
+
+            # Check to see if it has a sub div, which represents the label with the video length displayed, this will be done 
+            # because videos are represented by pictures in the main dialogue, so we need to click through and grab the video URL
+            if (lmpics.find_elements(By.CSS_SELECTOR,'div.fontLabelMedium.e5A3N')) :
+                ext='.mp4'
+                lmpics.click()
+                time.sleep(2)
+                # After we click the right side is rendered in an inframe, Store iframe web element
+                iframe = driver.find_element(By.TAG_NAME, "iframe")
+                # switch to selected iframe
+                driver.switch_to.frame(iframe)
+                # Now find button and click on button
+                video_elements = driver.find_elements(By.XPATH ,'//video') #.get_attribute('src')
+                urlmedia = str((video_elements[0]).get_attribute("src"))
+                # return back away from iframe
+                driver.switch_to.default_content()
+            else:
+                # The default path if it is not a video link
+                ext='.jpg' 
+            # Add the correct extension to the file name
+            filename = filename+ext
+            # Test to see if file already exists, and if it does not grab the media and store it in location folder
+            picsLocalpath = build_review_media_path(name, visitdate, filename)
+            if not os.path.isfile(picsLocalpath):
+                urlretrieve(urlmedia, picsLocalpath)
+            # Store the local path to be used in the excel document
+            pics2.append(picsLocalpath)
+        dictPostComplete= {'google':1,'web':0,'yelp':0,'facebook':0,'xtwitter':0,'Instagram':0,'tiktok':0}
+        lst_data.append([name , text, score,pics,pics2,"GoogleMaps",visitdate,address,dictPostComplete])
+    return lst_data
+
+# Grab a count of how far we need to scroll
+def counter(driver):
+    return count_google_review_pages(driver)
+
+# Do the scrolling
+def scrolling(driver, review_count):
+    return scroll_google_reviews(driver, review_count, label='scrolling...')
+
+def write_to_xlsx(data):
+    print('write to excel...')
+
+    cols = ["name", "comment", 'rating','picsURL','picsLocalpath','source','date','address','dictPostComplete']
+    df = pd.DataFrame(data, columns=cols)
+    df.to_excel('./Output/reviews.xlsx')
+
+
+if __name__ == "__main__":
+    print('starting...')
+    options = build_google_chrome_options(
+        ignore_certificate_errors=True,
+        ignore_ssl_errors=True,
+    )
+    # Setting the driver path and requesting a page 
+    driver = webdriver.Chrome(options=options) # Firefox(options=options)  
+    # Changing the property of the navigator value for webdriver to undefined 
+    set_stealth_driver(driver)
+    driver.get(URL)
+    time.sleep(5)
+
+    review_count = counter(driver)
+    scrolling(driver, review_count)
+
+    data = get_data(driver)
+    driver.close()
+
+    write_to_xlsx(data)
+    print('Done!')

WP2xls.py

@@ -0,0 +1,283 @@
+import ast
+import base64
+import pandas as pd
+import requests
+from openpyxl import Workbook, load_workbook
+from env import wpAPOurl, xls,user, password,wpAPI
+import datetime as dt
+import json
+import re
+#import socket
+import urllib3
+from datetime import datetime
+
+from review_workflow_utils import normalize_wordpress_date
+
+today = datetime.today().strftime('%Y-%m-%d')
+
+#print(urllib.request.urlopen("https://www.stackoverflow.com").getcode())
+
+def is_port_open(host, port):
+#    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    try:
+        # removehttps = re.sub(r'https://','', host)
+        # removetrailing = removehttps.split('/')[0]
+        isWebUp = urllib3.request("GET", host)
+        #sock.close()
+#        result = sock.connect_ex((host, port))
+        if isWebUp.status == 200:
+            return True
+    except Exception as error:
+        print ('Could not open port to website: ', host,  type(error))
+        return False        
+    
+# def to_json(obj):
+#      return json.dumps(obj, default=lambda obj: obj,__dict__ )
+
+def connect_and_auth_wp( headers):
+    # connect and  auth
+    data_string = f"{user}:{password}"
+    token = base64.b64encode(data_string.encode()).decode("utf-8")
+    headers = {"Authorization": f"Basic {token}"}
+    return (headers)
+
+def check_media(filename, headers):
+    # Regex gilename to format like in WordPress media name
+    #file_name_minus_extension = re.sub(r'\'|(....$)','', filename, flags=re.IGNORECASE)
+    file_name_minus_extension = filename
+    response = requests.get(wpAPI + "/media?search="+file_name_minus_extension, headers=headers)
+    try:
+        result = response.json()
+        file_id = int(result[0]['id'])
+        link = result[0]['guid']['rendered']
+        return file_id, link
+    except Exception:
+        print('    No existing media with same name in Wordpress media folder: ' + filename)
+        return False, False
+    
+def check_post(postname,postdate, headers):
+
+    response = requests.get(wpAPI+"/posts?search="+postname, headers=headers)
+    try:
+        result = response.json()
+        post_id = int(result[0]['id'])
+        post_date = result[0]['date']
+        if postdate == post_date:
+            return post_id
+    except Exception:
+        print('No existing post with same name: ' + postname)
+    return False
+def post_to_wp(title, content,  headers,date, rating,address, picslist):
+    # post
+    NewPost = False
+    countreview = False
+    addresshtml = re.sub(" ", ".",address)
+    googleadress = r"<a href=https://www.google.com/maps/dir/?api=1&destination="+addresshtml + r">"+address+r"</a>" # https://www.google.com/maps/dir/?api=1&destination=760+West+Genesee+Street+Syracuse+NY+13204
+    contentpics = ''
+    picl = picslist[1:-1] 
+    pic2 = picl.replace(",","")#re.sub(r',','',picl) #re.sub( r'[^a-zA-Z0-9]','',tempdate[1])
+    pic3= pic2.replace("'","")
+#    print (pic3)
+    pidchop = pic3.split(" ")
+    linkslist=[]
+    # linkslist.clear
+    print ('    Figuring out date of Post : ',title)
+    date = normalize_wordpress_date(date)
+    try:
+        post_id = check_post(title, str(date), headers)
+    except Exception as error:
+        print ('Could not check to see post already exists', type(error).__name__)
+        post_id = False
+    if ( post_id == False):        
+#    if ((title) != False):        
+        googleadress =  r"<a href=https://www.google.com/maps/dir/?api=1&destination="+addresshtml + r">"+address+r"</a>"
+        post_data = {
+            "title": title,
+    #        "content": address+'\n\n'+content+'\n'+rating+'\n\n' ,
+            "content": googleadress+'\n\n'+content+'\n'+rating ,
+            "status": "publish",  # Set to 'draft' if you want to save as a draft
+            "date": date,
+     #       "date": str(newdate)+'T22:00:00',
+        # "author":"joesteele" 
+        }
+        try: 
+        # response = requests.post(wpAPOurl, json=post_data, headers=headers)
+        # myobj = {'json': post_data, 'headers': headers }
+        # response = requests.post(wpAPOurl, json = myobj)
+            headers2 = headers
+            response = requests.post(wpAPOurl, json = post_data, headers=headers2)
+            if ( response.status_code != 201 ):
+                print ('Error: ',response)
+    #            print ('json:'+ post_data+' headers :'+headers)
+            # else:
+            #     for loop in linkslist:
+            #         print ('    Adding ', loop['link'], ' to posting')
+            #         try:
+            #             contentpics += r'<img src="'+ loop['link'] + r' alt="' + title +r'">' +'\n\n'
+            #         except Exception as error:
+            #             print("An error occurred:", type(error).__name__) # An error occurred:
+            #     post_data = ()
+            #     response2 = requests.post(wpAPI+"/media/" + response['id'], json = post_data, headers=headers2)
+            else:
+                NewPost = True
+                post_id_json = response.json()
+                post_id = post_id_json.get('id')
+                print ('    New post is has post_id = ',post_id)
+        except Exception as error:
+            print("An error occurred:", type(error).__name__) # An error occurred:
+        postneedsupdate = True
+    else:
+        print ('    Post already existed: Post ID : ',post_id)
+    for pic in pidchop:
+        picslice2 = pic.split("/")[-1]
+        picslice = picslice2.split(".")
+        picname = picslice[0] 
+        caption =title
+        description = title+"\n"+address
+        print ('  Found Picture: ',picname)
+        file_id, link = check_media(picname, headers)
+#        link = linknew['rendered']
+        if (file_id) is False:
+            print ('    '+picname+' was not already found in library, adding it')
+            countreview = True
+            image = {
+                "file": open(pic, "rb"),
+                "post": post_id,
+                "caption": caption,
+                "description": description
+            }
+            try:
+                image_response = requests.post(wpAPI + "/media", headers=headers, files=image)
+            except Exception as error:
+                print("    An error uploading picture ' + picname+ ' occurred:", type(error).__name__) # An error occurred:
+            if ( image_response.status_code != 201 ):
+                print ('    Error- Image ',picname,' was not successfully uploaded.  response: ',image_response)
+            else:
+                PicDict=image_response.json()
+                file_id= PicDict.get('id')
+                link = PicDict.get('guid').get("rendered")
+                print ('    ',picname,' was successfully uploaded to website with ID: ',file_id, link)
+            try:
+                linksDict = {'file_id' : file_id , 'link' : link}
+                linkslist.append(linksDict)
+            except Exception as error:
+                print("    An error adding to dictionary " , file_id , link , " occurred:", type(error).__name__) # An error occurred:
+        else:
+            print ('    Photo ',picname,' was already in library and added to post with ID: ',file_id,' : ',link)
+            try:
+                image_response = requests.post(wpAPI + "/media/" + str(file_id), headers=headers, data={"post" : post_id})
+            except Exception as error:
+                print ('    Error- Image ',picname,' was not attached to post.  response: ',image_response)
+            try:
+                post_response = requests.get(wpAPI + "/posts/" + str(post_id), headers=headers)
+                if (link in str(post_response.text)):
+                    print ('    Image link for ', picname, 'already in content of post: ',post_id, post_response.text, link)
+                else:
+                    linkslist.append({'file_id' : file_id , 'link' : link})
+                    countreview = True
+            except Exception as error:
+                print("    An error loading the metadata from the post " + post_response.title + ' occurred:", type(error).__name__) # An error occurred')
+    #ratinghtml = post_response.text
+    firstMP4 = True
+    for piclink in linkslist:
+        #for loop in linkslist:
+        print ('    Adding ', piclink['link'], ' to posting')
+        try:
+            ext = (piclink['link'].split( '.')[-1] )
+            if (ext == 'mp4'):
+                if (firstMP4):
+                    contentpics += '\n' +r'[evp_embed_video url="' + piclink['link'] + r'" autoplay="true"]'
+                    firstMP4 = False
+                else:
+                    contentpics += '\n' +r'[evp_embed_video url="' + piclink['link'] + r'"]'           
+                #[evp_embed_video url="http://example.com/wp-content/uploads/videos/vid1.mp4" autoplay="true"]
+            else:
+                contentpics += '\n '+r'<div class="col-xs-4"><img id="'+str(file_id)+r'"' + r'src="' + piclink['link'] + r'"></div>'
+#            contentpics += '\n '+r'<img src="'+ piclink['link'] + '> \n'
+            #contentpics += r'<img src="'+ piclink['link'] + r' alt="' + title +r'">' +'\n\n'
+        except Exception as error:
+            print("An error occurred:", type(error).__name__) # An error occurred:
+    try:
+        response_piclinks = requests.post(wpAPI+"/posts/"+ str(post_id), data={"content" : googleadress+'\n\n'+content+'\n'+rating  + contentpics, "featured_media" : file_id}, headers=headers)
+    except Exception as error:
+        print("    An error writing images to the post " + post_response.title + ' occurred:", type(error).__name__) # An error occurred')
+    return NewPost
+
+# def add_image(wpAPI,pic, picname,caption, description,headers):
+#     # loop thru
+#     image = {
+#         "file": open(pic, "rb"),
+#         "caption": caption,
+#         "description": description
+#     }
+#     try:
+#         image_response = requests.post(wpAPI + "/media", headers=headers, files=image)
+#     except Exception as error:
+#         print("An error occurred:", type(error).__name__) # An error occurred:
+#     if ( image_response != 201 ):
+#         print (image_response)
+#        # print ('json:'+ post_data+' headers :'+headers)
+#         print ('image response: ',image_response)
+#     return image_response
+
+# def read_from_xlsx(ws):
+#     print('read from excel...')
+#     return (ws)
+
+def process_reviews(ws, headers):
+    # Process
+    needreversed = False
+    totalcount = 2
+    count = 0
+    rows = list(ws.iter_rows(min_row=1, max_row=ws.max_row))
+    if needreversed:
+        rows = reversed(rows)
+    for processrow in rows:
+        if (count >= totalcount):
+            print ('Exceeded the number of posts per run, exiting')
+            break
+#    for processrow in (ws.rows):
+#        If reviewxls (comment has content) and (picsURL has connect)
+        if processrow[1].value != "name":  # Skip header line of xls sheet
+            print ("Processing : ",processrow[1].value)
+            # ast.literal_eval(deployments) = processrow[9].value
+            writtento = (ast.literal_eval(processrow[9].value))
+            # Check to see if the website has already been written to according to the xls sheet, if it has not... then process
+            if ((writtento["web"]) == 0) and (is_port_open(wpAPI, 443)) and (processrow[2].value != None) :
+                try: 
+                    NewPost = post_to_wp(processrow[1].value, processrow[2].value, headers ,processrow[7].value, processrow[3].value, processrow[8].value, processrow[5].value)
+                    try:
+                        writtento["web"] = 1
+                        #processrow[9] = writtento[9]
+#                        processrow[9].value = '"'+writtento+'"'
+                        processrow[9].value = str(writtento)
+                    except Exception as error:
+                        print("An error occurred writing value Excel file:", type(error).__name__) # An error occurred:
+                    print ('Success Posting: '+processrow[1].value)# ',processrow[1].value, processrow[2].value, headers,processrow[7].value, processrow[3].value,processrow[8].value, processrow[5].value, temp3["web"] )
+                    if NewPost == True:
+                        count +=1
+                    try: 
+                        wb.save(xls)
+                    except Exception as error:
+                        print("An error occurred writing Excel file:", type(error).__name__) # An error occurred:
+                except: 
+                    print ('Error writing post : ',processrow[1].value, processrow[2].value, headers,processrow[7].value, processrow[3].value,processrow[8].value, processrow[5].value, writtento["web"] )
+    return (headers)
+
+if __name__ == "__main__":
+    countreview = False
+    headers = {}
+    wb = load_workbook(filename = xls)
+    ws = wb['Sheet1']
+    print('starting...')
+   # read_from_xlsx(ws)
+    print('Connect and auth to Wordpress')
+    data_string = f"{user}:{password}"
+    token = base64.b64encode(data_string.encode()).decode("utf-8")
+    headers = {"Authorization": f"Basic {token}"}
+#    print ('Headers:',headers)
+    connect_and_auth_wp( headers)
+    print('Processing Reviews')
+#    print ('Headers:',headers)
+    process_reviews(ws, headers)  
+    print('Done!')

requirements.txt

@@ -1,13 +1,14 @@
-autopep8==1.5.7
+aiohttp
-et-xmlfile==1.1.0
+googlemaps
-numpy==1.21.2
+instagrapi
-openpyxl==3.0.9
+jsonpickle
-pandas==1.3.3
+moviepy
-pkg_resources==0.0.0
+mysqlclient
-pycodestyle==2.7.0
+openpyxl
-python-dateutil==2.8.2
+pandas
-pytz==2021.1
+python-dateutil
-selenium==3.141.0
+requests
-six==1.16.0
+selenium
-toml==0.10.2
+sqlalchemy
-urllib3==1.26.6
+tweepy
+urllib3

review_workflow_utils.py

@@ -0,0 +1,121 @@
+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')

tests/test_review_workflow_utils.py

@@ -0,0 +1,62 @@
+import unittest
+from datetime import datetime
+
+from review_workflow_utils import (
+    build_review_media_path,
+    count_google_review_pages,
+    normalize_google_media_url,
+    normalize_wordpress_date,
+    sanitize_review_name,
+)
+
+
+class FakeElement:
+    def __init__(self, text):
+        self.text = text
+
+
+class FakeDriver:
+    def __init__(self, text):
+        self.text = text
+
+    def find_element(self, by, value):
+        self.last_lookup = (by, value)
+        return FakeElement(self.text)
+
+
+class ReviewWorkflowUtilsTests(unittest.TestCase):
+    def test_count_google_review_pages_rounds_up_by_review_page(self):
+        driver = FakeDriver("25 reviews")
+
+        page_count = count_google_review_pages(driver)
+
+        self.assertEqual(page_count, 3)
+        self.assertEqual(driver.last_lookup, ("class name", "Qha3nb"))
+
+    def test_normalize_wordpress_date_handles_relative_weeks(self):
+        publish_date = normalize_wordpress_date("2 weeks ago", now=datetime(2026, 5, 27, 9, 30, 0))
+
+        self.assertEqual(publish_date, "2026-05-13T22:00:00")
+
+    def test_normalize_wordpress_date_handles_month_year_strings(self):
+        publish_date = normalize_wordpress_date("Jan 2025")
+
+        self.assertEqual(publish_date, "2025-01-01T22:00:00")
+
+    def test_normalize_google_media_url_strips_thumbnail_suffix(self):
+        style_text = 'background-image: url("https://example.com/photo=s1600-p-k-no");'
+
+        self.assertEqual(
+            normalize_google_media_url(style_text),
+            "https://example.com/photo=-no",
+        )
+
+    def test_build_review_media_path_sanitizes_review_name(self):
+        media_path = build_review_media_path("Joe's Diner!", "20260527", "front.jpg", base_dir="/tmp/review-tests")
+
+        self.assertEqual(media_path, "/tmp/review-tests/JoesDiner/20260527/front.jpg")
+        self.assertEqual(sanitize_review_name("Joe's Diner!"), "JoesDiner")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_social.py

@@ -0,0 +1,50 @@
+import unittest
+from unittest.mock import patch, MagicMock
+import social
+
+class SocialPostingFlowTests(unittest.TestCase):
+    def setUp(self):
+        # Patch env and Posts for isolation
+        self.env_patch = patch('social.env')
+        self.mock_env = self.env_patch.start()
+        self.mock_env.request_timeout = 1
+        self.mock_env.facebooksleep = 0
+        self.mock_env.forcegoogleupdate = False
+        self.mock_env.xls = 'dummy.xlsx'
+        self.mock_env.mariadb = False
+        self.mock_env.mariadbuser = 'user'
+        self.mock_env.mariadbpass = 'pass'
+        self.mock_env.mariadbserver = 'localhost'
+        self.mock_env.mariadbdb = 'db'
+        self.mock_env.needreversed = False
+        self.mock_env.postssession = MagicMock()
+        self.mock_env.posts = []
+        self.mock_env.xlsdf = MagicMock()
+        self.mock_env.posts = []
+        self.mock_env.postssession.query.return_value.count.return_value = 0
+        self.addCleanup(self.env_patch.stop)
+
+    @patch('social.requests.post')
+    @patch('builtins.open', new_callable=MagicMock)
+    def test_post_facebook_video_handles_success(self, mock_open, mock_post):
+        mock_post.return_value.json.return_value = {'id': '123'}
+        mock_open.return_value.__enter__.return_value = MagicMock()
+        result = social.post_facebook_video('groupid', ['video.mp4'], 'token', 'title', 'content', 'date', 'rating', 'address')
+        self.assertEqual(result, {'id': '123'})
+
+    @patch('social.requests.post', side_effect=AttributeError('fail'))
+    @patch('builtins.open', new_callable=MagicMock)
+    def test_post_facebook_video_handles_error(self, mock_open, mock_post):
+        mock_open.return_value.__enter__.return_value = MagicMock()
+        result = social.post_facebook_video('groupid', ['video.mp4'], 'token', 'title', 'content', 'date', 'rating', 'address')
+        self.assertFalse(result)
+
+    def test_write_to_database_handles_empty(self):
+        # Should not raise
+        data = []
+        local_outputs = {'xlsdf': MagicMock(), 'posts': [], 'postssession': MagicMock()}
+        result = social.write_to_database(data, local_outputs)
+        self.assertEqual(result, data)
+
+if __name__ == '__main__':
+    unittest.main()

text.py

@@ -1,3 +1,14 @@
+"""
+    Sends a text message to a phone number through the specified carrier.
+
+    Args:
+        phone_number_inside (str): The phone number to send the message to.
+        carrier_inside (str): The carrier of the phone number.
+        message_inside (str): The message content to be sent.
+
+    Returns:
+        None
+    """
 import smtplib
 import sys
 
@@ -11,13 +22,24 @@ CARRIERS = {
 EMAIL = "EMAIL"
 PASSWORD = "PASSWORD"
 
-def send_message(phone_number, carrier, message):
+def send_message(phone_number_inside, carrier_inside, message_inside):
-    recipient = phone_number + CARRIERS[carrier]
+    """
+    Sends a text message to a phone number through the specified carrier.
+
+    Args:
+        phone_number_inside (str): The phone number to send the message to.
+        carrier_inside (str): The carrier of the phone number.
+        message_inside (str): The message content to be sent.
+
+    Returns:
+        None
+    """
+    recipient = phone_number_inside + CARRIERS[carrier_inside]
     auth = (EMAIL, PASSWORD)
     server = smtplib.SMTP("smtp.gmail.com", 587)
     server.starttls()
     server.login(auth[0], auth[1])
-    server.sendmail(auth[0], recipient, message)
+    server.sendmail(auth[0], recipient, message_inside)
 
 if __name__ == "__main__":
     if len(sys.argv) < 4: