airflow_dag_run_status_monit.py

# -*- coding: utf-8 -*-

import time
from selenium import webdriver
from selenium.webdriver.common.desired_capabilities import DesiredCapabilities
import datetime

caps = DesiredCapabilities.FIREFOX

# Tell the Python bindings to use Marionette.
# This will not be necessary in the future,
# when Selenium will auto-detect what remote end
# it is talking to.
caps["marionette"] = True

# Path to Firefox DevEdition or Nightly.
# Firefox 47 (stable) is currently not supported,
# and may give you a suboptimal experience.
#
# On Mac OS you must point to the binary executable
# inside the application package, such as
# /Applications/FirefoxNightly.app/Contents/MacOS/firefox-bin
caps["binary"] = "/Applications/Firefox.app/Contents/MacOS/firefox-bin"
geckodriver = "/Users/xingzezhang/Code/python_playground/scrapy_westman/mep_site/geckodriver"

driver = webdriver.Firefox(capabilities=caps, executable_path=geckodriver)
driver.get("http://10.8.888.666:8088/")

driver.find_element_by_xpath("/html/body/div/div/div/div/div/form/input[1]").send_keys("admin")
driver.find_element_by_xpath("/html/body/div/div/div/div/div/form/input[2]").send_keys("2333333")
driver.find_element_by_xpath("/html/body/div/div/div/div/div/form/button").click()

fetch_date = driver.find_element_by_xpath("/html/body/div/div/div[2]/div[2]/div/table/tbody/tr[6]/td[7]/a").text
# fetch_date = driver.find_element_by_xpath("/html/body/div/div/div[2]/div[2]/div/table/tbody/tr[7]/td[7]/a").text


def gen_today():
    return datetime.datetime.now()


def gen_last_run_time():
    return gen_today() - datetime.timedelta(days=2)


def parse_string_to_date(fetch_date):
    """
        datetime.datetime.strptime("2018-06-30 16:05", "%Y-%m-%d %H:%S")
        datetime.datetime(2018, 6, 30, 16, 0, 5)
    """
    return datetime.datetime.strptime(str(fetch_date), "%Y-%m-%d %H:%S")


def compare_date():
    return (gen_today() - parse_string_to_date(fetch_date)).days


print("today is %s" % gen_today())
print("expect is %s" % str(gen_last_run_time()))
print("fetch is %s" % fetch_date)


if 2 == compare_date():
    print("Today dag is running.")
else:
    print("Something wrong")