import time
from prefect.engine.executors import DaskExecutor
import requests
from bs4 import BeautifulSoup
from prefect import task, Flow, Parameter
import pandas as pd
from datetime import timedelta
from dask.distributed import Client

@task(tags=["web"])
def retrieve_url(url):
    """
    Given a URL (string), retrieves html and
    returns the html as a string.
    """

    html = requests.get(url)
    if html.ok:
        return html.text
    else:
        raise ValueError("{} could not be retrieved.".format(url))


@task
def scrape_dialogue(episode_html):
    """
    Given a string of html representing an episode page,
    returns a tuple of (title, [(character, text)]) of the
    dialogue from that episode
    """

    episode = BeautifulSoup(episode_html, 'html.parser')

    title = episode.title.text.rstrip(' *').replace("'", "''")
    convos = episode.find_all('b') or episode.find_all('span', {'class': 'char'})
    dialogue = []
    for item in convos:
        who = item.text.rstrip(': ').rstrip(' *').replace("'", "''")
        what = str(item.next_sibling).rstrip(' *').replace("'", "''")
        dialogue.append((who, what))
    return (title, dialogue)


@task
def create_episode_list(base_url, main_html, bypass):
    """
    Given the main page html, creates a list of episode URLs
    """

    if bypass:
        return [base_url]

    main_page = BeautifulSoup(main_html, 'html.parser')

    episodes = []
    for link in main_page.find_all('a'):
        url = link.get('href')
        if 'transcrp/scrp' in (url or ''):
            episodes.append(base_url + url)

    return episodes

@task
def writeresults(dialogue):
    """
    Given the main page html, creates a list of episode URLs
    """

    resultslist = []
    resultslist.append(dialogue)
    df = pd.DataFrame(resultslist)
    filename = "interval_result " + str(time.strftime('%d-%m-%Y %H:%M:%S')) + ".csv"
    df.to_csv(filename, sep=',', encoding='utf-8',)


#daily_schedule = IntervalSchedule(interval=timedelta(minutes=2))

with Flow("xfiles") as flow:
    url = Parameter("url")
    bypass = Parameter("bypass", default=False, required=False)
    home_page = retrieve_url(url)
    episodes = create_episode_list(url, home_page, bypass=bypass)
    episode = retrieve_url.map(episodes)
    dialogue = scrape_dialogue.map(episode)
    results = writeresults(dialogue)


def run():
    client = Client()
    executor = DaskExecutor(local_processes=True, address=client.scheduler.address)
    start_time = time.time()
    flow.visualize()
    scraped_state = flow.run(parameters={"url": "http://www.insidethex.co.uk/"}, executor=executor)
    # state = flow.run(parameters={"url": "http://www.insidethex.co.uk/"}, executor=executor, task_states=scraped_state.result)
    print("\n\n --- %s seconds --- \n\n" % (time.time() - start_time))

    dialogue_state = scraped_state.result[dialogue]  # list of State objects
    print('\n'.join([f'{s.result[0]}: {s}' for s in dialogue_state.map_states[:5]]))
    print('\n\n\n end of the execution')


if __name__ == '__main__':
    run()