1import base64
2import json
3import random
4import secrets
5from collections import defaultdict
6from datetime import datetime as dt, timedelta
7from importlib import resources
8from typing import Literal, Union, List
9from uuid import UUID
10
11from jdatetime import date as jd
12from v2share import (
13 V2Data,
14 SingBoxConfig,
15 ClashConfig,
16 ClashMetaConfig,
17 XrayConfig,
18)
19from v2share.links import LinksConfig
20
21from app.config.env import (
22 XRAY_SUBSCRIPTION_TEMPLATE,
23 SINGBOX_SUBSCRIPTION_TEMPLATE,
24 CLASH_SUBSCRIPTION_TEMPLATE,
25 SUBSCRIPTION_PAGE_TEMPLATE,
26)
27from app.db import GetDB, crud
28from app.models.proxy import (
29 InboundHostSecurity,
30)
31from app.models.settings import SubscriptionSettings
32from app.models.user import UserResponse, UserExpireStrategy
33from app.templates import render_template
34from app.utils.keygen import gen_uuid, gen_password
35from app.utils.system import get_public_ip, readable_size
36
37SERVER_IP = get_public_ip()
38
39ACTIVITY_EMOJIS = {
40 True: "✅",
41 False: "❌",
42}
43
44subscription_handlers = {
45 "links": LinksConfig,
46 "xray": XrayConfig,
47 "clash-meta": ClashMetaConfig,
48 "clash": ClashConfig,
49 "sing-box": SingBoxConfig,
50}
51
52handlers_templates = {
53 LinksConfig: None,
54 XrayConfig: XRAY_SUBSCRIPTION_TEMPLATE
55 or resources.files("app.templates") / "xray.json",
56 ClashConfig: CLASH_SUBSCRIPTION_TEMPLATE,
57 ClashMetaConfig: CLASH_SUBSCRIPTION_TEMPLATE,
58 SingBoxConfig: SINGBOX_SUBSCRIPTION_TEMPLATE
59 or resources.files("app.templates") / "sing-box.json",
60}
61
62
63def generate_subscription_template(
64 db_user, subscription_settings: SubscriptionSettings
65):
66 print(subscription_settings.shuffle_configs)
67 links = generate_subscription(
68 user=db_user,
69 config_format="links",
70 use_placeholder=not db_user.is_active
71 and subscription_settings.placeholder_if_disabled,
72 placeholder_remark=subscription_settings.placeholder_remark,
73 shuffle=subscription_settings.shuffle_configs,
74 ).split()
75 return render_template(
76 SUBSCRIPTION_PAGE_TEMPLATE,
77 {"user": UserResponse.model_validate(db_user), "links": links},
78 )
79
80
81def generate_subscription(
82 user: "UserResponse",
83 config_format: Literal["links", "xray", "clash-meta", "clash", "sing-box"],
84 as_base64: bool = False,
85 use_placeholder: bool = False,
86 placeholder_remark: str = "disabled",
87 shuffle: bool = False,
88) -> str:
89 extra_data = UserResponse.model_validate(user).model_dump(
90 exclude={"subscription_url", "services", "inbounds"}
91 )
92 format_variables = setup_format_variables(extra_data)
93
94 if config_format not in subscription_handlers.keys():
95 raise ValueError(f'Unsupported format "{config_format}"')
96
97 subscription_handler_class = subscription_handlers[config_format]
98 if template_path := handlers_templates[subscription_handler_class]:
99 subscription_handler = subscription_handler_class(
100 template_path=template_path
101 )
102 else:
103 subscription_handler = subscription_handler_class()
104
105 if use_placeholder:
106 placeholder_config = V2Data(
107 "vmess",
108 placeholder_remark.format_map(format_variables),
109 "127.0.0.1",
110 80,
111 )
112 configs = [placeholder_config]
113
114 else:
115 configs = generate_user_configs(
116 user.inbounds,
117 user.key,
118 format_variables,
119 )
120
121 if shuffle:
122 random.shuffle(configs)
123
124 subscription_handler.add_proxies(configs)
125 config = subscription_handler.render()
126
127 return (
128 config if not as_base64 else base64.b64encode(config.encode()).decode()
129 )
130
131
132def format_time_left(seconds_left: int) -> str:
133 if not seconds_left or seconds_left <= 0:
134 return "∞"
135
136 minutes, seconds = divmod(seconds_left, 60)
137 hours, minutes = divmod(minutes, 60)
138 days, hours = divmod(hours, 24)
139 months, days = divmod(days, 30)
140
141 result = []
142 if months:
143 result.append(f"{months}m")
144 if days:
145 result.append(f"{days}d")
146 if hours and (days < 7):
147 result.append(f"{hours}h")
148 if minutes and not (months or days):
149 result.append(f"{minutes}m")
150 if seconds and not (months or days):
151 result.append(f"{seconds}s")
152 return " ".join(result)
153
154
155def setup_format_variables(extra_data: dict) -> dict:
156 expire_strategy = extra_data.get("expire_strategy")
157 expire_datetime = extra_data.get("expire_date")
158 usage_duration = extra_data.get("usage_duration")
159
160 if expire_strategy != UserExpireStrategy.START_ON_FIRST_USE:
161 if expire_datetime is not None:
162 seconds_left = (expire_datetime - dt.utcnow()).total_seconds()
163 expire_date = expire_datetime.date()
164 jalali_expire_date = jd.fromgregorian(
165 year=expire_date.year,
166 month=expire_date.month,
167 day=expire_date.day,
168 ).strftime("%Y-%m-%d")
169 days_left = (expire_datetime - dt.utcnow()).days + 1
170 time_left = format_time_left(seconds_left)
171 else:
172 days_left, time_left, expire_date, jalali_expire_date = "∞" * 4
173 else:
174 if usage_duration is not None and usage_duration >= 0:
175 days_left = timedelta(seconds=usage_duration).days
176 time_left = format_time_left(usage_duration)
177 expire_date = "-"
178 jalali_expire_date = "-"
179 else:
180 days_left, time_left, expire_date, jalali_expire_date = "∞" * 4
181
182 if extra_data.get("data_limit"):
183 data_limit = readable_size(extra_data["data_limit"])
184 data_left = extra_data["data_limit"] - extra_data["used_traffic"]
185 if data_left < 0:
186 data_left = 0
187 data_left = readable_size(data_left)
188 else:
189 data_limit, data_left = "∞" * 2
190
191 status_emoji = ACTIVITY_EMOJIS.get(extra_data.get("is_active")) or ""
192
193 format_variables = defaultdict(
194 lambda: "<missing>",
195 {
196 "SERVER_IP": SERVER_IP,
197 "USERNAME": extra_data.get("username", "{USERNAME}"),
198 "DATA_USAGE": readable_size(extra_data.get("used_traffic")),
199 "DATA_LIMIT": data_limit,
200 "DATA_LEFT": data_left,
201 "DAYS_LEFT": days_left,
202 "EXPIRE_DATE": expire_date,
203 "JALALI_EXPIRE_DATE": jalali_expire_date,
204 "TIME_LEFT": time_left,
205 "STATUS_EMOJI": status_emoji,
206 },
207 )
208
209 return format_variables
210
211
212def generate_user_configs(
213 inbounds: list,
214 key: str,
215 format_variables: dict,
216) -> Union[List, str]:
217
218 salt = secrets.token_hex(8)
219 configs = []
220
221 for inb in inbounds:
222 inb_id, inbound, protocol, tag = (
223 inb.id,
224 json.loads(inb.config),
225 inb.protocol,
226 inb.tag,
227 )
228
229 format_variables.update({"PROTOCOL": protocol.name})
230 if not inbound:
231 continue
232
233 format_variables.update({"TRANSPORT": inbound["network"]})
234 with GetDB() as db:
235 hosts = crud.get_inbound_hosts(db, inb_id)
236
237 for host in hosts:
238 if host.is_disabled:
239 continue
240 host_snis = host.sni.split(",") if host.sni else []
241 sni_list = host_snis or inbound["sni"]
242 if sni_list:
243 sni = random.choice(sni_list).replace("*", salt)
244 else:
245 sni = ""
246
247 host_hosts = host.host.split(",") if host.host else []
248 req_host_list = host_hosts or inbound["host"]
249 if req_host_list:
250 req_host = random.choice(req_host_list).replace("*", salt)
251 else:
252 req_host = ""
253
254 host_tls = (
255 None
256 if host.security == InboundHostSecurity.inbound_default
257 else host.security.value
258 )
259 data = V2Data(
260 protocol.value,
261 host.remark.format_map(format_variables),
262 host.address.format_map(format_variables),
263 host.port or inbound["port"],
264 transport_type=inbound["network"],
265 sni=sni,
266 host=req_host,
267 tls=host_tls or inbound["tls"],
268 header_type=inbound.get("header_type"),
269 alpn=host.alpn.value or None,
270 path=(
271 host.path.format_map(format_variables)
272 if host.path
273 else inbound.get("path")
274 ),
275 fingerprint=host.fingerprint.value or inbound.get("fp"),
276 reality_pbk=inbound.get("pbk"),
277 reality_sid=inbound.get("sid"),
278 flow=inbound.get("flow"),
279 allow_insecure=host.allowinsecure,
280 uuid=UUID(gen_uuid(key)),
281 password=gen_password(key),
282 enable_mux=host.mux,
283 http_headers={
284 "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.3"
285 },
286 shadowsocks_method="chacha20-ietf-poly1305",
287 )
288 if host.fragment:
289 data.fragment = True
290 data.fragment_packets = host.fragment["packets"]
291 data.fragment_length = host.fragment["length"]
292 data.fragment_interval = host.fragment["interval"]
293 configs.append(data)
294
295 return configs
296
297
298def encode_title(text: str) -> str:
299 return f"base64:{base64.b64encode(text.encode()).decode()}"