131 lines
4.7 KiB
Python
131 lines
4.7 KiB
Python
import os
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from urllib.parse import quote
|
|
|
|
|
|
def _bool_env(name: str, default: bool) -> bool:
|
|
value = os.getenv(name)
|
|
if value is None:
|
|
return default
|
|
return value.strip().lower() in {"1", "true", "yes", "on"}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Settings:
|
|
data_dir: Path
|
|
legacy_boundaries_file: Path
|
|
admin_api_key: str
|
|
client_api_key: str
|
|
public_base_url: str
|
|
max_upload_bytes: int
|
|
max_attempt_bytes: int
|
|
keep_attempt_audio: bool
|
|
moss_transcribe_url: str
|
|
moss_model: str
|
|
moss_timeout_seconds: float
|
|
moss_max_new_tokens: int
|
|
moss_end_padding_ms: int
|
|
pass_score: float
|
|
volcano_oss_access_key: str
|
|
volcano_oss_secret_key: str
|
|
volcano_oss_bucket: str
|
|
volcano_oss_endpoint: str
|
|
volcano_oss_region: str
|
|
volcano_oss_key_prefix: str
|
|
volcano_oss_public_base_url: str
|
|
|
|
@classmethod
|
|
def from_env(cls) -> "Settings":
|
|
package_dir = Path(__file__).resolve().parent
|
|
data_dir = Path(os.getenv("ORAL_TRAINER_DATA_DIR", str(package_dir / "data")))
|
|
return cls(
|
|
data_dir=data_dir,
|
|
legacy_boundaries_file=Path(
|
|
os.getenv(
|
|
"SENTENCE_BOUNDARIES_FILE",
|
|
str(package_dir / "data" / "sentence_boundaries.json"),
|
|
)
|
|
),
|
|
admin_api_key=os.getenv("ADMIN_API_KEY", ""),
|
|
client_api_key=os.getenv("CLIENT_API_KEY", ""),
|
|
public_base_url=os.getenv("PUBLIC_BASE_URL", "").rstrip("/"),
|
|
max_upload_bytes=int(os.getenv("MAX_VIDEO_UPLOAD_BYTES", str(12 * 1024**3))),
|
|
max_attempt_bytes=int(os.getenv("MAX_ATTEMPT_UPLOAD_BYTES", str(50 * 1024**2))),
|
|
keep_attempt_audio=_bool_env("KEEP_ATTEMPT_AUDIO", False),
|
|
moss_transcribe_url=os.getenv("MOSS_TRANSCRIBE_URL", "").strip(),
|
|
moss_model=os.getenv(
|
|
"MOSS_MODEL",
|
|
"OpenMOSS-Team/MOSS-Transcribe-Diarize",
|
|
),
|
|
moss_timeout_seconds=float(os.getenv("MOSS_TIMEOUT_SECONDS", "1800")),
|
|
moss_max_new_tokens=int(os.getenv("MOSS_MAX_NEW_TOKENS", "65536")),
|
|
moss_end_padding_ms=int(os.getenv("MOSS_END_PADDING_MS", "300")),
|
|
pass_score=float(os.getenv("ASSESSMENT_PASS_SCORE", "70")),
|
|
volcano_oss_access_key=os.getenv("VOLCANO_OSS_ACCESS_KEY", "").strip(),
|
|
volcano_oss_secret_key=os.getenv("VOLCANO_OSS_SECRET_KEY", "").strip(),
|
|
volcano_oss_bucket=os.getenv("VOLCANO_OSS_BUCKET", "").strip(),
|
|
volcano_oss_endpoint=os.getenv(
|
|
"VOLCANO_OSS_ENDPOINT", "tos-cn-beijing.volces.com"
|
|
).strip().strip("/"),
|
|
volcano_oss_region=os.getenv("VOLCANO_OSS_REGION", "cn-beijing").strip(),
|
|
volcano_oss_key_prefix=Settings._normalize_key_prefix(
|
|
os.getenv("VOLCANO_OSS_KEY_PREFIX", "videos/")
|
|
),
|
|
volcano_oss_public_base_url=os.getenv(
|
|
"VOLCANO_OSS_PUBLIC_BASE_URL", ""
|
|
).strip().rstrip("/"),
|
|
)
|
|
|
|
@property
|
|
def videos_dir(self) -> Path:
|
|
return self.data_dir / "v"
|
|
|
|
@property
|
|
def work_dir(self) -> Path:
|
|
return self.data_dir / "work"
|
|
|
|
@property
|
|
def attempts_dir(self) -> Path:
|
|
return self.data_dir / "attempts"
|
|
|
|
@property
|
|
def dub_shares_dir(self) -> Path:
|
|
return self.data_dir / "dub_shares"
|
|
|
|
@property
|
|
def database_path(self) -> Path:
|
|
return self.data_dir / "oral_trainer.sqlite3"
|
|
|
|
@property
|
|
def volcano_oss_enabled(self) -> bool:
|
|
return all(
|
|
(
|
|
self.volcano_oss_access_key,
|
|
self.volcano_oss_secret_key,
|
|
self.volcano_oss_bucket,
|
|
self.volcano_oss_endpoint,
|
|
self.volcano_oss_region,
|
|
)
|
|
)
|
|
|
|
def volcano_oss_public_url(self, object_key: str) -> str:
|
|
quoted_key = quote(object_key)
|
|
if self.volcano_oss_public_base_url:
|
|
return f"{self.volcano_oss_public_base_url}/{quoted_key}"
|
|
endpoint = self.volcano_oss_endpoint
|
|
if "://" in endpoint:
|
|
scheme, host = endpoint.split("://", 1)
|
|
else:
|
|
scheme, host = "https", endpoint
|
|
return f"{scheme}://{self.volcano_oss_bucket}.{host}/{quoted_key}"
|
|
|
|
@staticmethod
|
|
def _normalize_key_prefix(value: str) -> str:
|
|
normalized = value.strip().strip("/")
|
|
return f"{normalized}/" if normalized else ""
|
|
|
|
def ensure_directories(self) -> None:
|
|
for path in (self.data_dir, self.videos_dir, self.work_dir, self.attempts_dir, self.dub_shares_dir):
|
|
path.mkdir(parents=True, exist_ok=True)
|