Google SGE & Bing Conversational Search Video Capture Automation

#| default_exp chatshot
#| export

# Set your keyword search, query or prompt: 
prompt = "Write a Hello World program in Python"

headless = False
slow_mo = 200
width = 1280
height = 1800
short_delay = 2
long_delay = 25
delete_after_days_old = 7
min_file_size = 200 * 1024  #KBs
video_folder = "/home/ubuntu/repos/intro/videos"

#| export
import os


files = os.listdir(video_folder)

for file in files:
    file_path = os.path.join(video_folder, file)
    if os.path.isfile(file_path):
        os.remove(file_path)
        print(f"Deleted {file}")

print("Deletion of all files complete.")

#| export

import asyncio
import nest_asyncio
from playwright.async_api import async_playwright


# Create a new event loop (necessary for Playwright in Juptyer)
nest_asyncio.apply()
loop = asyncio.get_event_loop()
new_loop = asyncio.new_event_loop()
asyncio.set_event_loop(new_loop)

async def browser_automation():
    """Records Google and Bing conversational web search as video."""

    async with async_playwright() as p:
        playwright = await async_playwright().start()
        browser = await playwright.chromium.launch_persistent_context(
            accept_downloads=True,
            args=["--window-position=100,100"],
            channel="chrome",
            downloads_path="/home/ubuntu/Downloads",
            executable_path="/usr/bin/google-chrome",
            headless=headless,
            no_viewport=False,
            record_video_dir=video_folder,
            record_video_size={"width": width, "height": height},
            slow_mo=slow_mo,
            user_data_dir="/home/ubuntu/.config/google-chrome/",
            viewport={"width": width, "height": height},
        )

        # Open the browser window
        page = await browser.new_page()

        # Perform a Bing Search
        placeholder = "Ask me anything"
        await page.goto("https://www.bing.com/")
        await asyncio.sleep(short_delay)
        await page.get_by_placeholder(placeholder).click()
        # await page.get_by_placeholder(placeholder).fill(prompt)
        await page.get_by_placeholder(placeholder).type(prompt)
        await page.get_by_placeholder(placeholder).press("Escape")
        await page.locator("#search_icon svg").click()
        # try:
        #     await page.wait_for_load_state('networkidle', timeout=long_delay)
        # except:
        #     ...
        await asyncio.sleep(long_delay)
        
        # Perform a Google Search
        await page.goto("https://www.google.com/")
        await asyncio.sleep(short_delay)
        await page.get_by_label("Search", exact=True).click()
        # await page.get_by_label("Search", exact=True).fill(prompt)
        await page.get_by_label("Search", exact=True).type(prompt)
        await page.get_by_label("Search", exact=True).press("Escape")
        await page.get_by_role("button", name="Google Search").click()
        # try:
        #     await page.wait_for_load_state('networkidle', timeout=long_delay)
        # except:
        #     ...
        await asyncio.sleep(long_delay)

        await browser.close()

# Set the new event loop as the current event loop
asyncio.run(browser_automation())

files = [f for f in os.listdir(video_folder) if os.path.isfile(os.path.join(video_folder, f))]

# Deletes the smallest of multiple files.
if len(files) == 1:
    print("Already down to 1 file.")
else:
    smallest_file = min(files, key=lambda f: os.path.getsize(os.path.join(video_folder, f)))
    smallest_file_path = os.path.join(video_folder, smallest_file)
    os.remove(smallest_file_path)
    print(f"Deleted the smallest file: {smallest_file}")

# Find the newest .webm file
webm_files = [f for f in os.listdir(video_folder) if f.lower().endswith('.webm') and os.path.isfile(os.path.join(video_folder, f))]
newest_webm_file = max(webm_files, key=lambda f: os.path.getmtime(os.path.join(video_folder, f))) if webm_files else None
print(f"I will convert {newest_webm_file} to mp4.")

import subprocess
from datetime import date


def convert_webm_to_mp4(input_path, output_path):
    cmd = [
        "ffmpeg",
        "-i", input_path,
        "-c:v", "libx264",
        "-preset", "slow",
        "-crf", "40",  # Low quality (51 = lowest)
        "-r", "15",
        "-vf", f"setpts={1/2}*PTS",  # Adjust playback speed using setpts filter
        "-an",  # This disables audio
        output_path
    ]
    
    subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)

input_webm_file = f"/home/ubuntu/repos/intro/videos/{newest_webm_file}"
output_mp4_file = f"/home/ubuntu/repos/intro/videos/{date.today()}.mp4"

convert_webm_to_mp4(input_webm_file, output_mp4_file)

import smtplib
from pathlib import Path
from email import encoders
from datetime import datetime
from email.mime.base import MIMEBase
from email.mime.text import MIMEText
from email.mime.image import MIMEImage
from email.mime.multipart import MIMEMultipart


file_name = Path(output_mp4_file).name

# Create the plain-text email
msg_text = f'''{prompt}

Your video recording of a Google and Bing search for {file_name} is attached.

Regards'''

# Create the HTML email
msg_html = f'''<html><head></head><body><h1>{prompt}</h1>
<p>Your video recording of a Google and Bing search for {file_name} is attached.</p>
<h3>Regards</h3></body></html>'''

with open('/home/ubuntu/repos/intro/service/mail_from.txt') as fh:
    email, paswd = [x.strip() for x in fh.readlines()]

with open('/home/ubuntu/repos/intro/service/mail_to.txt') as fh:
    mail_to = [x.strip() for x in fh.readlines()]

server = smtplib.SMTP('smtp.gmail.com', 587)
server.ehlo()
server.starttls()
server.login(email, paswd)

msgdict = MIMEMultipart()
msgdict.preamble = 'This is a multi-part message in MIME format.'
msgdict['From'] = email
msgdict['To'] = ', '.join([x for x in mail_to])
msgdict['Subject'] = prompt

# Create plain text and HTML alternatives
msg_alts = MIMEMultipart('alternative')
msgdict.attach(msg_alts)
msg_alts.attach(MIMEText(msg_text))
msg_alts.attach(MIMEText(msg_html, 'html'))

mimecats = MIMEBase('application', 'octet-stream')
with open(Path(output_mp4_file), 'br') as zfh:
    mimecats.set_payload(zfh.read())
encoders.encode_base64(mimecats)
mimecats.add_header('Content-Disposition', f"video; filename={file_name}")
msgdict.attach(mimecats)

try:
    server.sendmail(email, mail_to, msgdict.as_string())
    print ('email sent')
except:
    print ('error sending mail')

server.quit()