Skills › Scientific Agent Skills › literature-review › search_databases.py
search_databases.py
266 lines · skills/literature-review/scripts/search_databases.py at 5f2bc03
1#!/usr/bin/env python3
2"""
3Literature Database Search Script
4Processes normalized search records locally; does not query databases.
5"""
7import argparse
8import json
9import math
10import re
11import sys
12from pathlib import Path
13from urllib.parse import unquote
14from typing import Dict, List
15from datetime import datetime
17def format_search_results(results: List[Dict], output_format: str = 'json') -> str:
18 """
19 Format search results for output.
21 Args:
22 results: List of search results
23 output_format: Format (json, markdown, or bibtex)
25 Returns:
26 Formatted string
27 """
28 if output_format == 'json':
29 return json.dumps(results, indent=2)
31 elif output_format == 'markdown':
32 md = f"# Literature Search Results\n\n"
33 md += f"**Export Generated**: {datetime.now().strftime('%Y-%m-%d %H:%M')}\n"
34 md += f"**Total Results**: {len(results)}\n\n"
36 for i, result in enumerate(results, 1):
37 md += f"## {i}. {result.get('title', 'Untitled')}\n\n"
38 md += f"**Authors**: {result.get('authors', 'Unknown')}\n\n"
39 md += f"**Year**: {result.get('year', 'N/A')}\n\n"
40 md += f"**Source**: {result.get('source', 'Unknown')}\n\n"
42 if result.get('abstract'):
43 md += f"**Abstract**: {result['abstract']}\n\n"
45 if result.get('doi'):
46 md += f"**DOI**: [{result['doi']}](https://doi.org/{result['doi']})\n\n"
48 if result.get('url'):
49 md += f"**URL**: {result['url']}\n\n"
51 if result.get('citations') is not None:
52 md += f"**Citations**: {result['citations']}\n\n"
54 md += "---\n\n"
56 return md
58 elif output_format == 'bibtex':
59 bibtex = ""
60 used_keys = set()
61 for i, result in enumerate(results, 1):
62 entry_type = result.get('type', 'article')
63 base_key = re.sub(r'[^a-zA-Z0-9_-]', '',
64 f"{result.get('first_author') or 'unknown'}{result.get('year') or '0000'}")
65 cite_key, suffix = base_key, 2
66 while cite_key in used_keys:
67 cite_key = f'{base_key}_{suffix}'
68 suffix += 1
69 used_keys.add(cite_key)
71 bibtex += f"@{entry_type}{{{cite_key},\n"
72 bibtex += f" title = {{{result.get('title', '')}}},\n"
73 bibtex += f" author = {{{result.get('authors', '')}}},\n"
74 bibtex += f" year = {{{result.get('year', '')}}},\n"
76 if result.get('journal'):
77 bibtex += f" journal = {{{result['journal']}}},\n"
79 if result.get('volume'):
80 bibtex += f" volume = {{{result['volume']}}},\n"
82 if result.get('pages'):
83 bibtex += f" pages = {{{result['pages']}}},\n"
85 if result.get('doi'):
86 bibtex += f" doi = {{{result['doi']}}},\n"
88 bibtex += "}\n\n"
90 return bibtex
92 else:
93 raise ValueError(f"Unknown format: {output_format}")
95def deduplicate_results(results: List[Dict]) -> List[Dict]:
96 """
97 Keep the first occurrence of a DOI; use title only among DOI-less records.
99 Exact title matches still require review. Preserve raw exports to recover
100 metadata from duplicate records; this helper does not identify studies.
102 Args:
103 results: List of search results
105 Returns:
106 Deduplicated list
107 """
108 seen_dois = set()
109 seen_titles = set()
110 unique_results = []
112 for result in results:
113 doi = unquote(re.sub(r'^(?:https?://(?:dx\.)?doi\.org/|doi:\s*)', '',
114 str(result.get('doi') or '').strip(), flags=re.I)).lower()
115 title = str(result.get('title') or '').lower().strip()
117 # Check DOI first (more reliable)
118 if doi and doi in seen_dois:
119 continue
121 # Check title as fallback
122 if not doi and title in seen_titles:
123 continue
125 # Add to results
126 if doi:
127 seen_dois.add(doi)
128 if title and not doi:
129 seen_titles.add(title)
131 unique_results.append(result)
133 return unique_results
135def rank_results(results: List[Dict], criteria: str = 'citations') -> List[Dict]:
136 """
137 Rank results by specified criteria.
139 Args:
140 results: List of search results
141 criteria: Ranking criteria (citations, year, relevance)
143 Returns:
144 Ranked list
145 """
146 field = {'citations': 'citations', 'year': 'year', 'relevance': 'relevance_score'}.get(criteria)
147 if field is None:
148 return results
149 def score(record):
150 try:
151 value = float(record.get(field))
152 return value if math.isfinite(value) else float('-inf')
153 except (ValueError, TypeError):
154 return float('-inf')
155 return sorted(results, key=score, reverse=True)
157def filter_by_year(results: List[Dict], start_year: int = None, end_year: int = None) -> List[Dict]:
158 """
159 Filter results by publication year range.
161 Args:
162 results: List of search results
163 start_year: Minimum year (inclusive)
164 end_year: Maximum year (inclusive)
166 Returns:
167 Filtered list
168 """
169 filtered = []
171 for result in results:
172 try:
173 year = int(result.get('year'))
174 if start_year and year < start_year:
175 continue
176 if end_year and year > end_year:
177 continue
178 filtered.append(result)
179 except (ValueError, TypeError):
180 # Include if year parsing fails
181 filtered.append(result)
183 return filtered
185def generate_search_summary(results: List[Dict]) -> Dict:
186 """
187 Generate summary statistics for search results.
189 Args:
190 results: List of search results
192 Returns:
193 Summary dictionary
194 """
195 summary = {
196 'total_results': len(results),
197 'sources': {},
198 'year_distribution': {},
199 'avg_citations': 0,
200 'total_citations': 0
201 }
203 citations = []
205 for result in results:
206 # Count by source
207 source = result.get('source', 'Unknown')
208 summary['sources'][source] = summary['sources'].get(source, 0) + 1
210 # Count by year
211 year = result.get('year', 'Unknown')
212 summary['year_distribution'][year] = summary['year_distribution'].get(year, 0) + 1
214 # Collect citations
215 if result.get('citations') is not None:
216 try:
217 value = int(result['citations'])
218 if value >= 0:
219 citations.append(value)
220 except (ValueError, TypeError):
221 pass
223 if citations:
224 summary['avg_citations'] = sum(citations) / len(citations)
225 summary['total_citations'] = sum(citations)
227 return summary
229def main():
230 """Process a JSON array; refuse raw provider envelopes and unknown options."""
231 parser = argparse.ArgumentParser(description=__doc__)
232 parser.add_argument('results_file', type=Path)
233 parser.add_argument('--format', choices=['json', 'markdown', 'bibtex'], default='markdown')
234 parser.add_argument('--output', type=Path)
235 parser.add_argument('--rank', choices=['citations', 'year', 'relevance'])
236 parser.add_argument('--year-start', type=int)
237 parser.add_argument('--year-end', type=int)
238 parser.add_argument('--deduplicate', action='store_true')
239 parser.add_argument('--summary', action='store_true')
240 args = parser.parse_args()
241 if args.year_start and args.year_end and args.year_start > args.year_end:
242 parser.error('--year-start must not exceed --year-end')
243 try:
244 results = json.loads(args.results_file.read_text(encoding='utf-8'))
245 if not isinstance(results, list) or any(not isinstance(r, dict) for r in results):
246 raise ValueError('input must be a normalized JSON array of record objects, not a provider response')
247 if args.deduplicate:
248 results = deduplicate_results(results)
249 if args.year_start is not None or args.year_end is not None:
250 results = filter_by_year(results, args.year_start, args.year_end)
251 if args.rank:
252 results = rank_results(results, args.rank)
253 if args.summary:
254 print(json.dumps(generate_search_summary(results), indent=2), file=sys.stderr)
255 output = format_search_results(results, args.format)
256 if args.output:
257 args.output.write_text(output, encoding='utf-8')
258 print(f'[OK] Results saved to: {args.output}', file=sys.stderr)
259 else:
260 print(output)
261 except (OSError, ValueError, TypeError) as exc:
262 parser.exit(1, f'[FAIL] {exc}\n')
265if __name__ == '__main__':
266 main()