225 lines
No EOL
8.5 KiB
Python
225 lines
No EOL
8.5 KiB
Python
# /// script
|
|
# requires-python = ">=3.12"
|
|
# dependencies = [
|
|
# "dump-things-pyclient @ https://hub.psychoinformatics.de/datalink/dump-things-pyclient.git",
|
|
# "icalendar",
|
|
# "rich",
|
|
# "rich-click",
|
|
# ]
|
|
# ///
|
|
from icalendar import Calendar
|
|
from os import environ
|
|
import json
|
|
import copy
|
|
import urllib.request
|
|
import rich_click as click
|
|
from dump_things_pyclient.communicate import (
|
|
collection_write_record,
|
|
collection_read_record_with_pid,
|
|
)
|
|
|
|
# example target format:
|
|
# https://hedgedoc.psychoinformatics.de/3cSouq0YSJ6m64_ArWpJEg?edit
|
|
# TODO:
|
|
# - create merit-based award in psyinf pool, add pid here
|
|
# - create record for this script in pool, add pid here
|
|
|
|
|
|
urls = {
|
|
'juniorgroup': 'https://webmail.fz-juelich.de/owa/calendar/d61ec0ce8d704cb293df97fbb3c8fe23@fz-juelich.de/2e3e5a7baff44a3780cb2f872bef77b07381212461382934376/calendar.ics',
|
|
'funding': 'https://webmail.fz-juelich.de/owa/calendar/d61ec0ce8d704cb293df97fbb3c8fe23@fz-juelich.de/a7ad1dee32cf49749872136f3a9223191125633822045414026/calendar.ics',
|
|
'award': 'https://webmail.fz-juelich.de/owa/calendar/d61ec0ce8d704cb293df97fbb3c8fe23@fz-juelich.de/2f6c8a7e45e5416f89c22258bba6ae0114099503221118509202/calendar.ics'
|
|
}
|
|
|
|
competition_types = {'funding': "xyzrins:competition-types/4e49ac7d-d6da-4131-806b-6425491e26fd",
|
|
'award': "xyzrins:competition-types/16e38ddd-6323-4fc7-abad-398609bf8541", # pid does not yet exist
|
|
'juniorgroup': "xyzrins:competition-types/4e49ac7d-d6da-4131-806b-6425491e26fd"}
|
|
|
|
|
|
|
|
class CalendarScraper(object):
|
|
def __init__(
|
|
self,
|
|
pool: str,
|
|
collection: str,
|
|
competition_type: str,
|
|
calendar_url: str,
|
|
) -> None:
|
|
self.pool = pool
|
|
self.collection = collection
|
|
self.comptetion_type = competition_type
|
|
self.calendar_url = calendar_url
|
|
# this list stores to-be-submitted records
|
|
self.to_submit = []
|
|
self.scriptpid = 'xyzrins:instruments/d3126ff5-623c-48af-ac48-4d921ef9b80d'
|
|
|
|
def get_calendar(self) -> None:
|
|
with urllib.request.urlopen(self.calendar_url) as f:
|
|
calsource = f.read().decode('utf-8')
|
|
self.cal = Calendar.from_ical(calsource)
|
|
|
|
def submit(self) -> None:
|
|
for record in self.to_submit:
|
|
print(f"submitting record with pid {record['pid']}"
|
|
f" to collection {self.collection}")
|
|
try:
|
|
collection_write_record(
|
|
service_url=self.pool,
|
|
collection=self.collection,
|
|
class_name='XYZCompetition',
|
|
record=record,
|
|
format='json',
|
|
token=environ['DTC_TOKEN']
|
|
)
|
|
except urllib.requests.exceptions.HTTPError:
|
|
print("SUBMISSION ERROR FOR RECORD: ")
|
|
print(json.dumps(record))
|
|
return
|
|
def _existing_or_new_record(self,
|
|
pid: str) -> (dict, (dict | None)):
|
|
record = collection_read_record_with_pid(
|
|
service_url=self.pool,
|
|
collection=self.collection,
|
|
pid=pid,
|
|
token=environ['DTC_TOKEN']
|
|
)
|
|
old_record = copy.deepcopy(record)
|
|
record=None
|
|
if record is None:
|
|
old_record = None
|
|
print(f"New record: {pid}")
|
|
record = {'schema_type': 'xyzri:XYZCompetition',
|
|
'pid': pid,
|
|
'kind': competition_types[self.comptetion_type]}
|
|
return record, old_record
|
|
|
|
def create_competitions(self) -> None:
|
|
self.get_calendar()
|
|
for event in self.cal.events:
|
|
pid = 'xyzrins:competition/' + event.get("UID").ical_value
|
|
self.assemble_record(pid, event)
|
|
self.submit()
|
|
|
|
def _check_if_mutable(
|
|
self,
|
|
record: dict,
|
|
k: str,
|
|
only_self_edits: bool = False,
|
|
predicate: str | None = None,
|
|
) -> bool:
|
|
"""If a record either does not already have the info, or the info is
|
|
annotated to be machine-generated, allow overwriting it.
|
|
params:
|
|
record: dict -> the metadata record
|
|
k: str -> record key to check
|
|
only_self_edits: True|False -> only allows overwriting existing infos if
|
|
they stem from the same script
|
|
"""
|
|
|
|
if k not in record.keys():
|
|
return True
|
|
infos = record[k]
|
|
if type(infos) == dict and 'annotations' in infos.keys():
|
|
importedBy = \
|
|
infos.get('annotations', {}).get('http://purl.org/pav/importedBy', 'nothing')
|
|
return self._is_machine_generated(importedBy, only_self_edits)
|
|
# if the key is a data property, infos is just a string.
|
|
attributes = record.get('attributes', [{}])
|
|
for attribute in attributes:
|
|
if attribute.get('predicate', None) == predicate:
|
|
return \
|
|
self._is_machine_generated(attribute.get('annotations', {}).get('http://purl.org/pav/importedBy', 'nothing'))
|
|
|
|
def _is_machine_generated(
|
|
self,
|
|
importedBy,
|
|
only_self_edits: bool = False,
|
|
):
|
|
if importedBy.startswith('xyzrins:instruments'):
|
|
if not only_self_edits:
|
|
return True
|
|
else:
|
|
return importedBy == self.scriptpid
|
|
return False
|
|
|
|
def assemble_record(
|
|
self,
|
|
pid: str,
|
|
event: dict,
|
|
) -> dict | None:
|
|
"""Extract information from a calendar event and create a metadata
|
|
record from it."""
|
|
annotations = {"http://purl.org/pav/importedBy": self.scriptpid,
|
|
"http://purl.org/pav/importedFrom": self.calendar_url}
|
|
record, old_record = self._existing_or_new_record(pid)
|
|
title, deadline, desc = None, None, None
|
|
# title
|
|
if self._check_if_mutable(record, 'title', predicate='dcterms:title'):
|
|
title = event.get('SUMMARY').ical_value
|
|
|
|
# application deadline
|
|
if self._check_if_mutable(record, 'application_deadline', predicate='dcterms:date'):
|
|
deadline = event.get("DTSTART").td.isoformat()
|
|
|
|
# description
|
|
if self._check_if_mutable(record, 'description', predicate='dcterms:description'):
|
|
# needs to be stripped from newlines to be valid
|
|
desc = event.get("DESCRIPTION").ical_value.replace('\n', ' ')
|
|
if 'Please note' in desc:
|
|
desc = desc.split('Please note')[0]
|
|
|
|
attributes = record.get('attributes', [])
|
|
for prop, slot, term in [(title, 'title', 'dcterms:title'),
|
|
(desc, 'description', 'dcterms:description'),
|
|
(deadline, 'application_deadline', 'dcterms:date')]:
|
|
if prop is not None:
|
|
record[slot] = prop
|
|
metadata = {'predicate': term,
|
|
'value': prop,
|
|
'annotations': annotations}
|
|
if attributes:
|
|
for i, attr in enumerate(attributes):
|
|
found = False
|
|
if attr.get('predicate', None) == term:
|
|
attributes[i] = metadata
|
|
if not found:
|
|
attributes.append(metadata)
|
|
else:
|
|
attributes.append(metadata)
|
|
record['attributes'] = attributes
|
|
if record == old_record:
|
|
import pdb; pdb.set_trace()
|
|
print(f"No change for existing cecord with PID {pid}.")
|
|
return
|
|
self.to_submit.append(record)
|
|
return
|
|
|
|
|
|
|
|
@click.command()
|
|
@click.option('--dtc-api-url', '-a', default='https://pool.psychoinformatics.de/api')
|
|
@click.option('--dtc-collection', '-c', default='public')
|
|
def main(
|
|
dtc_api_url: str = 'https://pool.psychoinformatics.de/api',
|
|
dtc_collection: str = 'public',
|
|
) -> None:
|
|
"""
|
|
Scrape the outlook funding calendars of the FZJ and write events as
|
|
Competition records into the knowledge pool at --dtc-api-url, into the
|
|
collection determined by --dtc-collection.
|
|
"""
|
|
if environ.get('DTC_TOKEN', None) is None:
|
|
print("DTC_TOKEN required in environment! Aborting.")
|
|
|
|
for competition_type, url in urls.items():
|
|
print(f'scraping for {competition_type}')
|
|
scraper = CalendarScraper(
|
|
pool=dtc_api_url,
|
|
collection=dtc_collection,
|
|
competition_type=competition_type,
|
|
calendar_url=url
|
|
)
|
|
scraper.create_competitions()
|
|
|
|
if __name__ == '__main__':
|
|
main() |