verify_citations.py · Scientific Agent Skills

1#!/usr/bin/env python3

2"""Check DOI registration and retrieve Crossref metadata for manual citation review.

3

4Registration does not establish that a citation supports a claim. Crossref does

5not cover all DOI registration agencies; unavailable metadata needs manual review.

6"""

7

8import argparse

9import json

10import re

11import time

12from pathlib import Path

13from typing import Dict, List, Tuple

14from urllib.parse import quote, unquote

15

16import requests

17

18

19def normalize_doi(value: str) -> str:

20 """Normalize a bare DOI or doi.org URL without stripping legal DOI punctuation."""

21 value = re.sub(r"^(?:https?://(?:dx\.)?doi\.org/|doi:\s*)", "", value.strip(), flags=re.I)

22 return unquote(value).lower()

23

24

25class CitationVerifier:

26 def __init__(self, email: str = None):

27 self.email = email

28 self.session = requests.Session()

29 self.session.headers.update({'User-Agent': 'LiteratureReviewCitationVerifier/1.1'})

30

31 def extract_dois(self, text: str) -> List[str]:

32 """Extract unique candidates; prose punctuation remains inherently ambiguous."""

33 found = []

34 for candidate in re.findall(r'10\.\d{4,9}/[^\s<>"{}]+', text, flags=re.I):

35 candidate = unquote(candidate).rstrip('.,;:')

36 # Keep balanced parentheses inside older DOI suffixes, remove Markdown closers.

37 while candidate and candidate[-1] in ')]':

38 closing = candidate[-1]

39 opening = '(' if closing == ')' else '['

40 if candidate.count(closing) <= candidate.count(opening):

41 break

42 candidate = candidate[:-1].rstrip('.,;:')

43 candidate = normalize_doi(candidate)

44 if candidate not in found:

45 found.append(candidate)

46 return found

47

48 def _get(self, url: str, **kwargs):

49 """Retry only transient read failures, with bounded backoff."""

50 for attempt in range(3):

51 response = self.session.get(url, timeout=20, **kwargs)

52 if response.status_code not in (429, 500, 502, 503, 504) or attempt == 2:

53 return response

54 try:

55 delay = float(response.headers.get('Retry-After', 2 ** attempt))

56 except (TypeError, ValueError):

57 delay = 2 ** attempt

58 # Long server-directed delays need a later run, not an early retry.

59 if delay > 60:

60 return response

61 time.sleep(max(0, delay))

62 raise AssertionError('unreachable')

63

64 def verify_doi(self, doi: str) -> Tuple[bool, Dict]:

65 """Return registration status and metadata; false includes inconclusive requests."""

66 doi = normalize_doi(doi)

67 if not re.fullmatch(r'10\.\d{4,9}/\S+', doi):

68 return False, {'doi': doi, 'status': 'invalid_syntax'}

69 try:

70 response = self._get(f"https://doi.org/api/handles/{quote(doi, safe='/')}")

71 payload = response.json()

72 code = payload.get('responseCode') if isinstance(payload, dict) else None

73 returned = payload.get('handle', '') if isinstance(payload, dict) else ''

74 if (code == 100 and response.status_code in (200, 404) and

75 isinstance(returned, str) and normalize_doi(returned) == doi):

76 return False, {'doi': doi, 'status': 'missing', 'handle_response_code': code}

77 if (response.status_code != 200 or code != 1 or

78 not isinstance(returned, str) or normalize_doi(returned) != doi):

79 return False, {'doi': doi, 'status': 'inconclusive',

80 'http_status': response.status_code, 'handle_response_code': code}

81 metadata = self._get_crossref_metadata(doi)

82 metadata.update({'doi': doi, 'status': 'registered'})

83 return True, metadata

84 except (requests.RequestException, ValueError) as exc:

85 return False, {'doi': doi, 'status': 'inconclusive', 'error': str(exc)}

86

87 def _get_crossref_metadata(self, doi: str) -> Dict:

88 """Fetch one Crossref work; no list pagination or key is needed."""

89 try:

90 response = self._get(f"https://api.crossref.org/works/{quote(doi, safe='/')}",

91 params={'mailto': self.email} if self.email else {})

92 if response.status_code != 200:

93 return {'metadata_status': 'unavailable', 'crossref_http_status': response.status_code}

94 message = response.json().get('message')

95 if not isinstance(message, dict) or normalize_doi(message.get('DOI', '')) != doi:

96 return {'metadata_status': 'unavailable', 'error': 'Crossref response DOI missing or mismatched'}

97 def first(field):

98 values = message.get(field) or []

99 return values[0] if isinstance(values, list) and values else ''

100 authors = message.get('author') or []

101 return {

102 'metadata_status': 'retrieved', 'title': first('title'),

103 'authors': self._format_authors(authors), 'author': authors,

104 'year': self._extract_year(message), 'journal': first('container-title'),

105 'volume': message.get('volume', ''), 'issue': message.get('issue', ''),

106 'pages': message.get('page', ''), 'doi': doi,

107 'crossref_message': message,

108 }

109 except (requests.RequestException, ValueError, AttributeError) as exc:

110 return {'metadata_status': 'unavailable', 'error': str(exc)}

111

112 def _format_authors(self, authors: List[Dict]) -> str:

113 """Preserve all authors for review; this is not a CSL citation formatter."""

114 names = []

115 for author in authors:

116 if not isinstance(author, dict):

117 continue

118 family, given = author.get('family', ''), author.get('given', '')

119 if family:

120 names.append(f'{family}, {given}' if given else family)

121 elif author.get('name'):

122 names.append(author['name'])

123 return '; '.join(names)

124

125 def _extract_year(self, message: Dict) -> str:

126 for field in ('published-print', 'published-online', 'published', 'issued'):

127 dates = (message.get(field) or {}).get('date-parts') or []

128 if dates and dates[0]:

129 return str(dates[0][0])

130 return ''

131

132 def verify_url(self, url: str) -> Tuple[bool, int]:

133 """Check accessibility only; a 403/HEAD failure is not proof of an invalid URL."""

134 try:

135 response = self.session.head(url, timeout=20, allow_redirects=True)

136 return response.status_code < 400, response.status_code

137 except requests.RequestException:

138 return False, 0

139

140 def verify_citations_in_file(self, filepath: str) -> Dict:

141 dois = self.extract_dois(Path(filepath).read_text(encoding='utf-8'))

142 report = {'total_dois': len(dois), 'verified': [], 'failed': [],

143 'metadata_unavailable': [], 'metadata': {}, 'errors': {}, 'missing': [], 'inconclusive': [],

144 'verification_scope': 'DOI registration and Crossref metadata, not claim support or retraction screening'}

145 for doi in dois:

146 print(f'Checking DOI: {doi}')

147 registered, metadata = self.verify_doi(doi)

148 if registered:

149 report['verified'].append(doi)

150 report['metadata'][doi] = metadata

151 if metadata.get('metadata_status') != 'retrieved':

152 report['metadata_unavailable'].append(doi)

153 else:

154 report['failed'].append(doi)

155 report['errors'][doi] = metadata

156 bucket = 'missing' if metadata.get('status') == 'missing' else 'inconclusive'

157 report[bucket].append(doi)

158 time.sleep(0.5) # Sequential, <=2 Crossref single-work requests/s.

159 return report

160

161 def format_citation_apa(self, metadata: Dict) -> str:

162 """Return a metadata preview, not a standards-compliant APA reference."""

163 return (f"{metadata.get('authors', '')} ({metadata.get('year') or 'n.d.'}). "

164 f"{metadata.get('title', '')}. {metadata.get('journal', '')}. "

165 f"https://doi.org/{metadata.get('doi', '')}")

166

167 def format_citation_nature(self, metadata: Dict) -> str:

168 """Return a metadata preview; use CSL for final Nature formatting."""

169 return (f"{metadata.get('authors', '')} {metadata.get('title', '')}. "

170 f"{metadata.get('journal', '')} {metadata.get('volume', '')}, "

171 f"{metadata.get('pages', '')} ({metadata.get('year', '')}).")

172

173

174def main():

175 parser = argparse.ArgumentParser(description=__doc__)

176 parser.add_argument('markdown_file', type=Path)

177 parser.add_argument('--email', help='Your contact email for the Crossref polite pool')

178 parser.add_argument('-o', '--output', type=Path, help='JSON report path')

179 args = parser.parse_args()

180 output = args.output or args.markdown_file.with_name(args.markdown_file.stem + '_citation_report.json')

181 if output.resolve() == args.markdown_file.resolve():

182 parser.error('report path must differ from the input file')

183 try:

184 verifier = CitationVerifier(email=args.email)

185 report = verifier.verify_citations_in_file(str(args.markdown_file))

186 output.write_text(json.dumps(report, indent=2), encoding='utf-8')

187 except (OSError, ValueError) as exc:

188 parser.exit(1, f'[FAIL] {exc}\n')

189 print(f"Registered: {len(report['verified'])}; unresolved: {len(report['failed'])}; "