File size: 3,520 Bytes
f440f03
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
105
106
107
108
109
110
111
112
113
114
115
116
"""Vides mainīgo helperi Maris runtime noklusējumiem."""

from __future__ import annotations

import os
import re

MARIS_MODEL_NAMESPACE = os.getenv("MARIS_MODEL_NAMESPACE", "MarisUK/")
HF_REPO_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*/[A-Za-z0-9][A-Za-z0-9._-]*$")


def get_env(name: str, default: str | None = None) -> str | None:
    value = os.getenv(name)
    if value is None:
        return default
    normalized = value.strip()
    if normalized:
        return normalized
    return default


def require_env(name: str) -> str:
    value = get_env(name)
    if value is None:
        raise RuntimeError(f"Trūkst obligātais vides mainīgais: {name}")
    return value


def get_env_or_default(name: str, default: str) -> str:
    return get_env(name, default) or default


def get_env_any(*names: str, default: str | None = None) -> str | None:
    for name in names:
        value = get_env(name)
        if value is not None:
            return value
    return default


def get_env_any_or_default(*names: str, default: str) -> str:
    return get_env_any(*names, default=default) or default


def get_hf_token() -> str | None:
    return get_env_any(
        "HF_TOKEN",
        "HUGGING_FACE_HUB_TOKEN",
        "HUGGINGFACEHUB_API_TOKEN",
        "HUGGINGFACE_API_KEY",
        "MARIS_REPO_TOKEN",
        "MARIS_TOKEN",
    )


def validate_maris_repo(repo_id: str, source: str, *, label: str = "repozitorijs") -> str:
    normalized = validate_hf_repo_id(repo_id, source, label=label)
    if not normalized.startswith(MARIS_MODEL_NAMESPACE):
        raise RuntimeError(f"{source} jānorāda uz Maris AI {label} ({MARIS_MODEL_NAMESPACE}...).")
    return normalized


def validate_hf_repo_id(repo_id: str, source: str, *, label: str = "repozitorijs") -> str:
    normalized = repo_id.strip()
    if not normalized:
        raise RuntimeError(f"Trūkst Maris AI {label} konfigurācija: {source}")
    if not HF_REPO_ID_PATTERN.fullmatch(normalized):
        raise RuntimeError(f"{source} jābūt owner/name formātā.")
    return normalized


def validate_maris_model(model_id: str, source: str) -> str:
    return validate_maris_repo(model_id, source, label="modeli")


def validate_hf_model(model_id: str, source: str) -> str:
    return validate_hf_repo_id(model_id, source, label="modeli")


def get_maris_model(*names: str, default: str | None = None) -> str:
    source = ", ".join(names)
    value = get_env_any(*names)
    if value is None:
        if default is None:
            raise RuntimeError(f"Trūkst Maris AI modeļa konfigurācija: {source}")
        value = default
    if value is None:
        raise RuntimeError(f"Trūkst Maris AI modeļa konfigurācija: {source}")
    return validate_maris_model(value, source)


def get_optional_maris_model(*names: str) -> str | None:
    source = ", ".join(names)
    value = get_env_any(*names)
    if value is None:
        return None
    return validate_maris_model(value, source)


def get_hf_model(*names: str, default: str | None = None) -> str:
    source = ", ".join(names)
    value = get_env_any(*names)
    if value is None:
        if default is None:
            raise RuntimeError(f"Trūkst Hugging Face modeļa konfigurācija: {source}")
        value = default
    return validate_hf_model(value, source)


def get_optional_hf_model(*names: str) -> str | None:
    source = ", ".join(names)
    value = get_env_any(*names)
    if value is None:
        return None
    return validate_hf_model(value, source)