74 lines
2.1 KiB
Python
74 lines
2.1 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from app.models import ProxyNode, SourceSnapshot
|
|
|
|
|
|
def source_auto_group_name(display_name: str) -> str:
|
|
return f"{display_name} 自动"
|
|
|
|
|
|
def proxy_dict_to_node(proxy: dict[str, Any]) -> ProxyNode | None:
|
|
name = str(proxy.get("name", "")).strip()
|
|
proxy_type = str(proxy.get("type", "")).strip()
|
|
if not name or not proxy_type:
|
|
return None
|
|
|
|
attrs = dict(proxy)
|
|
attrs.pop("name", None)
|
|
attrs.pop("type", None)
|
|
server = attrs.pop("server", None)
|
|
port = attrs.pop("port", None)
|
|
udp = bool(attrs.pop("udp", True))
|
|
return ProxyNode(
|
|
name=name,
|
|
type=proxy_type,
|
|
server=server,
|
|
port=port,
|
|
udp=udp,
|
|
attrs=attrs,
|
|
)
|
|
|
|
|
|
def dedupe_proxy_nodes(nodes: list[ProxyNode]) -> list[ProxyNode]:
|
|
seen: set[str] = set()
|
|
deduped: list[ProxyNode] = []
|
|
for node in nodes:
|
|
original = node.name
|
|
name = original
|
|
index = 2
|
|
while name in seen:
|
|
name = f"{original} #{index}"
|
|
index += 1
|
|
seen.add(name)
|
|
deduped.append(node.model_copy(update={"name": name}))
|
|
return deduped
|
|
|
|
|
|
def build_bundle_proxy_inventory(
|
|
snapshots: list[SourceSnapshot],
|
|
) -> tuple[list[ProxyNode], dict[str, list[str]]]:
|
|
all_nodes: list[ProxyNode] = []
|
|
source_nodes: dict[str, list[ProxyNode]] = {}
|
|
|
|
for snapshot in snapshots:
|
|
nodes: list[ProxyNode] = []
|
|
for proxy in snapshot.document.proxies:
|
|
node = proxy_dict_to_node(proxy)
|
|
if node is not None:
|
|
nodes.append(node)
|
|
source_nodes[snapshot.name] = nodes
|
|
all_nodes.extend(nodes)
|
|
|
|
deduped_all = dedupe_proxy_nodes(all_nodes)
|
|
source_proxy_names: dict[str, list[str]] = {}
|
|
cursor = 0
|
|
for snapshot in snapshots:
|
|
original_nodes = source_nodes[snapshot.name]
|
|
count = len(original_nodes)
|
|
source_proxy_names[snapshot.name] = [node.name for node in deduped_all[cursor : cursor + count]]
|
|
cursor += count
|
|
|
|
return deduped_all, source_proxy_names
|