check_latest_month.py

#!/usr/bin/env python3
"""
Check for latest available ClinVar data month
Compares what's available in the archive vs what we have locally
"""

import os
import re
import requests
from pathlib import Path
import json

# Configuration
ARCHIVE_URL = "https://ftp.ncbi.nlm.nih.gov/pub/clinvar/tab_delimited/archive/"
DATA_DIR = "data"

def get_latest_available_month():
    """Check archive directory for latest month with variant_summary files"""
    try:
        response = requests.get(ARCHIVE_URL, timeout=30)
        response.raise_for_status()

        # Parse FTP directory listing for year folders
        year_pattern = r'href="(\d{4})/"'
        years = re.findall(year_pattern, response.text)
        years = sorted([int(y) for y in years], reverse=True)

        if not years:
            return None

        # Check the most recent year first
        for year in years:
            year_url = f"{ARCHIVE_URL}{year}/"

            try:
                year_response = requests.get(year_url, timeout=30)
                year_response.raise_for_status()

                # Look for variant_summary files
                file_pattern = r'variant_summary_(\d{4})-(\d{2})-\d{2}\.txt\.gz'
                matches = re.findall(file_pattern, year_response.text)

                if matches:
                    # Sort by month and get the latest
                    months = sorted([(int(y), int(m)) for y, m in matches], reverse=True)
                    latest_year, latest_month = months[0]
                    return f"{latest_year}-{latest_month:02d}"

            except requests.RequestException:
                continue

        return None

    except requests.RequestException:
        return None

def get_latest_local_month():
    """Check what JSON files we have locally to determine latest processed month"""
    data_path = Path(DATA_DIR)

    if not data_path.exists():
        return None

    # Look for JSON files with pattern: variant_summary_YYYY-MM.json
    json_pattern = r'variant_summary_(\d{4})-(\d{2})\.json'
    json_files = []

    for file_path in data_path.glob("variant_summary_*.json"):
        match = re.match(json_pattern, file_path.name)
        if match:
            year, month = int(match.group(1)), int(match.group(2))
            json_files.append((year, month))

    if not json_files:
        return None

    # Get the latest
    latest_year, latest_month = max(json_files)
    return f"{latest_year}-{latest_month:02d}"

def set_github_output(name, value):
    """Set GitHub Actions output variable"""
    if os.getenv('GITHUB_ACTIONS'):
        with open(os.environ['GITHUB_OUTPUT'], 'a') as f:
            f.write(f"{name}={value}\n")

def main():
    """Main function"""
    latest_available = get_latest_available_month()
    latest_local = get_latest_local_month()

    if not latest_available:
        set_github_output("new_data_available", "false")
        set_github_output("error", "Could not check archive")
        return

    if not latest_local:
        set_github_output("new_data_available", "true")
        set_github_output("target_month", latest_available)
        return

    if latest_available > latest_local:
        set_github_output("new_data_available", "true")
        set_github_output("target_month", latest_available)
    else:
        set_github_output("new_data_available", "false")

if __name__ == "__main__":
    main()