- Data Pipeline

The pipeline automates the full process of collecting, processing, and uploading movie data into a vector search engine (Elasticsearch) and cloud storage (Google BigQuery). This pipeline ensures scalable and efficient ingestion of new movie data for recommendation.

- Workflow

TT Code Scraping
Uses Playwright to extract unique IMDb tt codes from genre-based search results. Simulates user scrolling and clicking "Show More" to dynamically load more movie IDs.
Metadata & Synopsis Extraction
Asynchronously visits movie and plot pages using aiohttp + BeautifulSoup, extracting:
Title, year, duration, rating, genres
Short synopsis, long synopsis, and user summaries
Data Cleaning
Removes rows with no narrative text and cleans extra characters, usernames, and unnecessary suffixes.
Data Splitting
Metadata File: structured movie info
Synopsis File: cleaned narrative texts, labeled by type (short, long, summary)
Cloud Upload
Metadata is uploaded to Google Drive via API, then ingested into Google BigQuery.
Chunking & Embedding
Synopsis text is chunked (~250 words) and embedded with SBERT into dense vectors using SentenceTransformer.
Output Generation
Each chunk + vector is saved as a .json file and archived into a ZIP for ingestion.
ETL Trigger
The ZIP is moved to etl/data/jsons/, where run_etl.sh automatically unzips and indexes all files into Elasticsearch.

- Main Pipeline

`run_pipeline()`

Executes the main pipeline for processing IMDb movie data. The pipeline consists of the following steps: 1. Scrape IMDb tt codes. 2. Scrape movie metadata and synopsis. 3. Split combined CSV into metadata and synopsis files. 4. Upload metadata to Google Drive. 5. Upload metadata from Google Drive to BigQuery. 6. Clean synopsis data. 7. Chunk and embed synopsis data. 8. Move the resulting ZIP file to the etl/data/jsons/ directory.

Source code in app/pipeline/main_pipeline.py

def run_pipeline():
    """
    Executes the main pipeline for processing IMDb movie data. The pipeline consists of the following steps:
    1. Scrape IMDb tt codes.
    2. Scrape movie metadata and synopsis.
    3. Split combined CSV into metadata and synopsis files.
    4. Upload metadata to Google Drive.
    5. Upload metadata from Google Drive to BigQuery.
    6. Clean synopsis data.
    7. Chunk and embed synopsis data.
    8. Move the resulting ZIP file to the `etl/data/jsons/` directory.
    """
    # Step 1: Scrape tt codes
    logger.info("- Step 1: Scraping tt codes from IMDb...")
    run_scraper()

    # Step 2: Scrape movie metadata and synopsis
    logger.info("- Step 2: Scraping movie metadata and synopsis...")
    asyncio.run(run_crawler(
        output_dir="imdb_data",
        num_movies=0
    ))

    # Step 3: Clean full data
    logger.info("- Step 3: Cleaning synopsis data...")
    clean_romance_synopsis(
        input_csv=None,
        output_csv=None
    )

    # Step 4: Split combined XLSX into metadata + synopsis
    logger.info("- Step 4: Splitting movie data into metadata and synopsis files...")
    metadata_path, synopsis_path = split_movie_xlsx()

    # Step 5: Upload metadata to Google Drive
    logger.info("- Step 5: Uploading metadata Excel to Google Drive...")
    upload_metadata_to_drive(file_path=None, file_name=None)

    # Step 6: Upload metadata from Drive to BigQuery
    logger.info("- Step 6: Uploading metadata from Drive to BigQuery...")
    upload_single_file_to_bigquery(
        target_file_name=None,
        dataset_id="romance_dataset",
        table_name="full_data_table"
    )

    # Step 7: Chunk and embed
    logger.info("- Step 7: Chunking and embedding synopsis...")
    run_chunk_and_embed_pipeline(
        input_excel=None,
        output_dir=None
    )

    # Step 7.5: Move ZIP to etl/data/jsons/
    src_zip = f"{JSONS_FOLDER}.zip"
    dst_zip = "../etl/data/jsons/" + src_zip

    # Ensure the destination directory exists and move the ZIP file
    os.makedirs("../etl/data/jsons", exist_ok=True)
    shutil.move(src_zip, dst_zip)
    logger.info(f"- Moved ZIP to: {dst_zip}")

    # === CLEANUP ===
    try:
        # Delete romance_newbatch.tsv
        if os.path.exists(TSV_FILENAME):
            os.remove(TSV_FILENAME)
            logger.info(f"- Removed file: {TSV_FILENAME}")

        # Delete imdb_data/ folder
        if os.path.exists("imdb_data"):
            shutil.rmtree("imdb_data")
            logger.info("- Removed folder: imdb_data/")

        # Delete romance_chunks_json/ folder
        if os.path.exists(JSONS_FOLDER):
            shutil.rmtree(JSONS_FOLDER)
            logger.info(f"- Removed folder: {JSONS_FOLDER}/")

    except Exception as e:
        logger.warning(f"- Cleanup issue: {e}")

    # Log pipeline completion
    logger.info("- Pipeline execution complete!")

- Submodules

`extract_tt_codes(html)` `async`

Extracts IMDb tt codes from the provided HTML content.

Parameters:

Name	Type	Description	Default
`html`	`str`	The HTML content of the IMDb page.	required

Returns:

Name	Type	Description
`set`	`Set[str]`	A set of unique tt codes extracted from the page.

Source code in app/pipeline/scrape_tt_codes.py

async def extract_tt_codes(html: str) -> Set[str]:
    """
    Extracts IMDb tt codes from the provided HTML content.

    Args:
        html (str): The HTML content of the IMDb page.

    Returns:
        set: A set of unique tt codes extracted from the page.
    """
    soup = BeautifulSoup(html, "html.parser")
    movie_list = soup.find_all("li", {"class": "ipc-metadata-list-summary-item"})

    tt_codes = set()
    for movie in movie_list:
        link = movie.find("a", {"class": "ipc-title-link-wrapper", "href": True})
        if link:
            match = re.search(r"/title/(tt\d+)/", link["href"])
            if match:
                tt_codes.add(match.group(1))  # Extract tt_code
    return tt_codes

`run_scraper()`

Runs the IMDb tt code scraper.

This function initializes the asynchronous scraping process and ensures that the scrape_imdb_movie_codes coroutine is executed.

Source code in app/pipeline/scrape_tt_codes.py

def run_scraper() -> None:
    """
    Runs the IMDb tt code scraper.

    This function initializes the asynchronous scraping process and ensures
    that the `scrape_imdb_movie_codes` coroutine is executed.
    """
    asyncio.run(scrape_imdb_movie_codes())

`scrape_imdb_movie_codes()` `async`

Scrapes IMDb movie tt codes by navigating through the IMDb page and extracting tt codes. The extracted tt codes are saved to a TSV file.

Steps

Opens the IMDb URL using Playwright.
Extracts tt codes from the current page.
Saves new tt codes to a TSV file.
Clicks the "Show More" button to load additional movies and repeats the process.

Raises:

Type	Description
`Exception`	If any error occurs during the scraping process.

Source code in app/pipeline/scrape_tt_codes.py

async def scrape_imdb_movie_codes() -> None:
    """
    Scrapes IMDb movie tt codes by navigating through the IMDb page and extracting tt codes.
    The extracted tt codes are saved to a TSV file.

    Steps:
        1. Opens the IMDb URL using Playwright.
        2. Extracts tt codes from the current page.
        3. Saves new tt codes to a TSV file.
        4. Clicks the "Show More" button to load additional movies and repeats the process.

    Raises:
        Exception: If any error occurs during the scraping process.
    """
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=False)  # Set to True for headless mode
        page = await browser.new_page()

        # Load IMDb page
        logger.info(f"Opening {URL}")
        await page.goto(URL, timeout=120000)
        await asyncio.sleep(5)  # Wait for initial page load

        saved_codes = set()

        # Load existing codes from CSV
        if Path(TSV_FILENAME).exists():
            with open(TSV_FILENAME, "r", newline="") as file:
                reader = csv.reader(file, delimiter="\t")
                next(reader, None)
                saved_codes.update(row[0] for row in reader)

        # Open CSV for writing
        with open(TSV_FILENAME, "w", newline="") as file:
            writer = csv.writer(file, delimiter="\t")
            writer.writerow(["tconst"])

            click_attempts = 0

            while click_attempts < MAX_CLICKS:
                logger.info(f"Scraping Page {click_attempts + 1}...")

                # Extract tt_codes from the current page
                html = await page.content()
                new_tt_codes = await extract_tt_codes(html)

                # Save only new tt_codes
                unique_tt_codes = new_tt_codes - saved_codes
                if unique_tt_codes:
                    for code in unique_tt_codes:
                        writer.writerow([code])
                        saved_codes.add(code)
                    logger.info(f"Saved {len(unique_tt_codes)} new tt_codes to CSV.")

                # Try clicking "Show More" button
                try:
                    show_more_button = await page.wait_for_selector("button.ipc-see-more__button", timeout=5000)
                    if show_more_button:
                        logger.info("Clicking 'Show More' button...")
                        await show_more_button.scroll_into_view_if_needed()
                        await asyncio.sleep(1)  # Small delay before clicking
                        await show_more_button.click()
                        await asyncio.sleep(5)  # Wait for more movies to load
                    else:
                        logger.info("No more 'Show More' button found. Stopping scrape.")
                        break
                except:
                    logger.info("No more 'Show More' button detected.")
                    break

                click_attempts += 1

        await browser.close()
        logger.info("Scraping completed successfully!")

`IMDbCrawler`

An asynchronous web crawler for scraping IMDb movie metadata and synopses.

Attributes:

Name	Type	Description
`output_dir`	`Path`	Directory to store output files.
`min_delay`	`float`	Minimum delay between requests (for rate limiting).
`max_delay`	`float`	Maximum delay between requests.
`max_retries`	`int`	Number of retry attempts for failed requests.
`concurrent_requests`	`int`	Maximum number of concurrent requests.
`headers`	`dict`	HTTP headers to mimic browser behavior.
`session`	`ClientSession`	Reusable HTTP session for requests.

Source code in app/pipeline/extract_movie_data.py

class IMDbCrawler:
    """
    An asynchronous web crawler for scraping IMDb movie metadata and synopses.

    Attributes:
        output_dir (Path): Directory to store output files.
        min_delay (float): Minimum delay between requests (for rate limiting).
        max_delay (float): Maximum delay between requests.
        max_retries (int): Number of retry attempts for failed requests.
        concurrent_requests (int): Maximum number of concurrent requests.
        headers (dict): HTTP headers to mimic browser behavior.
        session (aiohttp.ClientSession): Reusable HTTP session for requests.
    """
    def __init__(self, output_dir: str = "imdb_data",
                 min_delay: float = 2.0,
                 max_delay: float = 4.0,
                 max_retries: int = 3,
                 concurrent_requests: int = 3) -> None:

        # Initialize directory structure
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(exist_ok=True)

        # Set crawler parameters
        self.min_delay = min_delay
        self.max_delay = max_delay
        self.max_retries = max_retries
        self.concurrent_requests = concurrent_requests

        # Configure request headers to mimic browser
        self.headers = {
            'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36...',
            'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,...',
            'Accept-Language': 'en-US,en;q=0.9',
            'Accept-Encoding': 'gzip, deflate, br',
            'Connection': 'keep-alive',
            'Upgrade-Insecure-Requests': '1'
        }

        self.session = None

    async def init_session(self) -> None:
        """Initializes the aiohttp session with SSL context."""
        if not self.session:
            # Create SSL context with proper certificate verification
            ssl_context = ssl.create_default_context(cafile=certifi.where())
            # Configure TCP connector with SSL settings
            connector = aiohttp.TCPConnector(
                ssl=ssl_context,
                verify_ssl=True,
                force_close=True  # Ensure connections are closed after use
            )
            # Create the client session
            self.session = aiohttp.ClientSession(
                headers=self.headers,
                connector=connector
            )

    async def get_movies_from_tsv(self, tsv_file: str = TSV_FILENAME) -> List[str]:
        """
        Reads movie IDs from a TSV file and filters out already processed IDs.

        Args:
            tsv_file (str): Path to the input TSV file.

        Returns:
            List[str]: List of IMDb movie IDs to process.
        """
        try:
            # Read the input TSV file
            df = pd.read_csv(tsv_file, sep='\t')
            # Check for and handle already processed movies
            processed_file = self.output_dir / PROCESSED_IDS
            if processed_file.exists():
                processed_df = pd.read_csv(processed_file, sep='\t')
                processed_ids = set(processed_df['tconst'].tolist())
                # Remove already processed movies from the queue
                df = df[~df['tconst'].isin(processed_ids)]
            logger.info(f"Found {len(df)} movies to process")
            return df['tconst'].tolist()
        except Exception as e:
            logger.info(f"Error reading TSV file: {e}")
            return []

    async def record_processed_id(self, tconst: str) -> None:
        """
        Records a processed IMDb ID into a TSV log with timestamp.

        Args:
            tconst (str): IMDb title ID.
        """
        processed_file = self.output_dir / PROCESSED_IDS
        try:
            # Create DataFrame with movie ID and timestamp
            df = pd.DataFrame({
                'tconst': [tconst],
                'processed_time': [datetime.now().isoformat()]
            })
            # Append or create the processed IDs file
            if processed_file.exists():
                df.to_csv(processed_file, mode='a', header=False, index=False, sep='\t')
            else:
                df.to_csv(processed_file, index=False, sep='\t')
        except Exception as e:
            logger.info(f"Error recording processed ID: {e}")

    async def extract_movie_data(self, tconst: str) -> Dict[str, Union[str, List[str], None]]:
        """
        Extracts full metadata and synopsis info from IMDb for a single title.

        Args:
            tconst (str): IMDb title ID.

        Returns:
            Dict: Extracted movie data.
        """
        # Initialize movie data structure with metadata
        movie_data = {
            'tconst': tconst,
            'crawl_timestamp': datetime.now().isoformat()
        }
        # Construct main movie page URL
        if not tconst.startswith('/title/'):
            main_url = f"https://www.imdb.com/title/{tconst}/"
        else:
            main_url = f"https://www.imdb.com{tconst}"
        logger.info(f"Fetching movie data from: {main_url}")
        try:
            # Fetch and process main page
            async with self.session.get(main_url) as response:
                if response.status == 200:
                    html = await response.text()
                    soup = BeautifulSoup(html, 'html.parser')
                    # Extract all available data points
                    movie_data.update({
                        'movie_title': self._extract_title(soup),
                        'movie_year': self._extract_year(soup, tconst),
                        'age_rating': self._extract_age_rating(soup, tconst),
                        'duration': self._extract_runtime(soup),
                        'imdb_rating': self._extract_imdb_rating(soup),
                        'short_synopsis': self._extract_short_synopsis(soup),
                        'top_5_actors': self._extract_top_actors(soup),
                        'genres': self._extract_genres(soup),
                        'poster_url': self._extract_poster_url(soup),
                        'summaries': self._extract_summaries(soup),
                        'long_synopsis': self._extract_long_synopsis(soup),
                    })
                    # Implement rate limiting
                    await asyncio.sleep(random.uniform(self.min_delay, self.max_delay))
                else:
                    logger.info(f"Failed to get movie {tconst}: Status {response.status}")
        except Exception as e:
            logger.info(f"Error extracting main data for {tconst}: {str(e)}")
        # Fetch and process plot summary page
        plot_url = f"https://www.imdb.com/title/{tconst}/plotsummary/"
        try:
            async with self.session.get(plot_url) as response:
                if response.status == 200:
                    html = await response.text()
                    soup = BeautifulSoup(html, 'html.parser')
                    movie_data['long_synopsis'] = self._extract_long_synopsis(soup)
                    movie_data['summaries'] = self._extract_summaries(soup)
                    await asyncio.sleep(random.uniform(self.min_delay, self.max_delay))
        except Exception as e:
            logger.info(f"Error extracting plot data for {tconst}: {str(e)}")
        return movie_data

    def _extract_title(self, soup: BeautifulSoup) -> Optional[str]:
        """
        Extracts the movie title from the given HTML soup.

        Args:
            soup (BeautifulSoup): Parsed IMDb movie page.

        Returns:
            Optional[str]: Extracted movie title or default fallback.
        """
        try:
            # IMDb page structures can change, so we use multiple selector strategies
            title_tag = soup.find("span", {"data-testid": "hero__primary-text"})
            # Extract and return the title if found
            return title_tag.get_text(strip=True) if title_tag else "Title not found"
        except Exception as e:
            logger.info(f"Error extracting title: {e}")
            return None

    def _extract_long_synopsis(self, soup: BeautifulSoup) -> Optional[str]:
        """
        Extracts the long synopsis from the movie's IMDb plot page.

        Args:
            soup (BeautifulSoup): Parsed IMDb plot page HTML.

        Returns:
            Optional[str]: Full synopsis text or fallback message.
        """
        try:
            # Locate the full synopsis section
            synopsis_section = soup.find("div", {"data-testid": "sub-section-synopsis"})
            # Extract the full synopsis text if available
            movie_synopsis = (
                synopsis_section.find("div", {"class": "ipc-html-content-inner-div", "role": "presentation"}).get_text(
                    strip=True)
                if synopsis_section else "Synopsis not found"
            )
            return movie_synopsis
        except Exception as e:
            logger.info(f"Error extracting long synopsis: {e}")
            return None

    def _extract_year(self, soup: BeautifulSoup, tconst: str) -> Optional[str]:
        """
        Extracts the movie release year using the IMDb releaseinfo URL.

        Args:
            soup (BeautifulSoup): Parsed IMDb movie page.
            tconst (str): IMDb title ID.

        Returns:
            Optional[str]: Release year or fallback message.
        """
        try:
            # Generate the release info URL dynamically based on tconst
            release_info_href = f"/title/{tconst}/releaseinfo"
            # Find the anchor tag that links to the release info page
            year_tag = soup.find("a", {"class": "ipc-link ipc-link--baseAlt ipc-link--inherit-color"},
                                 href=lambda x: x and release_info_href in x)
            # Extract and return the year if found
            return year_tag.get_text(strip=True) if year_tag else "Year not found"
        except Exception as e:
            logger.info(f"Error extracting year for {tconst}: {e}")
            return None

    def _extract_age_rating(self, soup: BeautifulSoup, tconst: str) -> Optional[str]:
        """
        Extracts the parental age rating.

        Args:
            soup (BeautifulSoup): Parsed IMDb movie page.
            tconst (str): IMDb title ID.

        Returns:
            Optional[str]: Age rating or fallback message.
        """
        try:
            # Generate the parental guide URL dynamically based on tconst
            parental_guide_href = f"/title/{tconst}/parentalguide"
            # Find the anchor tag that links to the parental guide page
            rating_tag = soup.find("a", {"class": "ipc-link ipc-link--baseAlt ipc-link--inherit-color"},
                                   href=lambda x: x and parental_guide_href in x)
            # Extract and return the age rating if found
            return rating_tag.get_text(strip=True) if rating_tag else "Rating not found"
        except Exception as e:
            logger.info(f"Error extracting age rating for {tconst}: {e}")
            return None

    def _extract_runtime(self, soup: BeautifulSoup) -> Optional[str]:
        """
        Extracts the movie runtime.

        Args:
            soup (BeautifulSoup): Parsed IMDb movie page.

        Returns:
            Optional[str]: Runtime string or fallback.
        """
        try:
            # Locate the runtime metadata section
            runtime_tag = soup.find("li", {"role": "presentation", "class": "ipc-metadata-list__item",
                                           "data-testid": "title-techspec_runtime"})
            # Extract runtime from the correct div container
            if runtime_tag:
                runtime_span = runtime_tag.find("div", {"class": "ipc-metadata-list-item__content-container"})
                return runtime_span.get_text(strip=True) if runtime_span else "Runtime not found"
            return "Runtime not found"
        except Exception as e:
            logger.info(f"Error extracting runtime: {e}")
            return None

    def _extract_imdb_rating(self, soup: BeautifulSoup) -> Optional[str]:
        """
        Extracts the IMDb rating of the movie.

        Args:
            soup (BeautifulSoup): Parsed IMDb movie page.

        Returns:
            Optional[str]: IMDb rating or fallback.
        """
        try:
            # Locate the IMDb rating section
            rating_tag = soup.find("div", {"data-testid": "hero-rating-bar__aggregate-rating__score",
                                           "class": "sc-d541859f-2 kxphVf"})
            # Extract rating from the correct span container
            if rating_tag:
                rating_span = rating_tag.find("span", {"class": "sc-d541859f-1 imUuxf"})
                return rating_span.get_text(strip=True) if rating_span else "Rating not found"
            return "Rating not found"
        except Exception as e:
            logger.info(f"Error extracting IMDb rating: {e}")
            return None

    def _extract_top_actors(self, soup: BeautifulSoup) -> List[str]:
        """
        Extracts the top 5 actors listed in the cast section.

        Args:
            soup (BeautifulSoup): Parsed IMDb movie page.

        Returns:
            List[str]: Actor names or fallback.
        """
        try:
            # Locate the actors section
            actors_section = soup.find("div",
                                       {"class": "ipc-shoveler ipc-shoveler--base ipc-shoveler--page0 title-cast__grid",
                                        "role": "group", "data-testid": "shoveler"})
            # Extract all actor items
            actors = []
            if actors_section:
                actor_items = actors_section.find_all("a", {"data-testid": "title-cast-item__actor"})
                actors = [actor.get_text(strip=True) for actor in actor_items[:5]]  # Extract top 5 actors
            return actors if actors else ["Actors not found"]
        except Exception as e:
            logger.info(f"Error extracting actors: {e}")
            return ["Actors not found"]

    def _extract_genres(self, soup: BeautifulSoup) -> List[str]:
        """
        Extracts the list of genres from the IMDb movie page.

        Args:
            soup (BeautifulSoup): Parsed IMDb movie page.

        Returns:
            List[str]: Genres or fallback.
        """
        try:
            # Locate the genre section
            genre_section = soup.find("div", {"class": "ipc-chip-list__scroller"})
            # Extract all genre items
            genres = []
            if genre_section:
                genre_items = genre_section.find_all("span")  # Adjust based on structure
                genres = [genre.get_text(strip=True) for genre in genre_items]
            return genres if genres else ["Genres not found"]
        except Exception as e:
            logger.info(f"Error extracting genres: {e}")
            return ["Genres not found"]

    def _extract_poster_url(self, soup: BeautifulSoup) -> Optional[str]:
        """
        Extracts the movie poster image URL.

        Args:
            soup (BeautifulSoup): Parsed IMDb movie page.

        Returns:
            Optional[str]: Poster URL or fallback.
        """
        try:
            # Locate the poster container
            poster_container = soup.find("div", {
                "class": "ipc-media ipc-media--poster-27x40 ipc-image-media-ratio--poster-27x40 ipc-media--media-radius ipc-media--baseAlt ipc-media--poster-l ipc-poster__poster-image ipc-media__img"})
            # Extract image tag and its URL
            if poster_container:
                img_tag = poster_container.find("img")
                return img_tag['src'] if img_tag and img_tag.has_attr('src') else "Poster not found"
            return "Poster not found"
        except Exception as e:
            logger.info(f"Error extracting poster URL: {e}")
            return None

    def _extract_short_synopsis(self, soup: BeautifulSoup) -> Optional[str]:
        """
        Extracts the short synopsis (1-line summary).

        Args:
            soup (BeautifulSoup): Parsed IMDb movie page.

        Returns:
            Optional[str]: Short synopsis or fallback.
        """
        try:
            # Locate the short synopsis section
            synopsis_tag = soup.find("span", {"role": "presentation", "data-testid": "plot-l", "class":"sc-191592d9-1 kBRZRe"})
            # Extract and return the short synopsis
            return synopsis_tag.get_text(strip=True) if synopsis_tag else "No short sum found"
        except Exception as e:
            logger.info(f"Error extracting short synopsis: {e}")
            return None

    def _extract_summaries(self, soup: BeautifulSoup) -> List[str]:
        """
        Extracts a list of user-submitted plot summaries.

        Args:
            soup (BeautifulSoup): Parsed IMDb plot summary page.

        Returns:
            List[str]: Plot summaries or fallback.
        """
        try:
            summaries_section = soup.find("div", {"data-testid": "sub-section-summaries"})

            summaries = []
            if summaries_section:
                summary_items = summaries_section.find_all("li")  # Adjust based on IMDb structure
                summaries = [summary.get_text(strip=True) for summary in summary_items]

            return summaries if summaries else ["No summaries found"]
        except Exception as e:
            logger.info(f"Error extracting summaries: {e}")
            return ["No summaries found"]

    def _extract_text(self, element: Optional[Tag]) -> Optional[str]:
        """
        Extracts text from a BeautifulSoup element.

        Args:
            element (Tag): A BeautifulSoup element.

        Returns:
            Optional[str]: Cleaned text or None.
        """
        return element.get_text(strip=True) if element else None

    async def crawl_movies(self, num_movies: Optional[int] = None) -> None:
        """
        Orchestrates crawling multiple IMDb movies concurrently and writes results to CSV in batches.

        Args:
            num_movies (int, optional): Maximum number of movies to crawl. Defaults to all.
        """
        try:
            await self.init_session()
            logger.info(f"Reading movie IDs from TSV...")
            movie_ids = await self.get_movies_from_tsv()
            if not movie_ids:
                logger.info("No movies found in TSV!")
                return
            # Apply optional movie limit
            if num_movies:
                movie_ids = movie_ids[:num_movies]
            logger.info(f"Found {len(movie_ids)} movies to crawl")
            # Set up concurrency control
            semaphore = asyncio.Semaphore(self.concurrent_requests)
            all_movie_data = []  # Temporary storage for batch processing
            csv_file = self.output_dir / OUTPUT_FILE
            # Track CSV file existence for header management
            self.csv_exists = csv_file.exists()

            async def crawl_with_semaphore(tcont):
                nonlocal all_movie_data
                async with semaphore:
                    movie_data = await self.extract_movie_data(tcont)
                    if movie_data:
                        all_movie_data.append(movie_data)
                        await self.record_processed_id(tcont)
                        # Batch write to CSV every 100 movies
                        if len(all_movie_data) >= 100:
                            df = pd.DataFrame(all_movie_data)
                            if self.csv_exists:
                                df.to_csv(csv_file, mode='a', header=False, index=False)
                            else:
                                df.to_csv(csv_file, index=False)
                                self.csv_exists = True
                            all_movie_data.clear()
                            logger.info(f"Wrote batch of {len(df)} movies to CSV")
            # Create and execute crawling tasks
            tasks = [crawl_with_semaphore(tcont) for tcont in movie_ids]
            # Monitor progress
            for i, task in enumerate(asyncio.as_completed(tasks)):
                await task
                logger.info(f"Progress: {i + 1}/{len(movie_ids)} movies processed")
            # Handle any remaining movies in the batch
            if all_movie_data:
                df = pd.DataFrame(all_movie_data)
                if self.csv_exists:
                    df.to_csv(csv_file, mode='a', header=False, index=False)
                else:
                    df.to_csv(csv_file, index=False)
                logger.info(f"Wrote final batch of {len(df)} movies to CSV")
        except Exception as e:
            logger.info(f"Error during crawling: {str(e)}")
        finally:
            await self.close_session()

    async def close_session(self) -> None:
        if self.session:
            await self.session.close()
            self.session = None
            logger.info("Session closed.")

`crawl_movies(num_movies=None)` `async`

Orchestrates crawling multiple IMDb movies concurrently and writes results to CSV in batches.

Parameters:

Name	Type	Description	Default
`num_movies`	`int`	Maximum number of movies to crawl. Defaults to all.	`None`

Source code in app/pipeline/extract_movie_data.py

async def crawl_movies(self, num_movies: Optional[int] = None) -> None:
    """
    Orchestrates crawling multiple IMDb movies concurrently and writes results to CSV in batches.

    Args:
        num_movies (int, optional): Maximum number of movies to crawl. Defaults to all.
    """
    try:
        await self.init_session()
        logger.info(f"Reading movie IDs from TSV...")
        movie_ids = await self.get_movies_from_tsv()
        if not movie_ids:
            logger.info("No movies found in TSV!")
            return
        # Apply optional movie limit
        if num_movies:
            movie_ids = movie_ids[:num_movies]
        logger.info(f"Found {len(movie_ids)} movies to crawl")
        # Set up concurrency control
        semaphore = asyncio.Semaphore(self.concurrent_requests)
        all_movie_data = []  # Temporary storage for batch processing
        csv_file = self.output_dir / OUTPUT_FILE
        # Track CSV file existence for header management
        self.csv_exists = csv_file.exists()

        async def crawl_with_semaphore(tcont):
            nonlocal all_movie_data
            async with semaphore:
                movie_data = await self.extract_movie_data(tcont)
                if movie_data:
                    all_movie_data.append(movie_data)
                    await self.record_processed_id(tcont)
                    # Batch write to CSV every 100 movies
                    if len(all_movie_data) >= 100:
                        df = pd.DataFrame(all_movie_data)
                        if self.csv_exists:
                            df.to_csv(csv_file, mode='a', header=False, index=False)
                        else:
                            df.to_csv(csv_file, index=False)
                            self.csv_exists = True
                        all_movie_data.clear()
                        logger.info(f"Wrote batch of {len(df)} movies to CSV")
        # Create and execute crawling tasks
        tasks = [crawl_with_semaphore(tcont) for tcont in movie_ids]
        # Monitor progress
        for i, task in enumerate(asyncio.as_completed(tasks)):
            await task
            logger.info(f"Progress: {i + 1}/{len(movie_ids)} movies processed")
        # Handle any remaining movies in the batch
        if all_movie_data:
            df = pd.DataFrame(all_movie_data)
            if self.csv_exists:
                df.to_csv(csv_file, mode='a', header=False, index=False)
            else:
                df.to_csv(csv_file, index=False)
            logger.info(f"Wrote final batch of {len(df)} movies to CSV")
    except Exception as e:
        logger.info(f"Error during crawling: {str(e)}")
    finally:
        await self.close_session()

`extract_movie_data(tconst)` `async`

Extracts full metadata and synopsis info from IMDb for a single title.

Parameters:

Name	Type	Description	Default
`tconst`	`str`	IMDb title ID.	required

Returns:

Name	Type	Description
`Dict`	`Dict[str, Union[str, List[str], None]]`	Extracted movie data.

Source code in app/pipeline/extract_movie_data.py

async def extract_movie_data(self, tconst: str) -> Dict[str, Union[str, List[str], None]]:
    """
    Extracts full metadata and synopsis info from IMDb for a single title.

    Args:
        tconst (str): IMDb title ID.

    Returns:
        Dict: Extracted movie data.
    """
    # Initialize movie data structure with metadata
    movie_data = {
        'tconst': tconst,
        'crawl_timestamp': datetime.now().isoformat()
    }
    # Construct main movie page URL
    if not tconst.startswith('/title/'):
        main_url = f"https://www.imdb.com/title/{tconst}/"
    else:
        main_url = f"https://www.imdb.com{tconst}"
    logger.info(f"Fetching movie data from: {main_url}")
    try:
        # Fetch and process main page
        async with self.session.get(main_url) as response:
            if response.status == 200:
                html = await response.text()
                soup = BeautifulSoup(html, 'html.parser')
                # Extract all available data points
                movie_data.update({
                    'movie_title': self._extract_title(soup),
                    'movie_year': self._extract_year(soup, tconst),
                    'age_rating': self._extract_age_rating(soup, tconst),
                    'duration': self._extract_runtime(soup),
                    'imdb_rating': self._extract_imdb_rating(soup),
                    'short_synopsis': self._extract_short_synopsis(soup),
                    'top_5_actors': self._extract_top_actors(soup),
                    'genres': self._extract_genres(soup),
                    'poster_url': self._extract_poster_url(soup),
                    'summaries': self._extract_summaries(soup),
                    'long_synopsis': self._extract_long_synopsis(soup),
                })
                # Implement rate limiting
                await asyncio.sleep(random.uniform(self.min_delay, self.max_delay))
            else:
                logger.info(f"Failed to get movie {tconst}: Status {response.status}")
    except Exception as e:
        logger.info(f"Error extracting main data for {tconst}: {str(e)}")
    # Fetch and process plot summary page
    plot_url = f"https://www.imdb.com/title/{tconst}/plotsummary/"
    try:
        async with self.session.get(plot_url) as response:
            if response.status == 200:
                html = await response.text()
                soup = BeautifulSoup(html, 'html.parser')
                movie_data['long_synopsis'] = self._extract_long_synopsis(soup)
                movie_data['summaries'] = self._extract_summaries(soup)
                await asyncio.sleep(random.uniform(self.min_delay, self.max_delay))
    except Exception as e:
        logger.info(f"Error extracting plot data for {tconst}: {str(e)}")
    return movie_data

`get_movies_from_tsv(tsv_file=TSV_FILENAME)` `async`

Reads movie IDs from a TSV file and filters out already processed IDs.

Parameters:

Name	Type	Description	Default
`tsv_file`	`str`	Path to the input TSV file.	`TSV_FILENAME`

Returns:

Type	Description
`List[str]`	List[str]: List of IMDb movie IDs to process.

Source code in app/pipeline/extract_movie_data.py

async def get_movies_from_tsv(self, tsv_file: str = TSV_FILENAME) -> List[str]:
    """
    Reads movie IDs from a TSV file and filters out already processed IDs.

    Args:
        tsv_file (str): Path to the input TSV file.

    Returns:
        List[str]: List of IMDb movie IDs to process.
    """
    try:
        # Read the input TSV file
        df = pd.read_csv(tsv_file, sep='\t')
        # Check for and handle already processed movies
        processed_file = self.output_dir / PROCESSED_IDS
        if processed_file.exists():
            processed_df = pd.read_csv(processed_file, sep='\t')
            processed_ids = set(processed_df['tconst'].tolist())
            # Remove already processed movies from the queue
            df = df[~df['tconst'].isin(processed_ids)]
        logger.info(f"Found {len(df)} movies to process")
        return df['tconst'].tolist()
    except Exception as e:
        logger.info(f"Error reading TSV file: {e}")
        return []

`init_session()` `async`

Initializes the aiohttp session with SSL context.

Source code in app/pipeline/extract_movie_data.py

async def init_session(self) -> None:
    """Initializes the aiohttp session with SSL context."""
    if not self.session:
        # Create SSL context with proper certificate verification
        ssl_context = ssl.create_default_context(cafile=certifi.where())
        # Configure TCP connector with SSL settings
        connector = aiohttp.TCPConnector(
            ssl=ssl_context,
            verify_ssl=True,
            force_close=True  # Ensure connections are closed after use
        )
        # Create the client session
        self.session = aiohttp.ClientSession(
            headers=self.headers,
            connector=connector
        )

`record_processed_id(tconst)` `async`

Records a processed IMDb ID into a TSV log with timestamp.

Parameters:

Name	Type	Description	Default
`tconst`	`str`	IMDb title ID.	required

Source code in app/pipeline/extract_movie_data.py

async def record_processed_id(self, tconst: str) -> None:
    """
    Records a processed IMDb ID into a TSV log with timestamp.

    Args:
        tconst (str): IMDb title ID.
    """
    processed_file = self.output_dir / PROCESSED_IDS
    try:
        # Create DataFrame with movie ID and timestamp
        df = pd.DataFrame({
            'tconst': [tconst],
            'processed_time': [datetime.now().isoformat()]
        })
        # Append or create the processed IDs file
        if processed_file.exists():
            df.to_csv(processed_file, mode='a', header=False, index=False, sep='\t')
        else:
            df.to_csv(processed_file, index=False, sep='\t')
    except Exception as e:
        logger.info(f"Error recording processed ID: {e}")

`run_crawler(output_dir='imdb_data', tsv_file=None, num_movies=0)` `async`

Entrypoint to run the IMDbCrawler with optional input and batch limits.

Parameters:

Name	Type	Description	Default
`output_dir`	`str`	Output directory path.	`'imdb_data'`
`tsv_file`	`str`	Path to custom TSV file of movie IDs.	`None`
`num_movies`	`int`	Number of movies to crawl (0 = all).	`0`

Source code in app/pipeline/extract_movie_data.py

async def run_crawler(output_dir: str = "imdb_data", tsv_file: Optional[str] = None, num_movies: int = 0) -> None:
    """
    Entrypoint to run the IMDbCrawler with optional input and batch limits.

    Args:
        output_dir (str): Output directory path.
        tsv_file (str, optional): Path to custom TSV file of movie IDs.
        num_movies (int): Number of movies to crawl (0 = all).
    """
    try:
        crawler = IMDbCrawler(
            output_dir=output_dir,
            min_delay=2.0,
            max_delay=4.0,
            concurrent_requests=3
        )
        if tsv_file:
            crawler.tsv_file = tsv_file  # Optional override
        await crawler.crawl_movies(num_movies=num_movies)
        logger.info("Movie crawl complete.")
    except Exception as e:
        logger.info(f"Error running crawler: {e}")
    finally:
        if crawler.session:
            await crawler.close_session()

`clean_romance_synopsis(input_csv=None, output_csv=None)`

Cleans a CSV file containing romance movie synopses by removing rows with missing data and processing the summaries column. The cleaned data is saved to an Excel file.

Parameters:

Name	Type	Description	Default
`input_csv`	`str`	Path to the input CSV file containing the full movie data. Default is "imdb_data/romance_newbatch.csv".	`None`
`output_csv`	`str`	Path to the output CSV file where the cleaned data will be saved. Default is "imdb_data/romance_full_cleaned.csv".	`None`

Returns:

Type	Description
`None`	None

Source code in app/pipeline/clean_synopsis.py

def clean_romance_synopsis(input_csv: str = None, output_csv: str = None) -> None:
    """
    Cleans a CSV file containing romance movie synopses by removing rows with missing data
    and processing the summaries column. The cleaned data is saved to an Excel file.

    Args:
        input_csv (str): Path to the input CSV file containing the full movie data.
                         Default is "imdb_data/romance_newbatch.csv".
        output_csv (str): Path to the output CSV file where the cleaned data will be saved.
                            Default is "imdb_data/romance_full_cleaned.csv".

    Returns:
        None
    """

    if input_csv is None:
        input_csv = f"imdb_data/{OUTPUT_FILE}"
    if output_csv is None:
        output_csv = f"imdb_data/{CLEANED_OUTPUT_FILE}"

    # Load the input CSV file into a pandas DataFrame.
    df: pd.DataFrame = pd.read_csv(input_csv)

    # Filter out rows where all synopsis fields are missing or invalid.
    original_len: int = len(df)
    condition = (
        (df['short_synopsis'] == 'No short sum found') &
        (df['long_synopsis'] == 'Synopsis not found') &
        (df['summaries'].astype(str) == "['No summaries found']")
    )
    df_cleaned: pd.DataFrame = df[~condition].copy()
    removed_count: int = original_len - len(df_cleaned)

    def clean_summary_list(summary_list: Union[List[str], Any]) -> Union[List[str], Any]:
        """
        Cleans a list of summary strings by removing unwanted characters and truncating text.

        Args:
            summary_list (list or any): A list of summary strings to clean. If not a list, it is returned as-is.

        Returns:
            list or any: The cleaned list of summaries, or the original input if not a list.
        """
        if isinstance(summary_list, list):
            cleaned: List[str] = []
            for s in summary_list:
                if isinstance(s, str):
                    # Remove unwanted characters or truncate text based on specific patterns.
                    if '_x0014_' in s:
                        s = s.split('_x0014_')[0].strip()
                    if '—' in s:
                        s = s.rsplit('—', 1)[0].strip()
                cleaned.append(s)
            return cleaned
        return summary_list

    # Parse string representations of lists into actual Python lists and clean them.
    df_cleaned['summaries'] = df_cleaned['summaries'].apply(lambda x: ast.literal_eval(x) if isinstance(x, str) else x)
    df_cleaned['summaries'] = df_cleaned['summaries'].apply(clean_summary_list)

    def remove_backslashes(text: Any) -> Any:
        """
        Removes backslashes from a string.

        Args:
            text (str or any): The input text to process. If not a string, it is returned as-is.

        Returns:
            str or any: The processed string with backslashes removed, or the original input if not a string.
        """
        if isinstance(text, str):
            return text.replace("\\", "")
        return text

    # Remove backslashes from the summaries column.
    df_cleaned['summaries'] = df_cleaned['summaries'].apply(remove_backslashes)

    # Save the cleaned DataFrame to an CSV file.
    df_cleaned.to_csv(output_csv, index=False)
    logger.info(f"- Cleaned synopsis saved to: {output_csv}")
    logger.info(f"- Rows removed: {removed_count}")

`split_movie_xlsx(input_csv=None, output_dir='imdb_data')`

Splits an XLSX file containing movie data into two separate files: one for metadata and another for synopsis.

Parameters:

Name	Type	Description	Default
`input_csv`	`str`	Path to the input CSV file containing movie data.	`None`
`output_dir`	`str`	Directory where the output files will be saved. Defaults to "imdb_data".	`'imdb_data'`

Returns:

Name	Type	Description
`tuple`		A tuple containing the paths to the metadata and synopsis files. (metadata_path, synopsis_path)

Source code in app/pipeline/split_movie_data.py

def split_movie_xlsx(input_csv=None, output_dir="imdb_data"):
    """
    Splits an XLSX file containing movie data into two separate files:
    one for metadata and another for synopsis.

    Args:
        input_csv (str): Path to the input CSV file containing movie data.
        output_dir (str): Directory where the output files will be saved.
                          Defaults to "imdb_data".

    Returns:
        tuple: A tuple containing the paths to the metadata and synopsis files.
               (metadata_path, synopsis_path)
    """
    if input_csv is None:
        input_csv = f"{output_dir}/{CLEANED_OUTPUT_FILE}"
    # Read the input CSV file into a DataFrame
    df = pd.read_csv(input_csv)

    # Define the columns to be included in the metadata file
    metadata_cols = [
        'tconst',          # Unique identifier for the movie
        'movie_title',     # Title of the movie
        'movie_year',      # Release year of the movie
        'age_rating',      # Age rating of the movie
        'duration',        # Duration of the movie
        'imdb_rating',     # IMDb rating of the movie
        'top_5_actors',    # List of top 5 actors in the movie
        'poster_url'       # URL of the movie's poster
    ]

    # Define the columns to be included in the synopsis file
    synopsis_cols = [
        'tconst',          # Unique identifier for the movie
        'genres',          # List of genres associated with the movie
        'short_synopsis',  # Short synopsis of the movie
        'summaries',       # List of summaries for the movie
        'long_synopsis'    # Detailed synopsis of the movie
    ]

    # Create separate DataFrames for metadata and synopsis
    metadata_df = df[metadata_cols].copy()
    synopsis_df = df[synopsis_cols].copy()

    # Define the output file paths
    metadata_path = f"{output_dir}/{MOVIE_METADATA_FILE}"
    synopsis_path = f"{output_dir}/{MOVIE_SYNOPSIS_FILE}"

    # Save the metadata DataFrame to an Excel file
    metadata_df.to_excel(metadata_path, index=False)

    # Save the synopsis DataFrame to a CSV file
    synopsis_df.to_excel(synopsis_path, index=False)

    # Log the paths of the saved files
    logger.info(f"- Saved metadata to: {metadata_path}")
    logger.info(f"- Saved synopsis to: {synopsis_path}")

    # Return the paths of the saved files
    return metadata_path, synopsis_path

`upload_metadata_to_drive(file_path=None, file_name=None)`

Uploads a metadata file to Google Drive.

Parameters:

Name	Type	Description	Default
`file_path`	`str`	The local path to the file to be uploaded. Default is "imdb_data/romance_metadata.xlsx".	`None`
`file_name`	`str`	The name of the file as it will appear in Google Drive. Default is "romance_metadata.xlsx".	`None`

Returns:

Name	Type	Description
`str`	`str`	The ID of the uploaded file in Google Drive.

Source code in app/pipeline/drive_upload.py

def upload_metadata_to_drive(file_path: str = None, file_name: str = None) -> str:

    """
    Uploads a metadata file to Google Drive.

    Args:
        file_path (str): The local path to the file to be uploaded.
                         Default is "imdb_data/romance_metadata.xlsx".
        file_name (str): The name of the file as it will appear in Google Drive.
                         Default is "romance_metadata.xlsx".

    Returns:
        str: The ID of the uploaded file in Google Drive.
    """
    if file_path is None:
        file_path = f"imdb_data/{MOVIE_METADATA_FILE}"
    if file_name is None:
        file_name = MOVIE_METADATA_FILE

    # Define the MIME type for an Excel file.
    mime_type: str = 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'

    # Authenticate using the service account credentials and the specified scopes.
    credentials = service_account.Credentials.from_service_account_file(
        SERVICE_ACCOUNT_FILE, scopes=SCOPES)
    service = build('drive', 'v3', credentials=credentials)

    # Define the metadata for the file to be uploaded, including its name and parent folder ID.
    file_metadata: dict = {
        'name': file_name,
        'parents': [FOLDER_ID]
    }

    # Create a MediaFileUpload object to handle the file upload.
    media = MediaFileUpload(file_path, mimetype=mime_type)

    # Upload the file to Google Drive and retrieve the response containing the file ID.
    response: dict = service.files().create(
        body=file_metadata,
        media_body=media,
        fields='id'
    ).execute()

    # Extract the file ID from the response and log the upload success.
    file_id: str = response.get("id")
    logger.info(f"- Uploaded '{file_name}' with file ID: {file_id}")
    return file_id

`upload_single_file_to_bigquery(target_file_name=None, folder_id='1sJcfq2h5NCOO2DNz2RxeIgSq-HYhl9Kg', dataset_id='romance_dataset', table_name='full_data_table')`

Uploads a single Excel file from Google Drive to a BigQuery table.

Parameters:

Name	Type	Description	Default
`target_file_name`	`str`	The name of the target Excel file to search for in Google Drive.	`None`
`folder_id`	`str`	The ID of the Google Drive folder containing the file.	`'1sJcfq2h5NCOO2DNz2RxeIgSq-HYhl9Kg'`
`dataset_id`	`str`	The ID of the BigQuery dataset where the table resides.	`'romance_dataset'`
`table_name`	`str`	The name of the BigQuery table to upload data to.	`'full_data_table'`

Returns:

Type	Description
`Optional[None]`	None

Source code in app/pipeline/bigquery_upload.py

def upload_single_file_to_bigquery(
    target_file_name: Optional[str] = None,
    folder_id="1sJcfq2h5NCOO2DNz2RxeIgSq-HYhl9Kg",
    dataset_id="romance_dataset",
    table_name="full_data_table"
) -> Optional[None]:
    """
    Uploads a single Excel file from Google Drive to a BigQuery table.

    Args:
        target_file_name (str): The name of the target Excel file to search for in Google Drive.
        folder_id (str): The ID of the Google Drive folder containing the file.
        dataset_id (str): The ID of the BigQuery dataset where the table resides.
        table_name (str): The name of the BigQuery table to upload data to.

    Returns:
        None
    """
    if target_file_name is None:
        target_file_name = MOVIE_METADATA_FILE

    # Query Google Drive for the file with the specified name and MIME type in the given folder.
    query: str = f"'{folder_id}' in parents and name='{target_file_name}' and mimeType='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'"
    results = drive_service.files().list(q=query, fields="files(id, name)").execute()
    files = results.get('files', [])

    if not files:
        logger.info(f"- File '{target_file_name}' not found in folder.")
        return None

    file: dict = files[0]
    file_id: str = file['id']
    logger.info(f"- Found file: {file['name']} (ID: {file_id})")

    # Use the Google Drive API to download the file content into a memory buffer.
    request = drive_service.files().get_media(fileId=file_id)
    file_stream = io.BytesIO()
    downloader = MediaIoBaseDownload(file_stream, request)
    done = False
    while not done:
        status, done = downloader.next_chunk()
        logger.info(f"- Download {int(status.progress() * 100)}% complete.")
    file_stream.seek(0)

    # Read the Excel file into a pandas DataFrame and remove duplicate rows based on the "tconst" column.
    xls = pd.ExcelFile(file_stream)
    df = xls.parse(xls.sheet_names[0])
    df = df.drop_duplicates(subset=["tconst"])  # Remove internal duplicates

    # Query the BigQuery table for existing rows and filter them out from the DataFrame.
    full_table_id = f"{bq_client.project}.{dataset_id}.{table_name}"
    query = f"SELECT tconst FROM {full_table_id}"
    existing = bq_client.query(query).to_dataframe()
    existing_set = set(existing['tconst'])

    original_len = len(df)
    df = df[~df['tconst'].isin(existing_set)]
    logger.info(f"- Removed {original_len - len(df)} rows already in BigQuery.")

    if df.empty:
        logger.info("- No new rows to upload.")
        return None

    # Add a "staging_raw_id" column to the DataFrame if it does not already exist.
    if "staging_raw_id" not in df.columns:
        df.insert(0, "staging_raw_id", range(1, len(df) + 1))

    # Convert specific columns to string type to ensure compatibility with BigQuery schema.
    columns_to_stringify = ['movie_year', 'imdb_rating']

    for col in columns_to_stringify:
        if col in df.columns:
            df[col] = df[col].astype(str)

    # Configure the BigQuery load job to append data to the table and upload the DataFrame.
    job_config = bigquery.LoadJobConfig(
        autodetect=True,
        write_disposition=bigquery.WriteDisposition.WRITE_APPEND  # Append new rows to the table
    )

    load_job = bq_client.load_table_from_dataframe(df, full_table_id, job_config=job_config)
    load_job.result()  # Wait for the job to complete

    logger.info(f"- Uploaded {len(df)} new row(s) to BigQuery table {full_table_id}")
    return None

`run_chunk_and_embed_pipeline(input_excel=None, output_dir=None)`

Processes an Excel file containing movie data, chunks text fields, embeds the chunks using a SentenceTransformer model, and saves the results as JSON files.

Parameters:

Name	Type	Description	Default
`input_excel`	`str`	Path to the input Excel file containing movie data. Default is "imdb_data/romance_synopsis_cleaned.xlsx".	`None`
`output_dir`	`str`	Directory where the chunked JSON files will be saved. Default is "chunked_jsons/romance_chunks_json".	`None`

Returns:

Type	Description
`None`	None

Source code in app/pipeline/chunk_and_embed.py

def run_chunk_and_embed_pipeline(
    input_excel: str = None,
    output_dir: str = None
) -> None:
    """
    Processes an Excel file containing movie data, chunks text fields, embeds the chunks using a
    SentenceTransformer model, and saves the results as JSON files.

    Args:
        input_excel (str): Path to the input Excel file containing movie data.
                           Default is "imdb_data/romance_synopsis_cleaned.xlsx".
        output_dir (str): Directory where the chunked JSON files will be saved.
                          Default is "chunked_jsons/romance_chunks_json".

    Returns:
        None
    """
    if input_excel is None:
        input_excel = f"imdb_data/{MOVIE_SYNOPSIS_FILE}"
    if output_dir is None:
        output_dir = JSONS_FOLDER

    # Load the SentenceTransformer model for embedding text.
    model: SentenceTransformer = SentenceTransformer('bert-base-nli-mean-tokens')
    os.makedirs(output_dir, exist_ok=True)

    # Read the input Excel file into a pandas DataFrame.
    df: pd.DataFrame = pd.read_excel(input_excel, engine='openpyxl')

    def smart_split_sentences(text: str) -> List[str]:
        """
        Splits a text into sentences while preserving abbreviations.

        Args:
            text (str): The input text to split.

        Returns:
            list: A list of sentences.
        """
        abbreviations = {
            "Dr.": "DR_ABBR", "Mr.": "MR_ABBR", "Mrs.": "MRS_ABBR", "Ms.": "MS_ABBR",
            "Jr.": "JR_ABBR", "Sr.": "SR_ABBR", "St.": "ST_ABBR", "Prof.": "PROF_ABBR",
            "Inc.": "INC_ABBR", "Ltd.": "LTD_ABBR", "vs.": "VS_ABBR",
            "e.g.": "EG_ABBR", "i.e.": "IE_ABBR"
        }
        for abbr, token in abbreviations.items():
            text = text.replace(abbr, token)
        sentences = re.split(r'(?<=[.!?])\s+', text)
        restored = []
        for sentence in sentences:
            for abbr, token in abbreviations.items():
                sentence = sentence.replace(token, abbr)
            restored.append(sentence.strip())
        return restored

    def chunk_text(text: str, min_chunk_size: int = 100, max_chunk_size: int = 275, target_chunk_size: int = 240) -> List[str]:
        """
        Splits a text into chunks of specified sizes.

        Args:
            text (str): The input text to chunk.
            min_chunk_size (int): Minimum size of a chunk in words. Default is 100.
            max_chunk_size (int): Maximum size of a chunk in words. Default is 275.
            target_chunk_size (int): Target size of a chunk in words. Default is 240.

        Returns:
            list: A list of text chunks.
        """
        words = text.split()
        if len(words) <= target_chunk_size:
            return [text.strip()]
        sentences = smart_split_sentences(text)
        chunks = []
        current_chunk = []
        current_length = 0
        for sentence in sentences:
            sentence_length = len(sentence.split())
            if current_length + sentence_length > max_chunk_size:
                if current_chunk:
                    chunks.append(" ".join(current_chunk).strip())
                current_chunk = [sentence]
                current_length = sentence_length
            else:
                current_chunk.append(sentence)
                current_length += sentence_length
        if current_chunk:
            if chunks and len(" ".join(current_chunk).split()) < min_chunk_size:
                chunks[-1] += " " + " ".join(current_chunk)
            else:
                chunks.append(" ".join(current_chunk).strip())
        return chunks

    def smart_split_summaries(text: str) -> List[str]:
        """
        Splits a summary text into individual summaries.

        Args:
            text (str): The input summary text.

        Returns:
            list: A list of cleaned summary strings.
        """
        text = text.strip()
        if text.startswith("[") and text.endswith("]"):
            text = text[1:-1]
        split_candidates = re.split(r"""(?<!\\)['"]\s*,\s*['"]""", text)
        cleaned = []
        for chunk in split_candidates:
            chunk = chunk.strip().strip('"').strip("'").strip()
            if chunk and chunk.lower() != "no summaries found":
                cleaned.append(chunk)
        return cleaned

    # Initialize lists to store documents and text for embedding.
    all_documents: List[dict] = []
    texts_to_embed: List[str] = []

    # Iterate over each row in the DataFrame.
    for index, row in df.iterrows():
        movie_id: str = row["tconst"]
        try:
            genres: List[str] = eval(row["genres"]) if pd.notna(row["genres"]) and isinstance(row["genres"], str) else []
        except:
            genres = []
        short_synopsis: str = row["short_synopsis"] if pd.notna(row["short_synopsis"]) else ""
        long_synopsis: str = row["long_synopsis"] if pd.notna(row["long_synopsis"]) else ""
        summaries: Union[str, List[str]] = row["summaries"]

        # Process short synopsis if available.
        if short_synopsis.strip() and short_synopsis.strip() != "No short sum found":
            doc = {
                "movie_id": movie_id,
                "genres": genres,
                "type": "short",
                "chunk_id": f"{movie_id}-sh-1",
                "text": short_synopsis.strip()
            }
            all_documents.append(doc)
            texts_to_embed.append(doc["text"])

        # Process summaries if available.
        if pd.notna(summaries):
            if isinstance(summaries, str):
                summaries = smart_split_summaries(summaries)
            elif isinstance(summaries, list):
                summaries = [s for s in summaries if isinstance(s, str)]
            else:
                summaries = []
        else:
            summaries = []

        for j, summary in enumerate(summaries):
            if summary.strip():
                chunks = chunk_text(summary)
                for i, chunk in enumerate(chunks):
                    doc = {
                        "movie_id": movie_id,
                        "genres": genres,
                        "type": "summary",
                        "chunk_id": f"{movie_id}-summary-{j + 1}-{i + 1}",
                        "text": chunk
                    }
                    all_documents.append(doc)
                    texts_to_embed.append(chunk)

        # Process long synopsis if available.
        if long_synopsis.strip() and long_synopsis.strip() != "Synopsis not found":
            long_chunks = chunk_text(long_synopsis)
            for i, chunk in enumerate(long_chunks):
                doc = {
                    "movie_id": movie_id,
                    "genres": genres,
                    "type": "long",
                    "chunk_id": f"{movie_id}-lon-{i + 1}",
                    "text": chunk
                }
                all_documents.append(doc)
                texts_to_embed.append(chunk)

        # Embed the text chunks using the SentenceTransformer model.
        logger.info(f"- Embedding {len(texts_to_embed)} chunks on GPU...")
        vectors = model.encode(texts_to_embed, batch_size=32, show_progress_bar=True)

        # Add the embedding vectors to the corresponding documents.
        for i, vector in enumerate(vectors):
            all_documents[i]["vector"] = vector.tolist()

        # Save each document as a JSON file in the output directory.
        for doc in all_documents:
            filename = f"{doc['chunk_id']}.json"
            path = os.path.join(output_dir, filename)
            with open(path, "w", encoding="utf-8") as f:
                json.dump(doc, f, indent=4)

        logger.info(f"- {len(all_documents)} documents saved to: {output_dir}")

        # Create a ZIP archive of the output directory.
        shutil.make_archive(output_dir, 'zip', output_dir)
        logger.info(f"- Archive created: {output_dir}.zip")

- Data Pipeline

- Workflow

- Main Pipeline

run_pipeline()

- Submodules

extract_tt_codes(html) async

run_scraper()

scrape_imdb_movie_codes() async

IMDbCrawler

crawl_movies(num_movies=None) async

extract_movie_data(tconst) async

get_movies_from_tsv(tsv_file=TSV_FILENAME) async

init_session() async

record_processed_id(tconst) async

run_crawler(output_dir='imdb_data', tsv_file=None, num_movies=0) async

clean_romance_synopsis(input_csv=None, output_csv=None)

split_movie_xlsx(input_csv=None, output_dir='imdb_data')

upload_metadata_to_drive(file_path=None, file_name=None)

upload_single_file_to_bigquery(target_file_name=None, folder_id='1sJcfq2h5NCOO2DNz2RxeIgSq-HYhl9Kg', dataset_id='romance_dataset', table_name='full_data_table')

run_chunk_and_embed_pipeline(input_excel=None, output_dir=None)

`run_pipeline()`

`extract_tt_codes(html)` `async`

`run_scraper()`

`scrape_imdb_movie_codes()` `async`

`IMDbCrawler`

`crawl_movies(num_movies=None)` `async`

`extract_movie_data(tconst)` `async`

`get_movies_from_tsv(tsv_file=TSV_FILENAME)` `async`

`init_session()` `async`

`record_processed_id(tconst)` `async`

`run_crawler(output_dir='imdb_data', tsv_file=None, num_movies=0)` `async`

`clean_romance_synopsis(input_csv=None, output_csv=None)`

`split_movie_xlsx(input_csv=None, output_dir='imdb_data')`

`upload_metadata_to_drive(file_path=None, file_name=None)`

`upload_single_file_to_bigquery(target_file_name=None, folder_id='1sJcfq2h5NCOO2DNz2RxeIgSq-HYhl9Kg', dataset_id='romance_dataset', table_name='full_data_table')`

`run_chunk_and_embed_pipeline(input_excel=None, output_dir=None)`