150 lines
5.7 KiB
Python
150 lines
5.7 KiB
Python
from sqlalchemy import select, update, delete
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from models.stream import Stream, StreamStatus
|
|
from services.media_service import media_service
|
|
from services.load_balancer import lb
|
|
from services.multicast_detector import multicast_detector
|
|
from services.transcoder import transcode_mgr
|
|
from config import settings
|
|
import json
|
|
from datetime import datetime
|
|
|
|
|
|
class StreamService:
|
|
def __init__(self, db: AsyncSession):
|
|
self.db = db
|
|
|
|
async def create_stream(self, title: str, description: str | None = None,
|
|
cover_url: str | None = None,
|
|
scheduled_at: str | None = None,
|
|
replay_file: str | None = None,
|
|
chapters: list[dict] | None = None,
|
|
transcode_profile: str = "standard",
|
|
homepage_visible: bool = True,
|
|
remote_control_enabled: bool = False) -> Stream:
|
|
import uuid
|
|
scheduled_datetime = None
|
|
if scheduled_at:
|
|
try:
|
|
scheduled_datetime = datetime.fromisoformat(scheduled_at.replace("Z", "+00:00"))
|
|
except ValueError as exc:
|
|
raise ValueError("预定开播时间格式无效") from exc
|
|
stream = Stream(
|
|
title=title,
|
|
description=description,
|
|
cover_url=cover_url,
|
|
replay_file=replay_file,
|
|
chapters_json=json.dumps(chapters or [], ensure_ascii=False),
|
|
transcode_profile=transcode_profile,
|
|
homepage_visible=homepage_visible,
|
|
remote_control_enabled=remote_control_enabled,
|
|
stream_key=uuid.uuid4().hex,
|
|
status=StreamStatus.pending,
|
|
scheduled_at=scheduled_datetime,
|
|
)
|
|
self.db.add(stream)
|
|
await self.db.commit()
|
|
await self.db.refresh(stream)
|
|
return stream
|
|
|
|
async def get_all_streams(self):
|
|
result = await self.db.execute(
|
|
select(Stream).order_by(Stream.created_at.desc())
|
|
)
|
|
return result.scalars().all()
|
|
|
|
async def get_public_streams(self):
|
|
result = await self.db.execute(
|
|
select(Stream)
|
|
.where(Stream.homepage_visible.is_(True))
|
|
.order_by(Stream.created_at.desc())
|
|
)
|
|
return result.scalars().all()
|
|
|
|
async def get_stream(self, stream_id: int):
|
|
result = await self.db.execute(
|
|
select(Stream).where(Stream.id == stream_id)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
async def get_stream_by_key(self, stream_key: str):
|
|
result = await self.db.execute(
|
|
select(Stream).where(Stream.stream_key == stream_key)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
async def update_stream(self, stream_id: int, **kwargs):
|
|
stmt = update(Stream).where(Stream.id == stream_id).values(**kwargs)
|
|
await self.db.execute(stmt)
|
|
await self.db.commit()
|
|
|
|
async def delete_stream(self, stream_id: int):
|
|
stream = await self.get_stream(stream_id)
|
|
if stream:
|
|
transcode_mgr.stop_transcode(stream.stream_key)
|
|
await self.db.delete(stream)
|
|
await self.db.commit()
|
|
|
|
async def set_stream_live(self, stream_key: str):
|
|
stream = await self.get_stream_by_key(stream_key)
|
|
if stream:
|
|
from datetime import datetime
|
|
await self.update_stream(
|
|
stream.id,
|
|
status=StreamStatus.live,
|
|
started_at=datetime.utcnow(),
|
|
)
|
|
rtmp_url = media_service.get_rtmp_url(stream_key)
|
|
transcode_mgr.start_transcode(stream_key, rtmp_url, stream.transcode_profile)
|
|
|
|
async def set_stream_ended(self, stream_key: str):
|
|
stream = await self.get_stream_by_key(stream_key)
|
|
if stream:
|
|
from datetime import datetime
|
|
await self.update_stream(
|
|
stream.id,
|
|
status=StreamStatus.ended,
|
|
ended_at=datetime.utcnow(),
|
|
)
|
|
transcode_mgr.stop_transcode(stream_key)
|
|
|
|
async def get_watch_urls(self, stream_key: str, client_ip: str | None = None):
|
|
stream = await self.get_stream_by_key(stream_key)
|
|
if not stream:
|
|
return {"lines": [], "qualities": [], "multicast_available": False, "selected_node": None}
|
|
qualities = settings.transcode_profiles.get(stream.transcode_profile, settings.qualities)
|
|
best_node = lb.select_best_node(client_ip)
|
|
if not best_node:
|
|
best_node = lb.get_enabled_nodes()[0] if lb.get_enabled_nodes() else None
|
|
|
|
lines = []
|
|
if best_node:
|
|
base_hls = f"/hls/{stream_key}"
|
|
line_info = {
|
|
"node_id": best_node["id"],
|
|
"node_name": best_node["name"],
|
|
"weight": best_node["weight"],
|
|
"type": "hls",
|
|
"url": f"{base_hls}/index.m3u8",
|
|
"qualities": [],
|
|
}
|
|
for quality in qualities:
|
|
label = quality["label"]
|
|
line_info["qualities"].append({
|
|
"label": label,
|
|
"width": quality["width"],
|
|
"height": quality["height"],
|
|
"url": f"{base_hls}/{label}/index.m3u8",
|
|
})
|
|
lines.append(line_info)
|
|
|
|
return {
|
|
"lines": lines,
|
|
"qualities": [{"label": q["label"], "width": q["width"], "height": q["height"]} for q in qualities],
|
|
"multicast_available": multicast_detector.is_available(),
|
|
"selected_node": {
|
|
"id": best_node["id"],
|
|
"name": best_node["name"],
|
|
} if best_node else None,
|
|
}
|