1import re
2from collections import defaultdict
3
4from fastapi import APIRouter
5from fastapi import Header, HTTPException, Path, Request, Response
6from starlette.responses import HTMLResponse
7
8from app.db import crud
9from app.db.models import Settings
10from app.dependencies import DBDep, SubUserDep, StartDateDep, EndDateDep
11from app.models.settings import SubscriptionSettings
12from app.models.user import UserResponse
13from app.utils.share import (
14 encode_title,
15 generate_subscription,
16 generate_subscription_template,
17)
18
19router = APIRouter(prefix="/sub", tags=["Subscription"])
20
21
22config_mimetype = defaultdict(
23 lambda: "text/plain",
24 {
25 "links": "text/plain",
26 "base64-links": "text/plain",
27 "sing-box": "application/json",
28 "xray": "application/json",
29 "clash": "text/yaml",
30 "clash-meta": "text/yaml",
31 "template": "text/html",
32 "block": "text/plain",
33 },
34)
35
36
37def get_subscription_user_info(user: UserResponse) -> dict:
38 return {
39 "upload": 0,
40 "download": user.used_traffic,
41 "total": user.data_limit or 0,
42 "expire": (
43 int(user.expire_date.timestamp())
44 if user.expire_strategy == "fixed_date"
45 else 0
46 ),
47 }
48
49
50@router.get("/{username}/{key}")
51def user_subscription(
52 db_user: SubUserDep,
53 request: Request,
54 db: DBDep,
55 user_agent: str = Header(default=""),
56):
57 """
58 Subscription link, result format depends on subscription settings
59 """
60
61 user: UserResponse = UserResponse.model_validate(db_user)
62
63 crud.update_user_sub(db, db_user, user_agent)
64
65 subscription_settings = SubscriptionSettings.model_validate(
66 db.query(Settings.subscription).first()[0]
67 )
68
69 if (
70 subscription_settings.template_on_acceptance
71 and "text/html" in request.headers.get("Accept", [])
72 ):
73 return HTMLResponse(
74 generate_subscription_template(db_user, subscription_settings)
75 )
76
77 response_headers = {
78 "content-disposition": f'attachment; filename="{user.username}"',
79 "profile-web-page-url": str(request.url),
80 "support-url": subscription_settings.support_link,
81 "profile-title": encode_title(subscription_settings.profile_title),
82 "profile-update-interval": str(subscription_settings.update_interval),
83 "subscription-userinfo": "; ".join(
84 f"{key}={val}"
85 for key, val in get_subscription_user_info(user).items()
86 ),
87 }
88
89 for rule in subscription_settings.rules:
90 if re.match(rule.pattern, user_agent):
91 if rule.result.value == "template":
92 return HTMLResponse(
93 generate_subscription_template(
94 db_user, subscription_settings
95 )
96 )
97 elif rule.result.value == "block":
98 raise HTTPException(404)
99 elif rule.result.value == "base64-links":
100 b64 = True
101 config_format = "links"
102 else:
103 b64 = False
104 config_format = rule.result.value
105
106 conf = generate_subscription(
107 user=db_user,
108 config_format=config_format,
109 as_base64=b64,
110 use_placeholder=not user.is_active
111 and subscription_settings.placeholder_if_disabled,
112 placeholder_remark=subscription_settings.placeholder_remark,
113 shuffle=subscription_settings.shuffle_configs,
114 )
115 return Response(
116 content=conf,
117 media_type=config_mimetype[rule.result],
118 headers=response_headers,
119 )
120
121
122@router.get("/{username}/{key}/info", response_model=UserResponse)
123def user_subscription_info(db_user: SubUserDep):
124 return db_user
125
126
127@router.get("/{username}/{key}/usage")
128def user_get_usage(
129 db_user: SubUserDep,
130 db: DBDep,
131 start_date: StartDateDep,
132 end_date: EndDateDep,
133):
134 usages = crud.get_user_usages(db, db_user, start_date, end_date)
135
136 return {"usages": usages, "username": db_user.username}
137
138
139client_type_mime_type = {
140 "sing-box": "application/json",
141 "clash-meta": "text/yaml",
142 "clash": "text/yaml",
143 "xray": "application/json",
144 "v2ray": "text/plain",
145}
146
147
148@router.get("/{username}/{key}/{client_type}")
149def user_subscription_with_client_type(
150 db: DBDep,
151 db_user: SubUserDep,
152 request: Request,
153 client_type: str = Path(regex="sing-box|clash-meta|clash|xray|v2ray"),
154):
155 """
156 Subscription by client type; v2ray, xray, sing-box, clash and clash-meta formats supported
157 """
158
159 user: UserResponse = UserResponse.model_validate(db_user)
160
161 subscription_settings = SubscriptionSettings.model_validate(
162 db.query(Settings.subscription).first()[0]
163 )
164
165 response_headers = {
166 "content-disposition": f'attachment; filename="{user.username}"',
167 "profile-web-page-url": str(request.url),
168 "support-url": subscription_settings.support_link,
169 "profile-title": encode_title(subscription_settings.profile_title),
170 "profile-update-interval": str(subscription_settings.update_interval),
171 "subscription-userinfo": "; ".join(
172 f"{key}={val}"
173 for key, val in get_subscription_user_info(user).items()
174 ),
175 }
176
177 conf = generate_subscription(
178 user=db_user,
179 config_format="links" if client_type == "v2ray" else client_type,
180 as_base64=client_type == "v2ray",
181 use_placeholder=not user.is_active
182 and subscription_settings.placeholder_if_disabled,
183 placeholder_remark=subscription_settings.placeholder_remark,
184 shuffle=subscription_settings.shuffle_configs,
185 )
186 return Response(
187 content=conf,
188 media_type=client_type_mime_type[client_type],
189 headers=response_headers,
190 )