import os import re import requests import threading import time as _time from distutils.version import LooseVersion # ─── Routing Resolver ─────────────────────────────────────────────────────────── # Получает .DEEPLINK из GitHub с кешем 10 минут и thread-safe блокировкой. # # Переменные окружения: # CUSTOM_ROUTING_SOURCE = default | jsonsub | whitelist | custom # CUSTOM_ROUTING_CUSTOM = <ваша happ:// или incy:// ссылка> # ───────────────────────────────────────────────────────────────────────────────── _ROUTING_URLS = { "default": "https://raw.githubusercontent.com/kaiketsu-moe/krouting/main/HAPP/DEFAULT.DEEPLINK", "jsonsub": "https://raw.githubusercontent.com/kaiketsu-moe/krouting/main/HAPP/JSONSUB.DEEPLINK", "whitelist": "https://raw.githubusercontent.com/kaiketsu-moe/krouting/main/HAPP/WHITELIST.DEEPLINK", } class _RoutingResolver: def __init__(self, default_source: str): self._lock = threading.Lock() self._value = "" self._fetched_at = 0.0 self._last_fail = 0.0 self._source = os.environ.get("CUSTOM_ROUTING_SOURCE", default_source).strip().lower() self._custom = os.environ.get("CUSTOM_ROUTING_CUSTOM", "").strip() def get(self) -> str: if self._source == "custom": return self._custom url = _ROUTING_URLS.get(self._source) if not url: return self._custom now = _time.monotonic() if self._value and (now - self._fetched_at) < 600: return self._value if self._last_fail and (now - self._last_fail) < 30: return self._value with self._lock: now = _time.monotonic() if self._value and (now - self._fetched_at) < 600: return self._value try: r = requests.get(url, timeout=4) r.raise_for_status() self._value = r.text.strip() self._fetched_at = now self._last_fail = 0.0 except Exception: self._last_fail = now return self._value routing_resolver = _RoutingResolver("default") from fastapi import APIRouter, Depends, Header, Path, Request, Response from fastapi.responses import HTMLResponse from app.db import Session, crud, get_db from app.dependencies import get_validated_sub, validate_dates from app.models.user import SubscriptionUserResponse, UserResponse from app.subscription.share import encode_title, generate_subscription from app.templates import render_template from config import ( SUB_PROFILE_TITLE, SUB_SUPPORT_URL, SUB_UPDATE_INTERVAL, SUBSCRIPTION_PAGE_TEMPLATE, USE_CUSTOM_JSON_DEFAULT, USE_CUSTOM_JSON_FOR_HAPP, USE_CUSTOM_JSON_FOR_STREISAND, USE_CUSTOM_JSON_FOR_V2RAYN, USE_CUSTOM_JSON_FOR_V2RAYNG, XRAY_SUBSCRIPTION_PATH, ) client_config = { "clash-meta": {"config_format": "clash-meta", "media_type": "text/yaml", "as_base64": False, "reverse": False}, "sing-box": {"config_format": "sing-box", "media_type": "application/json", "as_base64": False, "reverse": False}, "clash": {"config_format": "clash", "media_type": "text/yaml", "as_base64": False, "reverse": False}, "v2ray": {"config_format": "v2ray", "media_type": "text/plain", "as_base64": True, "reverse": False}, "outline": {"config_format": "outline", "media_type": "application/json", "as_base64": False, "reverse": False}, "v2ray-json": {"config_format": "v2ray-json", "media_type": "application/json", "as_base64": False, "reverse": False} } router = APIRouter(tags=['Subscription'], prefix=f'/{XRAY_SUBSCRIPTION_PATH}') def get_subscription_user_info(user: UserResponse) -> dict: return { "upload": 0, "download": user.used_traffic, "total": user.data_limit if user.data_limit is not None else 0, "expire": user.expire if user.expire is not None else 0, } @router.get("/{token}/") @router.get("/{token}", include_in_schema=False) def user_subscription( request: Request, db: Session = Depends(get_db), dbuser: UserResponse = Depends(get_validated_sub), user_agent: str = Header(default="") ): user: UserResponse = UserResponse.model_validate(dbuser) accept_header = request.headers.get("Accept", "") if "text/html" in accept_header: return HTMLResponse( render_template( SUBSCRIPTION_PAGE_TEMPLATE, {"user": user} ) ) crud.update_user_sub(db, dbuser, user_agent) response_headers = { "content-disposition": f'attachment; filename="{user.username}"', "profile-web-page-url": str(request.url), "support-url": SUB_SUPPORT_URL, "profile-title": encode_title(SUB_PROFILE_TITLE), "profile-update-interval": SUB_UPDATE_INTERVAL, "subscription-userinfo": "; ".join( f"{key}={val}" for key, val in get_subscription_user_info(user).items() ) } _routing = routing_resolver.get() if _routing: response_headers["routing"] = _routing response_headers["routing-enable"] = "true" if re.match(r'^([Cc]lash-verge|[Cc]lash[-\.]?[Mm]eta|[Ff][Ll][Cc]lash|[Mm]ihomo)', user_agent): conf = generate_subscription(user=user, config_format="clash-meta", as_base64=False, reverse=False) return Response(content=conf, media_type="text/yaml", headers=response_headers) elif re.match(r'^([Cc]lash|[Ss]tash)', user_agent): conf = generate_subscription(user=user, config_format="clash", as_base64=False, reverse=False) return Response(content=conf, media_type="text/yaml", headers=response_headers) elif re.match(r'^(SFA|SFI|SFM|SFT|[Kk]aring|[Hh]iddify[Nn]ext)', user_agent): conf = generate_subscription(user=user, config_format="sing-box", as_base64=False, reverse=False) return Response(content=conf, media_type="application/json", headers=response_headers) elif re.match(r'^(SS|SSR|SSD|SSS|Outline|Shadowsocks|SSconf)', user_agent): conf = generate_subscription(user=user, config_format="outline", as_base64=False, reverse=False) return Response(content=conf, media_type="application/json", headers=response_headers) elif (USE_CUSTOM_JSON_DEFAULT or USE_CUSTOM_JSON_FOR_V2RAYN) and re.match(r'^v2rayN/(\d+\.\d+)', user_agent): version_str = re.match(r'^v2rayN/(\d+\.\d+)', user_agent).group(1) if LooseVersion(version_str) >= LooseVersion("6.40"): conf = generate_subscription(user=user, config_format="v2ray-json", as_base64=False, reverse=False) return Response(content=conf, media_type="application/json", headers=response_headers) else: conf = generate_subscription(user=user, config_format="v2ray", as_base64=True, reverse=False) return Response(content=conf, media_type="text/plain", headers=response_headers) elif (USE_CUSTOM_JSON_DEFAULT or USE_CUSTOM_JSON_FOR_V2RAYNG) and re.match(r'^v2rayNG/(\d+\.\d+\.\d+)', user_agent): version_str = re.match(r'^v2rayNG/(\d+\.\d+\.\d+)', user_agent).group(1) if LooseVersion(version_str) >= LooseVersion("1.8.29"): conf = generate_subscription(user=user, config_format="v2ray-json", as_base64=False, reverse=False) return Response(content=conf, media_type="application/json", headers=response_headers) elif LooseVersion(version_str) >= LooseVersion("1.8.18"): conf = generate_subscription(user=user, config_format="v2ray-json", as_base64=False, reverse=True) return Response(content=conf, media_type="application/json", headers=response_headers) else: conf = generate_subscription(user=user, config_format="v2ray", as_base64=True, reverse=False) return Response(content=conf, media_type="text/plain", headers=response_headers) elif re.match(r'^[Ss]treisand', user_agent): if USE_CUSTOM_JSON_DEFAULT or USE_CUSTOM_JSON_FOR_STREISAND: conf = generate_subscription(user=user, config_format="v2ray-json", as_base64=False, reverse=False) return Response(content=conf, media_type="application/json", headers=response_headers) else: conf = generate_subscription(user=user, config_format="v2ray", as_base64=True, reverse=False) return Response(content=conf, media_type="text/plain", headers=response_headers) elif (USE_CUSTOM_JSON_DEFAULT or USE_CUSTOM_JSON_FOR_HAPP) and re.match(r'^Happ/(\d+\.\d+\.\d+)', user_agent): version_str = re.match(r'^Happ/(\d+\.\d+\.\d+)', user_agent).group(1) if LooseVersion(version_str) >= LooseVersion("1.63.1"): conf = generate_subscription(user=user, config_format="v2ray-json", as_base64=False, reverse=False) return Response(content=conf, media_type="application/json", headers=response_headers) else: conf = generate_subscription(user=user, config_format="v2ray", as_base64=True, reverse=False) return Response(content=conf, media_type="text/plain", headers=response_headers) else: conf = generate_subscription(user=user, config_format="v2ray", as_base64=True, reverse=False) return Response(content=conf, media_type="text/plain", headers=response_headers) @router.get("/{token}/info", response_model=SubscriptionUserResponse) def user_subscription_info( dbuser: UserResponse = Depends(get_validated_sub), ): return dbuser @router.get("/{token}/usage") def user_get_usage( dbuser: UserResponse = Depends(get_validated_sub), start: str = "", end: str = "", db: Session = Depends(get_db) ): start, end = validate_dates(start, end) usages = crud.get_user_usages(db, dbuser, start, end) return {"usages": usages, "username": dbuser.username} @router.get("/{token}/{client_type}") def user_subscription_with_client_type( request: Request, dbuser: UserResponse = Depends(get_validated_sub), client_type: str = Path(..., regex="sing-box|clash-meta|clash|outline|v2ray|v2ray-json"), db: Session = Depends(get_db), user_agent: str = Header(default="") ): user: UserResponse = UserResponse.model_validate(dbuser) response_headers = { "content-disposition": f'attachment; filename="{user.username}"', "profile-web-page-url": str(request.url), "support-url": SUB_SUPPORT_URL, "profile-title": encode_title(SUB_PROFILE_TITLE), "profile-update-interval": SUB_UPDATE_INTERVAL, "subscription-userinfo": "; ".join( f"{key}={val}" for key, val in get_subscription_user_info(user).items() ) } _routing = routing_resolver.get() if _routing: response_headers["routing"] = _routing response_headers["routing-enable"] = "true" config = client_config.get(client_type) conf = generate_subscription(user=user, config_format=config["config_format"], as_base64=config["as_base64"], reverse=config["reverse"]) return Response(content=conf, media_type=config["media_type"], headers=response_headers)