Research Lookup

scripts/manuscript_packet.py

"""Pure helpers for building manuscript-ready research packets."""

from __future__ import annotations

import json
import re
from collections import Counter
from pathlib import Path
from typing import Any, Iterable
from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit


DOI_PATTERN = re.compile(
    r"(?:doi[:\s]*|https?://(?:dx\.)?doi\.org/)?"
    r"(10\.\d{4,9}/[-._;()/:A-Z0-9]+)",
    re.IGNORECASE,
)
PMID_PATTERN = re.compile(
    r"(?:pubmed\.ncbi\.nlm\.nih\.gov/|pmid[:\s]*)(\d{6,9})",
    re.IGNORECASE,
)
YEAR_PATTERN = re.compile(r"\b(19\d{2}|20\d{2})\b")
SAMPLE_SIZE_PATTERN = re.compile(
    r"\b(?:n\s*=\s*|sample(?:\s+size)?\s+(?:of\s+)?)([\d,]+)\b",
    re.IGNORECASE,
)
EFFECT_PATTERN = re.compile(
    r"(?:\b\d+(?:\.\d+)?\s*%|\bp\s*[<=>]\s*0?\.\d+|"
    r"\b(?:OR|RR|HR|MD|SMD)\s*[=:]\s*-?\d+(?:\.\d+)?|"
    r"\b95\s*%\s*CI\b)",
    re.IGNORECASE,
)

SCHOLARLY_DOMAINS = {
    "pubmed.ncbi.nlm.nih.gov",
    "pmc.ncbi.nlm.nih.gov",
    "europepmc.org",
    "crossref.org",
    "api.crossref.org",
    "openalex.org",
    "semanticscholar.org",
    "arxiv.org",
    "biorxiv.org",
    "medrxiv.org",
    "nature.com",
    "science.org",
    "cell.com",
    "pnas.org",
    "nejm.org",
    "thelancet.com",
    "jamanetwork.com",
    "bmj.com",
    "springer.com",
    "link.springer.com",
    "wiley.com",
    "onlinelibrary.wiley.com",
    "sciencedirect.com",
    "ieee.org",
    "ieeexplore.ieee.org",
    "acm.org",
    "dl.acm.org",
    "nih.gov",
    "who.int",
}

PREPRINT_DOMAINS = {"arxiv.org", "biorxiv.org", "medrxiv.org"}

PUBLICATION_TYPES: tuple[tuple[str, tuple[str, ...]], ...] = (
    ("systematic review", ("systematic review",)),
    ("meta-analysis", ("meta-analysis", "meta analysis")),
    (
        "randomized controlled trial",
        ("randomized controlled trial", "randomised controlled trial", " rct "),
    ),
    ("clinical trial", ("clinical trial",)),
    ("cohort study", ("cohort study", "prospective cohort", "retrospective cohort")),
    ("case-control study", ("case-control", "case control")),
    ("cross-sectional study", ("cross-sectional", "cross sectional")),
    ("methods/protocol", ("protocol", "benchmark", "methodology", "methods paper")),
    ("case report/series", ("case report", "case series")),
    ("review", ("review",)),
)

EVIDENCE_WEIGHTS = {
    "systematic review": 5,
    "meta-analysis": 5,
    "randomized controlled trial": 4,
    "clinical trial": 4,
    "cohort study": 3,
    "case-control study": 3,
    "cross-sectional study": 2,
    "methods/protocol": 2,
    "review": 2,
    "case report/series": 1,
    "primary/other": 2,
}


def canonicalize_url(url: str) -> str:
    """Return a stable URL for deduplication without tracking parameters."""
    if not url:
        return ""
    parts = urlsplit(url.strip())
    query = [
        (key, value)
        for key, value in parse_qsl(parts.query, keep_blank_values=True)
        if not key.lower().startswith("utm_")
        and key.lower() not in {"ref", "source", "campaign"}
    ]
    path = parts.path.rstrip("/") or "/"
    return urlunsplit(
        (parts.scheme.lower(), parts.netloc.lower(), path, urlencode(query), "")
    )


def normalize_title(title: str) -> str:
    """Normalize a title for conservative duplicate detection."""
    return re.sub(r"[^a-z0-9]+", " ", (title or "").lower()).strip()


def source_text(source: dict[str, Any]) -> str:
    """Join source title and excerpts into searchable text."""
    excerpts = source.get("excerpts") or []
    if isinstance(excerpts, str):
        excerpts = [excerpts]
    return "\n".join(
        part.strip()
        for part in [str(source.get("title") or ""), *map(str, excerpts)]
        if part and str(part).strip()
    )


def extract_doi(text: str) -> str:
    """Extract the first DOI from text and strip sentence punctuation."""
    match = DOI_PATTERN.search(text or "")
    if not match:
        return ""
    return match.group(1).rstrip(".,;:)]}").lower()


def extract_pmid(text: str) -> str:
    """Extract the first PubMed identifier from text."""
    match = PMID_PATTERN.search(text or "")
    return match.group(1) if match else ""


def extract_year(source: dict[str, Any], text: str) -> str:
    """Extract a plausible publication year."""
    publish_date = str(source.get("publish_date") or "")
    match = YEAR_PATTERN.search(publish_date)
    if match:
        return match.group(1)
    match = YEAR_PATTERN.search(text)
    return match.group(1) if match else ""


def extract_label(text: str, labels: Iterable[str]) -> str:
    """Extract a simple labeled metadata value from source text."""
    joined = "|".join(re.escape(label) for label in labels)
    match = re.search(
        rf"(?:^|\n)(?:{joined})\s*[:\-]\s*([^\n]{{2,300}})",
        text,
        re.IGNORECASE,
    )
    return match.group(1).strip() if match else ""


def split_sentences(text: str) -> list[str]:
    """Split excerpts into useful, reasonably bounded sentences."""
    compact = re.sub(r"\s+", " ", text or "").strip()
    if not compact:
        return []
    sentences = re.split(r"(?<=[.!?])\s+(?=[A-Z0-9])", compact)
    return [sentence.strip() for sentence in sentences if 30 <= len(sentence) <= 800]


def classify_publication(text: str) -> str:
    """Classify publication type from explicit wording."""
    lowered = f" {text.lower()} "
    for publication_type, terms in PUBLICATION_TYPES:
        if any(term in lowered for term in terms):
            return publication_type
    return "primary/other"


def evidence_quality(
    publication_type: str, *, preprint: bool, retracted: bool
) -> tuple[str, str]:
    """Return a transparent evidence tier and rationale."""
    if retracted:
        return "exclude", "Source is marked as retracted or withdrawn."
    weight = EVIDENCE_WEIGHTS.get(publication_type, 1)
    if preprint:
        weight = max(1, weight - 1)
    label = {5: "high", 4: "high", 3: "moderate", 2: "contextual"}.get(
        weight, "low"
    )
    rationale = f"Classified as {publication_type}"
    if preprint:
        rationale += "; preprint status lowers confidence pending peer review"
    return label, rationale + "."


def relevance_tags(text: str, publication_type: str) -> list[str]:
    """Map external evidence to manuscript sections without using Results."""
    lowered = text.lower()
    tags = {"introduction", "discussion"}
    if publication_type == "methods/protocol" or any(
        term in lowered
        for term in (
            "method",
            "protocol",
            "assay",
            "measure",
            "instrument",
            "model",
            "analysis",
            "benchmark",
        )
    ):
        tags.add("methods-rationale")
    return sorted(tags)


def _host(url: str) -> str:
    return urlsplit(url).netloc.lower().removeprefix("www.")


def is_scholarly_url(url: str) -> bool:
    host = _host(url)
    return any(host == domain or host.endswith(f".{domain}") for domain in SCHOLARLY_DOMAINS)


def is_preprint_url(url: str) -> bool:
    host = _host(url)
    return any(host == domain or host.endswith(f".{domain}") for domain in PREPRINT_DOMAINS)


def deduplicate_sources(sources: Iterable[dict[str, Any]]) -> list[dict[str, Any]]:
    """Merge duplicate search/extract records by DOI, PMID, URL, or title."""
    merged: list[dict[str, Any]] = []
    key_to_index: dict[str, int] = {}

    for raw_source in sources:
        source = dict(raw_source)
        source["url"] = canonicalize_url(str(source.get("url") or ""))
        text = source_text(source)
        doi = str(source.get("doi") or extract_doi(f"{source['url']}\n{text}"))
        pmid = str(source.get("pmid") or extract_pmid(f"{source['url']}\n{text}"))
        title_key = normalize_title(str(source.get("title") or ""))
        keys = [
            f"doi:{doi}" if doi else "",
            f"pmid:{pmid}" if pmid else "",
            f"url:{source['url']}" if source["url"] else "",
            f"title:{title_key}" if len(title_key) >= 20 else "",
        ]
        keys = [key for key in keys if key]
        existing = next((key_to_index[key] for key in keys if key in key_to_index), None)

        if existing is None:
            source["doi"] = doi
            source["pmid"] = pmid
            source["facets"] = sorted(set(source.get("facets") or []))
            source["excerpts"] = list(dict.fromkeys(source.get("excerpts") or []))
            merged.append(source)
            index = len(merged) - 1
            for key in keys:
                key_to_index[key] = index
            continue

        current = merged[existing]
        if not current.get("title") and source.get("title"):
            current["title"] = source["title"]
        if not current.get("publish_date") and source.get("publish_date"):
            current["publish_date"] = source["publish_date"]
        if not current.get("doi") and doi:
            current["doi"] = doi
        if not current.get("pmid") and pmid:
            current["pmid"] = pmid
        current["excerpts"] = list(
            dict.fromkeys([*(current.get("excerpts") or []), *(source.get("excerpts") or [])])
        )
        current["facets"] = sorted(
            set(current.get("facets") or []) | set(source.get("facets") or [])
        )
        if source.get("extracted"):
            current["extracted"] = True
        for key in keys:
            key_to_index[key] = existing

    return merged


def normalize_reference(source: dict[str, Any], index: int) -> dict[str, Any]:
    """Convert a source into a structured evidence-matrix record."""
    text = source_text(source)
    url = canonicalize_url(str(source.get("url") or ""))
    doi = str(source.get("doi") or extract_doi(f"{url}\n{text}"))
    pmid = str(source.get("pmid") or extract_pmid(f"{url}\n{text}"))
    publication_type = classify_publication(text)
    preprint = is_preprint_url(url) or "preprint" in text.lower()
    retracted = any(
        marker in text.lower()
        for marker in ("retracted", "retraction notice", "withdrawn")
    )
    corrected = any(marker in text.lower() for marker in ("correction", "erratum"))
    quality, quality_rationale = evidence_quality(
        publication_type, preprint=preprint, retracted=retracted
    )
    sentences = split_sentences(text)
    findings = [
        sentence
        for sentence in sentences
        if any(
            term in sentence.lower()
            for term in (
                "found",
                "showed",
                "demonstrated",
                "associated",
                "increased",
                "decreased",
                "effect",
                "result",
                "concluded",
            )
        )
    ][:3]
    if not findings:
        findings = sentences[:2]
    quantitative = [sentence for sentence in sentences if EFFECT_PATTERN.search(sentence)][:5]
    limitations = [
        sentence
        for sentence in sentences
        if any(
            term in sentence.lower()
            for term in ("limitation", "limited by", "bias", "uncertain", "caution")
        )
    ][:3]
    sample_match = SAMPLE_SIZE_PATTERN.search(text)
    authors = extract_label(text, ("authors", "author"))
    venue = extract_label(text, ("journal", "venue", "published in"))
    methods = extract_label(text, ("methods", "methodology", "design"))
    outcomes = extract_label(text, ("outcomes", "outcome", "endpoints", "endpoint"))
    verification_status = (
        "extracted"
        if source.get("extracted")
        else "identifier-verified"
        if doi or pmid
        else "search-only"
    )

    return {
        "reference_id": f"ref-{index:03d}",
        "title": str(source.get("title") or "Untitled source").strip(),
        "authors": authors,
        "year": extract_year(source, text),
        "venue": venue,
        "url": url,
        "doi": doi,
        "pmid": pmid,
        "publication_type": publication_type,
        "study_design": publication_type,
        "population_or_system": extract_label(
            text, ("population", "participants", "subjects", "system")
        ),
        "sample_size": sample_match.group(1).replace(",", "") if sample_match else "",
        "methods": methods,
        "intervention_or_exposure": extract_label(
            text, ("intervention", "exposure", "treatment")
        ),
        "comparator": extract_label(text, ("comparator", "control")),
        "outcomes": outcomes,
        "key_findings": findings,
        "quantitative_findings": quantitative,
        "limitations": limitations,
        "evidence_quality": quality,
        "quality_rationale": quality_rationale,
        "preprint": preprint,
        "retracted": retracted,
        "corrected": corrected,
        "verification_status": verification_status,
        "manuscript_sections": relevance_tags(text, publication_type),
        "supporting_excerpts": list(source.get("excerpts") or [])[:5],
        "facets": list(source.get("facets") or []),
        "scholarly_source": is_scholarly_url(url),
    }


def reference_score(reference: dict[str, Any]) -> tuple[int, int, int, int]:
    """Sort by exclusion status, verification, quality, and recency."""
    quality_score = {
        "high": 4,
        "moderate": 3,
        "contextual": 2,
        "low": 1,
        "exclude": 0,
    }.get(str(reference.get("evidence_quality")), 0)
    verification_score = {
        "extracted": 2,
        "identifier-verified": 1,
        "search-only": 0,
    }.get(str(reference.get("verification_status")), 0)
    try:
        year = int(reference.get("year") or 0)
    except (TypeError, ValueError):
        year = 0
    return (
        0 if reference.get("retracted") else 1,
        verification_score,
        quality_score,
        year,
    )


def _claim_map(references: list[dict[str, Any]]) -> list[dict[str, Any]]:
    claims: list[dict[str, Any]] = []
    for reference in references:
        if reference.get("retracted"):
            continue
        for finding in reference.get("key_findings") or []:
            claims.append(
                {
                    "claim": finding,
                    "reference_ids": [reference["reference_id"]],
                    "supporting_excerpts": (reference.get("supporting_excerpts") or [])[:2],
                    "status": "single-source",
                }
            )
    return claims


def _synthesis(references: list[dict[str, Any]]) -> dict[str, Any]:
    conflict_terms = (
        "conflict",
        "contradict",
        "no significant",
        "null result",
        "did not",
        "failed to",
        "inconsistent",
    )
    consensus: list[dict[str, str]] = []
    conflicts: list[dict[str, str]] = []
    for reference in references:
        for finding in reference.get("key_findings") or []:
            item = {"reference_id": reference["reference_id"], "finding": finding}
            if any(term in finding.lower() for term in conflict_terms):
                conflicts.append(item)
            elif len(consensus) < 20:
                consensus.append(item)

    design_counts = Counter(
        str(reference.get("publication_type") or "unknown")
        for reference in references
    )
    gaps: list[str] = []
    if not any(reference.get("evidence_quality") == "high" for reference in references):
        gaps.append("No high-tier synthesis or trial evidence was verified.")
    if not conflicts:
        gaps.append(
            "No explicit contradictory or null evidence was identified; run a targeted "
            "negative-results search before asserting consensus."
        )
    if not any(reference.get("year") for reference in references):
        gaps.append("Publication years remain incomplete.")
    return {
        "consensus_evidence": consensus,
        "conflicting_evidence": conflicts,
        "methodological_patterns": dict(design_counts),
        "research_gaps": gaps,
    }


def _section_briefs(references: list[dict[str, Any]]) -> dict[str, Any]:
    briefs: dict[str, Any] = {
        "introduction": {
            "purpose": "Established background, significance, and unresolved gap.",
            "reference_ids": [],
            "candidate_evidence": [],
        },
        "methods-rationale": {
            "purpose": "Published precedent for measures, protocols, models, and analyses.",
            "reference_ids": [],
            "candidate_evidence": [],
        },
        "discussion": {
            "purpose": "Supporting and conflicting studies, mechanisms, limits, and implications.",
            "reference_ids": [],
            "candidate_evidence": [],
        },
    }
    for reference in references:
        for section in reference.get("manuscript_sections") or []:
            brief = briefs.get(section)
            if not brief:
                continue
            brief["reference_ids"].append(reference["reference_id"])
            if reference.get("key_findings"):
                brief["candidate_evidence"].append(
                    {
                        "reference_id": reference["reference_id"],
                        "finding": reference["key_findings"][0],
                    }
                )
    for brief in briefs.values():
        brief["reference_ids"] = brief["reference_ids"][:30]
        brief["candidate_evidence"] = brief["candidate_evidence"][:20]
    return briefs


def _coverage(
    references: list[dict[str, Any]], target_references: int
) -> dict[str, Any]:
    verified = [
        reference
        for reference in references
        if reference.get("verification_status") != "search-only"
        and not reference.get("retracted")
    ]
    years = Counter(
        str(reference["year"])
        for reference in references
        if reference.get("year")
    )
    quality = Counter(
        str(reference.get("evidence_quality") or "unknown")
        for reference in references
    )
    verification = Counter(
        str(reference.get("verification_status") or "unknown")
        for reference in references
    )
    return {
        "requested_references": target_references,
        "total_unique_references": len(references),
        "verified_references": len(verified),
        "shortfall": max(0, target_references - len(verified)),
        "evidence_quality_mix": dict(quality),
        "verification_mix": dict(verification),
        "publication_years": dict(sorted(years.items())),
        "scholarly_sources": sum(
            1 for reference in references if reference.get("scholarly_source")
        ),
        "preprints": sum(1 for reference in references if reference.get("preprint")),
        "retracted_or_withdrawn": sum(
            1 for reference in references if reference.get("retracted")
        ),
        "corrections_or_errata": sum(
            1 for reference in references if reference.get("corrected")
        ),
        "missing_doi_or_pmid": sum(
            1
            for reference in references
            if not reference.get("doi") and not reference.get("pmid")
        ),
        "full_text_note": (
            "Extraction verifies available public page content; paywalled full text may "
            "remain unavailable and must not be represented as reviewed."
        ),
    }


def build_manuscript_packet(
    *,
    query: str,
    sources: list[dict[str, Any]],
    search_ledger: list[dict[str, Any]],
    target_references: int = 60,
    manuscript_context: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Build a structured packet without adding facts beyond source excerpts."""
    deduplicated = deduplicate_sources(sources)
    references = [
        normalize_reference(source, index)
        for index, source in enumerate(deduplicated, start=1)
    ]
    references.sort(key=reference_score, reverse=True)
    for index, reference in enumerate(references, start=1):
        reference["reference_id"] = f"ref-{index:03d}"

    coverage = _coverage(references, target_references)
    warnings: list[str] = []
    if coverage["shortfall"]:
        warnings.append(
            f"Verified {coverage['verified_references']} of "
            f"{target_references} requested references; the list was not padded."
        )
    if coverage["retracted_or_withdrawn"]:
        warnings.append(
            "Retracted or withdrawn records are marked for exclusion and must not support claims."
        )
    if not manuscript_context:
        warnings.append(
            "No structured manuscript context was supplied; section briefs are broad."
        )

    return {
        "schema_version": "1.0",
        "query": query,
        "manuscript_context": manuscript_context or {},
        "target_references": target_references,
        "references": references,
        "evidence_matrix": references,
        "claim_source_map": _claim_map(references),
        "synthesis": _synthesis(references),
        "section_briefs": _section_briefs(references),
        "coverage": coverage,
        "search_ledger": search_ledger,
        "warnings": warnings,
    }


def citation_text(reference: dict[str, Any]) -> str:
    """Render a conservative citation without inventing missing metadata."""
    authors = str(reference.get("authors") or "").strip()
    title = str(reference.get("title") or "Untitled source").strip()
    year = str(reference.get("year") or "n.d.")
    venue = str(reference.get("venue") or "").strip()
    identifier = (
        f"https://doi.org/{reference['doi']}"
        if reference.get("doi")
        else str(reference.get("url") or "")
    )
    lead = f"{authors} ({year}). " if authors else f"({year}). "
    venue_text = f" {venue}." if venue else ""
    return f"{lead}{title}.{venue_text} {identifier}".strip()


def packet_markdown(packet: dict[str, Any]) -> str:
    """Render the packet as a concise, source-linked Markdown artifact."""
    coverage = packet["coverage"]
    lines = [
        "# Manuscript Research Packet",
        "",
        f"**Query:** {packet['query']}",
        f"**Verified references:** {coverage['verified_references']} / "
        f"{coverage['requested_references']}",
        "",
    ]
    if packet.get("warnings"):
        lines.extend(["## Warnings", ""])
        lines.extend(f"- {warning}" for warning in packet["warnings"])
        lines.append("")

    lines.extend(["## References", ""])
    for reference in packet["references"]:
        status = reference.get("verification_status", "unknown")
        quality = reference.get("evidence_quality", "unknown")
        lines.append(
            f"- **{reference['reference_id']}** {citation_text(reference)} "
            f"_[{status}; {quality}]_"
        )

    lines.extend(["", "## Evidence Synthesis", "", "### Consensus candidates", ""])
    consensus = packet["synthesis"]["consensus_evidence"]
    lines.extend(
        f"- [{item['reference_id']}] {item['finding']}" for item in consensus
    )
    if not consensus:
        lines.append("- No consensus statements could be extracted.")

    lines.extend(["", "### Conflicting or null evidence", ""])
    conflicts = packet["synthesis"]["conflicting_evidence"]
    lines.extend(
        f"- [{item['reference_id']}] {item['finding']}" for item in conflicts
    )
    if not conflicts:
        lines.append("- No explicit conflicting evidence was identified.")

    lines.extend(["", "### Research gaps", ""])
    gaps = packet["synthesis"]["research_gaps"]
    lines.extend(f"- {gap}" for gap in gaps)
    if not gaps:
        lines.append("- No automatic gap signal was detected; expert review remains required.")

    lines.extend(["", "## Section Briefs", ""])
    for section, brief in packet["section_briefs"].items():
        lines.extend([f"### {section.replace('-', ' ').title()}", "", brief["purpose"], ""])
        for item in brief["candidate_evidence"]:
            lines.append(f"- [{item['reference_id']}] {item['finding']}")
        if not brief["candidate_evidence"]:
            lines.append("- No section-specific evidence extracted.")
        lines.append("")
    return "\n".join(lines).rstrip() + "\n"


def bibtex_text(references: list[dict[str, Any]]) -> str:
    """Render minimal BibTeX records using only verified metadata."""
    records: list[str] = []
    for reference in references:
        if reference.get("retracted"):
            continue
        key_base = re.sub(
            r"[^A-Za-z0-9]+",
            "",
            f"{reference.get('authors') or 'source'}"
            f"{reference.get('year') or 'nd'}"
            f"{reference.get('reference_id')}",
        )
        fields = {
            "title": reference.get("title"),
            "author": reference.get("authors"),
            "year": reference.get("year"),
            "journal": reference.get("venue"),
            "doi": reference.get("doi"),
            "url": reference.get("url"),
        }
        lines = [f"@article{{{key_base},"]
        for name, value in fields.items():
            if value:
                escaped = str(value).replace("\\", "\\\\").replace("{", "\\{").replace("}", "\\}")
                lines.append(f"  {name} = {{{escaped}}},")
        lines.append("}")
        records.append("\n".join(lines))
    return "\n\n".join(records) + ("\n" if records else "")


def save_packet(packet: dict[str, Any], directory: str | Path) -> dict[str, str]:
    """Write reproducible packet artifacts and return their paths."""
    destination = Path(directory)
    destination.mkdir(parents=True, exist_ok=True)
    artifacts = {
        "packet_json": destination / "packet.json",
        "packet_markdown": destination / "packet.md",
        "references_json": destination / "references.json",
        "references_bib": destination / "references.bib",
        "evidence_matrix": destination / "evidence-matrix.json",
        "claim_source_map": destination / "claim-source-map.json",
        "synthesis": destination / "synthesis.json",
        "section_briefs": destination / "section-briefs.json",
        "coverage": destination / "coverage.json",
        "search_ledger": destination / "search-ledger.json",
    }
    payloads: dict[str, Any] = {
        "packet_json": packet,
        "references_json": packet["references"],
        "evidence_matrix": packet["evidence_matrix"],
        "claim_source_map": packet["claim_source_map"],
        "synthesis": packet["synthesis"],
        "section_briefs": packet["section_briefs"],
        "coverage": packet["coverage"],
        "search_ledger": packet["search_ledger"],
    }
    for name, payload in payloads.items():
        artifacts[name].write_text(
            json.dumps(payload, indent=2, ensure_ascii=False) + "\n",
            encoding="utf-8",
        )
    artifacts["packet_markdown"].write_text(packet_markdown(packet), encoding="utf-8")
    artifacts["references_bib"].write_text(
        bibtex_text(packet["references"]), encoding="utf-8"
    )
    return {name: str(path) for name, path in artifacts.items()}