8614062ecd
report.status_translation in YAML config overrides/extends the builtin
STATUS_TRANSLATION dictionary; without the section behavior is unchanged.
- AppConfig.report_status_translation + _resolve_str_dict (warns and
skips non-scalar values, resolves ${VAR} references)
- Config.get_status_translation() returns merged copy (lazy import,
builtin dict never mutated)
- build_grouped_report() accepts optional status_translation parameter
- --init-config template gains commented example, docs/CONFIG.md updated
Closes #65
535 lines
19 KiB
Python
535 lines
19 KiB
Python
import logging
|
||
import os
|
||
import sys
|
||
from dataclasses import dataclass, field
|
||
from datetime import date, timedelta
|
||
from pathlib import Path
|
||
from typing import Dict, Union
|
||
|
||
import yaml
|
||
from dotenv import load_dotenv
|
||
|
||
from .yaml_config import check_file_permissions, resolve_env_vars
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
DEFAULT_REDMINE_VERIFY: Union[bool, str] = True
|
||
FALSE_VALUES = {"0", "false", "no", "off"}
|
||
TRUE_VALUES = {"1", "true", "yes", "on"}
|
||
|
||
|
||
@dataclass
|
||
class SmtpConfig:
|
||
host: str = ""
|
||
port: int = 587
|
||
user: str = ""
|
||
password: str = ""
|
||
tls: bool = True
|
||
|
||
|
||
@dataclass
|
||
class EmailConfig:
|
||
smtp: SmtpConfig = field(default_factory=SmtpConfig)
|
||
from_: str = ""
|
||
to: list = field(default_factory=list)
|
||
cc: list = field(default_factory=list)
|
||
bcc: list = field(default_factory=list)
|
||
subject: str = "Отчёт {author} за {period}"
|
||
body_text: str = "Во вложении отчёт."
|
||
attach: bool = True
|
||
html: bool = False
|
||
|
||
|
||
@dataclass
|
||
class PeriodLastUsed:
|
||
from_: str = ""
|
||
to: str = ""
|
||
|
||
|
||
@dataclass
|
||
class AppConfig:
|
||
redmine_url: str = ""
|
||
redmine_api_key: str = ""
|
||
redmine_author: str = ""
|
||
redmine_verify: Union[bool, str] = DEFAULT_REDMINE_VERIFY
|
||
period_precision: str = "date"
|
||
period_default_from: str = ""
|
||
period_default_to: str = ""
|
||
period_dynamic: bool = False
|
||
period_last_used_from: str = ""
|
||
period_last_used_to: str = ""
|
||
output_dir: str = ""
|
||
output_filename: str = "{author}_{from}_{to}.{ext}"
|
||
output_default_format: str = "xlsx"
|
||
report_no_time: bool = False
|
||
report_status_translation: Dict[str, str] = field(default_factory=dict)
|
||
email: EmailConfig = field(default_factory=EmailConfig)
|
||
|
||
@classmethod
|
||
def from_yaml(cls, path: Union[str, Path]) -> "AppConfig":
|
||
"""Загружает настройки из YAML-файла.
|
||
|
||
Если файл не существует — возвращает конфиг со значениями по умолчанию.
|
||
Неизвестные ключи верхнего уровня логируются с warning.
|
||
"""
|
||
path = Path(path)
|
||
if not path.exists():
|
||
return cls()
|
||
|
||
# Проверяем права файла
|
||
for warning in check_file_permissions(path):
|
||
logger.warning(warning)
|
||
|
||
with open(path, "r", encoding="utf-8") as fh:
|
||
raw = yaml.safe_load(fh) or {}
|
||
|
||
# Предупреждаем о неизвестных ключах верхнего уровня
|
||
known_keys = {
|
||
"redmine",
|
||
"period",
|
||
"output",
|
||
"report",
|
||
"email",
|
||
}
|
||
for key in raw:
|
||
if key not in known_keys:
|
||
logger.warning("Unknown top-level key in config: %s", key)
|
||
|
||
return cls(
|
||
redmine_url=cls._resolve_str(raw, "redmine", "url"),
|
||
redmine_api_key=cls._resolve_str(raw, "redmine", "api_key"),
|
||
redmine_author=cls._resolve_str(raw, "redmine", "author"),
|
||
redmine_verify=cls._resolve_verify(raw),
|
||
period_precision=cls._resolve_str(raw, "period", "precision") or "date",
|
||
period_default_from=cls._resolve_str(raw, "period", "default_from"),
|
||
period_default_to=cls._resolve_str(raw, "period", "default_to"),
|
||
period_dynamic=cls._resolve_bool(raw, "period", "dynamic"),
|
||
period_last_used_from=cls._resolve_str(raw, "period", "last_used", "from"),
|
||
period_last_used_to=cls._resolve_str(raw, "period", "last_used", "to"),
|
||
output_dir=cls._resolve_str(raw, "output", "dir"),
|
||
output_filename=cls._resolve_str(raw, "output", "filename")
|
||
or "{author}_{from}_{to}.{ext}",
|
||
output_default_format=cls._resolve_str(raw, "output", "default_format")
|
||
or "xlsx",
|
||
report_no_time=cls._resolve_bool(raw, "report", "no_time"),
|
||
report_status_translation=cls._resolve_str_dict(
|
||
raw, "report", "status_translation"
|
||
),
|
||
email=cls._resolve_email(raw),
|
||
)
|
||
|
||
@classmethod
|
||
def from_env(cls) -> "AppConfig":
|
||
"""Загружает настройки из переменных окружения."""
|
||
return cls(
|
||
redmine_url=os.getenv("REDMINE_URL", "").strip().rstrip("/"),
|
||
redmine_api_key=os.getenv("REDMINE_API_KEY", "").strip(),
|
||
redmine_author=os.getenv("REDMINE_AUTHOR", "").strip(),
|
||
period_default_from=os.getenv("DEFAULT_FROM_DATE", "").strip(),
|
||
period_default_to=os.getenv("DEFAULT_TO_DATE", "").strip(),
|
||
)
|
||
|
||
@staticmethod
|
||
def _resolve_str(raw: dict, section: str, *keys: str) -> str:
|
||
value = raw.get(section)
|
||
for key in keys:
|
||
if isinstance(value, dict):
|
||
value = value.get(key)
|
||
else:
|
||
value = None
|
||
break
|
||
if isinstance(value, str):
|
||
return resolve_env_vars(value)
|
||
if value is None:
|
||
return ""
|
||
return str(value)
|
||
|
||
@staticmethod
|
||
def _resolve_str_dict(raw: dict, section: str, key: str) -> Dict[str, str]:
|
||
"""Читает вложенный mapping строк из YAML-секции.
|
||
|
||
Не-mapping значение логируется с warning и игнорируется.
|
||
Ключи и значения приводятся к str, ${VAR} в значениях резолвится.
|
||
Записи со значением null или не-scalar (dict/list) пропускаются
|
||
с warning — иначе null стал бы строкой "None", а dict — repr.
|
||
"""
|
||
section_value = raw.get(section)
|
||
if not isinstance(section_value, dict):
|
||
return {}
|
||
value = section_value.get(key)
|
||
if value is None:
|
||
return {}
|
||
if not isinstance(value, dict):
|
||
logger.warning(
|
||
"Config %s.%s must be a mapping, got %s — ignoring",
|
||
section,
|
||
key,
|
||
type(value).__name__,
|
||
)
|
||
return {}
|
||
result: Dict[str, str] = {}
|
||
for k, v in value.items():
|
||
if v is None or isinstance(v, (dict, list)):
|
||
logger.warning(
|
||
"Config %s.%s entry %r must be a scalar, got %s — skipping",
|
||
section,
|
||
key,
|
||
k,
|
||
type(v).__name__,
|
||
)
|
||
continue
|
||
result[str(k)] = resolve_env_vars(str(v))
|
||
return result
|
||
|
||
@staticmethod
|
||
def _resolve_bool(raw: dict, section: str, key: str) -> bool:
|
||
value = raw.get(section, {}).get(key)
|
||
if isinstance(value, bool):
|
||
return value
|
||
if isinstance(value, str):
|
||
normalized = value.lower()
|
||
if normalized in TRUE_VALUES:
|
||
return True
|
||
if normalized in FALSE_VALUES:
|
||
return False
|
||
return False
|
||
|
||
@classmethod
|
||
def _resolve_verify(cls, raw: dict) -> Union[bool, str]:
|
||
"""Семантика verify_ssl, единая с REDMINE_VERIFY из env:
|
||
|
||
- true (bool или строка) → True — стандартная проверка TLS (requests);
|
||
- false (bool или строка) → False — проверка отключена;
|
||
- любая другая строка → путь к CA-bundle как есть.
|
||
"""
|
||
value = raw.get("redmine", {}).get("verify_ssl")
|
||
if value is None:
|
||
return DEFAULT_REDMINE_VERIFY
|
||
if isinstance(value, bool):
|
||
return value
|
||
if isinstance(value, str):
|
||
normalized = value.lower()
|
||
if normalized in FALSE_VALUES:
|
||
return False
|
||
if normalized in TRUE_VALUES:
|
||
return True
|
||
return resolve_env_vars(value)
|
||
return DEFAULT_REDMINE_VERIFY
|
||
|
||
@classmethod
|
||
def _resolve_email(cls, raw: dict) -> EmailConfig:
|
||
email_raw = raw.get("email", {}) or {}
|
||
smtp_raw = email_raw.get("smtp", {}) or {}
|
||
|
||
return EmailConfig(
|
||
smtp=SmtpConfig(
|
||
host=cls._safe_str(smtp_raw.get("host")),
|
||
port=cls._safe_int(smtp_raw.get("port"), 587),
|
||
user=cls._safe_str(smtp_raw.get("user")),
|
||
password=resolve_env_vars(cls._safe_str(smtp_raw.get("password"))),
|
||
tls=cls._safe_bool(smtp_raw.get("tls"), True),
|
||
),
|
||
from_=cls._safe_str(email_raw.get("from")),
|
||
to=cls._safe_list(email_raw.get("to")),
|
||
cc=cls._safe_list(email_raw.get("cc")),
|
||
bcc=cls._safe_list(email_raw.get("bcc")),
|
||
subject=resolve_env_vars(
|
||
cls._safe_str(email_raw.get("subject")) or "Отчёт {author} за {period}"
|
||
),
|
||
body_text=resolve_env_vars(
|
||
cls._safe_str(email_raw.get("body_text")) or "Во вложении отчёт."
|
||
),
|
||
attach=cls._safe_bool(email_raw.get("attach"), True),
|
||
html=cls._safe_bool(email_raw.get("html"), False),
|
||
)
|
||
|
||
@staticmethod
|
||
def _safe_str(value) -> str:
|
||
if value is None:
|
||
return ""
|
||
if isinstance(value, str):
|
||
return value
|
||
return str(value)
|
||
|
||
@staticmethod
|
||
def _safe_int(value, default: int = 0) -> int:
|
||
if value is None:
|
||
return default
|
||
try:
|
||
return int(value)
|
||
except (TypeError, ValueError):
|
||
return default
|
||
|
||
@staticmethod
|
||
def _safe_bool(value, default: bool = False) -> bool:
|
||
if isinstance(value, bool):
|
||
return value
|
||
if isinstance(value, str):
|
||
normalized = value.lower()
|
||
if normalized in TRUE_VALUES:
|
||
return True
|
||
if normalized in FALSE_VALUES:
|
||
return False
|
||
return default
|
||
|
||
@staticmethod
|
||
def _safe_list(value) -> list:
|
||
if value is None:
|
||
return []
|
||
if isinstance(value, list):
|
||
return [str(v) for v in value]
|
||
return [str(value)]
|
||
|
||
|
||
class Config:
|
||
_cli_url: str | None = None
|
||
_cli_api_key: str | None = None
|
||
_app: AppConfig | None = None
|
||
|
||
@classmethod
|
||
def load_yaml(cls, path: str) -> None:
|
||
"""Загружает конфигурацию приложения.
|
||
|
||
Сначала подгружает .env из текущей директории (override=False:
|
||
переменные окружения не перебиваются), затем читает YAML-конфиг.
|
||
Не бросает исключений при отсутствии файла.
|
||
"""
|
||
load_dotenv(override=False)
|
||
cls._app = AppConfig.from_yaml(path)
|
||
|
||
@classmethod
|
||
def load_config(cls, path: str) -> None:
|
||
"""Загружает переменные из указанного .env-файла с override."""
|
||
load_dotenv(path, override=True)
|
||
|
||
@classmethod
|
||
def set_redmine_url(cls, url: str) -> None:
|
||
cls._cli_url = url.strip().rstrip("/") if url else None
|
||
|
||
@classmethod
|
||
def set_redmine_api_key(cls, key: str) -> None:
|
||
cls._cli_api_key = key.strip() if key else None
|
||
|
||
@classmethod
|
||
def get_redmine_url(cls) -> str:
|
||
if cls._cli_url is not None:
|
||
return cls._cli_url
|
||
env = os.getenv("REDMINE_URL", "").strip().rstrip("/")
|
||
if env:
|
||
return env
|
||
if cls._app:
|
||
return cls._app.redmine_url.strip().rstrip("/")
|
||
return ""
|
||
|
||
@classmethod
|
||
def get_redmine_api_key(cls) -> str:
|
||
if cls._cli_api_key is not None:
|
||
return cls._cli_api_key
|
||
env = os.getenv("REDMINE_API_KEY", "").strip()
|
||
if env:
|
||
return env
|
||
if cls._app:
|
||
return cls._app.redmine_api_key
|
||
return ""
|
||
|
||
@classmethod
|
||
def get_redmine_user(cls) -> str:
|
||
return os.getenv("REDMINE_USER", "").strip()
|
||
|
||
@classmethod
|
||
def get_redmine_password(cls) -> str:
|
||
return os.getenv("REDMINE_PASSWORD", "").strip()
|
||
|
||
@classmethod
|
||
def get_redmine_verify(cls) -> Union[bool, str]:
|
||
value = os.getenv("REDMINE_VERIFY", "").strip()
|
||
if value:
|
||
normalized = value.lower()
|
||
if normalized in FALSE_VALUES:
|
||
return False
|
||
if normalized in TRUE_VALUES:
|
||
return True
|
||
return value
|
||
if cls._app:
|
||
return cls._app.redmine_verify
|
||
return DEFAULT_REDMINE_VERIFY
|
||
|
||
@classmethod
|
||
def get_author(cls, cli_author: str = "") -> str:
|
||
if cli_author:
|
||
return cli_author
|
||
env = os.getenv("REDMINE_AUTHOR", "").strip()
|
||
if env:
|
||
return env
|
||
if cls._app:
|
||
return cls._app.redmine_author
|
||
return ""
|
||
|
||
@classmethod
|
||
def get_period_precision(cls) -> str:
|
||
if cls._app:
|
||
return cls._app.period_precision or "date"
|
||
return "date"
|
||
|
||
@classmethod
|
||
def get_last_used_from(cls) -> str:
|
||
if cls._app:
|
||
return cls._app.period_last_used_from
|
||
return ""
|
||
|
||
@classmethod
|
||
def get_last_used_to(cls) -> str:
|
||
if cls._app:
|
||
return cls._app.period_last_used_to
|
||
return ""
|
||
|
||
@classmethod
|
||
def get_output_dir(cls) -> str:
|
||
if cls._app:
|
||
return cls._app.output_dir
|
||
return ""
|
||
|
||
@classmethod
|
||
def get_output_filename(cls) -> str:
|
||
if cls._app:
|
||
return cls._app.output_filename or "{author}_{from}_{to}.{ext}"
|
||
return "{author}_{from}_{to}.{ext}"
|
||
|
||
@classmethod
|
||
def get_output_default_format(cls) -> str:
|
||
if cls._app:
|
||
return cls._app.output_default_format or "xlsx"
|
||
return "xlsx"
|
||
|
||
@classmethod
|
||
def get_report_no_time(cls) -> bool:
|
||
"""Возвращает report.no_time из YAML-конфига (по умолчанию False)."""
|
||
if cls._app:
|
||
return cls._app.report_no_time
|
||
return False
|
||
|
||
@classmethod
|
||
def get_status_translation(cls) -> Dict[str, str]:
|
||
"""Полный словарь перевода статусов: встроенный + переопределения из YAML.
|
||
|
||
Возвращает новый словарь — модульный STATUS_TRANSLATION не мутируется.
|
||
Импорт ленивый: config.py не должен тянуть redminelib при импорте.
|
||
"""
|
||
from .report_builder import STATUS_TRANSLATION
|
||
|
||
overrides = cls._app.report_status_translation if cls._app else {}
|
||
return {**STATUS_TRANSLATION, **overrides}
|
||
|
||
@classmethod
|
||
def get_email_config(cls) -> "EmailConfig | None":
|
||
"""Возвращает EmailConfig из YAML-конфига или None, если не настроен."""
|
||
if cls._app is None:
|
||
return None
|
||
email = cls._app.email
|
||
if not email.smtp.host:
|
||
return None
|
||
return email
|
||
|
||
@classmethod
|
||
def get_default_date_range(cls) -> str:
|
||
from_env = os.getenv("DEFAULT_FROM_DATE", "").strip()
|
||
to_env = os.getenv("DEFAULT_TO_DATE", "").strip()
|
||
today = date.today()
|
||
today_str = today.isoformat()
|
||
|
||
if from_env:
|
||
return f"{from_env}--{to_env or today_str}"
|
||
|
||
if (
|
||
cls._app
|
||
and cls._app.period_dynamic
|
||
and cls._app.period_last_used_from
|
||
and cls._app.period_last_used_to
|
||
):
|
||
nf, nt = compute_next_period(
|
||
cls._app.period_last_used_from,
|
||
cls._app.period_last_used_to,
|
||
cls._app.period_precision or "date",
|
||
)
|
||
return f"{nf}--{nt}"
|
||
|
||
if cls._app and cls._app.period_default_from:
|
||
default_to = cls._app.period_default_to or today_str
|
||
return f"{cls._app.period_default_from}--{default_to}"
|
||
|
||
start = today.replace(day=1)
|
||
return f"{start.isoformat()}--{today_str}"
|
||
|
||
@classmethod
|
||
def validate(cls) -> None:
|
||
url = cls.get_redmine_url()
|
||
if not url:
|
||
raise ValueError("REDMINE_URL is required (set via env or .env)")
|
||
if not url.lower().startswith("https://"):
|
||
raise ValueError(
|
||
"REDMINE_URL must use HTTPS: the API key is sent in request "
|
||
"headers and requires TLS"
|
||
)
|
||
if cls.get_redmine_verify() is False:
|
||
print(
|
||
"⚠️ TLS certificate verification is disabled "
|
||
"(REDMINE_VERIFY=false / verify_ssl: false): connection is "
|
||
"vulnerable to MITM attacks",
|
||
file=sys.stderr,
|
||
)
|
||
if cls.get_redmine_api_key():
|
||
return
|
||
if not (cls.get_redmine_user() and cls.get_redmine_password()):
|
||
raise ValueError(
|
||
"REDMINE_API_KEY is required, or set both REDMINE_USER and REDMINE_PASSWORD"
|
||
)
|
||
|
||
|
||
def compute_next_period(
|
||
last_from: str, last_to: str, precision: str
|
||
) -> tuple[str, str]:
|
||
"""Compute the next report period based on the last committed period.
|
||
|
||
- For a full calendar month → next full calendar month.
|
||
- For an arbitrary range → same duration, starting the day after last_to.
|
||
- For datetime precision → same duration, starting 1 second after last_to.
|
||
"""
|
||
from datetime import datetime as dt_mod
|
||
|
||
if precision == "datetime":
|
||
fmt = "%Y-%m-%dT%H:%M:%S"
|
||
from_dt = dt_mod.fromisoformat(last_from.replace("Z", "+00:00"))
|
||
to_dt = dt_mod.fromisoformat(last_to.replace("Z", "+00:00"))
|
||
duration = to_dt - from_dt
|
||
next_from_dt = to_dt + timedelta(seconds=1)
|
||
next_to_dt = next_from_dt + duration
|
||
return next_from_dt.strftime(fmt), next_to_dt.strftime(fmt)
|
||
|
||
from_parts = last_from.split("-")
|
||
to_parts = last_to.split("-")
|
||
from_d = date(int(from_parts[0]), int(from_parts[1]), int(from_parts[2]))
|
||
to_d = date(int(to_parts[0]), int(to_parts[1]), int(to_parts[2]))
|
||
|
||
first_of_month = from_d.replace(day=1)
|
||
if from_d == first_of_month:
|
||
if to_d.month == 12:
|
||
last_of_month = date(to_d.year, 12, 31)
|
||
else:
|
||
next_first = date(to_d.year, to_d.month + 1, 1)
|
||
last_of_month = next_first - timedelta(days=1)
|
||
if to_d == last_of_month:
|
||
if to_d.month == 12:
|
||
nstart = date(to_d.year + 1, 1, 1)
|
||
else:
|
||
nstart = date(to_d.year, to_d.month + 1, 1)
|
||
if nstart.month == 12:
|
||
nend = date(nstart.year, 12, 31)
|
||
else:
|
||
nend = date(nstart.year, nstart.month + 1, 1) - timedelta(days=1)
|
||
return nstart.isoformat(), nend.isoformat()
|
||
|
||
duration = to_d - from_d
|
||
next_from = to_d + timedelta(days=1)
|
||
next_to = next_from + duration
|
||
return next_from.isoformat(), next_to.isoformat()
|