Запустите python класс только один раз и сохраните вывод для дальнейшего использования - PullRequest
0 голосов
/ 13 апреля 2020

Итак, у меня есть проект скребка craigslist, над которым я работаю, и я столкнулся с потенциальной проблемой. У меня есть этот файл url.py, который имеет класс UrlObj и геттеры и сеттеры. В моем файле main.py я создаю экземпляр этого объекта и возвращаю заполненный URL-адрес для отправки в класс Job в моем файле main.py, чтобы выполнить его очистку.

Я хотел бы развернуть его в облаке в будущем и запускать его через определенный промежуток времени (т.е. каждый день, 4 часа и т. д. c), но я заметил проблему. Каждый раз, когда запускается эта программа, вызывается класс UrlObj, предлагающий пользователю ввести соответствующие данные для построения URL. Поскольку это будет в облаке, работающем в фоновом режиме, никто не сможет вводить подсказки каждый раз, когда он создается и запускается.

Я хочу, чтобы url.py и UrlObj вызывались только один раз, в начале, чтобы пользователь мог вводить и заполнять необходимые поля для создания URL. Затем, каждый раз, когда программа собирается и запускается, следует использовать URL-адрес, созданный пользователем в начале, не вызывая url.py и UrlObj, чтобы снова предложить пользователю ввести данные, так как она будет работать в облаке и на интервал времени.

Не слишком ли наивно думать, чтобы устанавливать условия около url = UrlObj().url, чтобы убедиться, что он выполняется один раз. Как оператор if или while l oop?

url.py:

class UrlObj:
    def __init__(self):
        self.location = self.get_location()  # Location(i.e. City) being searched
        self.postal_code = self.get_postal_code()  # Postal code of location being searched
        self.query = self.get_query()  # Search for the type of items that will be searched
        self.max_price = self.get_max_price()  # Max price of the items that will be searched
        self.radius = self.get_radius()  # Radius of the area searched derived from the postal code given previously

        self.url = f"https://{self.location}.craigslist.org/search/sss?&max_price={self.max_price}&postal={self.postal_code}&query={self.query}&20card&search_distance={self.radius}"

    def get_location(self):
        location = input("Please enter the location: ")
        return location

    def get_postal_code(self):
        postal_code = input("Please enter the postal code: ")
        return postal_code

    def get_query(self):
        query = input("Please enter the item: ")
        return query

    def get_max_price(self):
        max_price = input("Please enter the max price: ")
        return max_price

    def get_radius(self):
        radius = input("Please enter the radius: ")
        return radius

main.py:

from selenium import webdriver
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.common.by import By
import pandas as pd
from url import *


class Job():

    def __init__(self):
        self.driver = webdriver.Chrome(r"C:\Program Files\chromedriver")  # Path of Chrome web driver
        self.delay = 5  # The delay the driver gives when loading the web page

    # Load up the web page
    # Gets all relevant data on the page
    # Goes to next page until we are at the last page
    def load_craigslist_url(self, url):
        data = []
        self.driver.get(url)
        while True:
            try:
                wait = WebDriverWait(self.driver, self.delay)
                wait.until(EC.presence_of_element_located((By.ID, "searchform")))
                data.append(self.extract_post_titles())
                WebDriverWait(self.driver, 2).until(
                    EC.element_to_be_clickable((By.XPATH, '//*[@id="searchform"]/div[3]/div[3]/span[2]/a[3]'))).click()
            except:
                break
        return data

    # # Extracts all relevant information from the web-page and returns them as individual lists
    def extract_post_titles(self):
        all_posts = self.driver.find_elements_by_class_name("result-row")

        dates_list = []
        titles_list = []
        prices_list = []
        distance_list = []

        for post in all_posts:

            title = post.text.split("$")

            if title[0] == '':
                title = title[1]
            else:
                title = title[0]

            title = title.split("\n")
            price = title[0]
            title = title[-1]
            title = title.split(" ")
            month = title[0]
            day = title[1]
            title = ' '.join(title[2:])
            date = month + " " + day

            if not price[:1].isdigit():
                price = "0"
            int(price)

            raw_distance = post.find_element_by_class_name(
                'maptag').text
            distance = raw_distance[:-2]

            titles_list.append(title)
            prices_list.append(price)
            dates_list.append(date)
            distance_list.append(distance)

        return titles_list, prices_list, dates_list, distance_list

    # # Kills browser
    def kill(self):
        self.driver.close()

    @staticmethod
    def organizeResults(results):
        titles_list = results[0][0]
        prices_list = list(map(int, results[0][1]))
        dates_list = results[0][2]
        distance_list = list(map(float, results[0][3]))

        list_of_attributes = []

        for i in range(len(titles_list)):
            content = {'Listing': titles_list[i], 'Price': prices_list[i], 'Date posted': dates_list[i],
                       'Distance from zip': distance_list[i]}
            list_of_attributes.append(content)

        list_of_attributes.sort(key=lambda x: x['Distance from zip'])

        return list_of_attributes

    @staticmethod
    def to_csv(dictionary):
        df = pd.DataFrame(dictionary)
        df.to_csv('data.csv', index=False)


if __name__ == '__main__':

    # This should be called only once!!!
    # Then the 'url' should be used every time main.py is built and ran, and not be constructed again by calling 'UrlObj().url'
    url = UrlObj().url


    scraper = Job()
    results = scraper.load_craigslist_url(url)
    scraper.kill()
    dictionary_of_listings = scraper.organizeResults(results)
    scraper.to_csv(dictionary_of_listings)
Добро пожаловать на сайт PullRequest, где вы можете задавать вопросы и получать ответы от других членов сообщества.
...