"""
post_handlers.py
==========
Contains different handlers that trigger post registration
"""
import os.path
import textwrap
from datetime import datetime
import structlog
from typing import TYPE_CHECKING
from xml.sax.saxutils import escape
if TYPE_CHECKING:
from pystac import Item
from .utils import isoformat
logger = structlog.getLogger(__name__)
[docs]class ReportingPostHandler:
def __init__(self, service_url: str, reporting_dir: str):
self.service_url = service_url
self.reporting_dir = reporting_dir
[docs] def sanitize_path(self, path):
invalid_windows_chars = '<>:"/\\|?* '
for char in invalid_windows_chars:
path = path.replace(char, "_")
return path
def __call__(self, config: dict, item: "Item"):
inserted = datetime.now()
timestamp = inserted.strftime("%Y%m%dT%H%M%S")
sanitized_filename = self.sanitize_path(f"item_{timestamp}_{item.id}.xml")
filename = os.path.join(self.reporting_dir, sanitized_filename)
logger.info("Generating report", product=item.id, filename=filename)
identifier = escape(item.id)
availability_time = escape(isoformat(inserted))
wcs_capabilities_url = escape(
'%s/ows?service=wcs&request=GetCapabilities&cql=identifier="%s_%s"'
% (
self.service_url,
item.id,
[i for i in list(item.assets.keys()) if i != "gsc_metadata"][0],
)
)
wms_capabilities_url = escape(
'%s/ows?service=wms&request=GetCapabilities&cql=identifier="%s"'
% (self.service_url, item.id)
)
with open(filename, "w") as f:
f.write(
textwrap.dedent(
f"""\
<?xml version="1.0" encoding="UTF-8"?>
<DataAccessItem xmlns="http://www.telespazio.com/CSCDA/CDD/PDAS">
<identifier>{identifier}</identifier>
<BROWSE_AVAILABILITY_DATETIME>{availability_time}</BROWSE_AVAILABILITY_DATETIME>
<URL>
<Service>WCS</Service>
<URL>{wcs_capabilities_url}</URL>
</URL>
<URL>
<Service>WMS</Service>
<URL>{wms_capabilities_url}</URL>
</URL>
</DataAccessItem>"""
)
)