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