feat: /raw path for 3rd-party apps

Author: M03ED

app/operation/subscription.py

@@ -286,13 +286,9 @@ async def user_subscription(
             return HTMLResponse(
                 render_template(
                     template,
-                    {
-                        "user": user,
-                        "links": links,
-                        "announce": formatted_announce,
-                        "announce_url": sub_settings.announce_url,
-                        "apps": self._make_apps_import_urls(sub_settings.applications, format_variables),
-                    },
+                    self._build_raw_subscription_payload(
+                        user, links, formatted_announce, sub_settings, format_variables
+                    ),
                 )
             )
         else:
@@ -374,6 +370,34 @@ async def user_subscription_with_client_type(
         # Create response headers
         return Response(content=conf, media_type=media_type, headers=response_headers)
 
+    def _build_raw_subscription_payload(
+        self,
+        user: UsersResponseWithInbounds,
+        links: list[str],
+        formatted_announce: str,
+        sub_settings: SubSettings,
+        format_variables: dict,
+    ) -> dict[str, Any]:
+        return {
+            "user": user,
+            "links": links,
+            "announce": formatted_announce,
+            "announce_url": sub_settings.announce_url,
+            "apps": self._make_apps_import_urls(sub_settings.applications, format_variables),
+        }
+
+    async def user_subscription_raw(self, db: AsyncSession, token: str):
+        sub_settings: SubSettings = await subscription_settings()
+        db_user = await self.get_validated_sub(db, token)
+        user = await self.validated_user(db_user)
+        links = []
+        if sub_settings.allow_browser_config:
+            conf, _ = await self.fetch_config(user, ConfigFormat.links)
+            links = conf.splitlines()
+        format_variables = await self.get_format_variables(user)
+        formatted_announce = self._format_announce(sub_settings, format_variables)
+        return self._build_raw_subscription_payload(user, links, formatted_announce, sub_settings, format_variables)
+
     async def user_subscription_by_user(
         self,
         db_user: User,

app/routers/subscription.py

@@ -42,6 +42,11 @@ async def user_subscription_info(request: Request, token: str, db: AsyncSession
     return JSONResponse(content=user_data.model_dump(mode="json"), headers=response_headers)
 
 
+@router.get("/{token}/raw")
+async def user_subscription_raw(token: str, db: AsyncSession = Depends(get_db)):
+    return await subscription_operator.user_subscription_raw(db, token=token)
+
+
 @router.get("/{token}/apps", response_model=list[Application])
 async def user_subscription_apps(token: str, db: AsyncSession = Depends(get_db)):
     """

app/subscription/base.py

@@ -7,7 +7,6 @@
 from urllib.parse import quote, urlencode
 
 from app.models.subscription import SubscriptionInboundData
-from app.templates import render_template_string
 
 
 class BaseSubscription:
@@ -17,13 +16,13 @@ def __init__(
         grpc_user_agent_template_content: str | None = None,
     ):
         self.proxy_remarks = []
-        user_agent_data = json.loads(render_template_string(user_agent_template_content))
+        user_agent_data = json.loads(user_agent_template_content) if user_agent_template_content else {}
         if "list" in user_agent_data and isinstance(user_agent_data["list"], list):
             self.user_agent_list = user_agent_data["list"]
         else:
             self.user_agent_list = []
 
-        grpc_user_agent_data = json.loads(render_template_string(grpc_user_agent_template_content))
+        grpc_user_agent_data = json.loads(grpc_user_agent_template_content) if grpc_user_agent_template_content else {}
 
         if "list" in grpc_user_agent_data and isinstance(grpc_user_agent_data["list"], list):
             self.grpc_user_agent_data = grpc_user_agent_data["list"]

app/subscription/singbox.py

@@ -9,7 +9,6 @@
     TLSConfig,
     WebSocketTransportConfig,
 )
-from app.templates import render_template_string
 from app.utils.helpers import UUIDEncoder
 
 from . import BaseSubscription
@@ -26,7 +25,7 @@ def __init__(
             user_agent_template_content=user_agent_template_content,
             grpc_user_agent_template_content=grpc_user_agent_template_content,
         )
-        self.config = json.loads(render_template_string(singbox_template_content))
+        self.config = json.loads(singbox_template_content) if singbox_template_content else {}
         self.config.setdefault("endpoints", [])
         self.config.setdefault("outbounds", [])
 
@@ -59,6 +58,10 @@ def add_endpoint(self, endpoint_data):
         self.config["endpoints"].append(endpoint_data)
 
     def render(self):
+        self._finalize_config()
+        return json.dumps(self.config, indent=4, cls=UUIDEncoder)
+
+    def _finalize_config(self):
         urltest_types = ["vmess", "vless", "trojan", "shadowsocks", "hysteria2", "tuic", "http", "ssh"]
         urltest_tags = [outbound["tag"] for outbound in self.config["outbounds"] if outbound["type"] in urltest_types]
         selector_types = [
@@ -85,8 +88,6 @@ def render(self):
             if outbound.get("type") == "selector":
                 outbound["outbounds"] = selector_tags
 
-        return json.dumps(self.config, indent=4, cls=UUIDEncoder)
-
     def add(self, remark: str, address: str, inbound: SubscriptionInboundData, settings: dict):
         """Add outbound using registry pattern"""
         # Not supported by sing-box

app/subscription/xray.py

@@ -11,7 +11,6 @@
     WebSocketTransportConfig,
     XHTTPTransportConfig,
 )
-from app.templates import render_template_string
 from app.utils.helpers import UUIDEncoder
 
 from . import BaseSubscription
@@ -29,7 +28,7 @@ def __init__(
             grpc_user_agent_template_content=grpc_user_agent_template_content,
         )
         self.config = []
-        self.template = render_template_string(xray_template_content)
+        self.template = json.loads(xray_template_content) if xray_template_content else {}
 
         # Registry for transport handlers
         self.transport_handlers = {
@@ -57,8 +56,7 @@ def __init__(
         }
 
     def add_config(self, remarks, outbounds, template_content: str | None = None):
-        rendered_template = render_template_string(template_content) if template_content is not None else self.template
-        json_template = json.loads(rendered_template)
+        json_template = json.loads(template_content) if template_content is not None else self.template.copy()
         json_template["remarks"] = remarks
         json_template["outbounds"] = outbounds + json_template["outbounds"]
         self.config.append(json_template)