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,
42 "expire": (
43 user.expire_date if user.expire_strategy == "fixed_date" else None
44 ),
45 }
46
47
48@router.get("/{username}/{key}")
49def user_subscription(
50 db_user: SubUserDep,
51 request: Request,
52 db: DBDep,
53 user_agent: str = Header(default=""),
54):
55 """
56 Subscription link, result format depends on subscription settings
57 """
58
59 user: UserResponse = UserResponse.model_validate(db_user)
60
61 crud.update_user_sub(db, db_user, user_agent)
62
63 subscription_settings = SubscriptionSettings.model_validate(
64 db.query(Settings.subscription).first()[0]
65 )
66
67 if (
68 subscription_settings.template_on_acceptance
69 and "text/html" in request.headers.get("Accept", [])
70 ):
71 return HTMLResponse(
72 generate_subscription_template(db_user, subscription_settings)
73 )
74
75 response_headers = {
76 "content-disposition": f'attachment; filename="{user.username}"',
77 "profile-web-page-url": str(request.url),
78 "support-url": subscription_settings.support_link,
79 "profile-title": encode_title(subscription_settings.profile_title),
80 "profile-update-interval": str(subscription_settings.update_interval),
81 "subscription-userinfo": "; ".join(
82 f"{key}={val}"
83 for key, val in get_subscription_user_info(user).items()
84 if val is not None
85 ),
86 }
87
88 for rule in subscription_settings.rules:
89 if re.match(rule.pattern, user_agent):
90 if rule.result.value == "template":
91 return HTMLResponse(
92 generate_subscription_template(
93 db_user, subscription_settings
94 )
95 )
96 elif rule.result.value == "block":
97 raise HTTPException(404)
98 elif rule.result.value == "base64-links":
99 b64 = True
100 config_format = "links"
101 else:
102 b64 = False
103 config_format = rule.result.value
104
105 conf = generate_subscription(
106 user=db_user,
107 config_format=config_format,
108 as_base64=b64,
109 use_placeholder=not user.is_active
110 and subscription_settings.placeholder_if_disabled,
111 placeholder_remark=subscription_settings.placeholder_remark,
112 shuffle=subscription_settings.shuffle_configs,
113 )
114 return Response(
115 content=conf,
116 media_type=config_mimetype[rule.result],
117 headers=response_headers,
118 )
119
120
121@router.get("/{username}/{key}/info", response_model=UserResponse)
122def user_subscription_info(db_user: SubUserDep):
123 return db_user
124
125
126@router.get("/{username}/{key}/usage")
127def user_get_usage(
128 db_user: SubUserDep,
129 db: DBDep,
130 start_date: StartDateDep,
131 end_date: EndDateDep,
132):
133 usages = crud.get_user_usages(db, db_user, start_date, end_date)
134
135 return {"usages": usages, "username": db_user.username}
136
137
138client_type_mime_type = {
139 "sing-box": "application/json",
140 "clash-meta": "text/yaml",
141 "clash": "text/yaml",
142 "xray": "application/json",
143 "v2ray": "text/plain",
144}
145
146
147@router.get("/{username}/{key}/{client_type}")
148def user_subscription_with_client_type(
149 db: DBDep,
150 db_user: SubUserDep,
151 request: Request,
152 client_type: str = Path(regex="sing-box|clash-meta|clash|xray|v2ray"),
153):
154 """
155 Subscription by client type; v2ray, xray, sing-box, clash and clash-meta formats supported
156 """
157
158 user: UserResponse = UserResponse.model_validate(db_user)
159
160 subscription_settings = SubscriptionSettings.model_validate(
161 db.query(Settings.subscription).first()[0]
162 )
163
164 response_headers = {
165 "content-disposition": f'attachment; filename="{user.username}"',
166 "profile-web-page-url": str(request.url),
167 "support-url": subscription_settings.support_link,
168 "profile-title": encode_title(subscription_settings.profile_title),
169 "profile-update-interval": str(subscription_settings.update_interval),
170 "subscription-userinfo": "; ".join(
171 f"{key}={val}"
172 for key, val in get_subscription_user_info(user).items()
173 if val is not None
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 )