# app/routers/note.py import json import os import uuid from typing import Optional from urllib.parse import urlparse from fastapi import APIRouter, HTTPException, BackgroundTasks, UploadFile, File from pydantic import BaseModel, validator from dataclasses import asdict from app.db.video_task_dao import get_task_by_video from app.enmus.note_enums import DownloadQuality from app.services.note import NoteGenerator, logger from app.utils.response import ResponseWrapper as R from app.utils.url_parser import extract_video_id from app.validators.video_url_validator import is_supported_video_url from fastapi import APIRouter, Request, HTTPException from fastapi.responses import StreamingResponse import httpx from app.enmus.task_status_enums import TaskStatus # from app.services.downloader import download_raw_audio # from app.services.whisperer import transcribe_audio router = APIRouter() class RecordRequest(BaseModel): video_id: str platform: str class VideoRequest(BaseModel): video_url: str platform: str quality: DownloadQuality screenshot: Optional[bool] = False link: Optional[bool] = False model_name:str provider_id:str task_id: Optional[str] = None format:Optional[list]=[] style:str=None extras:Optional[str] @validator("video_url") def validate_supported_url(cls, v): url = str(v) parsed = urlparse(url) if parsed.scheme in ("http", "https"): # 是网络链接，继续用原有平台校验 if not is_supported_video_url(url): raise ValueError("暂不支持该视频平台或链接格式无效") else: # 是本地路径，检测一下文件是否存在 if not url.startswith('/uploads') and not os.path.exists(url): raise ValueError("本地文件路径不存在") return v NOTE_OUTPUT_DIR = "note_results" UPLOAD_DIR = "uploads" def save_note_to_file(task_id: str, note): os.makedirs(NOTE_OUTPUT_DIR, exist_ok=True) with open(os.path.join(NOTE_OUTPUT_DIR, f"{task_id}.json"), "w", encoding="utf-8") as f: json.dump(asdict(note), f, ensure_ascii=False, indent=2) def run_note_task(task_id: str, video_url: str, platform: str, quality: DownloadQuality, link: bool = False,screenshot: bool = False,model_name:str=None,provider_id:str=None, _format:list=None,style:str=None,extras:str=None): try: if not model_name or not provider_id: raise HTTPException(status_code=400, detail="请选择模型和提供者") note = NoteGenerator().generate( video_url=video_url, platform=platform, quality=quality, task_id=task_id, model_name=model_name, provider_id=provider_id, link=link, _format=_format, style=style, extras=extras, screenshot=screenshot ) print('Note 结果',note) save_note_to_file(task_id, note) except Exception as e: save_note_to_file(task_id, {"error": str(e)}) @router.post('/delete_task') def delete_task(data:RecordRequest): try: NoteGenerator().delete_note(video_id=data.video_id,platform=data.platform) return R.success(msg='删除成功') except Exception as e: return R.error(msg=e) @router.post("/upload") async def upload(file: UploadFile = File(...)): os.makedirs(UPLOAD_DIR, exist_ok=True) file_location = os.path.join(UPLOAD_DIR, file.filename) with open(file_location, "wb+") as f: f.write(await file.read()) # 假设你静态目录挂载了 /uploads return R.success({"url": f"/uploads/{file.filename}"}) @router.post("/generate_note") def generate_note(data: VideoRequest, background_tasks: BackgroundTasks): try: video_id = extract_video_id(data.video_url, data.platform) # if not video_id: # raise HTTPException(status_code=400, detail="无法提取视频 ID") # existing = get_task_by_video(video_id, data.platform) # if existing: # return R.error( # msg='笔记已生成，请勿重复发起', # # ) if data.task_id: # 如果传了task_id，说明是重试！ task_id = data.task_id # 更新之前的状态 NoteGenerator.update_task_status(task_id, TaskStatus.PENDING) logger.info(f"重试模式，复用已有 task_id={task_id}") else: # 正常新建任务 task_id = str(uuid.uuid4()) background_tasks.add_task(run_note_task, task_id, data.video_url, data.platform, data.quality,data.link ,data.screenshot,data.model_name,data.provider_id,data.format,data.style,data.extras) return R.success({"task_id": task_id}) except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @router.get("/task_status/{task_id}") def get_task_status(task_id: str): status_path = os.path.join(NOTE_OUTPUT_DIR, f"{task_id}.status.json") result_path = os.path.join(NOTE_OUTPUT_DIR, f"{task_id}.json") # 优先读状态文件 if os.path.exists(status_path): with open(status_path, "r", encoding="utf-8") as f: status_content = json.load(f) status = status_content.get("status") message = status_content.get("message", "") if status == TaskStatus.SUCCESS.value: # 成功状态的话，继续读取最终笔记内容 if os.path.exists(result_path): with open(result_path, "r", encoding="utf-8") as rf: result_content = json.load(rf) return R.success({ "status": status, "result": result_content, "message": message, "task_id": task_id }) else: # 理论上不会出现，保险处理 return R.success({ "status": TaskStatus.PENDING.value, "message": "任务完成，但结果文件未找到", "task_id": task_id }) if status == TaskStatus.FAILED.value: return R.error(message or "任务失败", code=500) # 处理中状态 return R.success({ "status": status, "message": message, "task_id": task_id }) # 没有状态文件，但有结果 if os.path.exists(result_path): with open(result_path, "r", encoding="utf-8") as f: result_content = json.load(f) return R.success({ "status": TaskStatus.SUCCESS.value, "result": result_content, "task_id": task_id }) # 什么都没有，默认PENDING return R.success({ "status": TaskStatus.PENDING.value, "message": "任务排队中", "task_id": task_id }) @router.get("/image_proxy") async def image_proxy(request: Request, url: str): headers = { "Referer": "https://www.bilibili.com/", # 模拟B站来源 "User-Agent": request.headers.get("User-Agent", ""), } try: async with httpx.AsyncClient(timeout=10.0) as client: resp = await client.get(url, headers=headers) if resp.status_code != 200: raise HTTPException(status_code=resp.status_code, detail="图片获取失败") content_type = resp.headers.get("Content-Type", "image/jpeg") return StreamingResponse(resp.aiter_bytes(), media_type=content_type) except Exception as e: raise HTTPException(status_code=500, detail=str(e))