File size: 2,748 Bytes
22741d9 | 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 | from __future__ import annotations
import os
from dataclasses import dataclass
from huggingface_hub import HfApi
_OWNER_ALIASES = {
'jack': 'jackoatmon',
'jackoatmon': 'jackoatmon',
'icarus': 'icarus112',
'icarus112': 'icarus112',
}
def _normalize_owner(value: str | None) -> str | None:
if not value:
return None
normalized = value.strip().lower().lstrip('@')
if not normalized:
return None
return _OWNER_ALIASES.get(normalized, normalized)
def _owner_from_env() -> str | None:
for key in ('FEATHER_HF_OWNER', 'FEATHER_HF_NAMESPACE_OWNER', 'FEATHER_HF_PROFILE'):
owner = _normalize_owner(os.environ.get(key))
if owner:
return owner
return None
def resolve_owner(token: str | None = None) -> str:
"""Resolve active HF owner in a collaborator-safe way.
Resolution precedence:
1) explicit env owner override (FEATHER_HF_OWNER/...)
2) Hugging Face `whoami` from HF_TOKEN (unless disabled)
3) default to jackoatmon
"""
owner = _owner_from_env()
if owner:
return owner
if os.environ.get('FEATHER_HF_DISABLE_WHOAMI', '0') != '1':
active_token = token or os.environ.get('HF_TOKEN')
if active_token:
try:
info = HfApi(token=active_token).whoami(token=active_token)
if isinstance(info, dict):
whoami_owner = _normalize_owner(info.get('name'))
if whoami_owner:
return whoami_owner
except Exception:
# Fail open to deterministic defaults for offline/dry-run tests.
pass
return 'jackoatmon'
@dataclass(frozen=True)
class HfRouting:
owner: str
space_repo: str
output_repo: str
retina_cache_repo: str
job_namespace: str
def resolve_routing(token: str | None = None) -> HfRouting:
owner = resolve_owner(token=token)
space_name = os.environ.get('FEATHER_HF_SPACE_NAME', 'feather-runtime')
output_name = os.environ.get('FEATHER_HF_OUTPUT_REPO_NAME', 'feather-pretrain-checkpoints')
retina_name = os.environ.get('FEATHER_HF_RETINA_REPO_NAME', 'feather-retina-cache')
space_repo = os.environ.get('FEATHER_HF_SPACE_REPO') or f'{owner}/{space_name}'
output_repo = os.environ.get('FEATHER_HF_OUTPUT_REPO') or f'{owner}/{output_name}'
retina_cache_repo = os.environ.get('FEATHER_HF_RETINA_CACHE_REPO') or f'{owner}/{retina_name}'
job_namespace = os.environ.get('FEATHER_HF_JOB_NAMESPACE') or owner
return HfRouting(
owner=owner,
space_repo=space_repo,
output_repo=output_repo,
retina_cache_repo=retina_cache_repo,
job_namespace=job_namespace,
)
|