Skills › Scientific Agent Skills › peer-review › audit_citations.py
audit_citations.py
207 lines · skills/peer-review/scripts/audit_citations.py at 5f2bc03
1#!/usr/bin/env python3
2"""Audit Markdown citation keys against a local reference CSV without network use."""
4from __future__ import annotations
6import argparse
7import re
8from collections import Counter
9from typing import Any
11from _common import (
12 ValidationError,
13 error_exit,
14 issue,
15 read_csv_records,
16 read_markdown,
17 require_doi,
18 require_enum,
19 require_identifier,
20 require_text,
21 require_unique,
22 require_url,
23 write_json_report,
24)
26REFERENCE_FIELDS = (
27 "reference_id",
28 "title",
29 "authors",
30 "year",
31 "doi",
32 "url",
33 "verification_status",
34)
35VERIFICATION_STATUSES = {
36 "verified_primary",
37 "verified_secondary",
38 "not_verified",
39}
40CITATION_GROUP_RE = re.compile(r"\[([^\]\n]{1,1000})\]")
41CITATION_KEY_RE = re.compile(r"@([A-Za-z][A-Za-z0-9._:-]{0,95})")
42YEAR_RE = re.compile(r"^(?:1[5-9]\d{2}|20\d{2}|2100)$")
45def load_references(raw_path: str) -> list[dict[str, str]]:
46 rows = read_csv_records(
47 raw_path,
48 required_fields=REFERENCE_FIELDS,
49 allowed_fields=REFERENCE_FIELDS,
50 )
51 parsed: list[dict[str, str]] = []
52 ids: list[str] = []
53 for line_number, row in enumerate(rows, start=2):
54 context = f"references row {line_number}"
55 reference_id = require_identifier(
56 row["reference_id"], f"{context}.reference_id"
57 )
58 ids.append(reference_id)
59 year = require_text(row["year"], f"{context}.year", maximum=4)
60 if not YEAR_RE.fullmatch(year):
61 raise ValidationError(f"{context}.year must be a four-digit year")
62 parsed.append(
63 {
64 "reference_id": reference_id,
65 "title": require_text(
66 row["title"], f"{context}.title", minimum=3, maximum=2_000
67 ),
68 "authors": require_text(
69 row["authors"], f"{context}.authors", minimum=2, maximum=2_000
70 ),
71 "year": year,
72 "doi": require_doi(
73 row["doi"], f"{context}.doi", allow_empty=True
74 ),
75 "url": require_url(
76 row["url"], f"{context}.url", allow_empty=True
77 ),
78 "verification_status": require_enum(
79 row["verification_status"],
80 VERIFICATION_STATUSES,
81 f"{context}.verification_status",
82 ),
83 }
84 )
85 require_unique(ids, "reference IDs")
86 return parsed
89def extract_citations(markdown: str) -> tuple[dict[str, list[int]], list[int]]:
90 citations: dict[str, list[int]] = {}
91 malformed_lines: list[int] = []
92 for line_number, line in enumerate(markdown.splitlines(), start=1):
93 matched_starts = 0
94 for group in CITATION_GROUP_RE.finditer(line):
95 content = group.group(1)
96 if "@" not in content:
97 continue
98 matched_starts += content.count("@")
99 keys = CITATION_KEY_RE.findall(content)
100 if not keys:
101 malformed_lines.append(line_number)
102 continue
103 for key in keys:
104 citations.setdefault(key, []).append(line_number)
105 if line.count("[@") > matched_starts:
106 malformed_lines.append(line_number)
107 return citations, sorted(set(malformed_lines))
110def audit(markdown: str, references: list[dict[str, str]]) -> dict[str, Any]:
111 citations, malformed_lines = extract_citations(markdown)
112 by_id = {row["reference_id"]: row for row in references}
113 cited_ids = set(citations)
114 reference_ids = set(by_id)
115 missing_ids = sorted(cited_ids - reference_ids)
116 uncited_ids = sorted(reference_ids - cited_ids)
117 unverified_ids = sorted(
118 reference_id
119 for reference_id in cited_ids.intersection(reference_ids)
120 if by_id[reference_id]["verification_status"] == "not_verified"
121 )
122 no_locator_ids = sorted(
123 row["reference_id"]
124 for row in references
125 if not row["doi"] and not row["url"]
126 )
127 errors: list[dict[str, str]] = [
128 issue("CITATION_WITHOUT_REFERENCE", reference_id)
129 for reference_id in missing_ids
130 ]
131 errors.extend(
132 issue("MALFORMED_CITATION_SYNTAX", f"line:{line_number}")
133 for line_number in malformed_lines
134 )
135 warnings: list[dict[str, str]] = [
136 issue("REFERENCE_NOT_CITED", reference_id) for reference_id in uncited_ids
137 ]
138 warnings.extend(
139 issue("CITED_REFERENCE_NOT_VERIFIED", reference_id)
140 for reference_id in unverified_ids
141 )
142 warnings.extend(
143 issue("REFERENCE_HAS_NO_PERSISTENT_LOCATOR", reference_id)
144 for reference_id in no_locator_ids
145 )
146 verification_counts = Counter(
147 row["verification_status"] for row in references
148 )
149 return {
150 "schema_version": "2.0",
151 "valid": not errors,
152 "status": "VALID" if not errors else "CITATION_INCONSISTENCIES",
153 "errors": errors,
154 "warnings": warnings,
155 "citation_occurrence_count": sum(len(lines) for lines in citations.values()),
156 "cited_reference_count": len(cited_ids),
157 "reference_count": len(references),
158 "missing_reference_ids": missing_ids,
159 "missing_reference_line_numbers": {
160 reference_id: sorted(set(citations[reference_id]))
161 for reference_id in missing_ids
162 },
163 "uncited_reference_ids": uncited_ids,
164 "unverified_cited_reference_ids": unverified_ids,
165 "references_without_persistent_locator": no_locator_ids,
166 "verification_counts": dict(sorted(verification_counts.items())),
167 "malformed_citation_line_numbers": malformed_lines,
168 "notice": (
169 "This local audit checks structured citation-key consistency and "
170 "identifier format only. It does not query registries, verify that a "
171 "source exists, confirm that a citation supports a claim, or echo "
172 "manuscript/reference prose."
173 ),
174 }
177def build_parser() -> argparse.ArgumentParser:
178 parser = argparse.ArgumentParser(
179 description=(
180 "Audit Pandoc-style Markdown citations such as [@ref-id] against a "
181 "strict local reference CSV. No network calls are made."
182 )
183 )
184 parser.add_argument("manuscript", help="Local Markdown file")
185 parser.add_argument("references", help="Local reference CSV")
186 parser.add_argument("-o", "--output", help="Optional local JSON report")
187 parser.add_argument(
188 "--force", action="store_true", help="Replace an existing output file"
189 )
190 return parser
193def main() -> int:
194 args = build_parser().parse_args()
195 try:
196 report = audit(
197 read_markdown(args.manuscript),
198 load_references(args.references),
199 )
200 write_json_report(report, args.output, force=args.force)
201 return 0 if report["valid"] else 1
202 except ValidationError as exc:
203 return error_exit(exc)
206if __name__ == "__main__":
207 raise SystemExit(main())