from datetime import datetime, timedelta

import pendulum
import prefect
from prefect import Flow, task
from prefect.run_configs import DockerRun
from prefect.schedules import CronSchedule
from prefect.storage import GitHub
from scrapy.crawler import CrawlerProcess
from sqlmodel import SQLModel, create_engine

from imdb_rating.dependencies.spiders import IMDBSpider


@task
def scrap_movies_from_imdb():
    """
    Scrap movies from IMDB and store them into a PostgreSQL database using SQLModel.
    Run a scrapy crawler process to launch a spider.
    """
    logger = prefect.context.get("logger")

    # engine = create_engine('postgresql://postgres:postgres@localhost:5432/imdb')
    engine = create_engine("sqlite:///imdb.db")
    SQLModel.metadata.create_all(engine)

    start = datetime.today() - timedelta(days=90)
    end = datetime.today() + timedelta(days=30)

    process = CrawlerProcess()

    process.crawl(IMDBSpider, start=start, end=end, engine=engine)
    process.start()


schedule = CronSchedule("0 0 * * *", start_date=pendulum.now(tz="Europe/Paris"))

storage = GitHub(repo="PeregHer/imdb-rating-predictions", path="imdb_rating/workflow/flow.py")

run_config = DockerRun(image="imdb-scraping:latest")


with Flow(
    "imdb_scraping", schedule=schedule, storage=storage, run_config=run_config
) as flow:
    scrap_movies_from_imdb()


# flow.register(project_name="imdb-scraping", tags=["imdb-scraping"])
# flow.run()
sqlmodel.SQLModel.metadata.create_all pendulum.now prefect.run_configs.DockerRun datetime.timedelta prefect.storage.GitHub scrapy.crawler.CrawlerProcess datetime.datetime.today prefect.Flow sqlmodel.create_engine prefect.context.get
[((1089, 1178), 'prefect.storage.GitHub', 'GitHub', ([], {'repo': '"""PeregHer/imdb-rating-predictions"""', 'path': '"""imdb_rating/workflow/flow.py"""'}), "(repo='PeregHer/imdb-rating-predictions', path=\n    'imdb_rating/workflow/flow.py')\n", (1095, 1178), False, 'from prefect.storage import GitHub\n'), ((1188, 1227), 'prefect.run_configs.DockerRun', 'DockerRun', ([], {'image': '"""imdb-scraping:latest"""'}), "(image='imdb-scraping:latest')\n", (1197, 1227), False, 'from prefect.run_configs import DockerRun\n'), ((573, 602), 'prefect.context.get', 'prefect.context.get', (['"""logger"""'], {}), "('logger')\n", (592, 602), False, 'import prefect\n'), ((700, 734), 'sqlmodel.create_engine', 'create_engine', (['"""sqlite:///imdb.db"""'], {}), "('sqlite:///imdb.db')\n", (713, 734), False, 'from sqlmodel import SQLModel, create_engine\n'), ((739, 775), 'sqlmodel.SQLModel.metadata.create_all', 'SQLModel.metadata.create_all', (['engine'], {}), '(engine)\n', (767, 775), False, 'from sqlmodel import SQLModel, create_engine\n'), ((890, 906), 'scrapy.crawler.CrawlerProcess', 'CrawlerProcess', ([], {}), '()\n', (904, 906), False, 'from scrapy.crawler import CrawlerProcess\n'), ((1235, 1320), 'prefect.Flow', 'Flow', (['"""imdb_scraping"""'], {'schedule': 'schedule', 'storage': 'storage', 'run_config': 'run_config'}), "('imdb_scraping', schedule=schedule, storage=storage, run_config=run_config\n    )\n", (1239, 1320), False, 'from prefect import Flow, task\n'), ((789, 805), 'datetime.datetime.today', 'datetime.today', ([], {}), '()\n', (803, 805), False, 'from datetime import datetime, timedelta\n'), ((808, 826), 'datetime.timedelta', 'timedelta', ([], {'days': '(90)'}), '(days=90)\n', (817, 826), False, 'from datetime import datetime, timedelta\n'), ((837, 853), 'datetime.datetime.today', 'datetime.today', ([], {}), '()\n', (851, 853), False, 'from datetime import datetime, timedelta\n'), ((856, 874), 'datetime.timedelta', 'timedelta', ([], {'days': '(30)'}), '(days=30)\n', (865, 874), False, 'from datetime import datetime, timedelta\n'), ((1045, 1076), 'pendulum.now', 'pendulum.now', ([], {'tz': '"""Europe/Paris"""'}), "(tz='Europe/Paris')\n", (1057, 1076), False, 'import pendulum\n')]