Further meating
This commit is contained in:
parent
ef8b8f1fd2
commit
d5d7b6626b
1
.gitignore
vendored
1
.gitignore
vendored
@ -1,4 +1,5 @@
|
|||||||
env_conf
|
env_conf
|
||||||
|
include
|
||||||
|
|
||||||
# ---> Python
|
# ---> Python
|
||||||
# Byte-compiled / optimized / DLL files
|
# Byte-compiled / optimized / DLL files
|
||||||
|
@ -1,2 +1,7 @@
|
|||||||
|
certifi==2024.2.2
|
||||||
|
charset-normalizer==3.3.2
|
||||||
|
idna==3.6
|
||||||
lxml==5.1.0
|
lxml==5.1.0
|
||||||
python-docx-2023==0.2.17
|
python-docx-2023==0.2.17
|
||||||
|
requests==2.31.0
|
||||||
|
urllib3==2.2.1
|
||||||
|
@ -0,0 +1,2 @@
|
|||||||
|
import requests
|
||||||
|
|
@ -1,9 +1,15 @@
|
|||||||
import parse_docx, pis_find
|
import parse_docx, pis_find, owlboard_connector
|
||||||
|
|
||||||
import os
|
import os, sys
|
||||||
|
|
||||||
def start():
|
def start():
|
||||||
print("Local mode activated")
|
print("Running OwlBoard Diagram Parser in local mode")
|
||||||
|
if not owlboard_connector.check_connection():
|
||||||
|
print("Exiting")
|
||||||
|
sys.exit(1)
|
||||||
|
#if not gitea_connector.check_connection():
|
||||||
|
# print("Exiting")
|
||||||
|
# sys.exit(1)
|
||||||
working_directory = os.getcwd()
|
working_directory = os.getcwd()
|
||||||
print("Working directory: ", working_directory)
|
print("Working directory: ", working_directory)
|
||||||
|
|
||||||
|
@ -2,31 +2,43 @@
|
|||||||
|
|
||||||
### AUTHENTICATION MUST BE COMPLETED, REGISTERING FOR THE API IF NECCESSARY
|
### AUTHENTICATION MUST BE COMPLETED, REGISTERING FOR THE API IF NECCESSARY
|
||||||
### THIS NEGATES THE ABILITY TO USE LOCAL MODE - MAILBOX MODE ONLY AS
|
### THIS NEGATES THE ABILITY TO USE LOCAL MODE - MAILBOX MODE ONLY AS
|
||||||
### MAILBOX ACCESS IS NEEDED FOR REGISTRATION
|
### MAILBOX ACCESS IS NEEDED FOR REGISTRATION... DON'T BE A FUCKING IDIOT... I CAN JUST PASS
|
||||||
|
### A UUID IN TO THE PROGRAM!! TWAT.
|
||||||
|
|
||||||
from imap_connector import IMAPConnector
|
import requests, os
|
||||||
|
|
||||||
imap_connector = IMAPConnector()
|
OB_BASE_URL = "https://owlboard.info/api/v2/pis/byCode/"
|
||||||
|
OB_TEST_URL = OB_BASE_URL + "5001"
|
||||||
|
UUID = os.environ.get('DGP_OB_UUID')
|
||||||
|
HEADERS = {
|
||||||
|
'user-agent': 'owlboard-diagram-parser',
|
||||||
|
'uuid': UUID
|
||||||
|
}
|
||||||
|
|
||||||
def check_registration():
|
def check_connection():
|
||||||
## First check env var `DGP_OB_UUID`
|
if not UUID:
|
||||||
## Then run a test query, if not unauthorized return TRUE
|
print("'DGP_OB_UUID' must be set in the environment")
|
||||||
## else FALSE.
|
|
||||||
uuid = os.environ.get('DGP_OB_UUID')
|
|
||||||
if not uuid:
|
|
||||||
return False
|
return False
|
||||||
|
|
||||||
|
res = requests.get(OB_TEST_URL, headers=HEADERS)
|
||||||
|
if res.status_code == 401:
|
||||||
|
print("Error - Unauthorised. The UUID is not valid. STATUS: ", res.status_code, "UUID: ", UUID)
|
||||||
|
return False
|
||||||
|
elif res.status_code != 200:
|
||||||
|
print("Error - Unable to reach OwlBoard. STATUS: ", res.status_code)
|
||||||
|
return False
|
||||||
|
return True
|
||||||
|
|
||||||
### Now run test query and if return not 200 return false.
|
def find_pis_code(code):
|
||||||
return
|
print("Searching for PIS Code: ", code)
|
||||||
|
url = OB_BASE_URL + code
|
||||||
def register():
|
res = requests.get(url, headers=HEADERS)
|
||||||
## Send registration Req
|
if res.status_code == 200:
|
||||||
## Keep checking IMAP until reg email received
|
json_response = res.json()
|
||||||
## Submit Reg confirmation
|
if json_response and isinstance(json_response, list):
|
||||||
## Save UUID to env var `DGP_OB_UUID`
|
return json_response
|
||||||
return
|
else:
|
||||||
|
return False
|
||||||
## INIT
|
else:
|
||||||
is_registered = check_registration()
|
print("Unable to reach OwlBoard. STATUS: ", res.status_code)
|
||||||
if not is_registered:
|
return True
|
||||||
register()
|
|
||||||
|
@ -9,6 +9,13 @@ import re
|
|||||||
### and find a service with valid stopping pattern,
|
### and find a service with valid stopping pattern,
|
||||||
### then the PIS codes can be generated for review.
|
### then the PIS codes can be generated for review.
|
||||||
|
|
||||||
|
### I think that I need to match each page. I need to search each page for the
|
||||||
|
### days of week the diagram is valid for, eg FSX, FO. Then I need to find the
|
||||||
|
### dates the diagram is valid for. Then use that to search for trains.
|
||||||
|
|
||||||
|
### Alternatively, I could ensure that only daily diagrams are sent and the date
|
||||||
|
### is contained within the body/subject of the email.
|
||||||
|
|
||||||
PIS_PATTERN = re.compile(r'PIS code\s*:\s*(\d{4})')
|
PIS_PATTERN = re.compile(r'PIS code\s*:\s*(\d{4})')
|
||||||
HEADCODE_PATTERN = re.compile(r'(\d{1}[A-Z]\d{2})')
|
HEADCODE_PATTERN = re.compile(r'(\d{1}[A-Z]\d{2})')
|
||||||
|
|
||||||
|
@ -1,3 +1,4 @@
|
|||||||
|
import owlboard_connector
|
||||||
import requests
|
import requests
|
||||||
|
|
||||||
def run(data_list):
|
def run(data_list):
|
||||||
@ -16,18 +17,15 @@ def dedup(data_list):
|
|||||||
|
|
||||||
## AUTH REQUIRED!!! Move to owlboard_connector.py
|
## AUTH REQUIRED!!! Move to owlboard_connector.py
|
||||||
def find_missing(data_list):
|
def find_missing(data_list):
|
||||||
BASEURL = 'http://localhost:8460/api/v2/pis/byCode/'
|
#BASEURL = 'http://localhost:8460/api/v2/pis/byCode/'
|
||||||
#BASEURL = 'https://owlboard.info/api/v2/pis/byCode/'
|
BASEURL = 'https://owlboard.info/api/v2/pis/byCode/'
|
||||||
missing_data = []
|
missing_data = []
|
||||||
|
|
||||||
for item in data_list:
|
for item in data_list:
|
||||||
pis_code = item.get('pis')
|
pis_code = item.get('pis')
|
||||||
if pis_code:
|
if pis_code:
|
||||||
url = BASEURL + pis_code
|
pis_code_res = owlboard_connector.find_pis_code(pis_code)
|
||||||
response = requests.get(url)
|
if not pis_code_res:
|
||||||
if response.status_code == 200:
|
print("PIS Code ", pis_code, " not found in existing data")
|
||||||
json_response = response.json()
|
missing_data.append(pis_code)
|
||||||
if json_response and isinstance(json_response, list):
|
print(missing_data)
|
||||||
missing_data.append(item)
|
|
||||||
else:
|
|
||||||
print(f"Request failed for PIS {pis_code}. Status: {response.status_code}")
|
|
Reference in New Issue
Block a user