93 lines
4.2 KiB
Python
93 lines
4.2 KiB
Python
import os
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import List, Optional
|
|
from pydantic import BaseModel, Field
|
|
from dotenv import load_dotenv
|
|
|
|
BASE_DIR = Path(__file__).resolve().parent
|
|
ENV_FILE = BASE_DIR / ".env"
|
|
|
|
if ENV_FILE.exists():
|
|
load_dotenv(ENV_FILE, override=True)
|
|
else:
|
|
load_dotenv(override=True)
|
|
|
|
class Settings(BaseModel):
|
|
# Telegram Bot Settings
|
|
telegram_bot_token: str = Field(default_factory=lambda: os.getenv("TELEGRAM_BOT_TOKEN", "").strip())
|
|
allowed_user_ids: List[int] = Field(default_factory=lambda: [
|
|
int(uid.strip()) for uid in os.getenv("ALLOWED_USER_IDS", "").split(",") if uid.strip().isdigit()
|
|
])
|
|
admin_user_ids: List[int] = Field(default_factory=lambda: [
|
|
int(uid.strip()) for uid in os.getenv("ADMIN_USER_IDS", "").split(",") if uid.strip().isdigit()
|
|
])
|
|
auth_password: Optional[str] = Field(default_factory=lambda: os.getenv("BOT_AUTH_PASSWORD", "").strip() or None)
|
|
|
|
# AGY Agent Settings
|
|
default_model: str = Field(default_factory=lambda: os.getenv("DEFAULT_MODEL", "gemini-3.7-flash-auto"))
|
|
default_effort: str = Field(default_factory=lambda: os.getenv("DEFAULT_EFFORT", "high"))
|
|
default_workspace: str = Field(default_factory=lambda: os.getenv("DEFAULT_WORKSPACE", "/root"))
|
|
default_language: str = Field(default_factory=lambda: os.getenv("DEFAULT_LANGUAGE", "fa"))
|
|
|
|
# Performance & Streaming tuning
|
|
stream_edit_interval: float = Field(default=1.2) # Rate-limit friendly streaming edits
|
|
max_message_length: int = Field(default=4000)
|
|
enable_thinking_display: bool = Field(default=True)
|
|
enable_tool_notifications: bool = Field(default=True)
|
|
enable_context_length_message: bool = Field(default_factory=lambda: os.getenv("ENABLE_CONTEXT_LENGTH_MESSAGE", "true").lower() in ("true", "1", "yes"))
|
|
|
|
# Gitea / Git Settings
|
|
gitea_url: str = Field(default_factory=lambda: os.getenv("GITEA_URL", "https://git.msa.artacloud.ir").rstrip("/"))
|
|
gitea_internal_url: str = Field(default_factory=lambda: os.getenv("GITEA_INTERNAL_URL", "http://127.0.0.1:3000").rstrip("/"))
|
|
gitea_admin_user: str = Field(default_factory=lambda: os.getenv("GITEA_ADMIN_USER", "root"))
|
|
gitea_admin_token: str = Field(default_factory=lambda: os.getenv("GITEA_ADMIN_TOKEN", "04b1dfa936c22bfd6ec11a6ddf412f4094c334c9"))
|
|
gitea_auto_commit: bool = Field(default_factory=lambda: os.getenv("GITEA_AUTO_COMMIT", "true").lower() in ("true", "1", "yes"))
|
|
gitea_auto_sync: bool = Field(default_factory=lambda: os.getenv("GITEA_AUTO_SYNC", "true").lower() in ("true", "1", "yes"))
|
|
|
|
# Logging
|
|
log_level: str = Field(default_factory=lambda: os.getenv("LOG_LEVEL", "INFO"))
|
|
|
|
def is_user_authorized(self, user_id: int) -> bool:
|
|
if self.is_admin(user_id):
|
|
return True
|
|
if self.allowed_user_ids:
|
|
return user_id in self.allowed_user_ids
|
|
# If no whitelist is set and no auth password, only admins or authorized
|
|
return False
|
|
|
|
def is_admin(self, user_id: int) -> bool:
|
|
if not self.admin_user_ids:
|
|
return False
|
|
return user_id in self.admin_user_ids
|
|
|
|
def add_authorized_user(self, user_id: int):
|
|
if user_id not in self.allowed_user_ids:
|
|
self.allowed_user_ids.append(user_id)
|
|
self._persist_allowed_users()
|
|
|
|
def remove_authorized_user(self, user_id: int) -> bool:
|
|
if user_id in self.allowed_user_ids:
|
|
self.allowed_user_ids.remove(user_id)
|
|
self._persist_allowed_users()
|
|
return True
|
|
return False
|
|
|
|
def _persist_allowed_users(self):
|
|
try:
|
|
ids_str = ",".join(str(uid) for uid in self.allowed_user_ids)
|
|
lines = []
|
|
if ENV_FILE.exists():
|
|
with open(ENV_FILE, "r", encoding="utf-8") as f:
|
|
for line in f:
|
|
if line.startswith("ALLOWED_USER_IDS="):
|
|
continue
|
|
lines.append(line)
|
|
lines.append(f"ALLOWED_USER_IDS={ids_str}\n")
|
|
with open(ENV_FILE, "w", encoding="utf-8") as f:
|
|
f.writelines(lines)
|
|
except Exception as e:
|
|
logging.error(f"Failed to persist allowed users: {e}")
|
|
|
|
settings = Settings()
|