codey/api
1
0
Fork 0
Code Issues 1 Pull Requests Activity

rm test.conf

Browse Source
This commit is contained in:
codey
2025-12-18 07:30:39 -05:00
parent 041de95698
commit bc8b407a91
12 changed files with 348 additions and 303 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
base.py
View File

@@ -141,7 +141,8 @@ End Blacklisted Routes
Actionable Routes
"""
_routes.update({
_routes.update(
{
"randmsg": importlib.import_module("endpoints.rand_msg").RandMsg(
app, util, constants
),
@@ -158,7 +159,8 @@ _routes.update({
"lighting": importlib.import_module("endpoints.lighting").Lighting(
app, util, constants
),
})
}
)
# Misc endpoint depends on radio endpoint instance
radio_endpoint = _routes.get("radio")

100
endpoints/lighting.py
View File

@@ -42,6 +42,7 @@ logger = logging.getLogger(__name__)
@dataclass
class CyncConnectionState:
"""Track the state of our Cync connection."""
session: Optional[aiohttp.ClientSession] = None
auth: Optional[Auth] = None
cync_api: Optional[Cync] = None
@@ -73,8 +74,7 @@ class Lighting:
# Redis for state persistence
self.redis_client = redis.Redis(
password=private.REDIS_PW,
decode_responses=True
password=private.REDIS_PW, decode_responses=True
)
self.lighting_key = "lighting:state"
@@ -208,7 +208,7 @@ class Lighting:
if self._state.cync_api:
try:
# pycync's command client has a shut_down method
client = getattr(self._state.cync_api, '_command_client', None)
client = getattr(self._state.cync_api, "_command_client", None)
if client:
await client.shut_down()
except Exception as e:
@@ -242,7 +242,7 @@ class Lighting:
if not self._state.user:
return True
expires_at = getattr(self._state.user, 'expires_at', 0)
expires_at = getattr(self._state.user, "expires_at", 0)
return expires_at < (time.time() + self.TOKEN_EXPIRY_BUFFER)
async def _wait_for_connection_ready(self) -> None:
@@ -255,19 +255,19 @@ class Lighting:
if not self._state.cync_api:
raise RuntimeError("Cync API not initialized")
client = getattr(self._state.cync_api, '_command_client', None)
client = getattr(self._state.cync_api, "_command_client", None)
if not client:
logger.warning("Could not access command client")
return
tcp_manager = getattr(client, '_tcp_manager', None)
tcp_manager = getattr(client, "_tcp_manager", None)
if not tcp_manager:
logger.warning("Could not access TCP manager")
return
# Wait for login to be acknowledged
start = time.time()
while not getattr(tcp_manager, '_login_acknowledged', False):
while not getattr(tcp_manager, "_login_acknowledged", False):
if time.time() - start > self.CONNECTION_READY_TIMEOUT:
raise TimeoutError("Timed out waiting for Cync login acknowledgment")
await asyncio.sleep(0.2)
@@ -341,8 +341,7 @@ class Lighting:
try:
loop = asyncio.get_event_loop()
twofa_code = await asyncio.wait_for(
loop.run_in_executor(None, input, "2FA Code: "),
timeout=60.0
loop.run_in_executor(None, input, "2FA Code: "), timeout=60.0
)
twofa_code = twofa_code.strip()
except asyncio.TimeoutError:
@@ -365,7 +364,7 @@ class Lighting:
def _is_user_token_expired(self, user: User) -> bool:
"""Check if a user's token is expired."""
expires_at = getattr(user, 'expires_at', 0)
expires_at = getattr(user, "expires_at", 0)
return expires_at < (time.time() + self.TOKEN_EXPIRY_BUFFER)
def _load_cached_token(self) -> Optional[User]:
@@ -374,15 +373,15 @@ class Lighting:
if not os.path.exists(self.token_cache_path):
return None
with open(self.token_cache_path, 'r') as f:
with open(self.token_cache_path, "r") as f:
data = json.load(f)
return User(
access_token=data['access_token'],
refresh_token=data['refresh_token'],
authorize=data['authorize'],
user_id=data['user_id'],
expires_at=data['expires_at'],
access_token=data["access_token"],
refresh_token=data["refresh_token"],
authorize=data["authorize"],
user_id=data["user_id"],
expires_at=data["expires_at"],
)
except Exception as e:
logger.warning(f"Failed to load cached token: {e}")
@@ -392,13 +391,13 @@ class Lighting:
"""Save authentication token to disk."""
try:
data = {
'access_token': user.access_token,
'refresh_token': user.refresh_token,
'authorize': user.authorize,
'user_id': user.user_id,
'expires_at': user.expires_at,
"access_token": user.access_token,
"refresh_token": user.refresh_token,
"authorize": user.authorize,
"user_id": user.user_id,
"expires_at": user.expires_at,
}
with open(self.token_cache_path, 'w') as f:
with open(self.token_cache_path, "w") as f:
json.dump(data, f)
logger.debug("Saved Cync token to disk")
except Exception as e:
@@ -450,12 +449,12 @@ class Lighting:
raise RuntimeError("No devices found")
device = next(
(d for d in devices if getattr(d, 'name', None) == self.cync_device_name),
None
(d for d in devices if getattr(d, "name", None) == self.cync_device_name),
None,
)
if not device:
available = [getattr(d, 'name', 'unnamed') for d in devices]
available = [getattr(d, "name", "unnamed") for d in devices]
raise RuntimeError(
f"Device '{self.cync_device_name}' not found. Available: {available}"
)
@@ -506,7 +505,9 @@ class Lighting:
async def get_lighting_state(self, user=Depends(get_current_user)) -> JSONResponse:
"""Get the current lighting state from Redis."""
if "lighting" not in user.get("roles", []) and "admin" not in user.get("roles", []):
if "lighting" not in user.get("roles", []) and "admin" not in user.get(
"roles", []
):
raise HTTPException(status_code=403, detail="Insufficient permissions")
try:
state = self.redis_client.get(self.lighting_key)
@@ -514,27 +515,34 @@ class Lighting:
return JSONResponse(content=json.loads(str(state)))
# Default state
return JSONResponse(content={
return JSONResponse(
content={
"power": "off",
"brightness": 50,
"color": {"r": 255, "g": 255, "b": 255},
})
}
)
except Exception as e:
logger.error(f"Error getting lighting state: {e}")
raise HTTPException(status_code=500, detail="Internal server error")
async def set_lighting_state(self, request: Request,
user=Depends(get_current_user)) -> JSONResponse:
async def set_lighting_state(
self, request: Request, user=Depends(get_current_user)
) -> JSONResponse:
"""Set the lighting state and apply to Cync device."""
try:
if "lighting" not in user.get("roles", []) and "admin" not in user.get("roles", []):
if "lighting" not in user.get("roles", []) and "admin" not in user.get(
"roles", []
):
raise HTTPException(status_code=403, detail="Insufficient permissions")
state = await request.json()
logger.info(f"Lighting request: {state}")
# Validate
if not isinstance(state, dict):
raise HTTPException(status_code=400, detail="State must be a JSON object")
raise HTTPException(
status_code=400, detail="State must be a JSON object"
)
power, brightness, rgb = self._parse_state(state)
@@ -544,11 +552,15 @@ class Lighting:
# Apply to device with retries
await self._apply_state_with_retry(power, brightness, rgb)
logger.info(f"Successfully applied state: power={power}, brightness={brightness}, rgb={rgb}")
return JSONResponse(content={
logger.info(
f"Successfully applied state: power={power}, brightness={brightness}, rgb={rgb}"
)
return JSONResponse(
content={
"message": "Lighting state updated",
"state": state,
})
}
)
except HTTPException:
raise
@@ -568,13 +580,19 @@ class Lighting:
if "brightness" in state:
brightness = state["brightness"]
if not isinstance(brightness, (int, float)) or not (0 <= brightness <= 100):
raise HTTPException(status_code=400, detail=f"Invalid brightness: {brightness}")
raise HTTPException(
status_code=400, detail=f"Invalid brightness: {brightness}"
)
brightness = int(brightness)
# Color
rgb = None
color = state.get("color")
if color and isinstance(color, dict) and all(k in color for k in ("r", "g", "b")):
if (
color
and isinstance(color, dict)
and all(k in color for k in ("r", "g", "b"))
):
rgb = (color["r"], color["g"], color["b"])
elif all(k in state for k in ("red", "green", "blue")):
rgb = (state["red"], state["green"], state["blue"])
@@ -582,7 +600,9 @@ class Lighting:
if rgb:
for i, name in enumerate(("red", "green", "blue")):
if not isinstance(rgb[i], int) or not (0 <= rgb[i] <= 255):
raise HTTPException(status_code=400, detail=f"Invalid {name}: {rgb[i]}")
raise HTTPException(
status_code=400, detail=f"Invalid {name}: {rgb[i]}"
)
return power, brightness, rgb
@@ -615,7 +635,9 @@ class Lighting:
except Exception as e:
last_error = e
logger.warning(f"Error on attempt {attempt + 1}: {type(e).__name__}: {e}")
logger.warning(
f"Error on attempt {attempt + 1}: {type(e).__name__}: {e}"
)
# Wait before retry (exponential backoff)
if attempt < self.MAX_RETRIES - 1:

12
endpoints/lyric_search.py
View File

@@ -96,9 +96,11 @@ class LyricSearch(FastAPI):
handler,
methods=["POST"],
include_in_schema=_schema_include,
dependencies=[Depends(RateLimiter(times=times, seconds=seconds))]
dependencies=(
[Depends(RateLimiter(times=times, seconds=seconds))]
if not endpoint == "typeahead/lyrics"
else None,
else None
),
)
async def typeahead_handler(self, data: ValidTypeAheadRequest) -> JSONResponse:
@@ -243,9 +245,9 @@ class LyricSearch(FastAPI):
if i + line_count <= len(lyric_lines):
# Combine consecutive lines with space separator
combined_lines = []
line_positions: list[
tuple[int, int]
] = [] # Track where each line starts in combined text
line_positions: list[tuple[int, int]] = (
[]
) # Track where each line starts in combined text
combined_text_parts: list[str] = []
for j in range(line_count):

31
endpoints/rip.py
View File

@@ -75,7 +75,11 @@ class RIP(FastAPI):
app.add_api_route(
f"/{endpoint}",
handler,
methods=["GET"] if endpoint not in ("trip/bulk_fetch", "trip/auth/check") else ["POST"],
methods=(
["GET"]
if endpoint not in ("trip/bulk_fetch", "trip/auth/check")
else ["POST"]
),
include_in_schema=False,
dependencies=dependencies,
)
@@ -131,9 +135,11 @@ class RIP(FastAPI):
"started_at": job.started_at,
"ended_at": job.ended_at,
"progress": progress,
"tracks": f"{succeeded_tracks} / {tracks_in}"
"tracks": (
f"{succeeded_tracks} / {tracks_in}"
if isinstance(tracks_in, int)
else tracks_out,
else tracks_out
),
"target": job.meta.get("target"),
"quality": job.meta.get("quality", "Unknown"),
}
@@ -434,7 +440,9 @@ class RIP(FastAPI):
- **JSONResponse**: Contains device_code and verification_url.
"""
try:
if "trip" not in user.get("roles", []) and "admin" not in user.get("roles", []):
if "trip" not in user.get("roles", []) and "admin" not in user.get(
"roles", []
):
raise HTTPException(status_code=403, detail="Insufficient permissions")
device_code, verification_url = await self.trip_util.start_device_auth()
# Store device code for this session
@@ -469,7 +477,9 @@ class RIP(FastAPI):
device_code = self._pending_device_codes.get(user.get("sub", "default"))
if not device_code:
return JSONResponse(
content={"error": "No pending authorization. Call /trip/auth/start first."},
content={
"error": "No pending authorization. Call /trip/auth/start first."
},
status_code=400,
)
@@ -479,11 +489,18 @@ class RIP(FastAPI):
# Clear the pending code
self._pending_device_codes.pop(user.get("sub", "default"), None)
return JSONResponse(
content={"success": True, "message": "Tidal authorization complete!"}
content={
"success": True,
"message": "Tidal authorization complete!",
}
)
elif error == "pending":
return JSONResponse(
content={"success": False, "pending": True, "message": "Waiting for user to authorize..."}
content={
"success": False,
"pending": True,
"message": "Waiting for user to authorize...",
}
)
else:
return JSONResponse(

1
endpoints/yt.py
View File

@@ -6,6 +6,7 @@ from typing import Optional, Union
from utils.yt_utils import sign_video_id
from .constructors import ValidYTSearchRequest
class YT(FastAPI):
"""
YT Endpoints

4
lyric_search/models.py
View File

@@ -99,7 +99,9 @@ POSTGRES_PASSWORD = os.getenv("POSTGRES_PASSWORD", "")
# URL-encode the password to handle special characters
encoded_password = urllib.parse.quote_plus(POSTGRES_PASSWORD)
DATABASE_URL: str = f"postgresql+asyncpg://{POSTGRES_USER}:{encoded_password}@{POSTGRES_HOST}:{POSTGRES_PORT}/{POSTGRES_DB}"
DATABASE_URL: str = (
f"postgresql+asyncpg://{POSTGRES_USER}:{encoded_password}@{POSTGRES_HOST}:{POSTGRES_PORT}/{POSTGRES_DB}"
)
async_engine: AsyncEngine = create_async_engine(
DATABASE_URL, pool_size=20, max_overflow=10, pool_pre_ping=True, echo=False
)

16
lyric_search/sources/cache.py
View File

@@ -91,8 +91,10 @@ class Cache:
logging.debug(
"Checking whether %s is already stored", artistsong.replace("\n", " - ")
)
check_query: str = 'SELECT id, artist, song FROM lyrics WHERE editdist3((lower(artist) || " " || lower(song)), (? || " " || ?))\
check_query: str = (
'SELECT id, artist, song FROM lyrics WHERE editdist3((lower(artist) || " " || lower(song)), (? || " " || ?))\
<= 410 ORDER BY editdist3((lower(artist) || " " || lower(song)), ?) ASC LIMIT 1'
)
artistsong_split = artistsong.split("\n", maxsplit=1)
artist = artistsong_split[0].lower()
song = artistsong_split[1].lower()
@@ -213,10 +215,8 @@ class Cache:
lyrics = regex.sub(r"(<br>|\n|\r\n)", " / ", lyr_result.lyrics.strip())
lyrics = regex.sub(r"\s{2,}", " ", lyrics)
insert_query = (
"INSERT INTO lyrics (src, date_retrieved, artist, song, artistsong, confidence, lyrics)\
insert_query = "INSERT INTO lyrics (src, date_retrieved, artist, song, artistsong, confidence, lyrics)\
VALUES(?, ?, ?, ?, ?, ?, ?)"
)
params = (
lyr_result.src,
time.time(),
@@ -260,8 +260,10 @@ class Cache:
if artist == "!" and song == "!":
random_search = True
search_query: str = "SELECT id, artist, song, lyrics, src, confidence\
search_query: str = (
"SELECT id, artist, song, lyrics, src, confidence\
FROM lyrics ORDER BY RANDOM() LIMIT 1"
)
logging.info("Searching %s - %s on %s", artist, song, self.label)
@@ -320,9 +322,11 @@ class Cache:
self.cache_pre_query
) as _db_cursor:
if not random_search:
search_query: str = 'SELECT id, artist, song, lyrics, src, confidence FROM lyrics\
search_query: str = (
'SELECT id, artist, song, lyrics, src, confidence FROM lyrics\
WHERE editdist3((lower(artist) || " " || lower(song)), (? || " " || ?))\
<= 410 ORDER BY editdist3((lower(artist) || " " || lower(song)), ?) ASC LIMIT 10'
)
search_params: tuple = (
artist.strip(),
song.strip(),

4
lyric_search/utils.py
View File

@@ -111,8 +111,7 @@ class DataUtils:
"""
def __init__(self) -> None:
self.lrc_regex = (
regex.compile( # capture mm:ss and optional .xxx, then the lyric text
self.lrc_regex = regex.compile( # capture mm:ss and optional .xxx, then the lyric text
r"""
\[ # literal “[”
( # 1st (and only) capture group:
@@ -126,7 +125,6 @@ class DataUtils:
""",
regex.VERBOSE,
)
)
self.scrub_regex_1: Pattern = regex.compile(r"(\[.*?\])(\s){0,}(\:){0,1}")
self.scrub_regex_2: Pattern = regex.compile(
r"(\d?)(Embed\b)", flags=regex.IGNORECASE

4
utils/meme_util.py
View File

@@ -127,7 +127,9 @@ class MemeUtil:
db_conn.row_factory = sqlite3.Row
rows_per_page: int = 10
offset: int = (page - 1) * rows_per_page
query: str = "SELECT id, timestamp FROM memes ORDER BY timestamp DESC LIMIT 10 OFFSET ?"
query: str = (
"SELECT id, timestamp FROM memes ORDER BY timestamp DESC LIMIT 10 OFFSET ?"
)
async with await db_conn.execute(query, (offset,)) as db_cursor:
results = await db_cursor.fetchall()
for result in results:

75
utils/sr_wrapper.py
View File

@@ -15,11 +15,11 @@ import time
# Monkey-patch streamrip's Tidal client credentials BEFORE importing TidalClient
import streamrip.client.tidal as _tidal_module # type: ignore # noqa: E402
_tidal_module.CLIENT_ID = "fX2JxdmntZWK0ixT"
_tidal_module.CLIENT_SECRET = "1Nn9AfDAjxrgJFJbKNWLeAyKGVGmINuXPPLHVXAvxAg="
_tidal_module.AUTH = aiohttp.BasicAuth(
login=_tidal_module.CLIENT_ID,
password=_tidal_module.CLIENT_SECRET
login=_tidal_module.CLIENT_ID, password=_tidal_module.CLIENT_SECRET
)
from streamrip.client import TidalClient # type: ignore # noqa: E402
@@ -157,7 +157,9 @@ class SRUtil:
await self._login_and_persist(force=True)
logging.info("Tidal keepalive: Session refresh successful")
except Exception as e:
logging.warning("Tidal keepalive: Session refresh failed: %s", e)
logging.warning(
"Tidal keepalive: Session refresh failed: %s", e
)
continue
# Make a lightweight API call to keep the session alive
@@ -195,7 +197,9 @@ class SRUtil:
tidal.access_token = cached.get("access_token", "")
tidal.refresh_token = cached.get("refresh_token", "")
tidal.token_expiry = cached.get("token_expiry", "")
tidal.country_code = cached.get("country_code", os.getenv("tidal_country_code", ""))
tidal.country_code = cached.get(
"country_code", os.getenv("tidal_country_code", "")
)
else:
tidal.user_id = os.getenv("tidal_user_id", "")
tidal.access_token = os.getenv("tidal_access_token", "")
@@ -212,7 +216,9 @@ class SRUtil:
with open(TIDAL_TOKEN_CACHE_PATH, "r") as f:
data = json.load(f)
# Validate required fields exist
if all(k in data for k in ("access_token", "refresh_token", "token_expiry")):
if all(
k in data for k in ("access_token", "refresh_token", "token_expiry")
):
logging.info("Loaded Tidal tokens from cache")
return data
except Exception as e:
@@ -252,7 +258,10 @@ class SRUtil:
Returns:
tuple: (device_code, verification_url) - User should visit the URL to authorize.
"""
if not hasattr(self.streamrip_client, 'session') or not self.streamrip_client.session:
if (
not hasattr(self.streamrip_client, "session")
or not self.streamrip_client.session
):
self.streamrip_client.session = await self.streamrip_client.get_session()
device_code, verification_url = await self.streamrip_client._get_device_code()
@@ -300,7 +309,8 @@ class SRUtil:
# token_expiry is typically an ISO timestamp string
if isinstance(token_expiry, str):
from datetime import datetime
expiry_dt = datetime.fromisoformat(token_expiry.replace('Z', '+00:00'))
expiry_dt = datetime.fromisoformat(token_expiry.replace("Z", "+00:00"))
expiry_ts = expiry_dt.timestamp()
else:
expiry_ts = float(token_expiry)
@@ -325,7 +335,7 @@ class SRUtil:
self.streamrip_client.logged_in = False
# Close existing session if present
if hasattr(self.streamrip_client, 'session') and self.streamrip_client.session:
if hasattr(self.streamrip_client, "session") and self.streamrip_client.session:
try:
if not self.streamrip_client.session.closed:
await self.streamrip_client.session.close()
@@ -333,7 +343,7 @@ class SRUtil:
logging.warning("Error closing old session: %s", e)
# Use object.__setattr__ to bypass type checking for session reset
try:
object.__setattr__(self.streamrip_client, 'session', None)
object.__setattr__(self.streamrip_client, "session", None)
except Exception:
pass # Session will be recreated on next login
@@ -345,7 +355,9 @@ class SRUtil:
logging.info("Fresh Tidal login successful")
return True
except Exception as e:
logging.warning("Forced Tidal login failed: %s - device re-auth may be required", e)
logging.warning(
"Forced Tidal login failed: %s - device re-auth may be required", e
)
return False
async def _login_and_persist(self, force: bool = False) -> None:
@@ -387,7 +399,9 @@ class SRUtil:
self._save_cached_tokens()
logging.info("Tidal login/refresh successful")
except Exception as e:
logging.warning("Tidal login/refresh failed: %s - device re-auth may be required", e)
logging.warning(
"Tidal login/refresh failed: %s - device re-auth may be required", e
)
# Don't mark as logged in on failure - let subsequent calls retry
async def rate_limited_request(self, func, *args, **kwargs):
@@ -402,7 +416,9 @@ class SRUtil:
try:
await self._login_and_persist()
except Exception as e:
logging.warning("Pre-request login failed in rate_limited_request: %s", e)
logging.warning(
"Pre-request login failed in rate_limited_request: %s", e
)
result = await func(*args, **kwargs)
self.LAST_METADATA_REQUEST = time.time()
@@ -432,7 +448,10 @@ class SRUtil:
try:
await self._login_and_persist()
except Exception as login_err:
logging.warning("Pre-request login failed: %s (continuing anyway)", login_err)
logging.warning(
"Pre-request login failed: %s (continuing anyway)",
login_err,
)
result = await func(*args, **kwargs)
# Track successful request
@@ -441,7 +460,12 @@ class SRUtil:
except AttributeError as e:
# Probably missing/closed client internals: try re-login once
last_exc = e
logging.warning("AttributeError in API call (attempt %d/%d): %s", attempt + 1, retries, e)
logging.warning(
"AttributeError in API call (attempt %d/%d): %s",
attempt + 1,
retries,
e,
)
try:
await self._force_fresh_login()
except Exception:
@@ -475,7 +499,10 @@ class SRUtil:
# Treat 401 (Unauthorized) as an auth failure: force a fresh re-login then retry
is_401_error = (
(isinstance(e, aiohttp.ClientResponseError) and getattr(e, "status", None) == 401)
(
isinstance(e, aiohttp.ClientResponseError)
and getattr(e, "status", None) == 401
)
or "401" in msg
or "unauthorized" in msg.lower()
)
@@ -491,7 +518,9 @@ class SRUtil:
if login_success:
logging.info("Forced re-login after 401 successful")
else:
logging.warning("Forced re-login after 401 failed - may need device re-auth")
logging.warning(
"Forced re-login after 401 failed - may need device re-auth"
)
except Exception as login_exc:
logging.warning("Forced login after 401 failed: %s", login_exc)
if attempt < retries - 1:
@@ -550,9 +579,7 @@ class SRUtil:
title_match = self.is_fuzzy_match(expected_title, found_title, threshold)
return artist_match and album_match and title_match
def dedupe_by_key(
self, key: str | list[str], entries: list[dict]
) -> list[dict]:
def dedupe_by_key(self, key: str | list[str], entries: list[dict]) -> list[dict]:
"""Return entries de-duplicated by one or more keys."""
keys = [key] if isinstance(key, str) else list(key)
@@ -679,9 +706,11 @@ class SRUtil:
"upc": album_json.get("upc"),
"album_copyright": album_json.get("copyright"),
"album_cover_id": album_json.get("cover"),
"album_cover_url": f"https://resources.tidal.com/images/{album_json.get('cover')}/1280x1280.jpg"
"album_cover_url": (
f"https://resources.tidal.com/images/{album_json.get('cover')}/1280x1280.jpg"
if album_json.get("cover")
else None,
else None
),
}
# Track-level (overrides or adds to album info)
@@ -813,7 +842,9 @@ class SRUtil:
return None
if not metadata:
return None
albums = self.dedupe_by_key(["title", "releaseDate"], metadata.get("albums", []))
albums = self.dedupe_by_key(
["title", "releaseDate"], metadata.get("albums", [])
)
albums_out = [
{
"artist": ", ".join(artist["name"] for artist in album["artists"]),

35
utils/test.conf
View File

@@ -1,35 +0,0 @@
# -----------------------
# /m/m2/ PHP handler
location ~ ^/m/m2/(.+\.php)$ {
alias /storage/music2/completed/;
include fastcgi_params;
fastcgi_pass unix:/run/php/php8.2-fpm.sock;
fastcgi_param SCRIPT_FILENAME /storage/music2/completed/$1;
fastcgi_param DOCUMENT_ROOT /storage/music2/completed;
fastcgi_param SCRIPT_NAME /m/m2/$1;
}
# /m/m2/ static files
location /m/m2/ {
alias /storage/music2/completed/;
index index.php;
try_files $uri $uri/ /index.php$is_args$args;
}
# -----------------------
# /m/ PHP handler
location ~ ^/m/(.+\.php)$ {
root /var/www/codey.lol/new/public;
include fastcgi_params;
fastcgi_pass unix:/run/php/php8.2-fpm.sock;
fastcgi_param SCRIPT_FILENAME $document_root/$1;
fastcgi_param DOCUMENT_ROOT $document_root;
fastcgi_param SCRIPT_NAME /m/$1;
}
# /m/ static files
location /m/ {
root /var/www/codey.lol/new/public;
index index.php;
try_files $uri $uri/ /m/index.php$is_args$args;
}

5
utils/yt_utils.py
View File

@@ -7,6 +7,7 @@ import os
VIDEO_PROXY_SECRET = os.environ.get("VIDEO_PROXY_SECRET", "").encode()
def sign_video_id(video_id: Optional[str | bool]) -> str:
"""Generate a signed token for a video ID."""
if not VIDEO_PROXY_SECRET or not video_id:
@@ -15,9 +16,7 @@ def sign_video_id(video_id: Optional[str|bool]) -> str:
timestamp = int(time.time() * 1000) # milliseconds to match JS Date.now()
payload = f"{video_id}:{timestamp}"
signature = hmac.new(
VIDEO_PROXY_SECRET,
payload.encode(),
hashlib.sha256
VIDEO_PROXY_SECRET, payload.encode(), hashlib.sha256
).hexdigest()
token_data = f"{payload}:{signature}"