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 host_snis = host.sni.split(",") if host.sni else []
239 sni_list = host_snis or inbound["sni"]
240 if sni_list:
241 sni = random.choice(sni_list).replace("*", salt)
242 else:
243 sni = ""
244
245 host_hosts = host.host.split(",") if host.host else []
246 req_host_list = host_hosts or inbound["host"]
247 if req_host_list:
248 req_host = random.choice(req_host_list).replace("*", salt)
249 else:
250 req_host = ""
251
252 host_tls = (
253 None
254 if host.security == InboundHostSecurity.inbound_default
255 else host.security.value
256 )
257 data = V2Data(
258 protocol.value,
259 host.remark.format_map(format_variables),
260 host.address.format_map(format_variables),
261 host.port or inbound["port"],
262 transport_type=inbound["network"],
263 sni=sni,
264 host=req_host,
265 tls=host_tls or inbound["tls"],
266 header_type=inbound.get("header_type"),
267 alpn=host.alpn.value or None,
268 path=(
269 host.path.format_map(format_variables)
270 if host.path
271 else inbound.get("path")
272 ),
273 fingerprint=host.fingerprint.value or inbound.get("fp"),
274 reality_pbk=inbound.get("pbk"),
275 reality_sid=inbound.get("sid"),
276 flow=inbound.get("flow"),
277 allow_insecure=host.allowinsecure,
278 uuid=UUID(gen_uuid(key)),
279 password=gen_password(key),
280 enable_mux=host.mux,
281 http_headers={
282 "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"
283 },
284 shadowsocks_method="chacha20-ietf-poly1305",
285 )
286 if host.fragment:
287 data.fragment = True
288 data.fragment_packets = host.fragment["packets"]
289 data.fragment_length = host.fragment["length"]
290 data.fragment_interval = host.fragment["interval"]
291 configs.append(data)
292
293 return configs
294
295
296def encode_title(text: str) -> str:
297 return f"base64:{base64.b64encode(text.encode()).decode()}"