main.py

import asyncio

import pandas as pd
import aiohttp
from fastapi import FastAPI

from bigquery_pkg.big_query_class import BigQuery
from async_functions.async_functs import AsyncFuncts
from settings.configs import logger


BATCH_SIZE = 5
DELAY_BETWEEN_BATCHES = 15  # seconds

async def run_batch(o_async: AsyncFuncts, session: aiohttp.ClientSession, batch: list):
    tasks = [
        o_async.scrape_data(session=session, property_id=pid, url=url, product_id=prod_id)
        for pid, url, prod_id in batch
    ]
    return await asyncio.gather(*tasks)

app = FastAPI()

@app.get("/")
async def main() -> tuple:

    # Initializing bigquery
    o_bq = BigQuery()

    df_vrbo_urls = o_bq.get_info_from_bq()
    dfs = [df_vrbo_urls.iloc[i:i+BATCH_SIZE] for i in range(0, len(df_vrbo_urls), BATCH_SIZE)]
    no_of_bacthes = len(dfs)

    # Scraping process
    o_async = AsyncFuncts()

    async with aiohttp.ClientSession() as session:
        progress = 0
        for df in dfs:
            progress += 1
            try:
                batch = [
                    (row["property_id"], row["Source_URL"], row["product_id"])
                    for _, row in df.iterrows()
                ]
                results = []
                batch_results = await run_batch(o_async, session, batch)
                results.extend(batch_results)

                flattened = [review for batch in results for review in (batch or [])]
                df_batch = pd.DataFrame(flattened)

                o_bq.insert_data_into_bq(df=df_batch)
                logger.info("Batch inserted into BQ")
                logger.info(f"Completed: {round((progress / no_of_bacthes) * 100, 2)} %")
                logger.info(f"Batch completed, waiting {DELAY_BETWEEN_BATCHES}s...\n")
                await asyncio.sleep(DELAY_BETWEEN_BATCHES)
            except Exception as e:
                logger.info(e)
                logger.info(f"Completed: {round((progress / no_of_bacthes) * 100, 2)} %\n")
                continue

    return "Task completed!", 200