init

.vscode/settings.json

@@ -0,0 +1,16 @@
+{
+    "python.languageServer": "None",
+    "cSpell.words": [
+        "traefik"
+    ],
+    "editor.formatOnSave": true,
+    "[python]": {
+        "editor.codeActionsOnSave": {
+            "source.organizeImports": "explicit",
+            "source.unusedImports": "explicit"
+        }
+    },
+    "terminal.integrated.env.linux": {
+        "PYTHONPATH": "${workspaceFolder}/src"
+    },
+}

src/compose.py

@@ -0,0 +1,250 @@
+from collections.abc import Collection, Iterator
+from dataclasses import dataclass
+from functools import reduce
+from pathlib import Path
+from shutil import copyfile
+from typing import Literal, NotRequired, Protocol, cast, final
+
+
+from Ts import (
+    CfgData,
+    Compose,
+    ComposeNet,
+    ComposeNetArgs,
+    HasServices,
+    OrgData,
+    TraefikComposeDict,
+    TraefikNet,
+    TraefikNetName,
+)
+from util import merge_dicts, read_yml, to_yaml
+from val_objs import RecordCls
+
+# CFG_ROOT = Path("/data/cfg")
+
+
+# def replace(rec: RecordCls, string: str) -> str:
+#     return str.replace(string, rec.name, rec.val.to_str())
+
+
+# @final
+# @dataclass(frozen=True, slots=True)
+# class ComposeBuild:
+#     root_path: Path
+#     # def __init__(self, path: str) -> None:
+#     # self.env_path = self.path.joinpath(".env")
+
+
+# def compose_build_factory(root_path: str):
+#     path = Path("/data/cfg").joinpath(root_path)
+#     return ComposeBuild(path)
+# CFG_PATH = Path("/data/cfg")
+
+
+def compose_treafik(proxy_data: Iterator[tuple[str, str]]):
+    root_path = CFG_PATH.joinpath("treafik")
+    cfg_data = CfgData(
+        name="treafik",
+        files=["treafik.yml"],
+        orgs=[OrgData(org="util", url="treafik")],
+    )
+    services = build_compose(root_path, cfg_data)["services"]
+    networks: TraefikNet = dict()
+    for name, proxy in proxy_data:
+        networks[name] = TraefikNetName(name=proxy)
+    traefik_compose = TraefikComposeDict(
+        name="traefik",
+        services=services,
+        networks=networks,
+    )
+    mk_volumes = tuple(get_volumes(traefik_compose))
+    og_yaml = to_yaml(traefik_compose)
+    org = cfg_data["orgs"][0]
+
+
+# def build_compose(compose_rw: ComposeRWData, cfg_data: CfgData) -> Compose:
+#     compose_dict = get_compose_dict(compose_rw, cfg_data)
+#     insert_defaults(compose_dict)
+#     insert_traefik_labels(compose_dict)
+#     return compose_dict
+
+
+# def write_compose(write_data: ComposeData):
+#     write(write_data)
+#     mk_compose_dir(write_data)
+#     mk_compose_env(write_data.rw)
+
+
+# def gen_compose(path: str) -> Iterator[tuple[str, str]]:
+#     # compose_build = compose_build_factory(path)
+#     root_path = CFG_PATH.joinpath(path)
+#     cfg_data = get_config_data(root_path)
+#     compose_dict = build_compose(root_path, cfg_data)
+#     mk_volumes = tuple(get_volumes(compose_dict))
+#     proxy_net = set_networks(compose_dict)
+#     og_yaml = to_yaml(compose_dict)
+
+#     for _dict in cfg_data["orgs"]:
+#         compose_rw = compose_rw_factory(
+#             _dict.get("org"), cfg_data["name"], _dict["url"]
+#         )
+
+#         write(compose_rw, og_yaml)
+#         mk_compose_dir(compose_rw, mk_volumes)
+#         mk_compose_env(root_path, compose_rw.data_dir.to_path())
+#         if proxy_net is None:
+#             continue
+#         for net in proxy_net:
+
+#             def sub():
+#                 for kv in net:
+#                     yield replace(compose_rw.org_name, kv)
+
+#             yield tuple[str, str](sub())
+
+
+# def get_config_data(cfg_dir: Path) -> CfgData:
+#     cfg_path = cfg_dir.joinpath("cfg.yml")
+#     return cast(CfgData, read_yml(cfg_path))
+
+
+# def get_compose_dict(compose_rw: ComposeRWData, cfg_data: CfgData) -> Compose:
+#     def sub():
+#         for file in cfg_data["files"]:
+#             path = compose_rw.data_dir.joinpath(file)
+#             yield cast(Compose, read_yml(path))
+
+#     return reduce(merge_dicts, sub(), cast(Compose, {}))  # pyright: ignore[reportInvalidCast]
+
+
+# @final
+# class TraefikBuild:
+
+
+# def insert_defaults(data: Compose) -> None:
+#     data["name"] = "${_ORG_NAME}"
+#     for app_data in data["services"].values():
+#         app_data["restart"] = "unless-stopped"
+#         sec_opts = {"no-new-privileges:true"}
+#         app_data["security_opt"] = list(
+#             sec_opts.union(app_data.get("security_opt", set()))
+#         )
+
+
+# def insert_traefik_labels(data: Compose) -> None:
+#     for app_data in data["services"].values():
+#         traefik_labels = {
+#             "traefik.http.routers.${_ORG_NAME}.rule=Host(`${_URL}`)",
+#             "traefik.http.routers.${_ORG_NAME}.entrypoints=websecure",
+#             "traefik.docker.network=${_ORG_NAME}_proxy",
+#             "traefik.http.routers.${_ORG_NAME}.tls.certresolver=le",
+#         }
+#         if "labels" not in app_data:
+#             continue
+#         if "traefik.enable=true" not in app_data["labels"]:
+#             continue
+#         app_data["labels"] = list(traefik_labels.union(app_data["labels"]))
+
+
+# def get_volumes(data: HasServices) -> Iterator[str]:
+#     for app_data in data["services"].values():
+#         if "volumes" not in app_data:
+#             return
+#         for vol in app_data["volumes"]:
+#             if not vol.startswith(r"${_DATA}"):
+#                 continue
+#             yield vol.split(":", 1)[0]
+
+
+# def set_networks(data: Compose) -> tuple[str, str] | None:
+#     def sub() -> Iterator[Literal["proxy", "internal"]]:
+#         for app_data in data["services"].values():
+#             if "networks" not in app_data:
+#                 continue
+#             yield from app_data["networks"]
+
+#     nets = set(sub())
+#     net = ComposeNet()
+#     if "proxy" in nets:
+#         proxy = ComposeNetArgs(name="${_ORG_NAME}_proxy", external=True)
+#         net["proxy"] = proxy
+#         ret = "${_ORG_NAME}", "${_ORG_NAME}_proxy"
+#     else:
+#         ret = None
+#     if "internal" in nets:
+#         net["internal"] = ComposeNetArgs(name="${_ORG_NAME}")
+#     if not net:
+#         return
+#     data["networks"] = net
+#     return ret
+
+
+# def compose_rw_factory(org_raw: str | None, name_raw: str, sub_url: str):
+#     _org = OrgVal(org_raw)
+#     _name = NameVal(name_raw)
+#     # org = Record("ORG", _org)
+#     # name = Record("NAME", _name)
+#     org_name = (
+#         NameVal(f"{_org.to_str()}_{_name.to_str()}") if _org.is_valid() else _name
+#     )
+#     # org_name = Record("ORG_NAME", org_name)
+#     data_dir = Path("/data").joinpath(_org.to_str(), _name.to_str())
+#     # cfg_dir = CFG_PATH.joinpath(org_name.val.to_str())
+#     # data = Record("DATA", DataDir(data_dir))
+#     # url = Record("URL", Url(sub_url))
+#     # Path("/data").joinpath(_org.to_str(), _name.to_str())
+#     # compose_path = data_dir.to_path().joinpath("docker-compose.yml")
+#     # env_path = data_dir.to_path().joinpath(".env")
+#     return ComposeRWData(
+#         Record("ORG", _org),
+#         Record("NAME", _name),
+#         Record("ORG_NAME", org_name),
+#         Record("DATA", DataDir(data_dir)),
+#         Record("URL", Url(sub_url)),
+#         data_dir,
+#         CFG_PATH.joinpath(org_name.to_str()),
+#     )
+
+# def __init__(self, org: str | None, name: str, sub_url: str) -> None:
+#     super().__init__()
+#     self._org = OrgVal(org)
+#     self._name = NameVal(name)
+#     self.org = Record("ORG", self._org)
+#     self.name = Record("NAME", self._name)
+#     org_name = (
+#         NameVal(f"{self.org.val.to_str()}_{self.name.val.to_str()}")
+#         if self._org.is_valid()
+#         else self._name
+#     )
+#     self.org_name = Record("ORG_NAME", org_name)
+#     self.data_dir = DataDir(
+#         self._org,
+#         self._name,
+#     )
+#     self.data = Record("DATA", self.data_dir)
+#     self.url = Record("URL", Url(sub_url))
+
+# self.compose_path = self.data_dir.to_path().joinpath("docker-compose.yml")
+# self.env_path = self.data_dir.to_path().joinpath(".env")
+
+
+# def mk_compose_dir(compose_data: ComposeData) -> None:
+#     for vol in compose_data.volumes:
+#         mk_path = Path(replace(compose_data.rw.data, vol))
+#         if mk_path.exists():
+#             continue
+#         mk_path.mkdir(parents=True)
+
+# def mk_compose_env(compose_data: ComposeRWData) -> None:
+#     root_env_path = compose_data.cfg_dir.joinpath(".env")
+#     dest_env_path = compose_data.data_dir.joinpath(".env")
+#     if root_env_path.exists() and not dest_env_path.exists():
+#         _ = copyfile(root_env_path, dest_env_path)
+
+# def write(compose_data: ComposeData) -> None:
+#     string = reduce(lambda s, rec: replace(rec, s), compose_data.rw, compose_data.yaml)
+#     compose_path = compose_data.rw.data_dir.joinpath("docker-compose.yml")
+#     if not compose_path.exists():
+#         compose_path.parent.mkdir(parents=True)
+#     with compose_path.open("wt") as f:
+#         _ = f.write(string)

src/compose/compose_struct.py

@@ -0,0 +1,163 @@
+from collections.abc import Mapping
+from dataclasses import dataclass, asdict
+from typing import Literal, NotRequired, Self, TypedDict, final
+
+from util import to_yaml
+
+type nested_list = list[str | nested_list]
+type T_Primitive = bool | int | str
+type T_PrimVal = T_Primitive | list[T_Primitive] | T_PrimDict
+type T_PrimDict = Mapping[T_Primitive, T_PrimVal]
+type T_YamlVals = T_Primitive | list[T_Primitive | T_YamlDict] | T_YamlDict
+type T_YamlDict = dict[str, T_YamlVals]
+
+
+class ComposeNetArgsYaml(TypedDict):
+    name: str
+    external: NotRequired[bool]
+
+
+@final
+@dataclass(frozen=True, slots=True)
+class ComposeNetArgs:
+    name: str
+    external: bool | None
+
+    @classmethod
+    def from_dict(cls, data: ComposeNetArgsYaml) -> Self:
+        return cls(
+            data["name"],
+            data.get("external"),
+        )
+
+
+class ComposeNetYaml(TypedDict):
+    internal: NotRequired[ComposeNetArgsYaml]
+    proxy: NotRequired[ComposeNetArgsYaml]
+
+
+@final
+@dataclass(frozen=True, slots=True)
+class ComposeNet:
+    internal: ComposeNetArgs | None
+    proxy: ComposeNetArgs | None
+
+    @classmethod
+    def from_class(cls, data: ComposeNetYaml) -> Self:
+        internal = data.get("internal")
+        if internal is not None:
+            internal = ComposeNetArgs.from_dict(internal)
+        proxy = data.get("proxy")
+        if proxy is not None:
+            proxy = ComposeNetArgs.from_dict(proxy)
+        return cls(internal, proxy)
+
+
+class ComposeServiceYaml(TypedDict):
+    command: NotRequired[list[str]]
+    container_name: str
+    entrypoint: list[str]
+    environment: NotRequired[dict[str, T_Primitive]]
+    image: str
+    labels: NotRequired[list[str]]
+    logging: dict[str, str]
+    networks: NotRequired[list[Literal["proxy", "internal"]]]
+    restart: str
+    security_opt: NotRequired[list[str]]
+    user: NotRequired[str]
+    volumes: NotRequired[list[str]]
+
+
+@final
+@dataclass(slots=True)
+class ComposeService:
+    command: list[str] | None
+    container_name: str | None
+    entrypoint: list[str]
+    environment: dict[str, T_Primitive] | None
+    image: str
+    labels: set[str] | None
+    logging: dict[str, str] | None
+    networks: set[Literal["proxy", "internal"]] | None
+    restart: str | None
+    security_opt: set[str] | None
+    user: str | None
+    volumes: set[str] | None
+
+    @classmethod
+    def from_dict(cls, data: ComposeServiceYaml) -> Self:
+        labels = data.get("labels")
+        if labels is not None:
+            labels = set(labels)
+        sec = data.get("security_opt")
+        if sec is not None:
+            sec = set(sec)
+        vols = data.get("volumes")
+        if vols is not None:
+            vols = set(vols)
+        return cls(
+            data.get("command"),
+            None,  # data['container_name'],
+            data["entrypoint"],
+            data.get("environment"),
+            data["image"],
+            labels,
+            None,
+            data.get("netwoks"),
+            None,
+            sec,
+            data.get("user"),
+            vols,
+        )
+
+    def is_valid(self) -> bool:
+        attrs = (self.container_name, self.restart, self.logging)
+        for attr in attrs:
+            if attr is None:
+                return False
+        return True
+
+
+class ComposeYaml(TypedDict):
+    name: str
+    networks: NotRequired[ComposeNetYaml]
+    services: dict[str, ComposeServiceYaml]
+
+
+@final
+@dataclass(slots=True)
+class Compose:
+    name: str
+    networks: ComposeNet | None
+    services: dict[str, ComposeService]
+
+    @classmethod
+    def from_dict(cls, data: ComposeYaml) -> Self:
+        services = dict[str, ComposeService]()
+        for k, v in data["services"].items():
+            services[k] = ComposeService.from_dict(v)
+        return cls(
+            data["name"],
+            None,
+            services,
+        )
+
+    def as_yaml(self) -> str:
+        return to_yaml(asdict(self))
+
+
+class TraefikNetName(TypedDict):
+    name: str
+
+
+type TraefikNet = dict[str, TraefikNetName]
+
+
+class TraefikComposeDict(TypedDict):
+    name: str
+    networks: TraefikNet
+    services: dict[str, ComposeServiceYaml]
+
+
+class HasServices(TypedDict):
+    services: dict[str, ComposeServiceYaml]

src/compose/entities.py

@@ -0,0 +1,265 @@
+from collections.abc import Iterator
+
+from .compose_struct import Compose, ComposeService, ComposeServiceYaml, ComposeYaml
+from dataclasses import dataclass
+from functools import reduce
+from pathlib import Path
+from typing import Literal, NotRequired, Self, TypedDict, cast, final
+
+from util import merge_dicts, read_yml, to_yaml
+
+from .compose_struct import ComposeNet, ComposeNetArgs, HasServices
+from .val_objs import DataDir, NameVal, OrgVal, Record, RecordCls, RecordVal, Url
+
+
+def replace(rec: RecordCls[RecordVal], string: str) -> str:
+    return str.replace(string, rec.name, rec.val.to_str())
+
+
+@final
+@dataclass(frozen=True, slots=True)
+class SrcPaths:
+    data_dir: Path
+    cfg_file: Path
+    env_file: Path
+
+
+def src_path_factory(src: str) -> SrcPaths:
+    root = Path("/data/cfg")
+    dir = root.joinpath(src)
+    return SrcPaths(
+        data_dir=dir,
+        cfg_file=dir.joinpath("cfg.yml"),
+        env_file=dir.joinpath(".env"),
+    )
+
+
+class OrgDataYaml(TypedDict):
+    org: NotRequired[str]
+    url: NotRequired[str]
+
+
+class CfgDataYaml(TypedDict):
+    name: str
+    files: list[str]
+    orgs: list[OrgDataYaml]
+
+
+@final
+@dataclass(frozen=True, slots=True)
+class OrgData:
+    org: str | None
+    url: str | None
+
+
+@final
+@dataclass(frozen=True, slots=True)
+class CfgData:
+    name: str
+    files: tuple[Path, ...]
+    orgs: tuple[OrgData, ...]
+
+
+def config_data_factory(cfg_dir: SrcPaths) -> CfgData:
+    data = cast(CfgDataYaml, read_yml(cfg_dir.cfg_file))
+
+    def sub_files():
+        for path in data["files"]:
+            yield cfg_dir.data_dir.joinpath(path)
+
+    def sub_orgs():
+        for org_data in data["orgs"]:
+            yield OrgData(
+                org_data.get("org"),
+                org_data.get("url"),
+            )
+
+    return CfgData(
+        data["name"],
+        tuple(sub_files()),
+        tuple(sub_orgs()),
+    )
+
+
+@final
+@dataclass(frozen=True, slots=True)
+class ComposeTemplate:
+    org: RecordCls[OrgVal]
+    name: RecordCls[NameVal]
+    org_name: RecordCls[NameVal]
+    data: RecordCls[DataDir]
+    url: RecordCls[Url]
+
+    def __iter__(self):
+        yield self.org
+        yield self.name
+        yield self.org_name
+        yield self.data
+        yield self.url
+
+
+def compose_template_factory(cfg_data: CfgData) -> Iterator[ComposeTemplate]:
+    # def sub():
+    for org_data in cfg_data.orgs:
+        _org = OrgVal(org_data.org)
+        _name = NameVal(cfg_data.name)
+        org_name = (
+            NameVal(f"{_org.to_str()}_{_name.to_str()}") if _org.is_valid() else _name
+        )
+        data_dir = Path("/data").joinpath(_org.to_str(), _name.to_str())
+
+        yield ComposeTemplate(
+            Record("ORG", _org),
+            Record("NAME", _name),
+            Record("ORG_NAME", org_name),
+            Record("DATA", DataDir(data_dir)),
+            Record("URL", Url(org_data.url)),
+            # data_dir,
+            # CFG_PATH.joinpath(org_name.to_str()),
+        )
+
+    # return tuple(sub())
+
+
+@final
+@dataclass(frozen=True, slots=True)
+class DestPaths:
+    data_dir: Path
+    env_file: Path
+    compose_file: Path
+
+
+def dest_paths_factory(compose_template: ComposeTemplate) -> DestPaths:
+    data_dir = compose_template.data.val.path
+    return DestPaths(
+        data_dir, data_dir.joinpath(".env"), data_dir.joinpath("docker-compose.yml")
+    )
+
+@final
+@dataclass(frozen=True, slots=True)
+class ParsedCompose:
+    
+
+@final
+@dataclass(frozen=True, slots=True)
+class DestData:
+    paths: DestPaths
+    templates: ComposeTemplate
+    volumes: tuple[Path, ...]
+
+
+def get_volumes_raw(data: HasServices) -> Iterator[str]:
+    for app_data in data["services"].values():
+        if "volumes" not in app_data:
+            return
+        for vol in app_data["volumes"]:
+            if not vol.startswith(r"${_DATA}"):
+                continue
+            yield vol.split(":", 1)[0]
+
+
+def dest_data_factory(cfg_data: CfgData, compose: Compose) -> Iterator[DestData]:
+    # def sub():
+    vols_raw = tuple(get_volumes_raw(compose))
+    # templates = tuple(compose_template_factory(cfg_data))
+    # for template in templates:
+    #     for vol in vols_raw:
+
+    for template in compose_template_factory(cfg_data):
+        paths = dest_paths_factory(template)
+
+        def vol_sub():
+            for vol in vols_raw:
+                yield Path(replace(template.data.val, vol))
+
+        yield DestData(
+            paths,
+            template,
+            tuple(vol_sub()),
+        )
+
+    # return tuple(sub())
+
+
+@final
+@dataclass(frozen=True, slots=True)
+class ComposeData:
+    src_paths: SrcPaths
+    dest_data: tuple[DestData, ...]
+    Compose: Compose
+    proxy_net: tuple[str, str] | None
+    yaml: str
+
+
+def get_compose(cfg_data: CfgData) -> Compose:
+    def insert_defaults(data: Compose) -> None:
+        sec_opts = "no-new-privileges:true"
+        data.name = "${_ORG_NAME}"
+        for app_data in data.services.values():
+            app_data.restart = "unless-stopped"
+            if app_data.security_opt is None:
+                app_data.security_opt = {sec_opts}
+            else:
+                app_data.security_opt.add(sec_opts)
+
+    def insert_traefik_labels(data: Compose) -> None:
+        for app_data in data.services.values():
+            traefik_labels = {
+                "traefik.http.routers.${_ORG_NAME}.rule=Host(`${_URL}`)",
+                "traefik.http.routers.${_ORG_NAME}.entrypoints=websecure",
+                "traefik.docker.network=${_ORG_NAME}_proxy",
+                "traefik.http.routers.${_ORG_NAME}.tls.certresolver=le",
+            }
+            if app_data.labels is None:
+                continue
+            if "traefik.enable=true" not in app_data.labels:
+                continue
+            app_data.labels.update(traefik_labels)
+
+    def sub():
+        for path in cfg_data.files:
+            _dict = cast(ComposeServiceYaml, read_yml(path))
+            yield ComposeService.from_dict(_dict)
+
+    compose = reduce(merge_dicts, sub(), cast(Compose, {}))  # pyright: ignore[reportInvalidCast]
+    defaults = (insert_defaults, insert_traefik_labels)
+    for func in defaults:
+        func(compose)
+    return compose
+
+
+def set_networks(data: Compose) -> tuple[str, str] | None:
+    def sub() -> Iterator[Literal["proxy", "internal"]]:
+        for app_data in data["services"].values():
+            if "networks" not in app_data:
+                continue
+            yield from app_data["networks"]
+
+    nets = set(sub())
+    net = ComposeNet()
+    if "proxy" in nets:
+        proxy = ComposeNetArgs(name="${_ORG_NAME}_proxy", external=True)
+        net["proxy"] = proxy
+        ret = "${_ORG_NAME}", "${_ORG_NAME}_proxy"
+    else:
+        ret = None
+    if "internal" in nets:
+        net["internal"] = ComposeNetArgs(name="${_ORG_NAME}")
+    if not net:
+        return
+    data["networks"] = net
+    return ret
+
+
+def compose_data_factory(src: str):
+    src_path = src_path_factory(src)
+    cfg_data = config_data_factory(src_path)
+    compose = get_compose(cfg_data)
+    dest_data = dest_data_factory(cfg_data, compose)
+    return ComposeData(
+        src_path,
+        tuple(dest_data),
+        compose,
+        set_networks(compose),
+        to_yaml(compose),
+    )

src/compose/service.py

@@ -0,0 +1,35 @@
+from functools import reduce
+from pathlib import Path
+from shutil import copyfile
+
+from .entities import ComposeData, replace
+
+
+def mk_dir(path: Path):
+    if path.exists():
+        return
+    path.mkdir(parents=True)
+
+
+def mk_compose_dir(compose_data: ComposeData) -> None:
+    for dest in compose_data.dest_data:
+        mk_dir(dest.paths.data_dir)
+        for vol in dest.volumes:
+            mk_dir(vol)
+
+
+def mk_compose_env(compose_data: ComposeData) -> None:
+    src = compose_data.src_paths.env_file
+    for dest_data in compose_data.dest_data:
+        dest = dest_data.paths.env_file
+        if src.exists() and not dest.exists():
+            _ = copyfile(src, dest)
+
+
+def write(compose_data: ComposeData) -> None:
+    for dest_data in compose_data.dest_data:
+        string = reduce(
+            lambda s, rec: replace(rec, s), dest_data.templates, compose_data.yaml
+        )
+        with dest_data.paths.compose_file.open("wt") as f:
+            _ = f.write(string)

src/compose/val_objs.py

@@ -0,0 +1,65 @@
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Protocol, final
+
+
+class RecordVal(Protocol):
+    def to_str(self) -> str: ...
+
+
+@final
+@dataclass(frozen=True, slots=True)
+class RecordCls[T: RecordVal]:
+    name: str
+    val: T
+
+
+@final
+@dataclass(frozen=True, slots=True)
+class OrgVal:
+    val: str | None
+
+    def to_str(self) -> str:
+        if self.val is None:
+            return "personal"
+        return self.val
+
+    def is_valid(self) -> bool:
+        return self.val is not None
+
+
+@final
+@dataclass(frozen=True, slots=True)
+class NameVal:
+    val: str
+
+    def to_str(self) -> str:
+        return self.val
+
+
+@final
+@dataclass(frozen=True, slots=True)
+class DataDir:
+    path: Path
+
+    def to_str(self) -> str:
+        return str(self.path)
+
+
+@final
+@dataclass(frozen=True, slots=True)
+class Url:
+    sub_url: str | None
+
+    def to_str(self) -> str:
+        if self.sub_url is None:
+            return ""
+        return ".".join([self.sub_url, "ccamper7", "net"])
+
+
+# def get_replace_var(string: str) -> str:
+#     return f"${{_{name}}}"
+
+
+def Record[T: RecordVal](name: str, val: T) -> RecordCls[T]:
+    return RecordCls(f"${{_{name}}}", val)

src/main.py

@@ -0,0 +1,23 @@
+from typing import Any, Generator
+
+
+from collections.abc import Iterator
+from Ts import TraefikComposeDict, TraefikNet, TraefikNetName
+from compose import gen_compose
+
+
+if __name__ == "__main__":
+    paths: tuple[str, ...] = ("gitea", "opencloud", "jellyfin", "immich")
+
+    def sub() -> Iterator[tuple[str, str]]:
+        for path in paths:
+            yield from gen_compose(path)
+
+            # yield name, TraefikNetName(name=proxy)
+
+    networks: TraefikNet = dict(sub())
+    traefik = ComposeBuild("traefik").build()
+    traefik_compose = TraefikComposeDict(
+        name="traefik",
+        networks=networks,
+    )

src/util.py

@@ -0,0 +1,46 @@
+import re
+from collections.abc import Mapping
+from pathlib import Path
+from typing import Any, cast, override
+
+import yaml
+
+from compose.compose_struct import T_PrimDict, T_PrimVal, T_Primitive
+
+
+class VerboseSafeDumper(yaml.SafeDumper):
+    @override
+    def ignore_aliases(self, data: Any) -> bool:  # pyright: ignore[reportExplicitAny, reportAny]
+        return True
+
+
+def merge_dicts[T: Mapping[Any, Any]](dict1: T, dict2: T) -> T:
+    def _merge_dicts(dict1: T_PrimDict, dict2: T_PrimDict):
+        s1 = set(dict1.keys())
+        s2 = set(dict2.keys())
+        for k in s1.difference(s2):
+            yield k, dict1[k]
+        for k in s2.difference(s1):
+            yield k, dict2[k]
+        for k in s1.intersection(s2):
+            v1 = dict1[k]
+            v2 = dict2[k]
+            if isinstance(v1, dict) and isinstance(v2, dict):
+                yield k, dict[T_Primitive, T_PrimVal](_merge_dicts(v1, v2))
+                continue
+            if isinstance(v1, list) and isinstance(v2, list):
+                yield k, list(set(v1).union(v2))
+                continue
+            raise Exception("merge error")
+
+    return cast(T, dict(_merge_dicts(dict1, dict2)))
+
+
+def read_yml(path: Path) -> Mapping[Any, Any]:  # pyright: ignore[reportExplicitAny]
+    with path.open("rt") as f:
+        return yaml.safe_load(f)  # pyright: ignore[reportAny]
+
+
+def to_yaml(data: Mapping[Any, Any]) -> str:  # pyright: ignore[reportExplicitAny]
+    _yaml = yaml.dump(data, Dumper=VerboseSafeDumper)
+    return re.sub(r"(^\s*-)", r"  \g<1>", _yaml, flags=re.MULTILINE)