oarepo_oaipmh_harvester.tasks

def format_oaipmh_datestamp(since: datetime | str, base_url: str) -> str:
    """Format `since` as an OAI-PMH datestamp matching the server granularity.

    OAI-PMH repositories that advertise `YYYY-MM-DD` granularity (e.g. the
    ASEP ARL server) reject any time component, while `...+00:00` offsets
    produced by `datetime.isoformat()` are illegal in both granularities.
    Seconds-granularity servers get `YYYY-MM-DDThh:mm:ssZ` (UTC), others a
    bare `YYYY-MM-DD`.
    """
    if isinstance(since, str):
        import arrow

        since_dt = arrow.get(since).datetime
    else:
        since_dt = since

    if since_dt.utcoffset() is None:
        since_dt = since_dt.astimezone()  # interpret naive input as local time
    since_dt = since_dt.astimezone(UTC)
    if _oai_granularity(base_url) == _SECONDS_GRANULARITY:
        return since_dt.strftime("%Y-%m-%dT%H:%M:%SZ")
    return since_dt.date().isoformat()

@shared_task
def harvest_oaipmh_records(
    *,
    harvester_id: str,
    oai_ids: list[str] | None = None,
    since: datetime | None = None,
    batch_size: int | None = None,
) -> None:
    """Harvest OAI-PMH records for the given harvester, then mail a report to its harvest managers."""
    harvester = OAIHarvesterAggregate.get_record(harvester_id)
    report = HarvestReport()
    try:
        _harvest(harvester, report, oai_ids=oai_ids, since=since, batch_size=batch_size)
    except Exception as e:
        report.error = f"{type(e).__name__}: {e}"
        raise
    finally:
        report.finished_at = datetime.now(UTC)
        try:
            send_harvest_report(harvester, report)
        except Exception:
            # the report must not hide the harvest's own outcome
            current_app.logger.exception("Could not send the report of harvester %s", harvester_id)

def parse_config(config: str) -> tuple[str, dict]:
    """Parse a config string of the form type{args}."""
    # the config is a single line, like xml{props}
    # props are json5 style key-value pairs
    # the result is {"type": "oai-import", "args": {"model": "test"}}
    if "{" not in config:
        return config, {}
    type_, args_str = config.split("{", 1)
    args_str = "{" + args_str

    args = cast("dict", json5.loads(args_str))
    return type_, args

def create_readers(
    harvester: OAIHarvesterAggregate,
    since: datetime | str | None = None,
    oai_ids: list[str] | None = None,
) -> dict:
    """Create readers config for the harvester."""
    reader_type, reader_args = parse_config(harvester.loader or "oai-pmh")
    if "base_url" not in reader_args:
        reader_args["base_url"] = harvester.base_url
    if "metadata_prefix" not in reader_args and harvester.metadata_prefix:
        reader_args["metadata_prefix"] = harvester.metadata_prefix
    if "set" not in reader_args and harvester.setspec:
        reader_args["set"] = harvester.setspec
    if "from_date" not in reader_args and since:
        if isinstance(since, datetime | str):
            since = format_oaipmh_datestamp(since, reader_args["base_url"])
        reader_args["from_date"] = since
    if oai_ids is not None:
        reader_args["identifiers"] = oai_ids
    reader_args["harvester_id"] = harvester.id
    return {
        "type": reader_type,
        "args": reader_args,
    }

def create_transformers(harvester: OAIHarvesterAggregate) -> list[dict]:
    """Create transformers config for the harvester."""
    transformers = []
    for transformer_config in harvester.transformers or []:
        transformer_type, transformer_args = parse_config(transformer_config)
        transformer_args["harvester_id"] = harvester.id
        transformers.append(
            {
                "type": transformer_type,
                "args": transformer_args,
            }
        )
    return transformers

def create_writers(harvester: OAIHarvesterAggregate) -> list[dict]:
    """Create writers config for the harvester."""
    writers = []
    for writer_config in harvester.writers or []:
        writer_type, writer_args = parse_config(writer_config)
        writer_args["harvester_id"] = harvester.id
        writers.append(
            {
                "type": writer_type,
                "args": writer_args,
            }
        )
    return writers

@shared_task(ignore_result=True)
def run_all_harvesters() -> None:
    """Run all OAI-PMH harvester tasks."""
    harvesters = OAIHarvester.query.all()
    for harvester in harvesters:
        try:
            current_app.logger.info("Started harvesting %s", harvester.id)
            harvest_oaipmh_records(harvester_id=harvester.id)
            current_app.logger.info("Successfully harvested %s", harvester.id)
        except Exception:  # pragma: no cover
            current_app.logger.exception("Exception during harvesting %s", harvester.id)