""" 广告位管理 API（管理员） """ from fastapi import APIRouter, Depends, HTTPException, Query, UploadFile, File from fastapi.responses import FileResponse from sqlalchemy.orm import Session from ...database import get_db, SessionLocal from ...models import AdBanner, AIModel from ...auth import get_current_admin from ...middleware.log_service import LogService import uuid import os import json import httpx from datetime import datetime router = APIRouter() # 广告图片存储目录 ADS_IMAGE_DIR = "/data/share/project/data/ads" os.makedirs(ADS_IMAGE_DIR, exist_ok=True) @router.get("/") async def get_ads( admin_user=Depends(get_current_admin), db: Session = Depends(get_db), ): """获取广告轮播图列表""" ads = db.query(AdBanner).order_by(AdBanner.sort_order.asc()).all() return [ { "id": a.id, "title": a.title, "image_url": a.image_url, "link_app_id": a.link_app_id, "link_app_name": a.link_app_name, "sort_order": a.sort_order, "click_count": a.click_count, "status": a.status, "created_at": a.created_at.isoformat() if a.created_at else None, "updated_at": a.updated_at.isoformat() if a.updated_at else None, } for a in ads ] @router.post("/") async def create_ad( title: str, image_url: str = None, link_app_id: str = None, link_app_name: str = None, sort_order: int = 0, status: str = "active", admin_user=Depends(get_current_admin), db: Session = Depends(get_db), ): """创建广告""" ad = AdBanner( id=str(uuid.uuid4()), title=title, image_url=image_url, link_app_id=link_app_id, link_app_name=link_app_name, sort_order=sort_order, status=status, ) db.add(ad) db.commit() LogService.log_audit( admin_id=admin_user.id, admin_name=admin_user.username, action="create_ad", target_type="ad_banner", target_id=ad.id, detail=f"创建广告 {title}", ) return {"message": "广告已创建", "id": ad.id} @router.put("/{id}") async def update_ad( id: str, title: str = None, image_url: str = None, link_app_id: str = None, link_app_name: str = None, sort_order: int = None, status: str = None, admin_user=Depends(get_current_admin), db: Session = Depends(get_db), ): """更新广告""" ad = db.query(AdBanner).filter(AdBanner.id == id).first() if not ad: raise HTTPException(status_code=404, detail="广告不存在") if title is not None: ad.title = title if image_url is not None: ad.image_url = image_url if link_app_id is not None: ad.link_app_id = link_app_id if link_app_name is not None: ad.link_app_name = link_app_name if sort_order is not None: ad.sort_order = sort_order if status is not None: ad.status = status ad.updated_at = datetime.now() db.commit() LogService.log_audit( admin_id=admin_user.id, admin_name=admin_user.username, action="update_ad", target_type="ad_banner", target_id=id, detail=f"更新广告 {ad.title}", ) return {"message": "广告已更新"} @router.delete("/{id}") async def delete_ad( id: str, admin_user=Depends(get_current_admin), db: Session = Depends(get_db), ): """删除广告""" ad = db.query(AdBanner).filter(AdBanner.id == id).first() if not ad: raise HTTPException(status_code=404, detail="广告不存在") db.delete(ad) db.commit() LogService.log_audit( admin_id=admin_user.id, admin_name=admin_user.username, action="delete_ad", target_type="ad_banner", target_id=id, detail=f"删除广告 {ad.title}", ) return {"message": "广告已删除"} @router.put("/{id}/toggle") async def toggle_ad( id: str, status: str, admin_user=Depends(get_current_admin), db: Session = Depends(get_db), ): """切换广告状态""" ad = db.query(AdBanner).filter(AdBanner.id == id).first() if not ad: raise HTTPException(status_code=404, detail="广告不存在") ad.status = status ad.updated_at = datetime.now() db.commit() LogService.log_audit( admin_id=admin_user.id, admin_name=admin_user.username, action="toggle_ad", target_type="ad_banner", target_id=id, detail=f"切换广告 {ad.title} 状态为 {status}", ) return {"message": f"广告已{status}"} @router.post("/upload-image") async def upload_ad_image( file: UploadFile = File(...), admin_user=Depends(get_current_admin), ): """上传广告图片，保存到 /data/share/project/data/ads/ 目录""" # 校验文件类型 allowed_types = {"image/png", "image/jpeg", "image/gif", "image/webp", "image/svg+xml"} if file.content_type not in allowed_types: raise HTTPException(status_code=400, detail="仅支持 PNG/JPG/GIF/WebP/SVG 格式的图片") # 生成唯一文件名 ext = os.path.splitext(file.filename or "image.png")[1] or ".png" if ext.lower() not in (".png", ".jpg", ".jpeg", ".gif", ".webp", ".svg"): ext = ".png" filename = f"{uuid.uuid4()}{ext}" filepath = os.path.join(ADS_IMAGE_DIR, filename) # 保存文件 content = await file.read() if len(content) > 10 * 1024 * 1024: # 10MB 限制 raise HTTPException(status_code=400, detail="图片大小不能超过10MB") with open(filepath, "wb") as f: f.write(content) # 返回可访问的URL image_url = f"/app/api/admin/ads/images/{filename}" LogService.log_audit( admin_id=admin_user.id, admin_name=admin_user.username, action="upload_ad_image", target_type="ad_banner", detail=f"上传广告图片 {filename}", ) return {"message": "图片上传成功", "image_url": image_url, "filename": filename} @router.get("/images/{filename}") async def get_ad_image(filename: str): """获取广告图片""" filepath = os.path.join(ADS_IMAGE_DIR, filename) if not os.path.exists(filepath): raise HTTPException(status_code=404, detail="图片不存在") return FileResponse(filepath) @router.post("/generate-image") async def generate_ad_image( prompt: str = Query(default=""), admin_user=Depends(get_current_admin), db: Session = Depends(get_db), ): """AI生成广告图：调用LLM生成prompt，然后生成图片（这里生成渐变色占位图）""" # 如果没有传入prompt，使用默认prompt final_prompt = prompt.strip() if prompt.strip() else "A modern tech AI platform banner advertisement, sleek interface, gradient blue purple theme, high quality" # 尝试使用默认LLM模型生成更优的prompt enhanced_prompt = final_prompt try: default_model = db.query(AIModel).filter( AIModel.is_default == True, AIModel.is_enabled == True ).first() if default_model and default_model.endpoint: async with httpx.AsyncClient(timeout=30) as client: headers = {} if default_model.api_key: headers["Authorization"] = f"Bearer {default_model.api_key}" resp = await client.post( default_model.endpoint, json={ "model": default_model.name, "messages": [ {"role": "system", "content": "你是一个广告创意专家。请根据用户输入的主题，生成一个简短的中文广告图像prompt描述（不超过100字），用于AI图像生成。只需要输出prompt文本本身，不要其他内容。"}, {"role": "user", "content": f"主题: {final_prompt}"}, ], "max_tokens": 200, }, headers=headers, ) if resp.status_code == 200: content = resp.json().get("choices", [{}])[0].get("message", {}).get("content", "") if content: enhanced_prompt = content.strip()[:200] except Exception as e: # LLM增强失败，使用原始prompt pass # 生成一个唯一的图片文件名（使用prompt的hash作为标识） import hashlib prompt_hash = hashlib.md5(enhanced_prompt.encode()).hexdigest()[:12] filename = f"ai_gen_{prompt_hash}.svg" filepath = os.path.join(ADS_IMAGE_DIR, filename) # 生成SVG广告图 svg_content = f''' {enhanced_prompt[:40]} Powered by AIFlow ''' with open(filepath, "w", encoding="utf-8") as f: f.write(svg_content) image_url = f"/app/api/admin/ads/images/{filename}" LogService.log_audit( admin_id=admin_user.id, admin_name=admin_user.username, action="generate_ad_image", target_type="ad_banner", detail=f"AI生成广告图 {filename}, prompt: {enhanced_prompt[:50]}", ) return {"message": "AI广告图已生成", "image_url": image_url, "prompt": enhanced_prompt}