Skills › Scientific Agent Skills › research-lookup › manuscript_packet.py
manuscript_packet.py
776 lines · skills/research-lookup/scripts/manuscript_packet.py at 5f2bc03
1"""Pure helpers for building manuscript-ready research packets."""
3from __future__ import annotations
5import json
6import re
7from collections import Counter
8from pathlib import Path
9from typing import Any, Iterable
10from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
13DOI_PATTERN = re.compile(
14 r"(?:doi[:\s]*|https?://(?:dx\.)?doi\.org/)?"
15 r"(10\.\d{4,9}/[-._;()/:A-Z0-9]+)",
16 re.IGNORECASE,
17)
18PMID_PATTERN = re.compile(
19 r"(?:pubmed\.ncbi\.nlm\.nih\.gov/|pmid[:\s]*)(\d{6,9})",
20 re.IGNORECASE,
21)
22YEAR_PATTERN = re.compile(r"\b(19\d{2}|20\d{2})\b")
23SAMPLE_SIZE_PATTERN = re.compile(
24 r"\b(?:n\s*=\s*|sample(?:\s+size)?\s+(?:of\s+)?)([\d,]+)\b",
25 re.IGNORECASE,
26)
27EFFECT_PATTERN = re.compile(
28 r"(?:\b\d+(?:\.\d+)?\s*%|\bp\s*[<=>]\s*0?\.\d+|"
29 r"\b(?:OR|RR|HR|MD|SMD)\s*[=:]\s*-?\d+(?:\.\d+)?|"
30 r"\b95\s*%\s*CI\b)",
31 re.IGNORECASE,
32)
34SCHOLARLY_DOMAINS = {
35 "pubmed.ncbi.nlm.nih.gov",
36 "pmc.ncbi.nlm.nih.gov",
37 "europepmc.org",
38 "crossref.org",
39 "api.crossref.org",
40 "openalex.org",
41 "semanticscholar.org",
42 "arxiv.org",
43 "biorxiv.org",
44 "medrxiv.org",
45 "nature.com",
46 "science.org",
47 "cell.com",
48 "pnas.org",
49 "nejm.org",
50 "thelancet.com",
51 "jamanetwork.com",
52 "bmj.com",
53 "springer.com",
54 "link.springer.com",
55 "wiley.com",
56 "onlinelibrary.wiley.com",
57 "sciencedirect.com",
58 "ieee.org",
59 "ieeexplore.ieee.org",
60 "acm.org",
61 "dl.acm.org",
62 "nih.gov",
63 "who.int",
64}
66PREPRINT_DOMAINS = {"arxiv.org", "biorxiv.org", "medrxiv.org"}
68PUBLICATION_TYPES: tuple[tuple[str, tuple[str, ...]], ...] = (
69 ("systematic review", ("systematic review",)),
70 ("meta-analysis", ("meta-analysis", "meta analysis")),
71 (
72 "randomized controlled trial",
73 ("randomized controlled trial", "randomised controlled trial", " rct "),
74 ),
75 ("clinical trial", ("clinical trial",)),
76 ("cohort study", ("cohort study", "prospective cohort", "retrospective cohort")),
77 ("case-control study", ("case-control", "case control")),
78 ("cross-sectional study", ("cross-sectional", "cross sectional")),
79 ("methods/protocol", ("protocol", "benchmark", "methodology", "methods paper")),
80 ("case report/series", ("case report", "case series")),
81 ("review", ("review",)),
82)
84EVIDENCE_WEIGHTS = {
85 "systematic review": 5,
86 "meta-analysis": 5,
87 "randomized controlled trial": 4,
88 "clinical trial": 4,
89 "cohort study": 3,
90 "case-control study": 3,
91 "cross-sectional study": 2,
92 "methods/protocol": 2,
93 "review": 2,
94 "case report/series": 1,
95 "primary/other": 2,
96}
99def canonicalize_url(url: str) -> str:
100 """Return a stable URL for deduplication without tracking parameters."""
101 if not url:
102 return ""
103 parts = urlsplit(url.strip())
104 query = [
105 (key, value)
106 for key, value in parse_qsl(parts.query, keep_blank_values=True)
107 if not key.lower().startswith("utm_")
108 and key.lower() not in {"ref", "source", "campaign"}
109 ]
110 path = parts.path.rstrip("/") or "/"
111 return urlunsplit(
112 (parts.scheme.lower(), parts.netloc.lower(), path, urlencode(query), "")
113 )
116def normalize_title(title: str) -> str:
117 """Normalize a title for conservative duplicate detection."""
118 return re.sub(r"[^a-z0-9]+", " ", (title or "").lower()).strip()
121def source_excerpts(source: dict[str, Any]) -> list[str]:
122 """Prefer Extract evidence over candidate snippets after retrieval succeeds."""
123 excerpts = (
124 source.get("extraction_excerpts")
125 if source.get("extracted") and "extraction_excerpts" in source
126 else source.get("excerpts")
127 ) or []
128 if isinstance(excerpts, str):
129 excerpts = [excerpts]
130 return [excerpt for excerpt in excerpts if isinstance(excerpt, str) and excerpt.strip()]
133def source_text(source: dict[str, Any]) -> str:
134 """Join source title and excerpts into searchable text."""
135 return "\n".join(
136 part.strip()
137 for part in [str(source.get("title") or ""), *source_excerpts(source)]
138 if part and str(part).strip()
139 )
142def extract_doi(text: str) -> str:
143 """Extract the first DOI from text and strip sentence punctuation."""
144 match = DOI_PATTERN.search(text or "")
145 if not match:
146 return ""
147 return match.group(1).rstrip(".,;:)]}").lower()
150def extract_pmid(text: str) -> str:
151 """Extract the first PubMed identifier from text."""
152 match = PMID_PATTERN.search(text or "")
153 return match.group(1) if match else ""
156def extract_year(source: dict[str, Any], text: str) -> str:
157 """Extract a plausible publication year."""
158 publish_date = str(source.get("publish_date") or "")
159 match = YEAR_PATTERN.search(publish_date)
160 if match:
161 return match.group(1)
162 match = YEAR_PATTERN.search(text)
163 return match.group(1) if match else ""
166def extract_label(text: str, labels: Iterable[str]) -> str:
167 """Extract a simple labeled metadata value from source text."""
168 joined = "|".join(re.escape(label) for label in labels)
169 match = re.search(
170 rf"(?:^|\n)(?:{joined})\s*[:\-]\s*([^\n]{{2,300}})",
171 text,
172 re.IGNORECASE,
173 )
174 return match.group(1).strip() if match else ""
177def split_sentences(text: str) -> list[str]:
178 """Split excerpts into useful, reasonably bounded sentences."""
179 compact = re.sub(r"\s+", " ", text or "").strip()
180 if not compact:
181 return []
182 sentences = re.split(r"(?<=[.!?])\s+(?=[A-Z0-9])", compact)
183 return [sentence.strip() for sentence in sentences if 30 <= len(sentence) <= 800]
186def classify_publication(text: str) -> str:
187 """Classify publication type from explicit wording."""
188 lowered = f" {text.lower()} "
189 for publication_type, terms in PUBLICATION_TYPES:
190 if any(term in lowered for term in terms):
191 return publication_type
192 return "primary/other"
195def evidence_quality(
196 publication_type: str, *, preprint: bool, retracted: bool
197) -> tuple[str, str]:
198 """Return a transparent evidence tier and rationale."""
199 if retracted:
200 return "exclude", "Source is marked as retracted or withdrawn."
201 weight = EVIDENCE_WEIGHTS.get(publication_type, 1)
202 if preprint:
203 weight = max(1, weight - 1)
204 label = {5: "high", 4: "high", 3: "moderate", 2: "contextual"}.get(
205 weight, "low"
206 )
207 rationale = f"Classified as {publication_type}"
208 if preprint:
209 rationale += "; preprint status lowers confidence pending peer review"
210 return label, rationale + "."
213def relevance_tags(text: str, publication_type: str) -> list[str]:
214 """Map external evidence to manuscript sections without using Results."""
215 lowered = text.lower()
216 tags = {"introduction", "discussion"}
217 if publication_type == "methods/protocol" or any(
218 term in lowered
219 for term in (
220 "method",
221 "protocol",
222 "assay",
223 "measure",
224 "instrument",
225 "model",
226 "analysis",
227 "benchmark",
228 )
229 ):
230 tags.add("methods-rationale")
231 return sorted(tags)
234def _host(url: str) -> str:
235 return urlsplit(url).netloc.lower().removeprefix("www.")
238def is_scholarly_url(url: str) -> bool:
239 host = _host(url)
240 return any(host == domain or host.endswith(f".{domain}") for domain in SCHOLARLY_DOMAINS)
243def is_preprint_url(url: str) -> bool:
244 host = _host(url)
245 return any(host == domain or host.endswith(f".{domain}") for domain in PREPRINT_DOMAINS)
248def deduplicate_sources(sources: Iterable[dict[str, Any]]) -> list[dict[str, Any]]:
249 """Merge duplicate search/extract records by DOI, PMID, URL, or title."""
250 merged: list[dict[str, Any]] = []
251 key_to_index: dict[str, int] = {}
253 for raw_source in sources:
254 source = dict(raw_source)
255 source["url"] = canonicalize_url(str(source.get("url") or ""))
256 text = source_text(source)
257 doi = str(source.get("doi") or extract_doi(f"{source['url']}\n{text}"))
258 pmid = str(source.get("pmid") or extract_pmid(f"{source['url']}\n{text}"))
259 title_key = normalize_title(str(source.get("title") or ""))
260 keys = [
261 f"doi:{doi}" if doi else "",
262 f"pmid:{pmid}" if pmid else "",
263 f"url:{source['url']}" if source["url"] else "",
264 f"title:{title_key}" if len(title_key) >= 20 else "",
265 ]
266 keys = [key for key in keys if key]
267 existing = next((key_to_index[key] for key in keys if key in key_to_index), None)
269 if existing is None:
270 source["doi"] = doi
271 source["pmid"] = pmid
272 source["facets"] = sorted(set(source.get("facets") or []))
273 source["excerpts"] = list(dict.fromkeys(source.get("excerpts") or []))
274 merged.append(source)
275 index = len(merged) - 1
276 for key in keys:
277 key_to_index[key] = index
278 continue
280 current = merged[existing]
281 if not current.get("title") and source.get("title"):
282 current["title"] = source["title"]
283 if not current.get("publish_date") and source.get("publish_date"):
284 current["publish_date"] = source["publish_date"]
285 if not current.get("doi") and doi:
286 current["doi"] = doi
287 if not current.get("pmid") and pmid:
288 current["pmid"] = pmid
289 current["excerpts"] = list(
290 dict.fromkeys([*(current.get("excerpts") or []), *(source.get("excerpts") or [])])
291 )
292 current["facets"] = sorted(
293 set(current.get("facets") or []) | set(source.get("facets") or [])
294 )
295 if source.get("extracted"):
296 current["extracted"] = True
297 current["extraction_excerpts"] = list(
298 dict.fromkeys([
299 *(current.get("extraction_excerpts") or []), *source_excerpts(source)
300 ])
301 )
302 for key in keys:
303 key_to_index[key] = existing
305 return merged
308def normalize_reference(source: dict[str, Any], index: int) -> dict[str, Any]:
309 """Convert a source into a structured evidence-matrix record."""
310 text = source_text(source)
311 url = canonicalize_url(str(source.get("url") or ""))
312 doi = str(source.get("doi") or extract_doi(f"{url}\n{text}"))
313 pmid = str(source.get("pmid") or extract_pmid(f"{url}\n{text}"))
314 publication_type = classify_publication(text)
315 preprint = is_preprint_url(url) or "preprint" in text.lower()
316 retracted = any(
317 marker in text.lower()
318 for marker in ("retracted", "retraction notice", "withdrawn")
319 )
320 corrected = any(marker in text.lower() for marker in ("correction", "erratum"))
321 quality, quality_rationale = evidence_quality(
322 publication_type, preprint=preprint, retracted=retracted
323 )
324 sentences = split_sentences(text)
325 findings = [
326 sentence
327 for sentence in sentences
328 if any(
329 term in sentence.lower()
330 for term in (
331 "found",
332 "showed",
333 "demonstrated",
334 "associated",
335 "increased",
336 "decreased",
337 "effect",
338 "result",
339 "concluded",
340 )
341 )
342 ][:3]
343 if not findings:
344 findings = sentences[:2]
345 quantitative = [sentence for sentence in sentences if EFFECT_PATTERN.search(sentence)][:5]
346 limitations = [
347 sentence
348 for sentence in sentences
349 if any(
350 term in sentence.lower()
351 for term in ("limitation", "limited by", "bias", "uncertain", "caution")
352 )
353 ][:3]
354 sample_match = SAMPLE_SIZE_PATTERN.search(text)
355 authors = extract_label(text, ("authors", "author"))
356 venue = extract_label(text, ("journal", "venue", "published in"))
357 methods = extract_label(text, ("methods", "methodology", "design"))
358 outcomes = extract_label(text, ("outcomes", "outcome", "endpoints", "endpoint"))
359 verification_status = (
360 "extracted"
361 if source.get("extracted") and source_excerpts(source)
362 else "search-only"
363 )
365 return {
366 "reference_id": f"ref-{index:03d}",
367 "title": str(source.get("title") or "Untitled source").strip(),
368 "authors": authors,
369 "year": extract_year(source, text),
370 "venue": venue,
371 "url": url,
372 "doi": doi,
373 "pmid": pmid,
374 "publication_type": publication_type,
375 "study_design": publication_type,
376 "population_or_system": extract_label(
377 text, ("population", "participants", "subjects", "system")
378 ),
379 "sample_size": sample_match.group(1).replace(",", "") if sample_match else "",
380 "methods": methods,
381 "intervention_or_exposure": extract_label(
382 text, ("intervention", "exposure", "treatment")
383 ),
384 "comparator": extract_label(text, ("comparator", "control")),
385 "outcomes": outcomes,
386 "key_findings": findings,
387 "quantitative_findings": quantitative,
388 "limitations": limitations,
389 "evidence_quality": quality,
390 "quality_rationale": quality_rationale,
391 "preprint": preprint,
392 "retracted": retracted,
393 "corrected": corrected,
394 "verification_status": verification_status,
395 "manuscript_sections": relevance_tags(text, publication_type),
396 "supporting_excerpts": source_excerpts(source)[:5],
397 "facets": list(source.get("facets") or []),
398 "scholarly_source": is_scholarly_url(url),
399 }
402def reference_score(reference: dict[str, Any]) -> tuple[int, int, int, int]:
403 """Sort by exclusion status, verification, quality, and recency."""
404 quality_score = {
405 "high": 4,
406 "moderate": 3,
407 "contextual": 2,
408 "low": 1,
409 "exclude": 0,
410 }.get(str(reference.get("evidence_quality")), 0)
411 verification_score = {
412 "extracted": 2,
413 "search-only": 0,
414 }.get(str(reference.get("verification_status")), 0)
415 try:
416 year = int(reference.get("year") or 0)
417 except (TypeError, ValueError):
418 year = 0
419 return (
420 0 if reference.get("retracted") else 1,
421 verification_score,
422 quality_score,
423 year,
424 )
427def _claim_map(references: list[dict[str, Any]]) -> list[dict[str, Any]]:
428 claims: list[dict[str, Any]] = []
429 for reference in references:
430 if reference.get("retracted"):
431 continue
432 for finding in reference.get("key_findings") or []:
433 claims.append(
434 {
435 "claim": finding,
436 "reference_ids": [reference["reference_id"]],
437 "supporting_excerpts": (reference.get("supporting_excerpts") or [])[:2],
438 "status": (
439 "single-source" if reference.get("verification_status") == "extracted"
440 else "unverified-source"
441 ),
442 }
443 )
444 return claims
447def _synthesis(references: list[dict[str, Any]]) -> dict[str, Any]:
448 references = [reference for reference in references if not reference.get("retracted")]
449 conflict_terms = (
450 "conflict",
451 "contradict",
452 "no significant",
453 "null result",
454 "did not",
455 "failed to",
456 "inconsistent",
457 )
458 consensus: list[dict[str, str]] = []
459 conflicts: list[dict[str, str]] = []
460 for reference in references:
461 for finding in reference.get("key_findings") or []:
462 item = {"reference_id": reference["reference_id"], "finding": finding}
463 if any(term in finding.lower() for term in conflict_terms):
464 conflicts.append(item)
465 elif len(consensus) < 20:
466 consensus.append(item)
468 design_counts = Counter(
469 str(reference.get("publication_type") or "unknown")
470 for reference in references
471 )
472 gaps: list[str] = []
473 if not any(
474 reference.get("evidence_quality") == "high"
475 and reference.get("verification_status") == "extracted"
476 for reference in references
477 ):
478 gaps.append("No high-tier synthesis or trial evidence was verified.")
479 if not conflicts:
480 gaps.append(
481 "No explicit contradictory or null evidence was identified; run a targeted "
482 "negative-results search before asserting consensus."
483 )
484 if not any(reference.get("year") for reference in references):
485 gaps.append("Publication years remain incomplete.")
486 return {
487 "consensus_evidence": consensus,
488 "conflicting_evidence": conflicts,
489 "methodological_patterns": dict(design_counts),
490 "research_gaps": gaps,
491 }
494def _section_briefs(references: list[dict[str, Any]]) -> dict[str, Any]:
495 briefs: dict[str, Any] = {
496 "introduction": {
497 "purpose": "Established background, significance, and unresolved gap.",
498 "reference_ids": [],
499 "candidate_evidence": [],
500 },
501 "methods-rationale": {
502 "purpose": "Published precedent for measures, protocols, models, and analyses.",
503 "reference_ids": [],
504 "candidate_evidence": [],
505 },
506 "discussion": {
507 "purpose": "Supporting and conflicting studies, mechanisms, limits, and implications.",
508 "reference_ids": [],
509 "candidate_evidence": [],
510 },
511 }
512 for reference in references:
513 if reference.get("retracted"):
514 continue
515 for section in reference.get("manuscript_sections") or []:
516 brief = briefs.get(section)
517 if not brief:
518 continue
519 brief["reference_ids"].append(reference["reference_id"])
520 if reference.get("key_findings"):
521 brief["candidate_evidence"].append(
522 {
523 "reference_id": reference["reference_id"],
524 "finding": reference["key_findings"][0],
525 }
526 )
527 for brief in briefs.values():
528 brief["reference_ids"] = brief["reference_ids"][:30]
529 brief["candidate_evidence"] = brief["candidate_evidence"][:20]
530 return briefs
533def _coverage(
534 references: list[dict[str, Any]], target_references: int
535) -> dict[str, Any]:
536 verified = [
537 reference
538 for reference in references
539 if reference.get("verification_status") == "extracted"
540 and not reference.get("retracted")
541 ]
542 years = Counter(
543 str(reference["year"])
544 for reference in references
545 if reference.get("year")
546 )
547 quality = Counter(
548 str(reference.get("evidence_quality") or "unknown")
549 for reference in references
550 )
551 verification = Counter(
552 str(reference.get("verification_status") or "unknown")
553 for reference in references
554 )
555 return {
556 "requested_references": target_references,
557 "total_unique_references": len(references),
558 "verified_references": len(verified),
559 "shortfall": max(0, target_references - len(verified)),
560 "evidence_quality_mix": dict(quality),
561 "verification_mix": dict(verification),
562 "publication_years": dict(sorted(years.items())),
563 "scholarly_sources": sum(
564 1 for reference in references if reference.get("scholarly_source")
565 ),
566 "preprints": sum(1 for reference in references if reference.get("preprint")),
567 "retracted_or_withdrawn": sum(
568 1 for reference in references if reference.get("retracted")
569 ),
570 "corrections_or_errata": sum(
571 1 for reference in references if reference.get("corrected")
572 ),
573 "missing_doi_or_pmid": sum(
574 1
575 for reference in references
576 if not reference.get("doi") and not reference.get("pmid")
577 ),
578 "full_text_note": (
579 "Verified means nonempty excerpts were retrieved by Extract, not that "
580 "bibliographic identity or claims were manually verified. Paywalled full "
581 "text may remain unavailable and must not be represented as reviewed."
582 ),
583 }
586def build_manuscript_packet(
587 *,
588 query: str,
589 sources: list[dict[str, Any]],
590 search_ledger: list[dict[str, Any]],
591 target_references: int = 60,
592 manuscript_context: dict[str, Any] | None = None,
593) -> dict[str, Any]:
594 """Build a structured packet without adding facts beyond source excerpts."""
595 deduplicated = deduplicate_sources(sources)
596 references = [
597 normalize_reference(source, index)
598 for index, source in enumerate(deduplicated, start=1)
599 ]
600 references.sort(key=reference_score, reverse=True)
601 for index, reference in enumerate(references, start=1):
602 reference["reference_id"] = f"ref-{index:03d}"
604 coverage = _coverage(references, target_references)
605 warnings: list[str] = []
606 if coverage["shortfall"]:
607 warnings.append(
608 f"Verified {coverage['verified_references']} of "
609 f"{target_references} requested references; the list was not padded."
610 )
611 if coverage["retracted_or_withdrawn"]:
612 warnings.append(
613 "Retracted or withdrawn records are marked for exclusion and must not support claims."
614 )
615 if not manuscript_context:
616 warnings.append(
617 "No structured manuscript context was supplied; section briefs are broad."
618 )
620 return {
621 "schema_version": "1.0",
622 "query": query,
623 "manuscript_context": manuscript_context or {},
624 "target_references": target_references,
625 "references": references,
626 "evidence_matrix": references,
627 "claim_source_map": _claim_map(references),
628 "synthesis": _synthesis(references),
629 "section_briefs": _section_briefs(references),
630 "coverage": coverage,
631 "search_ledger": search_ledger,
632 "warnings": warnings,
633 }
636def citation_text(reference: dict[str, Any]) -> str:
637 """Render a conservative citation without inventing missing metadata."""
638 authors = str(reference.get("authors") or "").strip()
639 title = str(reference.get("title") or "Untitled source").strip()
640 year = str(reference.get("year") or "n.d.")
641 venue = str(reference.get("venue") or "").strip()
642 identifier = (
643 f"https://doi.org/{reference['doi']}"
644 if reference.get("doi")
645 else str(reference.get("url") or "")
646 )
647 lead = f"{authors} ({year}). " if authors else f"({year}). "
648 venue_text = f" {venue}." if venue else ""
649 return f"{lead}{title}.{venue_text} {identifier}".strip()
652def packet_markdown(packet: dict[str, Any]) -> str:
653 """Render the packet as a concise, source-linked Markdown artifact."""
654 coverage = packet["coverage"]
655 lines = [
656 "# Manuscript Research Packet",
657 "",
658 f"**Query:** {packet['query']}",
659 f"**Verified references:** {coverage['verified_references']} / "
660 f"{coverage['requested_references']}",
661 "",
662 ]
663 if packet.get("warnings"):
664 lines.extend(["## Warnings", ""])
665 lines.extend(f"- {warning}" for warning in packet["warnings"])
666 lines.append("")
668 lines.extend(["## References", ""])
669 for reference in packet["references"]:
670 status = reference.get("verification_status", "unknown")
671 quality = reference.get("evidence_quality", "unknown")
672 lines.append(
673 f"- **{reference['reference_id']}** {citation_text(reference)} "
674 f"_[{status}; {quality}]_"
675 )
677 lines.extend(["", "## Evidence Synthesis", "", "### Consensus candidates", ""])
678 consensus = packet["synthesis"]["consensus_evidence"]
679 lines.extend(
680 f"- [{item['reference_id']}] {item['finding']}" for item in consensus
681 )
682 if not consensus:
683 lines.append("- No consensus statements could be extracted.")
685 lines.extend(["", "### Conflicting or null evidence", ""])
686 conflicts = packet["synthesis"]["conflicting_evidence"]
687 lines.extend(
688 f"- [{item['reference_id']}] {item['finding']}" for item in conflicts
689 )
690 if not conflicts:
691 lines.append("- No explicit conflicting evidence was identified.")
693 lines.extend(["", "### Research gaps", ""])
694 gaps = packet["synthesis"]["research_gaps"]
695 lines.extend(f"- {gap}" for gap in gaps)
696 if not gaps:
697 lines.append("- No automatic gap signal was detected; expert review remains required.")
699 lines.extend(["", "## Section Briefs", ""])
700 for section, brief in packet["section_briefs"].items():
701 lines.extend([f"### {section.replace('-', ' ').title()}", "", brief["purpose"], ""])
702 for item in brief["candidate_evidence"]:
703 lines.append(f"- [{item['reference_id']}] {item['finding']}")
704 if not brief["candidate_evidence"]:
705 lines.append("- No section-specific evidence extracted.")
706 lines.append("")
707 return "\n".join(lines).rstrip() + "\n"
710def bibtex_text(references: list[dict[str, Any]]) -> str:
711 """Render minimal BibTeX records using only verified metadata."""
712 records: list[str] = []
713 for reference in references:
714 if reference.get("retracted"):
715 continue
716 key_base = re.sub(
717 r"[^A-Za-z0-9]+",
718 "",
719 f"{reference.get('authors') or 'source'}"
720 f"{reference.get('year') or 'nd'}"
721 f"{reference.get('reference_id')}",
722 )
723 fields = {
724 "title": reference.get("title"),
725 "author": reference.get("authors"),
726 "year": reference.get("year"),
727 "journal": reference.get("venue"),
728 "doi": reference.get("doi"),
729 "url": reference.get("url"),
730 }
731 lines = [f"@article{{{key_base},"]
732 for name, value in fields.items():
733 if value:
734 escaped = str(value).replace("\\", "\\\\").replace("{", "\\{").replace("}", "\\}")
735 lines.append(f" {name} = {{{escaped}}},")
736 lines.append("}")
737 records.append("\n".join(lines))
738 return "\n\n".join(records) + ("\n" if records else "")
741def save_packet(packet: dict[str, Any], directory: str | Path) -> dict[str, str]:
742 """Write reproducible packet artifacts and return their paths."""
743 destination = Path(directory)
744 destination.mkdir(parents=True, exist_ok=True)
745 artifacts = {
746 "packet_json": destination / "packet.json",
747 "packet_markdown": destination / "packet.md",
748 "references_json": destination / "references.json",
749 "references_bib": destination / "references.bib",
750 "evidence_matrix": destination / "evidence-matrix.json",
751 "claim_source_map": destination / "claim-source-map.json",
752 "synthesis": destination / "synthesis.json",
753 "section_briefs": destination / "section-briefs.json",
754 "coverage": destination / "coverage.json",
755 "search_ledger": destination / "search-ledger.json",
756 }
757 payloads: dict[str, Any] = {
758 "packet_json": packet,
759 "references_json": packet["references"],
760 "evidence_matrix": packet["evidence_matrix"],
761 "claim_source_map": packet["claim_source_map"],
762 "synthesis": packet["synthesis"],
763 "section_briefs": packet["section_briefs"],
764 "coverage": packet["coverage"],
765 "search_ledger": packet["search_ledger"],
766 }
767 for name, payload in payloads.items():
768 artifacts[name].write_text(
769 json.dumps(payload, indent=2, ensure_ascii=False) + "\n",
770 encoding="utf-8",
771 )
772 artifacts["packet_markdown"].write_text(packet_markdown(packet), encoding="utf-8")
773 artifacts["references_bib"].write_text(
774 bibtex_text(packet["references"]), encoding="utf-8"
775 )
776 return {name: str(path) for name, path in artifacts.items()}