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