654 lines
26 KiB
Python
654 lines
26 KiB
Python
"""Configuration settings for AI Software Factory."""
|
|
|
|
import json
|
|
import os
|
|
from typing import Optional
|
|
from pathlib import Path
|
|
from urllib.parse import urlparse
|
|
from pydantic import Field
|
|
from pydantic_settings import BaseSettings, SettingsConfigDict
|
|
|
|
|
|
def _normalize_service_url(value: str, default_scheme: str = "https") -> str:
|
|
"""Normalize service URLs so host-only values still become valid absolute URLs."""
|
|
normalized = (value or "").strip().rstrip("/")
|
|
if not normalized:
|
|
return ""
|
|
if "://" not in normalized:
|
|
normalized = f"{default_scheme}://{normalized}"
|
|
parsed = urlparse(normalized)
|
|
if not parsed.scheme or not parsed.netloc:
|
|
return ""
|
|
return normalized
|
|
|
|
|
|
EDITABLE_LLM_PROMPTS: dict[str, dict[str, str]] = {
|
|
'LLM_GUARDRAIL_PROMPT': {
|
|
'label': 'Global Guardrails',
|
|
'category': 'guardrail',
|
|
'description': 'Applied to every outbound external LLM call.',
|
|
},
|
|
'LLM_REQUEST_INTERPRETER_GUARDRAIL_PROMPT': {
|
|
'label': 'Request Interpretation Guardrails',
|
|
'category': 'guardrail',
|
|
'description': 'Constrains project routing and continuation selection.',
|
|
},
|
|
'LLM_CHANGE_SUMMARY_GUARDRAIL_PROMPT': {
|
|
'label': 'Change Summary Guardrails',
|
|
'category': 'guardrail',
|
|
'description': 'Constrains factual delivery summaries.',
|
|
},
|
|
'LLM_PROJECT_NAMING_GUARDRAIL_PROMPT': {
|
|
'label': 'Project Naming Guardrails',
|
|
'category': 'guardrail',
|
|
'description': 'Constrains project display names and repo slugs.',
|
|
},
|
|
'LLM_PROJECT_NAMING_SYSTEM_PROMPT': {
|
|
'label': 'Project Naming System Prompt',
|
|
'category': 'system_prompt',
|
|
'description': 'Guides the dedicated new-project naming stage.',
|
|
},
|
|
'LLM_PROJECT_ID_GUARDRAIL_PROMPT': {
|
|
'label': 'Project ID Guardrails',
|
|
'category': 'guardrail',
|
|
'description': 'Constrains stable project id generation.',
|
|
},
|
|
'LLM_PROJECT_ID_SYSTEM_PROMPT': {
|
|
'label': 'Project ID System Prompt',
|
|
'category': 'system_prompt',
|
|
'description': 'Guides the dedicated project id naming stage.',
|
|
},
|
|
}
|
|
|
|
EDITABLE_RUNTIME_SETTINGS: dict[str, dict[str, str]] = {
|
|
'HOME_ASSISTANT_BATTERY_ENTITY_ID': {
|
|
'label': 'Battery Entity ID',
|
|
'category': 'home_assistant',
|
|
'description': 'Home Assistant entity used for battery state-of-charge gating.',
|
|
'value_type': 'string',
|
|
},
|
|
'HOME_ASSISTANT_SURPLUS_ENTITY_ID': {
|
|
'label': 'Surplus Power Entity ID',
|
|
'category': 'home_assistant',
|
|
'description': 'Home Assistant entity used for export or surplus power gating.',
|
|
'value_type': 'string',
|
|
},
|
|
'HOME_ASSISTANT_BATTERY_FULL_THRESHOLD': {
|
|
'label': 'Battery Full Threshold',
|
|
'category': 'home_assistant',
|
|
'description': 'Minimum battery percentage required before queued prompts may run.',
|
|
'value_type': 'float',
|
|
},
|
|
'HOME_ASSISTANT_SURPLUS_THRESHOLD_WATTS': {
|
|
'label': 'Surplus Threshold Watts',
|
|
'category': 'home_assistant',
|
|
'description': 'Minimum surplus/export power required before queued prompts may run.',
|
|
'value_type': 'float',
|
|
},
|
|
'PROMPT_QUEUE_ENABLED': {
|
|
'label': 'Queue Telegram Prompts',
|
|
'category': 'prompt_queue',
|
|
'description': 'When enabled, Telegram prompts are queued and gated instead of processed immediately.',
|
|
'value_type': 'boolean',
|
|
},
|
|
'PROMPT_QUEUE_AUTO_PROCESS': {
|
|
'label': 'Auto Process Queue',
|
|
'category': 'prompt_queue',
|
|
'description': 'Let the background worker drain the queue automatically when the gate is open.',
|
|
'value_type': 'boolean',
|
|
},
|
|
'PROMPT_QUEUE_FORCE_PROCESS': {
|
|
'label': 'Force Queue Processing',
|
|
'category': 'prompt_queue',
|
|
'description': 'Bypass the Home Assistant energy gate for queued prompts.',
|
|
'value_type': 'boolean',
|
|
},
|
|
'PROMPT_QUEUE_POLL_INTERVAL_SECONDS': {
|
|
'label': 'Queue Poll Interval Seconds',
|
|
'category': 'prompt_queue',
|
|
'description': 'Polling interval for the background queue worker.',
|
|
'value_type': 'integer',
|
|
},
|
|
'PROMPT_QUEUE_MAX_BATCH_SIZE': {
|
|
'label': 'Queue Max Batch Size',
|
|
'category': 'prompt_queue',
|
|
'description': 'Maximum number of queued prompts processed in one batch.',
|
|
'value_type': 'integer',
|
|
},
|
|
}
|
|
|
|
|
|
def _get_persisted_llm_prompt_override(env_key: str) -> str | None:
|
|
"""Load one persisted LLM prompt override from the database when available."""
|
|
if env_key not in EDITABLE_LLM_PROMPTS:
|
|
return None
|
|
try:
|
|
try:
|
|
from .database import get_db_sync
|
|
from .agents.database_manager import DatabaseManager
|
|
except ImportError:
|
|
from database import get_db_sync
|
|
from agents.database_manager import DatabaseManager
|
|
|
|
db = get_db_sync()
|
|
if db is None:
|
|
return None
|
|
try:
|
|
return DatabaseManager(db).get_llm_prompt_override(env_key)
|
|
finally:
|
|
db.close()
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _resolve_llm_prompt_value(env_key: str, fallback: str) -> str:
|
|
"""Resolve one editable prompt from DB override first, then environment/defaults."""
|
|
override = _get_persisted_llm_prompt_override(env_key)
|
|
if override is not None:
|
|
return override.strip()
|
|
return (fallback or '').strip()
|
|
|
|
|
|
def _get_persisted_runtime_setting_override(key: str):
|
|
"""Load one persisted runtime-setting override from the database when available."""
|
|
if key not in EDITABLE_RUNTIME_SETTINGS:
|
|
return None
|
|
try:
|
|
try:
|
|
from .database import get_db_sync
|
|
from .agents.database_manager import DatabaseManager
|
|
except ImportError:
|
|
from database import get_db_sync
|
|
from agents.database_manager import DatabaseManager
|
|
|
|
db = get_db_sync()
|
|
if db is None:
|
|
return None
|
|
try:
|
|
return DatabaseManager(db).get_runtime_setting_override(key)
|
|
finally:
|
|
db.close()
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _coerce_runtime_setting_value(key: str, value, fallback):
|
|
"""Coerce a persisted runtime setting override into the expected scalar type."""
|
|
value_type = EDITABLE_RUNTIME_SETTINGS.get(key, {}).get('value_type')
|
|
if value is None:
|
|
return fallback
|
|
if value_type == 'boolean':
|
|
if isinstance(value, bool):
|
|
return value
|
|
normalized = str(value).strip().lower()
|
|
if normalized in {'1', 'true', 'yes', 'on'}:
|
|
return True
|
|
if normalized in {'0', 'false', 'no', 'off'}:
|
|
return False
|
|
return bool(fallback)
|
|
if value_type == 'integer':
|
|
try:
|
|
return int(value)
|
|
except Exception:
|
|
return int(fallback)
|
|
if value_type == 'float':
|
|
try:
|
|
return float(value)
|
|
except Exception:
|
|
return float(fallback)
|
|
return str(value).strip()
|
|
|
|
|
|
def _resolve_runtime_setting_value(key: str, fallback):
|
|
"""Resolve one editable runtime setting from DB override first, then environment/defaults."""
|
|
override = _get_persisted_runtime_setting_override(key)
|
|
return _coerce_runtime_setting_value(key, override, fallback)
|
|
|
|
|
|
class Settings(BaseSettings):
|
|
"""Application settings loaded from environment variables."""
|
|
|
|
model_config = SettingsConfigDict(
|
|
env_file=".env",
|
|
env_file_encoding="utf-8",
|
|
extra="ignore",
|
|
)
|
|
|
|
# Server settings
|
|
HOST: str = "0.0.0.0"
|
|
PORT: int = 8000
|
|
LOG_LEVEL: str = "INFO"
|
|
|
|
# Ollama settings computed from environment
|
|
OLLAMA_URL: str = "http://ollama:11434"
|
|
OLLAMA_MODEL: str = "llama3"
|
|
LLM_GUARDRAIL_PROMPT: str = (
|
|
"You are operating inside AI Software Factory. Follow the requested schema exactly, "
|
|
"treat provided tool outputs as authoritative, and do not invent repositories, issues, pull requests, or delivery facts."
|
|
)
|
|
LLM_REQUEST_INTERPRETER_GUARDRAIL_PROMPT: str = (
|
|
"For routing and request interpretation: never select archived projects, prefer tracked project IDs from tool outputs, and only reference issues that are explicit in the prompt or available tool data."
|
|
)
|
|
LLM_CHANGE_SUMMARY_GUARDRAIL_PROMPT: str = (
|
|
"For summaries: only describe facts present in the provided context and tool outputs. Never claim a repository, commit, or pull request exists unless it is present in the supplied data."
|
|
)
|
|
LLM_PROJECT_NAMING_GUARDRAIL_PROMPT: str = (
|
|
"For project naming: prefer clear, product-like names and repository slugs that match the user's concrete deliverable. Avoid abstract or instructional words such as purpose, project, system, app, tool, platform, solution, new, create, or test unless the request truly centers on that exact noun. Base the name on the actual artifact or workflow being built, and avoid copying sentence fragments from the prompt. Avoid reusing tracked project identities unless the request is clearly asking for an existing project."
|
|
)
|
|
LLM_PROJECT_NAMING_SYSTEM_PROMPT: str = (
|
|
"You name newly requested software projects. Return only JSON with keys project_name, repo_name, and rationale. Project names should be concise human-readable titles based on the real product, artifact, or workflow being created. Repo names should be lowercase kebab-case slugs derived from that title. Never return generic names like purpose, project, system, app, tool, platform, solution, harness, or test by themselves, and never return a repo_name that is a copied sentence fragment from the prompt. Prefer 2 to 4 specific words when possible."
|
|
)
|
|
LLM_PROJECT_ID_GUARDRAIL_PROMPT: str = (
|
|
"For project ids: produce short stable slugs for newly created projects. Avoid collisions with known project ids and keep ids lowercase with hyphens."
|
|
)
|
|
LLM_PROJECT_ID_SYSTEM_PROMPT: str = (
|
|
"You derive stable project ids for new projects. Return only JSON with keys project_id and rationale. project_id must be a short lowercase kebab-case slug without spaces."
|
|
)
|
|
LLM_TOOL_ALLOWLIST: str = "gitea_project_catalog,gitea_project_state,gitea_project_issues,gitea_pull_requests"
|
|
LLM_TOOL_CONTEXT_LIMIT: int = 5
|
|
LLM_LIVE_TOOL_ALLOWLIST: str = "gitea_lookup_issue,gitea_lookup_pull_request"
|
|
LLM_LIVE_TOOL_STAGE_ALLOWLIST: str = "request_interpretation,change_summary"
|
|
LLM_LIVE_TOOL_STAGE_TOOL_MAP: str = ""
|
|
LLM_MAX_TOOL_CALL_ROUNDS: int = 1
|
|
|
|
# Gitea settings
|
|
GITEA_URL: str = "https://gitea.yourserver.com"
|
|
GITEA_TOKEN: str = ""
|
|
GITEA_OWNER: str = "ai-software-factory"
|
|
GITEA_REPO: str = ""
|
|
|
|
# n8n settings
|
|
N8N_WEBHOOK_URL: str = ""
|
|
N8N_API_URL: str = ""
|
|
N8N_API_KEY: str = ""
|
|
N8N_TELEGRAM_CREDENTIAL_NAME: str = "AI Software Factory Telegram"
|
|
N8N_USER: str = ""
|
|
N8N_PASSWORD: str = ""
|
|
|
|
# Runtime integration settings
|
|
BACKEND_PUBLIC_URL: str = "http://localhost:8000"
|
|
PROJECTS_ROOT: str = ""
|
|
|
|
# Telegram settings
|
|
TELEGRAM_BOT_TOKEN: str = ""
|
|
TELEGRAM_CHAT_ID: str = ""
|
|
|
|
# Home Assistant and prompt queue settings
|
|
HOME_ASSISTANT_URL: str = ""
|
|
HOME_ASSISTANT_TOKEN: str = ""
|
|
HOME_ASSISTANT_BATTERY_ENTITY_ID: str = ""
|
|
HOME_ASSISTANT_SURPLUS_ENTITY_ID: str = ""
|
|
HOME_ASSISTANT_BATTERY_FULL_THRESHOLD: float = 95.0
|
|
HOME_ASSISTANT_SURPLUS_THRESHOLD_WATTS: float = 100.0
|
|
PROMPT_QUEUE_ENABLED: bool = False
|
|
PROMPT_QUEUE_AUTO_PROCESS: bool = True
|
|
PROMPT_QUEUE_FORCE_PROCESS: bool = False
|
|
PROMPT_QUEUE_POLL_INTERVAL_SECONDS: int = 60
|
|
PROMPT_QUEUE_MAX_BATCH_SIZE: int = 1
|
|
|
|
# PostgreSQL settings
|
|
POSTGRES_HOST: str = "localhost"
|
|
POSTGRES_PORT: int = 5432
|
|
POSTGRES_USER: str = "postgres"
|
|
POSTGRES_PASSWORD: str = ""
|
|
POSTGRES_DB: str = "ai_software_factory"
|
|
POSTGRES_TEST_DB: str = "ai_software_factory_test"
|
|
POSTGRES_URL: Optional[str] = None # Optional direct PostgreSQL connection URL
|
|
|
|
# SQLite settings for testing
|
|
USE_SQLITE: bool = True # Enable SQLite by default for testing
|
|
SQLITE_DB_PATH: str = "sqlite.db"
|
|
|
|
# Database connection pool settings (only for PostgreSQL)
|
|
DB_POOL_SIZE: int = 10
|
|
DB_MAX_OVERFLOW: int = 20
|
|
DB_POOL_RECYCLE: int = 3600
|
|
DB_POOL_TIMEOUT: int = 30
|
|
|
|
@property
|
|
def postgres_url(self) -> str:
|
|
"""Get PostgreSQL URL with trimmed whitespace."""
|
|
return (self.POSTGRES_URL or "").strip()
|
|
|
|
@property
|
|
def postgres_env_configured(self) -> bool:
|
|
"""Whether PostgreSQL was explicitly configured via environment variables."""
|
|
if self.postgres_url:
|
|
return True
|
|
postgres_env_keys = (
|
|
"POSTGRES_HOST",
|
|
"POSTGRES_PORT",
|
|
"POSTGRES_USER",
|
|
"POSTGRES_PASSWORD",
|
|
"POSTGRES_DB",
|
|
)
|
|
return any(bool(os.environ.get(key, "").strip()) for key in postgres_env_keys)
|
|
|
|
@property
|
|
def use_sqlite(self) -> bool:
|
|
"""Whether SQLite should be used as the active database backend."""
|
|
if not self.USE_SQLITE:
|
|
return False
|
|
return not self.postgres_env_configured
|
|
|
|
@property
|
|
def pool(self) -> dict:
|
|
"""Get database pool configuration."""
|
|
return {
|
|
"pool_size": self.DB_POOL_SIZE,
|
|
"max_overflow": self.DB_MAX_OVERFLOW,
|
|
"pool_recycle": self.DB_POOL_RECYCLE,
|
|
"pool_timeout": self.DB_POOL_TIMEOUT
|
|
}
|
|
|
|
@property
|
|
def database_url(self) -> str:
|
|
"""Get database connection URL."""
|
|
if self.use_sqlite:
|
|
return f"sqlite:///{self.SQLITE_DB_PATH}"
|
|
if self.postgres_url:
|
|
return self.postgres_url
|
|
return (
|
|
f"postgresql://{self.POSTGRES_USER}:{self.POSTGRES_PASSWORD}"
|
|
f"@{self.POSTGRES_HOST}:{self.POSTGRES_PORT}/{self.POSTGRES_DB}"
|
|
)
|
|
|
|
@property
|
|
def test_database_url(self) -> str:
|
|
"""Get test database connection URL."""
|
|
if self.use_sqlite:
|
|
return f"sqlite:///{self.SQLITE_DB_PATH}"
|
|
if self.postgres_url:
|
|
return self.postgres_url
|
|
return (
|
|
f"postgresql://{self.POSTGRES_USER}:{self.POSTGRES_PASSWORD}"
|
|
f"@{self.POSTGRES_HOST}:{self.POSTGRES_PORT}/{self.POSTGRES_TEST_DB}"
|
|
)
|
|
|
|
@property
|
|
def ollama_url(self) -> str:
|
|
"""Get Ollama URL with trimmed whitespace."""
|
|
return self.OLLAMA_URL.strip()
|
|
|
|
@property
|
|
def llm_guardrail_prompt(self) -> str:
|
|
"""Get the global guardrail prompt used for all external LLM calls."""
|
|
return _resolve_llm_prompt_value('LLM_GUARDRAIL_PROMPT', self.LLM_GUARDRAIL_PROMPT)
|
|
|
|
@property
|
|
def llm_request_interpreter_guardrail_prompt(self) -> str:
|
|
"""Get the request-interpretation specific guardrail prompt."""
|
|
return _resolve_llm_prompt_value('LLM_REQUEST_INTERPRETER_GUARDRAIL_PROMPT', self.LLM_REQUEST_INTERPRETER_GUARDRAIL_PROMPT)
|
|
|
|
@property
|
|
def llm_change_summary_guardrail_prompt(self) -> str:
|
|
"""Get the change-summary specific guardrail prompt."""
|
|
return _resolve_llm_prompt_value('LLM_CHANGE_SUMMARY_GUARDRAIL_PROMPT', self.LLM_CHANGE_SUMMARY_GUARDRAIL_PROMPT)
|
|
|
|
@property
|
|
def llm_project_naming_guardrail_prompt(self) -> str:
|
|
"""Get the project-naming specific guardrail prompt."""
|
|
return _resolve_llm_prompt_value('LLM_PROJECT_NAMING_GUARDRAIL_PROMPT', self.LLM_PROJECT_NAMING_GUARDRAIL_PROMPT)
|
|
|
|
@property
|
|
def llm_project_naming_system_prompt(self) -> str:
|
|
"""Get the project-naming system prompt."""
|
|
return _resolve_llm_prompt_value('LLM_PROJECT_NAMING_SYSTEM_PROMPT', self.LLM_PROJECT_NAMING_SYSTEM_PROMPT)
|
|
|
|
@property
|
|
def llm_project_id_guardrail_prompt(self) -> str:
|
|
"""Get the project-id naming specific guardrail prompt."""
|
|
return _resolve_llm_prompt_value('LLM_PROJECT_ID_GUARDRAIL_PROMPT', self.LLM_PROJECT_ID_GUARDRAIL_PROMPT)
|
|
|
|
@property
|
|
def llm_project_id_system_prompt(self) -> str:
|
|
"""Get the project-id naming system prompt."""
|
|
return _resolve_llm_prompt_value('LLM_PROJECT_ID_SYSTEM_PROMPT', self.LLM_PROJECT_ID_SYSTEM_PROMPT)
|
|
|
|
@property
|
|
def editable_llm_prompts(self) -> list[dict[str, str]]:
|
|
"""Return metadata for all LLM prompts that may be persisted and edited from the UI."""
|
|
prompts = []
|
|
for env_key, metadata in EDITABLE_LLM_PROMPTS.items():
|
|
prompts.append(
|
|
{
|
|
'key': env_key,
|
|
'label': metadata['label'],
|
|
'category': metadata['category'],
|
|
'description': metadata['description'],
|
|
'default_value': (getattr(self, env_key, '') or '').strip(),
|
|
'value': _resolve_llm_prompt_value(env_key, getattr(self, env_key, '')),
|
|
}
|
|
)
|
|
return prompts
|
|
|
|
@property
|
|
def editable_runtime_settings(self) -> list[dict]:
|
|
"""Return metadata for all DB-editable runtime settings."""
|
|
items = []
|
|
for key, metadata in EDITABLE_RUNTIME_SETTINGS.items():
|
|
default_value = getattr(self, key)
|
|
value = _resolve_runtime_setting_value(key, default_value)
|
|
items.append(
|
|
{
|
|
'key': key,
|
|
'label': metadata['label'],
|
|
'category': metadata['category'],
|
|
'description': metadata['description'],
|
|
'value_type': metadata['value_type'],
|
|
'default_value': default_value,
|
|
'value': value,
|
|
}
|
|
)
|
|
return items
|
|
|
|
@property
|
|
def llm_tool_allowlist(self) -> list[str]:
|
|
"""Get the allowed LLM tool names as a normalized list."""
|
|
return [item.strip() for item in self.LLM_TOOL_ALLOWLIST.split(',') if item.strip()]
|
|
|
|
@property
|
|
def llm_tool_context_limit(self) -> int:
|
|
"""Get the number of items to expose per mediated tool payload."""
|
|
return max(int(self.LLM_TOOL_CONTEXT_LIMIT), 1)
|
|
|
|
@property
|
|
def llm_live_tool_allowlist(self) -> list[str]:
|
|
"""Get the allowed live tool-call names for model-driven lookup requests."""
|
|
return [item.strip() for item in self.LLM_LIVE_TOOL_ALLOWLIST.split(',') if item.strip()]
|
|
|
|
@property
|
|
def llm_live_tool_stage_allowlist(self) -> list[str]:
|
|
"""Get the LLM stages where live tool requests are enabled."""
|
|
return [item.strip() for item in self.LLM_LIVE_TOOL_STAGE_ALLOWLIST.split(',') if item.strip()]
|
|
|
|
@property
|
|
def llm_live_tool_stage_tool_map(self) -> dict[str, list[str]]:
|
|
"""Get an optional per-stage live tool map that overrides the simple stage allowlist."""
|
|
raw = (self.LLM_LIVE_TOOL_STAGE_TOOL_MAP or '').strip()
|
|
if not raw:
|
|
return {}
|
|
try:
|
|
parsed = json.loads(raw)
|
|
except Exception:
|
|
return {}
|
|
if not isinstance(parsed, dict):
|
|
return {}
|
|
allowed_tools = set(self.llm_live_tool_allowlist)
|
|
normalized: dict[str, list[str]] = {}
|
|
for stage, tools in parsed.items():
|
|
if not isinstance(stage, str):
|
|
continue
|
|
if not isinstance(tools, list):
|
|
continue
|
|
normalized[stage.strip()] = [str(tool).strip() for tool in tools if str(tool).strip() in allowed_tools]
|
|
return normalized
|
|
|
|
def llm_live_tools_for_stage(self, stage: str) -> list[str]:
|
|
"""Return live tools enabled for a specific LLM stage."""
|
|
stage_map = self.llm_live_tool_stage_tool_map
|
|
if stage_map:
|
|
return stage_map.get(stage, [])
|
|
if stage not in set(self.llm_live_tool_stage_allowlist):
|
|
return []
|
|
return self.llm_live_tool_allowlist
|
|
|
|
@property
|
|
def llm_max_tool_call_rounds(self) -> int:
|
|
"""Get the maximum number of model-driven live tool-call rounds per LLM request."""
|
|
return max(int(self.LLM_MAX_TOOL_CALL_ROUNDS), 0)
|
|
|
|
@property
|
|
def gitea_url(self) -> str:
|
|
"""Get Gitea URL with trimmed whitespace."""
|
|
return _normalize_service_url(self.GITEA_URL)
|
|
|
|
@property
|
|
def gitea_token(self) -> str:
|
|
"""Get Gitea token with trimmed whitespace."""
|
|
return self.GITEA_TOKEN.strip()
|
|
|
|
@property
|
|
def gitea_owner(self) -> str:
|
|
"""Get Gitea owner/organization with trimmed whitespace."""
|
|
return self.GITEA_OWNER.strip()
|
|
|
|
@property
|
|
def gitea_repo(self) -> str:
|
|
"""Get the optional fixed Gitea repository name with trimmed whitespace."""
|
|
return self.GITEA_REPO.strip()
|
|
|
|
@property
|
|
def use_project_repositories(self) -> bool:
|
|
"""Whether the service should create one repository per generated project."""
|
|
return not bool(self.gitea_repo)
|
|
|
|
@property
|
|
def n8n_webhook_url(self) -> str:
|
|
"""Get n8n webhook URL with trimmed whitespace."""
|
|
return _normalize_service_url(self.N8N_WEBHOOK_URL, default_scheme="http")
|
|
|
|
@property
|
|
def n8n_api_url(self) -> str:
|
|
"""Get n8n API URL with trimmed whitespace."""
|
|
return _normalize_service_url(self.N8N_API_URL, default_scheme="http")
|
|
|
|
@property
|
|
def n8n_api_key(self) -> str:
|
|
"""Get n8n API key with trimmed whitespace."""
|
|
return self.N8N_API_KEY.strip()
|
|
|
|
@property
|
|
def n8n_telegram_credential_name(self) -> str:
|
|
"""Get the preferred n8n Telegram credential name."""
|
|
return self.N8N_TELEGRAM_CREDENTIAL_NAME.strip() or "AI Software Factory Telegram"
|
|
|
|
@property
|
|
def telegram_bot_token(self) -> str:
|
|
"""Get Telegram bot token with trimmed whitespace."""
|
|
return self.TELEGRAM_BOT_TOKEN.strip()
|
|
|
|
@property
|
|
def telegram_chat_id(self) -> str:
|
|
"""Get Telegram chat ID with trimmed whitespace."""
|
|
return self.TELEGRAM_CHAT_ID.strip()
|
|
|
|
@property
|
|
def backend_public_url(self) -> str:
|
|
"""Get backend public URL with trimmed whitespace."""
|
|
return _normalize_service_url(self.BACKEND_PUBLIC_URL, default_scheme="http")
|
|
|
|
@property
|
|
def home_assistant_url(self) -> str:
|
|
"""Get Home Assistant URL with trimmed whitespace."""
|
|
return _normalize_service_url(self.HOME_ASSISTANT_URL, default_scheme="http")
|
|
|
|
@property
|
|
def home_assistant_token(self) -> str:
|
|
"""Get Home Assistant token with trimmed whitespace."""
|
|
return self.HOME_ASSISTANT_TOKEN.strip()
|
|
|
|
@property
|
|
def home_assistant_battery_entity_id(self) -> str:
|
|
"""Get the Home Assistant battery state entity id."""
|
|
return str(_resolve_runtime_setting_value('HOME_ASSISTANT_BATTERY_ENTITY_ID', self.HOME_ASSISTANT_BATTERY_ENTITY_ID)).strip()
|
|
|
|
@property
|
|
def home_assistant_surplus_entity_id(self) -> str:
|
|
"""Get the Home Assistant surplus power entity id."""
|
|
return str(_resolve_runtime_setting_value('HOME_ASSISTANT_SURPLUS_ENTITY_ID', self.HOME_ASSISTANT_SURPLUS_ENTITY_ID)).strip()
|
|
|
|
@property
|
|
def home_assistant_battery_full_threshold(self) -> float:
|
|
"""Get the minimum battery SoC percentage for queue processing."""
|
|
return float(_resolve_runtime_setting_value('HOME_ASSISTANT_BATTERY_FULL_THRESHOLD', self.HOME_ASSISTANT_BATTERY_FULL_THRESHOLD))
|
|
|
|
@property
|
|
def home_assistant_surplus_threshold_watts(self) -> float:
|
|
"""Get the minimum export/surplus power threshold for queue processing."""
|
|
return float(_resolve_runtime_setting_value('HOME_ASSISTANT_SURPLUS_THRESHOLD_WATTS', self.HOME_ASSISTANT_SURPLUS_THRESHOLD_WATTS))
|
|
|
|
@property
|
|
def prompt_queue_enabled(self) -> bool:
|
|
"""Whether Telegram prompts should be queued instead of processed immediately."""
|
|
return bool(_resolve_runtime_setting_value('PROMPT_QUEUE_ENABLED', self.PROMPT_QUEUE_ENABLED))
|
|
|
|
@property
|
|
def prompt_queue_auto_process(self) -> bool:
|
|
"""Whether the background worker should automatically process queued prompts."""
|
|
return bool(_resolve_runtime_setting_value('PROMPT_QUEUE_AUTO_PROCESS', self.PROMPT_QUEUE_AUTO_PROCESS))
|
|
|
|
@property
|
|
def prompt_queue_force_process(self) -> bool:
|
|
"""Whether queued prompts should bypass the Home Assistant energy gate."""
|
|
return bool(_resolve_runtime_setting_value('PROMPT_QUEUE_FORCE_PROCESS', self.PROMPT_QUEUE_FORCE_PROCESS))
|
|
|
|
@property
|
|
def prompt_queue_poll_interval_seconds(self) -> int:
|
|
"""Get the queue polling interval for background processing."""
|
|
return max(int(_resolve_runtime_setting_value('PROMPT_QUEUE_POLL_INTERVAL_SECONDS', self.PROMPT_QUEUE_POLL_INTERVAL_SECONDS)), 5)
|
|
|
|
@property
|
|
def prompt_queue_max_batch_size(self) -> int:
|
|
"""Get the maximum number of queued prompts to process in one batch."""
|
|
return max(int(_resolve_runtime_setting_value('PROMPT_QUEUE_MAX_BATCH_SIZE', self.PROMPT_QUEUE_MAX_BATCH_SIZE)), 1)
|
|
|
|
@property
|
|
def projects_root(self) -> Path:
|
|
"""Get the root directory for generated project artifacts."""
|
|
if self.PROJECTS_ROOT.strip():
|
|
return Path(self.PROJECTS_ROOT).expanduser().resolve()
|
|
return Path(__file__).resolve().parent.parent / "test-project"
|
|
|
|
@property
|
|
def postgres_host(self) -> str:
|
|
"""Get PostgreSQL host."""
|
|
return self.POSTGRES_HOST.strip()
|
|
|
|
@property
|
|
def postgres_port(self) -> int:
|
|
"""Get PostgreSQL port as integer."""
|
|
return int(self.POSTGRES_PORT)
|
|
|
|
@property
|
|
def postgres_user(self) -> str:
|
|
"""Get PostgreSQL user."""
|
|
return self.POSTGRES_USER.strip()
|
|
|
|
@property
|
|
def postgres_password(self) -> str:
|
|
"""Get PostgreSQL password."""
|
|
return self.POSTGRES_PASSWORD.strip()
|
|
|
|
@property
|
|
def postgres_db(self) -> str:
|
|
"""Get PostgreSQL database name."""
|
|
return self.POSTGRES_DB.strip()
|
|
|
|
@property
|
|
def postgres_test_db(self) -> str:
|
|
"""Get test PostgreSQL database name."""
|
|
return self.POSTGRES_TEST_DB.strip()
|
|
|
|
# Create instance for module-level access
|
|
settings = Settings() |