1import re
2
3from fastapi import APIRouter
4from fastapi import Header, HTTPException, Path, Request, Response
5from fastapi.responses import HTMLResponse
6
7from app.db import crud
8from app.dependencies import DBDep, SubUserDep, StartDateDep, EndDateDep
9from app.models.user import UserResponse
10from app.templates import render_template
11from app.utils.share import encode_title, generate_subscription, generate_v2ray_links
12from config import (
13 SUB_PROFILE_TITLE,
14 SUB_SUPPORT_URL,
15 SUB_UPDATE_INTERVAL,
16 SUBSCRIPTION_PAGE_TEMPLATE,
17)
18
19router = APIRouter(prefix="/sub", tags=["Subscription"])
20
21
22def get_subscription_user_info(user: UserResponse) -> dict:
23 return {
24 "upload": 0,
25 "download": user.used_traffic,
26 "total": user.data_limit,
27 "expire": user.expire,
28 }
29
30
31@router.get("/{username}/{key}")
32def user_subscription(
33 db_user: SubUserDep,
34 request: Request,
35 db: DBDep,
36 user_agent: str = Header(default=""),
37):
38 """
39 Subscription link, V2ray and Clash supported
40 """
41 accept_header = request.headers.get("Accept", "")
42
43 user: UserResponse = UserResponse.model_validate(db_user)
44
45 if "text/html" in accept_header:
46 links = generate_v2ray_links(
47 user.inbounds,
48 user.key,
49 user.model_dump(
50 exclude={"subscription_url", "services", "inbounds", "links"}
51 ),
52 )
53 return HTMLResponse(
54 render_template(SUBSCRIPTION_PAGE_TEMPLATE, {"user": user, "links": links})
55 )
56
57 response_headers = {
58 "content-disposition": f'attachment; filename="{user.username}"',
59 "profile-web-page-url": str(request.url),
60 "support-url": SUB_SUPPORT_URL,
61 "profile-title": encode_title(SUB_PROFILE_TITLE),
62 "profile-update-interval": SUB_UPDATE_INTERVAL,
63 "subscription-userinfo": "; ".join(
64 f"{key}={val}"
65 for key, val in get_subscription_user_info(user).items()
66 if val is not None
67 ),
68 }
69
70 crud.update_user_sub(db, db_user, user_agent)
71
72 if re.match("^([Cc]lash-verge|[Cc]lash-?[Mm]eta)", user_agent):
73 conf = generate_subscription(
74 user=user, config_format="clash-meta", as_base64=False
75 )
76 return Response(content=conf, media_type="text/yaml", headers=response_headers)
77
78 elif re.match("^([Cc]lash|[Ss]tash)", user_agent):
79 conf = generate_subscription(user=user, config_format="clash", as_base64=False)
80 return Response(content=conf, media_type="text/yaml", headers=response_headers)
81
82 elif re.match("^(SFA|SFI|SFM|SFT)", user_agent):
83 conf = generate_subscription(
84 user=user, config_format="sing-box", as_base64=False
85 )
86 return Response(
87 content=conf, media_type="application/json", headers=response_headers
88 )
89
90 elif re.match("^(SS|SSR|SSD|SSS|Outline|Shadowsocks|SSconf)", user_agent):
91 conf = generate_subscription(
92 user=user, config_format="outline", as_base64=False
93 )
94 return Response(
95 content=conf, media_type="application/json", headers=response_headers
96 )
97
98 else:
99 conf = generate_subscription(user=user, config_format="v2ray", as_base64=True)
100 return Response(content=conf, media_type="text/plain", headers=response_headers)
101
102
103@router.get("/{username}/{key}/info", response_model=UserResponse)
104def user_subscription_info(db_user: SubUserDep):
105 return db_user
106
107
108@router.get("/{username}/{key}/usage")
109def user_get_usage(
110 db_user: SubUserDep, db: DBDep, start_date: StartDateDep, end_date: EndDateDep
111):
112 usages = crud.get_user_usages(db, db_user, start_date, end_date)
113
114 return {"usages": usages, "username": db_user.username}
115
116
117@router.get("/{username}/{key}/{client_type}")
118def user_subscription_with_client_type(
119 db_user: SubUserDep,
120 request: Request,
121 client_type: str = Path(..., regex="sing-box|clash-meta|clash|outline|v2ray"),
122):
123 """
124 Subscription link, v2ray, clash, sing-box, outline and clash-meta supported
125 """
126
127 user: UserResponse = UserResponse.model_validate(db_user)
128
129 response_headers = {
130 "content-disposition": f'attachment; filename="{user.username}"',
131 "profile-web-page-url": str(request.url),
132 "support-url": SUB_SUPPORT_URL,
133 "profile-title": encode_title(SUB_PROFILE_TITLE),
134 "profile-update-interval": SUB_UPDATE_INTERVAL,
135 "subscription-userinfo": "; ".join(
136 f"{key}={val}"
137 for key, val in get_subscription_user_info(user).items()
138 if val is not None
139 ),
140 }
141
142 if client_type == "clash-meta":
143 conf = generate_subscription(
144 user=user, config_format="clash-meta", as_base64=False
145 )
146 return Response(content=conf, media_type="text/yaml", headers=response_headers)
147
148 elif client_type == "sing-box":
149 conf = generate_subscription(
150 user=user, config_format="sing-box", as_base64=False
151 )
152 return Response(
153 content=conf, media_type="application/json", headers=response_headers
154 )
155
156 elif client_type == "clash":
157 conf = generate_subscription(user=user, config_format="clash", as_base64=False)
158 return Response(content=conf, media_type="text/yaml", headers=response_headers)
159
160 elif client_type == "v2ray":
161 conf = generate_subscription(user=user, config_format="v2ray", as_base64=True)
162 return Response(content=conf, media_type="text/plain", headers=response_headers)
163
164 elif client_type == "outline":
165 conf = generate_subscription(
166 user=user, config_format="outline", as_base64=False
167 )
168 return Response(
169 content=conf, media_type="application/json", headers=response_headers
170 )
171
172 else:
173 raise HTTPException(status_code=400, detail="Invalid subscription type")