Spaces:
Sleeping
Sleeping
File size: 3,738 Bytes
0d7be17 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 | #!/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
@dataclass
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
|