import os from typing import Tuple from app.utils.config import server_config from app.utils.playlist_merge import SyncMode, sync_all_playlists, TEST_PLAYLIST_DIR from fastapi import FastAPI, Request, Form from fastapi.responses import HTMLResponse, RedirectResponse from fastapi.templating import Jinja2Templates from fastapi.staticfiles import StaticFiles from app.utils.plex_client import plex_client from app.utils.logger import logger from app.utils.local_playlist import scan_local_playlists app = FastAPI() templates = Jinja2Templates( directory=os.path.join(os.path.dirname(__file__), "templates") ) # mount static files # 这里的路径是相对于 main.py 文件所在的目录 app.mount( "/static", StaticFiles(directory=os.path.join(os.path.dirname(__file__), "static")), name="static", ) SYNC_MODE_OPTIONS = [ { "value": SyncMode.LOCAL_FORCE.value, "label": "完全本地优先（local_force）", "description": "单向同步，本地覆盖云端且顺序以本地为准。", }, { "value": SyncMode.REMOTE_FORCE.value, "label": "完全云端优先（remote_force）", "description": "单向同步，云端覆盖本地且顺序以云端为准。", }, { "value": SyncMode.MERGE_LOCAL_PRIMARY.value, "label": "双向合并（本地优先）", "description": "三方合并，冲突时选择本地版本。", }, { "value": SyncMode.MERGE_REMOTE_PRIMARY.value, "label": "双向合并（云端优先）", "description": "三方合并，冲突时选择云端版本。", }, ] def _get_cloud_playlists() -> Tuple[list[dict], str, dict]: """Fetch playlists and connection state from the remote Plex server.""" server_config.load() playlists: list[dict] = [] status = "unset" server_info = { "name": "未设置", "domain": server_config.url or "未设置", } # no server url configured if not server_config.url: return playlists, status, server_info status = "failed" try: plex_client.connect( token=server_config.token, scheme=server_config.scheme, url=server_config.url, port=server_config.port, ) status = "connected" if plex_client.connected else "failed" server_info.update( { "name": getattr(plex_client.server, "friendlyName", "未命名服务器"), "domain": server_config.url, } ) for playlist in plex_client.server.playlists(): track_count = getattr(playlist, "itemCount", None) if track_count is None: try: track_count = len(playlist.items()) except Exception: track_count = 0 playlists.append( { "name": playlist.title, "track_count": track_count, } ) except Exception as exc: logger.warning(f"Failed to fetch cloud playlists: {exc}") status = "failed" playlists.sort(key=lambda item: item["name"].lower()) return playlists, status, server_info def _build_home_context( request: Request, local_path: str, message: str | None = None, message_type: str | None = None, sync_result: dict | None = None, selected_mode: str | None = None, ): server_config.load() local_playlists = scan_local_playlists(local_path) cloud_playlists, connection_status, server_info = _get_cloud_playlists() return { "request": request, "theme": server_config.theme, "path": "/", "local_playlists": local_playlists, "local_path": local_path, "cloud_playlists": cloud_playlists, "connection_status": connection_status, "server_info": server_info, "sync_modes": SYNC_MODE_OPTIONS, "selected_mode": selected_mode, "message": message, "message_type": message_type, "sync_result": sync_result, } # 显示主页 @app.get("/", response_class=HTMLResponse) async def home(request: Request, local_path: str = "playlist"): context = _build_home_context(request, local_path) return templates.TemplateResponse("home.html", context) @app.post("/sync", response_class=HTMLResponse) async def trigger_sync(request: Request, mode: str = Form(...), local_path: str = Form("playlist")): try: sync_mode = SyncMode(mode) except ValueError: context = _build_home_context( request, local_path, message=f"未知的同步策略：{mode}", message_type="danger", selected_mode=mode, ) return templates.TemplateResponse("home.html", context) try: results = sync_all_playlists( local_dir=local_path, mode=sync_mode, test_folder=TEST_PLAYLIST_DIR, ) merged_count = sum(len(item.merged_paths) for item in results) conflict_count = sum(len(item.conflicts) for item in results) deleted_count = sum(1 for item in results if item.action == "deleted") context = _build_home_context( request, local_path, message="同步完成，输出已写入测试目录用于验证。", message_type="success", sync_result={ "mode": sync_mode.value, "mode_label": next( (item["label"] for item in SYNC_MODE_OPTIONS if item["value"] == sync_mode.value), sync_mode.value, ), "merged_count": merged_count, "conflict_count": conflict_count, "delete_count": deleted_count, "playlist_count": len(results), "output_dir": TEST_PLAYLIST_DIR, }, selected_mode=sync_mode.value, ) return templates.TemplateResponse("home.html", context) except Exception as exc: logger.warning(f"Sync failed: {exc}") context = _build_home_context( request, local_path, message=f"同步失败：{exc}", message_type="danger", selected_mode=sync_mode.value, ) return templates.TemplateResponse("home.html", context) # 登录页面和处理 @app.get("/login", response_class=HTMLResponse) async def login_page(request: Request): server_config.load() return templates.TemplateResponse( "login.html", { "request": request, "theme": server_config.theme, "path": "/login", "token": server_config.token, "scheme": server_config.scheme, "server_url": server_config.url, "port": server_config.port, }, ) @app.post("/login", response_class=HTMLResponse) async def login( request: Request, user: str = Form(...), pw: str = Form(...), token: str = Form(...), scheme: str = Form("https"), url: str = Form(...), port: str = Form("32400"), ): # 尝试连接到 Plex 服务器 try: # 优先使用 token 连接，如果 token 为空则使用用户名和密码连接 _, token_success = plex_client.connect( username=user, password=pw, token=token, scheme=scheme, url=url, port=port, ) # 成功连接后保存配置到配置文件 if plex_client.connected: server_config.set_and_save_config( token=token_success, scheme=scheme, url=url, port=port ) return templates.TemplateResponse( "login.html", { "request": request, "message": "连接成功", "message_type": "success", "theme": server_config.theme, "path": "/login", "token": server_config.token, "scheme": server_config.scheme, "server_url": server_config.url, "port": server_config.port, }, ) except Exception as e: return templates.TemplateResponse( "login.html", { "request": request, "message": f"连接失败：{str(e)}", "message_type": "danger", "theme": server_config.theme, "path": "/login", "scheme": scheme, "server_url": url, "port": port, }, ) @app.get("/playlist", response_class=HTMLResponse) async def get_playlist(request: Request): return templates.TemplateResponse( "playlist.html", {"request": request, "theme": server_config.theme, "path": "/playlist"}, ) @app.post("/playlist", response_class=HTMLResponse) async def set_playlist( request: Request, address: str = Form(...), interval: str = Form(...) ): # demo：返回提交的设置 return templates.TemplateResponse( "playlist.html", { "request": request, "message": f"设置成功：地址 {address}，间隔 {interval} 分钟", "message_type": "info", "theme": server_config.theme, "path": "/playlist", }, ) @app.post("/set-theme") async def set_theme(theme: str = Form(...)): server_config.set_and_save_config(theme=theme) return RedirectResponse("/login", status_code=303)