import os import re import requests import threading import time as _time from collections import defaultdict # ─── 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 from fastapi import Header, HTTPException, Path, Request, Response from starlette.responses import HTMLResponse from app.db import crud from app.db.models import Settings from app.dependencies import DBDep, SubUserDep, StartDateDep, EndDateDep from app.models.settings import SubscriptionSettings from app.models.system import TrafficUsageSeries from app.models.user import UserResponse from app.utils.share import ( encode_title, generate_subscription, generate_subscription_template, ) router = APIRouter(prefix="/sub", tags=["Subscription"]) config_mimetype = defaultdict( lambda: "text/plain", { "links": "text/plain", "base64-links": "text/plain", "sing-box": "application/json", "xray": "application/json", "clash": "text/yaml", "clash-meta": "text/yaml", "template": "text/html", "block": "text/plain", }, ) def get_subscription_user_info(user: UserResponse) -> dict: return { "upload": 0, "download": user.used_traffic, "total": user.data_limit or 0, "expire": ( int(user.expire_date.timestamp()) if user.expire_strategy == "fixed_date" else 0 ), } @router.get("/{username}/{key}") def user_subscription( db_user: SubUserDep, request: Request, db: DBDep, user_agent: str = Header(default=""), ): user: UserResponse = UserResponse.model_validate(db_user) crud.update_user_sub(db, db_user, user_agent) subscription_settings = SubscriptionSettings.model_validate( db.query(Settings.subscription).first()[0] ) if ( subscription_settings.template_on_acceptance and "text/html" in request.headers.get("Accept", []) ): try: template_content = generate_subscription_template(db, db_user, subscription_settings) except TypeError: template_content = generate_subscription_template(db_user, subscription_settings) return HTMLResponse(template_content) response_headers = { "content-disposition": f'attachment; filename="{user.username}"', "profile-web-page-url": str(request.url), "support-url": subscription_settings.support_link, "profile-title": encode_title(subscription_settings.profile_title), "profile-update-interval": str(subscription_settings.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" for rule in subscription_settings.rules: if re.match(rule.pattern, user_agent): if rule.result.value == "template": try: template_content = generate_subscription_template(db, db_user, subscription_settings) except TypeError: template_content = generate_subscription_template(db_user, subscription_settings) return HTMLResponse(template_content) elif rule.result.value == "block": raise HTTPException(404) elif rule.result.value == "base64-links": b64 = True config_format = "links" else: b64 = False config_format = rule.result.value try: conf = generate_subscription( db, user=db_user, config_format=config_format, as_base64=b64, use_placeholder=not user.is_active and subscription_settings.placeholder_if_disabled, placeholder_remark=subscription_settings.placeholder_remark, shuffle=subscription_settings.shuffle_configs, ) except TypeError: conf = generate_subscription( user=db_user, config_format=config_format, as_base64=b64, use_placeholder=not user.is_active and subscription_settings.placeholder_if_disabled, placeholder_remark=subscription_settings.placeholder_remark, shuffle=subscription_settings.shuffle_configs, ) return Response( content=conf, media_type=config_mimetype[rule.result], headers=response_headers, ) @router.get("/{username}/{key}/info", response_model=UserResponse) def user_subscription_info(db_user: SubUserDep): return db_user @router.get("/{username}/{key}/usage", response_model=TrafficUsageSeries) def user_get_usage( db_user: SubUserDep, db: DBDep, start_date: StartDateDep, end_date: EndDateDep, ): per_day = (end_date - start_date).total_seconds() > 3 * 86400 return crud.get_user_total_usage( db, db_user, start_date, end_date, per_day=per_day ) client_type_mime_type = { "sing-box": "application/json", "wireguard": "application/json", "clash-meta": "text/yaml", "clash": "text/yaml", "xray": "application/json", "v2ray": "text/plain", "links": "text/plain", } @router.get("/{username}/{key}/{client_type}") def user_subscription_with_client_type( db: DBDep, db_user: SubUserDep, request: Request, client_type: str = Path( regex="^(sing-box|clash-meta|clash|xray|v2ray|links|wireguard)$" ), ): user: UserResponse = UserResponse.model_validate(db_user) subscription_settings = SubscriptionSettings.model_validate( db.query(Settings.subscription).first()[0] ) response_headers = { "content-disposition": f'attachment; filename="{user.username}"', "profile-web-page-url": str(request.url), "support-url": subscription_settings.support_link, "profile-title": encode_title(subscription_settings.profile_title), "profile-update-interval": str(subscription_settings.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" try: conf = generate_subscription( db, user=db_user, config_format="links" if client_type == "v2ray" else client_type, as_base64=client_type == "v2ray", use_placeholder=not user.is_active and subscription_settings.placeholder_if_disabled, placeholder_remark=subscription_settings.placeholder_remark, shuffle=subscription_settings.shuffle_configs, ) except TypeError: conf = generate_subscription( user=db_user, config_format="links" if client_type == "v2ray" else client_type, as_base64=client_type == "v2ray", use_placeholder=not user.is_active and subscription_settings.placeholder_if_disabled, placeholder_remark=subscription_settings.placeholder_remark, shuffle=subscription_settings.shuffle_configs, ) return Response( content=conf, media_type=client_type_mime_type[client_type], headers=response_headers, )