120 lines
3.1 KiB
Django/Jinja
Executable File
120 lines
3.1 KiB
Django/Jinja
Executable File
#!/usr/bin/python3
|
|
|
|
from datetime import datetime, timezone
|
|
import lxml.html
|
|
import sys
|
|
import json
|
|
|
|
import requests
|
|
from influxdb import InfluxDBClient
|
|
from loguru import logger
|
|
|
|
import urllib3
|
|
urllib3.disable_warnings()
|
|
|
|
if "-v" not in sys.argv:
|
|
logger.remove()
|
|
|
|
logger.add(
|
|
"{{ userlist[myusername].home }}/hospital.log",
|
|
rotation="50 MB",
|
|
compression="zip",
|
|
level="DEBUG" if "-v" in sys.argv else "INFO"
|
|
)
|
|
|
|
|
|
def get_metric_name(icelandic_desc):
|
|
mappings = {
|
|
'á bráða- og göngudeild Fossvogi': 'patients_er_and_walk_fossvogur',
|
|
'á bráðadeild Fossvogi': 'patients_icu_fossvogur',
|
|
'skurðaðgerð': 'surgeries_now',
|
|
'bráðamóttöku barna': 'patients_children',
|
|
'fæðst í dag': 'births_today',
|
|
'útskrifast': 'discharges_today',
|
|
'við vinnu þessa stundina': 'at_work_now',
|
|
'lagst inn': 'admitted_today',
|
|
'Blóðbanka': 'blood_donations'
|
|
}
|
|
for mapping_hint, metric in mappings.items():
|
|
|
|
if mapping_hint in icelandic_desc:
|
|
return metric
|
|
|
|
raise ValueError(f"metric name not found for '{icelandic_desc}'")
|
|
|
|
|
|
def landspitali():
|
|
url = 'https://landspitali.is'
|
|
r = requests.get(url, verify=False)
|
|
r.raise_for_status()
|
|
|
|
root = lxml.html.fromstring(r.text)
|
|
d = dict()
|
|
|
|
# theres one value that is highligted and isnt in the normal list
|
|
# its rotated randomly
|
|
# parsing it works sort of in reverse to the rest...
|
|
list2 = root.xpath('/html/body/div[2]/div[2]/div[5]/div/div/strong')
|
|
for onlyitem in list2:
|
|
icelandic_desc = onlyitem.xpath("div")[0].text
|
|
_text = onlyitem.text_content()
|
|
|
|
try:
|
|
value = _text[:-len(icelandic_desc)]
|
|
metric = get_metric_name(icelandic_desc)
|
|
|
|
d[metric] = int(value)
|
|
except ValueError as e:
|
|
logger.exception(e)
|
|
|
|
worklist = root.xpath('//div[@class="work__list_numbers"]/div')
|
|
|
|
for s in worklist:
|
|
value = s.xpath("span")[0].text
|
|
_text = s.text_content()
|
|
|
|
|
|
prefix = len(str(value))
|
|
icelandic_desc = _text[prefix:]
|
|
logger.debug("{} {}".format(value, icelandic_desc))
|
|
try:
|
|
metric = get_metric_name(icelandic_desc)
|
|
d[metric] = int(value)
|
|
except ValueError as e:
|
|
logger.exception(e)
|
|
|
|
return d
|
|
|
|
@logger.catch
|
|
def main():
|
|
client = InfluxDBClient(
|
|
host='{{ influxdb_url }}',
|
|
port=443,
|
|
username='hospital',
|
|
password="{{ influxdb_user_database['hospital'] }}",
|
|
ssl=True,
|
|
verify_ssl=True,
|
|
database='hospital'
|
|
)
|
|
|
|
time = datetime.utcnow().replace(tzinfo=timezone.utc).isoformat()
|
|
hospital = landspitali()
|
|
|
|
datapoints = [{
|
|
"measurement": key,
|
|
"tags": {
|
|
"source": "landspitali.is"
|
|
},
|
|
"time": time,
|
|
"fields": {
|
|
"value": value
|
|
}
|
|
} for key, value in hospital.items()]
|
|
|
|
client.write_points(datapoints, time_precision='m')
|
|
|
|
logger.debug(json.dumps(hospital, indent=2))
|
|
|
|
if __name__ == "__main__":
|
|
main()
|