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, Type
  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.base import BaseConfig
 20from v2share.links import LinksConfig
 21
 22from app.config.env import (
 23    XRAY_SUBSCRIPTION_TEMPLATE,
 24    SINGBOX_SUBSCRIPTION_TEMPLATE,
 25    CLASH_SUBSCRIPTION_TEMPLATE,
 26    SUBSCRIPTION_PAGE_TEMPLATE,
 27)
 28from app.db import GetDB, crud
 29from app.models.proxy import (
 30    InboundHostSecurity,
 31)
 32from app.models.settings import SubscriptionSettings
 33from app.models.user import UserResponse, UserExpireStrategy
 34from app.templates import render_template
 35from app.utils.keygen import gen_uuid, gen_password
 36from app.utils.system import get_public_ip, readable_size
 37
 38SERVER_IP = get_public_ip()
 39
 40ACTIVITY_EMOJIS = {
 41    True: "✅",
 42    False: "❌",
 43}
 44
 45subscription_handlers: dict[str, Type[BaseConfig]] = {
 46    "links": LinksConfig,
 47    "xray": XrayConfig,
 48    "clash-meta": ClashMetaConfig,
 49    "clash": ClashConfig,
 50    "sing-box": SingBoxConfig,
 51}
 52
 53handlers_templates = {
 54    LinksConfig: None,
 55    XrayConfig: XRAY_SUBSCRIPTION_TEMPLATE
 56    or resources.files("app.templates") / "xray.json",
 57    ClashConfig: CLASH_SUBSCRIPTION_TEMPLATE,
 58    ClashMetaConfig: CLASH_SUBSCRIPTION_TEMPLATE,
 59    SingBoxConfig: SINGBOX_SUBSCRIPTION_TEMPLATE
 60    or resources.files("app.templates") / "sing-box.json",
 61}
 62
 63
 64def generate_subscription_template(
 65    db_user, subscription_settings: SubscriptionSettings
 66):
 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    subscription_handler.add_proxies(configs)
122    config = subscription_handler.render(sort=True, shuffle=shuffle)
123
124    return (
125        config if not as_base64 else base64.b64encode(config.encode()).decode()
126    )
127
128
129def format_time_left(seconds_left: int) -> str:
130    if not seconds_left or seconds_left <= 0:
131        return "∞"
132
133    minutes, seconds = divmod(seconds_left, 60)
134    hours, minutes = divmod(minutes, 60)
135    days, hours = divmod(hours, 24)
136    months, days = divmod(days, 30)
137
138    result = []
139    if months:
140        result.append(f"{months}m")
141    if days:
142        result.append(f"{days}d")
143    if hours and (days < 7):
144        result.append(f"{hours}h")
145    if minutes and not (months or days):
146        result.append(f"{minutes}m")
147    if seconds and not (months or days):
148        result.append(f"{seconds}s")
149    return " ".join(result)
150
151
152def setup_format_variables(extra_data: dict) -> dict:
153    expire_strategy = extra_data.get("expire_strategy")
154    expire_datetime = extra_data.get("expire_date")
155    usage_duration = extra_data.get("usage_duration")
156
157    if expire_strategy != UserExpireStrategy.START_ON_FIRST_USE:
158        if expire_datetime is not None:
159            seconds_left = (expire_datetime - dt.utcnow()).total_seconds()
160            expire_date = expire_datetime.date()
161            jalali_expire_date = jd.fromgregorian(
162                year=expire_date.year,
163                month=expire_date.month,
164                day=expire_date.day,
165            ).strftime("%Y-%m-%d")
166            days_left = (expire_datetime - dt.utcnow()).days + 1
167            time_left = format_time_left(seconds_left)
168        else:
169            days_left, time_left, expire_date, jalali_expire_date = "∞" * 4
170    else:
171        if usage_duration is not None and usage_duration >= 0:
172            days_left = timedelta(seconds=usage_duration).days
173            time_left = format_time_left(usage_duration)
174            expire_date = "-"
175            jalali_expire_date = "-"
176        else:
177            days_left, time_left, expire_date, jalali_expire_date = "∞" * 4
178
179    if extra_data.get("data_limit"):
180        data_limit = readable_size(extra_data["data_limit"])
181        data_left = extra_data["data_limit"] - extra_data["used_traffic"]
182        if data_left < 0:
183            data_left = 0
184        data_left = readable_size(data_left)
185    else:
186        data_limit, data_left = "∞" * 2
187
188    status_emoji = ACTIVITY_EMOJIS.get(extra_data.get("is_active")) or ""
189
190    format_variables = defaultdict(
191        lambda: "<missing>",
192        {
193            "SERVER_IP": SERVER_IP,
194            "USERNAME": extra_data.get("username", "{USERNAME}"),
195            "DATA_USAGE": readable_size(extra_data.get("used_traffic")),
196            "DATA_LIMIT": data_limit,
197            "DATA_LEFT": data_left,
198            "DAYS_LEFT": days_left,
199            "EXPIRE_DATE": expire_date,
200            "JALALI_EXPIRE_DATE": jalali_expire_date,
201            "TIME_LEFT": time_left,
202            "STATUS_EMOJI": status_emoji,
203        },
204    )
205
206    return format_variables
207
208
209def generate_user_configs(
210    inbounds: list,
211    key: str,
212    format_variables: dict,
213) -> Union[List, str]:
214
215    salt = secrets.token_hex(8)
216    configs = []
217
218    for inb in inbounds:
219        inb_id, inbound, protocol, tag = (
220            inb.id,
221            json.loads(inb.config),
222            inb.protocol,
223            inb.tag,
224        )
225
226        format_variables.update({"PROTOCOL": protocol.name})
227        if not inbound:
228            continue
229
230        format_variables.update(
231            {"TRANSPORT": inbound.get("network", "<missing>")}
232        )
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.get("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.get("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.get("network"),
264                sni=sni,
265                host=req_host,
266                tls=host_tls or inbound.get("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                weight=host.weight,
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()}"