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