|
| 1 | +import requests |
| 2 | +import os |
| 3 | +import pandas as pd |
| 4 | + |
| 5 | +# URL Configuration |
| 6 | +TAGS_URL = "https://api.github.com/repos/cov-lineages/pango-designation/tags" |
| 7 | +CSV_URL = "https://raw.githubusercontent.com/cov-lineages/pango-designation/master/lineages.csv" |
| 8 | +BASE_DIR = "lineages_data" |
| 9 | + |
| 10 | + |
| 11 | +def get_latest_tag(): |
| 12 | + """Fetches the latest published tag from the repository.""" |
| 13 | + response = requests.get(TAGS_URL) |
| 14 | + if response.status_code == 200: |
| 15 | + tags = response.json() |
| 16 | + if tags: |
| 17 | + return tags[0]["name"] # Latest available tag |
| 18 | + raise Exception("Failed to fetch repository tags.") |
| 19 | + |
| 20 | + |
| 21 | +def clean_tag(tag): |
| 22 | + """Removes the leading 'v' if present to avoid duplicate names.""" |
| 23 | + return tag.lstrip("v") |
| 24 | + |
| 25 | + |
| 26 | +def get_local_versions(): |
| 27 | + """Lists locally stored versions in subdirectories.""" |
| 28 | + if not os.path.exists(BASE_DIR): |
| 29 | + return [] |
| 30 | + return [d for d in os.listdir(BASE_DIR) if os.path.isdir(os.path.join(BASE_DIR, d))] |
| 31 | + |
| 32 | + |
| 33 | +def download_new_version(tag): |
| 34 | + """Downloads the CSV file and stores it in a versioned subdirectory.""" |
| 35 | + clean_version = clean_tag(tag) # Clean the version to avoid "vv" |
| 36 | + version_dir = os.path.join(BASE_DIR, f"v{clean_version}") # Version folder |
| 37 | + csv_path = os.path.join(version_dir, f"lineages_v{clean_version}.csv") |
| 38 | + excel_path = os.path.join(version_dir, f"Lineages_Mutations_v{clean_version}.xlsx") |
| 39 | + |
| 40 | + if clean_version in [clean_tag(v) for v in get_local_versions()]: |
| 41 | + print(f"⚠️ The latest version is already stored in {csv_path}.") |
| 42 | + update_latest_symlinks(csv_path, excel_path) |
| 43 | + return |
| 44 | + |
| 45 | + print(f"📥 Downloading new version to {csv_path}...") |
| 46 | + |
| 47 | + os.makedirs(version_dir, exist_ok=True) # Create directory if it doesn't exist |
| 48 | + |
| 49 | + response = requests.get(CSV_URL) |
| 50 | + |
| 51 | + if response.status_code == 200: |
| 52 | + with open(csv_path, "w", encoding="utf-8") as f: |
| 53 | + f.write(f"# Version: {tag}\n") # Add version as the first line |
| 54 | + f.write(response.text) # Save CSV content |
| 55 | + print(f"✅ File saved at {csv_path}.") |
| 56 | + |
| 57 | + # 📌 🔥 Ensure Excel file is correctly generated 🔥 📌 |
| 58 | + if os.path.exists(csv_path): |
| 59 | + print(f"📊 Generating Excel file: {excel_path} ...") |
| 60 | + generate_excel(csv_path, excel_path, tag) |
| 61 | + else: |
| 62 | + print(f"❌ ERROR: CSV file not found at {csv_path}.") |
| 63 | + |
| 64 | + update_latest_symlinks(csv_path, excel_path) |
| 65 | + else: |
| 66 | + print("❌ Error downloading the file.") |
| 67 | + |
| 68 | + |
| 69 | +def generate_excel(csv_path, excel_path, version): |
| 70 | + """Generates an Excel file with unique lineages from the CSV and includes the version.""" |
| 71 | + try: |
| 72 | + print(f"📂 Reading CSV: {csv_path}") |
| 73 | + |
| 74 | + df = pd.read_csv(csv_path, comment="#") # Ignore version line |
| 75 | + print("📌 Detected columns in CSV:", df.columns) |
| 76 | + |
| 77 | + if "lineage" not in df.columns: |
| 78 | + print( |
| 79 | + "❌ ERROR: The 'lineage' column is missing in the CSV. Aborting Excel generation." |
| 80 | + ) |
| 81 | + return |
| 82 | + |
| 83 | + unique_lineages = df["lineage"].dropna().unique() |
| 84 | + unique_df = pd.DataFrame(unique_lineages, columns=["Unique lineages"]) |
| 85 | + |
| 86 | + # Create an Excel workbook with one sheet for lineages and another for the version |
| 87 | + with pd.ExcelWriter(excel_path, engine="openpyxl") as writer: |
| 88 | + unique_df.to_excel(writer, sheet_name="Lineages", index=False) |
| 89 | + |
| 90 | + # Add the version in a second sheet |
| 91 | + wb = writer.book |
| 92 | + ws_version = wb.create_sheet(title="Metadata") |
| 93 | + ws_version.append(["Version", version]) |
| 94 | + |
| 95 | + print(f"✅ Excel file generated at {excel_path} with version included.") |
| 96 | + |
| 97 | + except Exception as e: |
| 98 | + print(f"❌ ERROR generating Excel file: {e}") |
| 99 | + |
| 100 | + |
| 101 | +def update_latest_symlinks(latest_csv_path, latest_excel_path): |
| 102 | + """Updates symbolic links 'latest.csv' and 'latest.xlsx'.""" |
| 103 | + latest_csv_symlink = os.path.join(BASE_DIR, "latest.csv") |
| 104 | + latest_excel_symlink = os.path.join(BASE_DIR, "latest.xlsx") |
| 105 | + |
| 106 | + for symlink, target in [ |
| 107 | + (latest_csv_symlink, latest_csv_path), |
| 108 | + (latest_excel_symlink, latest_excel_path), |
| 109 | + ]: |
| 110 | + if os.path.exists(symlink) or os.path.islink(symlink): |
| 111 | + os.unlink(symlink) # Remove old symbolic link |
| 112 | + os.symlink(os.path.abspath(target), symlink) # Create new symbolic link |
| 113 | + print(f"🔗 Symbolic link '{symlink}' now points to {target}") |
| 114 | + |
| 115 | + |
| 116 | +if __name__ == "__main__": |
| 117 | + try: |
| 118 | + latest_tag = get_latest_tag() |
| 119 | + download_new_version(latest_tag) |
| 120 | + except Exception as e: |
| 121 | + print(f"❌ GENERAL ERROR: {e}") |
0 commit comments