Spaces:
Sleeping
Sleeping
| #!/usr/bin/env python3 | |
| """Compact Taxonomic Resolver for Hugging Face Spaces & Edge Deployments. | |
| Provides sub-millisecond, zero-cost taxonomic entity resolution and kingdom classification | |
| using the lightweight eco_core_backbone.db (~2.1 MB) with optional GBIF API fallback. | |
| """ | |
| import os | |
| import sqlite3 | |
| import urllib.request | |
| import urllib.parse | |
| import json | |
| from pathlib import Path | |
| from dataclasses import dataclass | |
| from typing import Optional, Dict, List | |
| class CompactTaxon: | |
| name: str | |
| canonical_name: str | |
| gbif_key: Optional[int] | |
| ncbi_taxid: Optional[str] | |
| kingdom: str | |
| source: str = "compact_backbone" | |
| class CompactTaxonomicResolver: | |
| """Ultra-lightweight resolver tailored for CPU-basic online spaces.""" | |
| def __init__(self, db_path: Optional[str] = None, allow_network_fallback: bool = True): | |
| if db_path and os.path.exists(db_path): | |
| self.db_path = db_path | |
| else: | |
| candidates = [ | |
| "eco_core_backbone.db", | |
| str(Path(__file__).resolve().parent / "eco_core_backbone.db") | |
| ] | |
| self.db_path = next((c for c in candidates if os.path.exists(c)), None) | |
| self.allow_network = allow_network_fallback | |
| self.conn = sqlite3.connect(self.db_path, check_same_thread=False) if self.db_path else None | |
| self.cache: Dict[str, Optional[CompactTaxon]] = {} | |
| def resolve(self, name: str) -> Optional[CompactTaxon]: | |
| """Resolves a scientific name to GBIF, NCBI, and Kingdom in <50 µs.""" | |
| if not name or len(name.strip()) < 3: | |
| return None | |
| clean_name = name.strip() | |
| name_lower = clean_name.lower() | |
| if name_lower in self.cache: | |
| return self.cache[name_lower] | |
| # 1. Local SQLite Lookup (<50 µs) | |
| if self.conn: | |
| cur = self.conn.cursor() | |
| row = cur.execute( | |
| "SELECT canonical_name, gbif_key, ncbi_taxid, kingdom FROM taxa WHERE name_lower = ?", | |
| (name_lower,) | |
| ).fetchone() | |
| if row: | |
| canon, gbif, taxid, kingdom = row | |
| taxon = CompactTaxon( | |
| name=clean_name, | |
| canonical_name=canon or clean_name, | |
| gbif_key=gbif, | |
| ncbi_taxid=taxid, | |
| kingdom=kingdom or "Unknown", | |
| source="compact_backbone" | |
| ) | |
| self.cache[name_lower] = taxon | |
| return taxon | |
| # 2. Network Fallback to GBIF API (if enabled) | |
| if self.allow_network: | |
| taxon = self._fallback_gbif(clean_name) | |
| if taxon: | |
| self.cache[name_lower] = taxon | |
| return taxon | |
| self.cache[name_lower] = None | |
| return None | |
| def _fallback_gbif(self, name: str) -> Optional[CompactTaxon]: | |
| try: | |
| url = f"https://api.gbif.org/v1/species/match?name={urllib.parse.quote(name)}&strict=true" | |
| req = urllib.request.Request(url, headers={"User-Agent": "EcoTagger/CompactResolver-v1"}) | |
| with urllib.request.urlopen(req, timeout=3.0) as resp: | |
| data = json.loads(resp.read().decode("utf-8")) | |
| if data.get("matchType") in ("EXACT", "FUZZY"): | |
| return CompactTaxon( | |
| name=name, | |
| canonical_name=data.get("canonicalName", name), | |
| gbif_key=data.get("usageKey"), | |
| ncbi_taxid=None, | |
| kingdom=data.get("kingdom", "Unknown"), | |
| source="gbif_api_fallback" | |
| ) | |
| except Exception: | |
| pass | |
| return None | |