You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
125 lines
4.1 KiB
Python
125 lines
4.1 KiB
Python
import argparse
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
|
if str(PROJECT_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(PROJECT_ROOT))
|
|
|
|
from app.services.integrations.events import SUPPORTED_EVENT_TYPES
|
|
from app.services.integrations.service import upsert_email_integration_route
|
|
|
|
|
|
def _deep_merge_dicts(base: dict[str, Any], extra: dict[str, Any]) -> dict[str, Any]:
|
|
merged = dict(base)
|
|
for key, value in extra.items():
|
|
current = merged.get(key)
|
|
if isinstance(current, dict) and isinstance(value, dict):
|
|
merged[key] = _deep_merge_dicts(current, value)
|
|
else:
|
|
merged[key] = value
|
|
return merged
|
|
|
|
|
|
def _parse_provider_config_json(raw_value: str | None) -> dict[str, Any]:
|
|
if not raw_value:
|
|
return {}
|
|
try:
|
|
parsed = json.loads(raw_value)
|
|
except ValueError as exc:
|
|
raise SystemExit(f"provider_config_json invalido: {exc}") from exc
|
|
if not isinstance(parsed, dict):
|
|
raise SystemExit("provider_config_json deve ser um objeto JSON.")
|
|
return parsed
|
|
|
|
|
|
def _parse_headers(values: list[str] | None) -> dict[str, str]:
|
|
headers: dict[str, str] = {}
|
|
for value in values or []:
|
|
key, separator, header_value = str(value or "").partition("=")
|
|
key = key.strip()
|
|
header_value = header_value.strip()
|
|
if not separator or not key or not header_value:
|
|
raise SystemExit(f"header invalido: {value}. Use o formato Chave=Valor.")
|
|
headers[key] = header_value
|
|
return headers
|
|
|
|
|
|
def _build_provider_config(args) -> dict[str, Any] | None:
|
|
provider_config = _parse_provider_config_json(args.provider_config_json)
|
|
convenience_config: dict[str, Any] = {}
|
|
|
|
if args.sender_email or args.sender_name:
|
|
convenience_config["sender"] = {
|
|
key: value
|
|
for key, value in {
|
|
"email": args.sender_email,
|
|
"name": args.sender_name,
|
|
}.items()
|
|
if value
|
|
}
|
|
|
|
if args.reply_to_email or args.reply_to_name:
|
|
convenience_config["reply_to"] = {
|
|
key: value
|
|
for key, value in {
|
|
"email": args.reply_to_email,
|
|
"name": args.reply_to_name,
|
|
}.items()
|
|
if value
|
|
}
|
|
|
|
if args.cc:
|
|
convenience_config["cc"] = args.cc
|
|
if args.bcc:
|
|
convenience_config["bcc"] = args.bcc
|
|
if args.tag:
|
|
convenience_config["tags"] = args.tag
|
|
if args.html_content:
|
|
convenience_config["html_content"] = args.html_content
|
|
|
|
headers = _parse_headers(args.header)
|
|
if headers:
|
|
convenience_config["headers"] = headers
|
|
|
|
merged = _deep_merge_dicts(provider_config, convenience_config)
|
|
return merged or None
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="Cria ou atualiza uma rota de integracao por email.")
|
|
parser.add_argument("--event", required=True, choices=SUPPORTED_EVENT_TYPES)
|
|
parser.add_argument("--recipient", required=True)
|
|
parser.add_argument("--name")
|
|
parser.add_argument("--subject-template")
|
|
parser.add_argument("--body-template")
|
|
parser.add_argument("--provider-config-json")
|
|
parser.add_argument("--sender-email")
|
|
parser.add_argument("--sender-name")
|
|
parser.add_argument("--reply-to-email")
|
|
parser.add_argument("--reply-to-name")
|
|
parser.add_argument("--cc", action="append")
|
|
parser.add_argument("--bcc", action="append")
|
|
parser.add_argument("--tag", action="append")
|
|
parser.add_argument("--header", action="append")
|
|
parser.add_argument("--html-content")
|
|
parser.add_argument("--disabled", action="store_true")
|
|
args = parser.parse_args()
|
|
|
|
route = upsert_email_integration_route(
|
|
event_type=args.event,
|
|
recipient_email=args.recipient,
|
|
recipient_name=args.name,
|
|
subject_template=args.subject_template,
|
|
body_template=args.body_template,
|
|
enabled=not args.disabled,
|
|
provider_config=_build_provider_config(args),
|
|
)
|
|
print(json.dumps(route, ensure_ascii=True, indent=2, sort_keys=True))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|