Skills › Scientific Agent Skills › scientific-writing › check_references.py
check_references.py
219 lines · skills/scientific-writing/scripts/check_references.py at 5f2bc03
1"""Check local reference identifiers and duplicates without resolving them."""
3from __future__ import annotations
5import argparse
6import re
7import unicodedata
8from typing import Any
10from _common import (
11 InputError,
12 Issue,
13 emit_report,
14 is_nonempty_string,
15 issue,
16 read_json,
17 require_list,
18 require_object,
19 run,
20)
21from validate_manifest import validate_source_manifest
23TOOL = "check_references"
24EVIDENCE_ID_RE = re.compile(r"^E[0-9]{3,8}$")
25DOI_RE = re.compile(r"^10\.[0-9]{4,9}/\S+$", re.IGNORECASE)
26PMID_RE = re.compile(r"^[1-9][0-9]{0,8}$")
27PMCID_RE = re.compile(r"^PMC[1-9][0-9]{0,8}$", re.IGNORECASE)
28URL_RE = re.compile(r"^https?://[^\s]+$", re.IGNORECASE)
31def normalize_doi(value: str) -> str:
32 normalized = value.strip().lower()
33 for prefix in ("https://doi.org/", "http://doi.org/", "doi:"):
34 if normalized.startswith(prefix):
35 normalized = normalized[len(prefix) :]
36 break
37 return normalized.rstrip(".,;")
40def normalize_title(value: str) -> str:
41 decomposed = unicodedata.normalize("NFKD", value).casefold()
42 return " ".join(re.findall(r"[a-z0-9]+", decomposed))
45def normalize_isbn(value: str) -> str:
46 return re.sub(r"[^0-9Xx]", "", value)
49def valid_isbn(value: str) -> bool:
50 normalized = normalize_isbn(value)
51 if len(normalized) == 10:
52 if not re.fullmatch(r"[0-9]{9}[0-9Xx]", normalized):
53 return False
54 total = sum(
55 (10 - index) * (10 if char.lower() == "x" else int(char))
56 for index, char in enumerate(normalized)
57 )
58 return total % 11 == 0
59 if len(normalized) == 13 and normalized.isdigit():
60 total = sum(
61 int(char) * (1 if index % 2 == 0 else 3)
62 for index, char in enumerate(normalized)
63 )
64 return total % 10 == 0
65 return False
68def _record_duplicate(
69 seen: dict[str, str],
70 normalized: str,
71 evidence_id: str,
72 code: str,
73 issues: list[Issue],
74) -> None:
75 if not normalized:
76 return
77 prior = seen.get(normalized)
78 if prior is not None and prior != evidence_id:
79 issues.append(
80 issue(
81 "error",
82 code,
83 location=prior,
84 item_id=evidence_id,
85 )
86 )
87 else:
88 seen[normalized] = evidence_id
91def check_sources(data: dict[str, Any]) -> tuple[list[Issue], int]:
92 issues = validate_source_manifest(data, require_verified=False)
93 sources = require_list(data.get("sources"), "sources")
94 evidence_ids: set[str] = set()
95 seen_dois: dict[str, str] = {}
96 seen_pmids: dict[str, str] = {}
97 seen_pmcids: dict[str, str] = {}
98 seen_isbns: dict[str, str] = {}
99 seen_titles: dict[str, str] = {}
101 for index, raw_source in enumerate(sources):
102 source = require_object(raw_source, f"sources[{index}]")
103 evidence_id = source.get("evidence_id")
104 if not isinstance(evidence_id, str) or not EVIDENCE_ID_RE.fullmatch(
105 evidence_id
106 ):
107 issues.append(
108 issue("error", "INVALID_EVIDENCE_ID", location=f"sources[{index}]")
109 )
110 continue
111 if evidence_id in evidence_ids:
112 issues.append(issue("error", "DUPLICATE_EVIDENCE_ID", item_id=evidence_id))
113 continue
114 evidence_ids.add(evidence_id)
116 title = source.get("title")
117 if is_nonempty_string(title):
118 _record_duplicate(
119 seen_titles,
120 normalize_title(str(title)),
121 evidence_id,
122 "POSSIBLE_DUPLICATE_TITLE",
123 issues,
124 )
126 identifiers = require_object(
127 source.get("identifiers"), f"sources[{index}].identifiers"
128 )
129 doi = identifiers.get("doi")
130 if is_nonempty_string(doi):
131 normalized_doi = normalize_doi(str(doi))
132 if not DOI_RE.fullmatch(normalized_doi):
133 issues.append(issue("error", "MALFORMED_DOI", item_id=evidence_id))
134 else:
135 _record_duplicate(
136 seen_dois,
137 normalized_doi,
138 evidence_id,
139 "DUPLICATE_DOI",
140 issues,
141 )
143 pmid = identifiers.get("pmid")
144 if is_nonempty_string(pmid):
145 normalized_pmid = str(pmid).strip()
146 if not PMID_RE.fullmatch(normalized_pmid):
147 issues.append(issue("error", "MALFORMED_PMID", item_id=evidence_id))
148 else:
149 _record_duplicate(
150 seen_pmids,
151 normalized_pmid,
152 evidence_id,
153 "DUPLICATE_PMID",
154 issues,
155 )
157 pmcid = identifiers.get("pmcid")
158 if is_nonempty_string(pmcid):
159 normalized_pmcid = str(pmcid).strip().upper()
160 if not PMCID_RE.fullmatch(normalized_pmcid):
161 issues.append(issue("error", "MALFORMED_PMCID", item_id=evidence_id))
162 else:
163 _record_duplicate(
164 seen_pmcids,
165 normalized_pmcid,
166 evidence_id,
167 "DUPLICATE_PMCID",
168 issues,
169 )
171 isbn = identifiers.get("isbn")
172 if is_nonempty_string(isbn):
173 normalized_isbn = normalize_isbn(str(isbn))
174 if not valid_isbn(normalized_isbn):
175 issues.append(issue("error", "MALFORMED_ISBN", item_id=evidence_id))
176 else:
177 _record_duplicate(
178 seen_isbns,
179 normalized_isbn.upper(),
180 evidence_id,
181 "DUPLICATE_ISBN",
182 issues,
183 )
185 url = identifiers.get("url")
186 if is_nonempty_string(url) and not URL_RE.fullmatch(str(url).strip()):
187 issues.append(issue("error", "MALFORMED_URL", item_id=evidence_id))
189 if not any(is_nonempty_string(value) for value in identifiers.values()):
190 issues.append(
191 issue("warning", "NO_IDENTIFIER_TO_CHECK", item_id=evidence_id)
192 )
193 return issues, len(sources)
196def build_parser() -> argparse.ArgumentParser:
197 parser = argparse.ArgumentParser(
198 description=(
199 "Validate DOI, PMID, PMCID, ISBN, and URL syntax and flag duplicates "
200 "in a local source manifest. No identifier is resolved over a network."
201 )
202 )
203 parser.add_argument("sources", help="UTF-8 JSON source manifest")
204 return parser
207def cli() -> int:
208 args = build_parser().parse_args()
209 data = require_object(read_json(args.sources), "source_manifest")
210 if data.get("schema_version") != "1.0":
211 raise InputError("unsupported source-manifest version")
212 issues, count = check_sources(data)
213 return emit_report(
214 TOOL, issues, summary={"sources_checked": count, "network_used": False}
215 )
218if __name__ == "__main__":
219 run(TOOL, cli)