1import base64
2import json
3import random
4import secrets
5from datetime import datetime as dt, timedelta
6from typing import TYPE_CHECKING, Literal, Union, List
7from uuid import UUID
8
9from jdatetime import date as jd
10from v2share import V2Data, SingBoxConfig, ClashConfig, ClashMetaConfig
11
12from app.db import GetDB, crud
13from app.models.proxy import (
14 InboundHostSecurity,
15)
16from app.utils.keygen import gen_uuid, gen_password
17from app.utils.system import get_public_ip, readable_size
18
19if TYPE_CHECKING:
20 from app.models.user import UserResponse
21
22SERVER_IP = get_public_ip()
23
24STATUS_EMOJIS = {
25 "active": "✅",
26 "expired": "⌛️",
27 "limited": "🪫",
28 "disabled": "❌",
29 "connect_to_start": "🔌",
30}
31
32
33def generate_v2ray_links(inbounds: list, key: str, extra_data: dict) -> list:
34 format_variables = setup_format_variables(extra_data)
35 return process_inbounds_and_tags(inbounds, key, format_variables, mode="v2ray")
36
37
38def generate_clash_subscription(
39 inbounds: list, key: str, extra_data: dict, is_meta: bool = False
40) -> str:
41 if is_meta is True:
42 conf = ClashMetaConfig()
43 else:
44 conf = ClashConfig()
45
46 format_variables = setup_format_variables(extra_data)
47 return process_inbounds_and_tags(
48 inbounds, key, format_variables, mode="clash", conf=conf
49 )
50
51
52def generate_singbox_subscription(inbounds: list, key: str, extra_data: dict) -> str:
53 conf = SingBoxConfig()
54
55 format_variables = setup_format_variables(extra_data)
56 return process_inbounds_and_tags(
57 inbounds, key, format_variables, mode="sing-box", conf=conf
58 )
59
60
61def generate_v2ray_subscription(links: list) -> str:
62 return base64.b64encode("\n".join(links).encode()).decode()
63
64
65def generate_subscription(
66 user: "UserResponse",
67 config_format: Literal["v2ray", "clash-meta", "clash", "sing-box"],
68 as_base64: bool,
69) -> str:
70 kwargs = {
71 # "proxies": user.proxies,
72 "inbounds": user.inbounds,
73 "key": user.key,
74 "extra_data": user.__dict__,
75 }
76
77 if config_format == "v2ray":
78 config = "\n".join(generate_v2ray_links(**kwargs))
79 elif config_format == "clash-meta":
80 config = generate_clash_subscription(**kwargs, is_meta=True)
81 elif config_format == "clash":
82 config = generate_clash_subscription(**kwargs)
83 elif config_format == "sing-box":
84 config = generate_singbox_subscription(**kwargs)
85 else:
86 raise ValueError(f'Unsupported format "{config_format}"')
87
88 if as_base64:
89 config = base64.b64encode(config.encode()).decode()
90
91 return config
92
93
94def format_time_left(seconds_left: int) -> str:
95 if not seconds_left or seconds_left <= 0:
96 return "∞"
97
98 minutes, seconds = divmod(seconds_left, 60)
99 hours, minutes = divmod(minutes, 60)
100 days, hours = divmod(hours, 24)
101 months, days = divmod(days, 30)
102
103 result = []
104 if months:
105 result.append(f"{months}m")
106 if days:
107 result.append(f"{days}d")
108 if hours and (days < 7):
109 result.append(f"{hours}h")
110 if minutes and not (months or days):
111 result.append(f"{minutes}m")
112 if seconds and not (months or days):
113 result.append(f"{seconds}s")
114 return " ".join(result)
115
116
117from app.models.user import UserStatus
118
119
120def setup_format_variables(extra_data: dict) -> dict:
121
122 user_status = extra_data.get("status")
123 expire_datetime = extra_data.get("expire")
124 on_hold_expire_duration = extra_data.get("on_hold_expire_duration")
125
126 if user_status != UserStatus.on_hold:
127 if expire_datetime is not None:
128 seconds_left = (expire_datetime - dt.utcnow()).total_seconds()
129 expire_date = expire_datetime.date()
130 jalali_expire_date = jd.fromgregorian(
131 year=expire_date.year, month=expire_date.month, day=expire_date.day
132 ).strftime("%Y-%m-%d")
133 days_left = (expire_datetime - dt.utcnow()).days + 1
134 time_left = format_time_left(seconds_left)
135 else:
136 days_left = "∞"
137 time_left = "∞"
138 expire_date = "∞"
139 jalali_expire_date = "∞"
140 else:
141 if on_hold_expire_duration is not None and on_hold_expire_duration >= 0:
142 days_left = timedelta(seconds=on_hold_expire_duration).days
143 time_left = format_time_left(on_hold_expire_duration)
144 expire_date = "-"
145 jalali_expire_date = "-"
146 else:
147 days_left = "∞"
148 time_left = "∞"
149 expire_date = "∞"
150 jalali_expire_date = "∞"
151
152 if extra_data.get("data_limit"):
153 data_limit = readable_size(extra_data["data_limit"])
154 data_left = extra_data["data_limit"] - extra_data["used_traffic"]
155 if data_left < 0:
156 data_left = 0
157 data_left = readable_size(data_left)
158 else:
159 data_limit = "∞"
160 data_left = "∞"
161
162 status_emoji = STATUS_EMOJIS.get(extra_data.get("status")) or ""
163
164 format_variables = {
165 "SERVER_IP": SERVER_IP,
166 "USERNAME": extra_data.get("username", "{USERNAME}"),
167 "DATA_USAGE": readable_size(extra_data.get("used_traffic")),
168 "DATA_LIMIT": data_limit,
169 "DATA_LEFT": data_left,
170 "DAYS_LEFT": days_left,
171 "EXPIRE_DATE": expire_date,
172 "JALALI_EXPIRE_DATE": jalali_expire_date,
173 "TIME_LEFT": time_left,
174 "STATUS_EMOJI": status_emoji,
175 }
176
177 return format_variables
178
179
180def process_inbounds_and_tags(
181 inbounds: list,
182 ukey: str,
183 format_variables: dict,
184 mode: str = "v2ray",
185 conf=None,
186) -> Union[List, str]:
187 salt = secrets.token_hex(8)
188 results = []
189
190 for inb in inbounds:
191 inb_id, inbound, protocol, tag = (
192 inb.id,
193 json.loads(inb.config),
194 inb.protocol,
195 inb.tag,
196 )
197
198 format_variables.update({"PROTOCOL": protocol.name})
199 if not inbound:
200 continue
201
202 format_variables.update({"TRANSPORT": inbound["network"]})
203 with GetDB() as db:
204 hosts = crud.get_inbound_hosts(db, inb_id)
205
206 for host in hosts:
207 host_snis = host.sni.split(",") if host.sni else []
208 sni_list = host_snis or inbound["sni"]
209 if sni_list:
210 sni = random.choice(sni_list).replace("*", salt)
211 else:
212 sni = ""
213
214 host_hosts = host.host.split(",") if host.host else []
215 req_host_list = host_hosts or inbound["host"]
216 if req_host_list:
217 req_host = random.choice(req_host_list).replace("*", salt)
218 else:
219 req_host = ""
220
221 host_tls = (
222 None
223 if host.security == InboundHostSecurity.inbound_default
224 else host.security.value
225 )
226 data = V2Data(
227 protocol.value,
228 host.remark.format_map(format_variables),
229 host.address.format_map(format_variables),
230 host.port or inbound["port"],
231 sni=sni,
232 host=req_host,
233 tls=host_tls or inbound["tls"],
234 alpn=host.alpn.value or inbound.get("alpn", ""),
235 path=(
236 host.path.format_map(format_variables)
237 if host.path
238 else inbound.get("path", "")
239 ),
240 fingerprint=host.fingerprint.value,
241 allow_insecure=host.allowinsecure,
242 uuid=UUID(gen_uuid(ukey)),
243 password=gen_password(ukey),
244 )
245 if mode == "v2ray":
246 results.append(data.to_link())
247 elif mode in ["clash", "sing-box"]:
248 conf.add_proxies([data])
249
250 if mode in ["clash", "sing-box"]:
251 return conf.render()
252 return results
253
254
255def encode_title(text: str) -> str:
256 return f"base64:{base64.b64encode(text.encode()).decode()}"