423 lines
15 KiB
Python
423 lines
15 KiB
Python
"""PPT/海报管理后台路由"""
|
||
import os
|
||
from flask import Blueprint, request, jsonify, send_file
|
||
from insurance.admin.ppt_admin_service import PptAdminService
|
||
from insurance.middleware.auth_middleware import permission_required, jwt_required
|
||
|
||
ppt_admin_bp = Blueprint("ppt_admin", __name__)
|
||
ppt_admin_service = PptAdminService()
|
||
|
||
|
||
def _is_within(root: str, path: str) -> bool:
|
||
"""检查文件是否位于指定目录内，兼容 Windows 不同盘符。"""
|
||
try:
|
||
return os.path.commonpath([root, path]) == root
|
||
except ValueError:
|
||
return False
|
||
|
||
|
||
# ---- 保司管理 ----
|
||
|
||
@ppt_admin_bp.route("/companies", methods=["GET"])
|
||
@jwt_required
|
||
def list_companies():
|
||
params = {
|
||
"page": request.args.get("page", 1, type=int),
|
||
"page_size": request.args.get("page_size", 20, type=int),
|
||
"status": request.args.get("status", type=int),
|
||
"keyword": request.args.get("keyword", ""),
|
||
}
|
||
return jsonify(ppt_admin_service.list_companies(params))
|
||
|
||
|
||
@ppt_admin_bp.route("/companies", methods=["POST"])
|
||
@permission_required("config_manage")
|
||
def create_company():
|
||
return jsonify(ppt_admin_service.create_company(request.get_json(force=True)))
|
||
|
||
|
||
@ppt_admin_bp.route("/companies/<company_id>", methods=["PUT"])
|
||
@permission_required("config_manage")
|
||
def update_company(company_id):
|
||
return jsonify(ppt_admin_service.update_company(company_id, request.get_json(force=True)))
|
||
|
||
|
||
@ppt_admin_bp.route("/companies/<company_id>/status", methods=["PUT"])
|
||
@permission_required("config_manage")
|
||
def update_company_status(company_id):
|
||
return jsonify(ppt_admin_service.update_company_status(company_id, request.get_json(force=True)))
|
||
|
||
|
||
# ---- 保司Logo管理 ----
|
||
|
||
@ppt_admin_bp.route("/companies/<company_id>/logos", methods=["GET"])
|
||
@jwt_required
|
||
def list_company_logos(company_id):
|
||
return jsonify(ppt_admin_service.list_company_logos(company_id))
|
||
|
||
|
||
@ppt_admin_bp.route("/companies/<company_id>/logos", methods=["POST"])
|
||
@permission_required("config_manage")
|
||
def upload_company_logos(company_id):
|
||
files = request.files.getlist("files")
|
||
if not files:
|
||
return jsonify({"code": 1001, "message": "请选择图片文件", "data": None})
|
||
return jsonify(ppt_admin_service.upload_company_logos(company_id, files))
|
||
|
||
|
||
@ppt_admin_bp.route("/companies/<company_id>/logos/<logo_id>/primary", methods=["PUT"])
|
||
@permission_required("config_manage")
|
||
def set_primary_logo(company_id, logo_id):
|
||
return jsonify(ppt_admin_service.set_primary_logo(company_id, logo_id))
|
||
|
||
|
||
@ppt_admin_bp.route("/companies/<company_id>/logos/order", methods=["PUT"])
|
||
@permission_required("config_manage")
|
||
def reorder_logos(company_id):
|
||
data = request.get_json(force=True)
|
||
return jsonify(ppt_admin_service.reorder_logos(company_id, data.get("order", [])))
|
||
|
||
|
||
@ppt_admin_bp.route("/companies/<company_id>/logos/<logo_id>", methods=["DELETE"])
|
||
@permission_required("config_manage")
|
||
def delete_company_logo(company_id, logo_id):
|
||
return jsonify(ppt_admin_service.delete_company_logo(company_id, logo_id))
|
||
|
||
|
||
@ppt_admin_bp.route("/assets/company-logos/<logo_id>", methods=["GET"])
|
||
def serve_company_logo(logo_id):
|
||
"""提供Logo图片文件访问。"""
|
||
from insurance.models.ppt_config import CompanyLogo
|
||
logo = CompanyLogo.query.get(logo_id)
|
||
if not logo or not logo.file_path or not os.path.exists(logo.file_path):
|
||
return jsonify({"code": 404, "message": "图片不存在", "data": None}), 404
|
||
from insurance.config import get_storage_root
|
||
logo_root = os.path.abspath(os.path.join(get_storage_root(), "uploads", "company-logos"))
|
||
logo_path = os.path.abspath(logo.file_path)
|
||
if not _is_within(logo_root, logo_path):
|
||
return jsonify({"code": 404, "message": "图片不存在", "data": None}), 404
|
||
return send_file(
|
||
logo_path,
|
||
mimetype=logo.mime_type or "image/png",
|
||
as_attachment=False,
|
||
)
|
||
|
||
|
||
# ---- 产品管理 ----
|
||
|
||
@ppt_admin_bp.route("/products", methods=["GET"])
|
||
@jwt_required
|
||
def list_products():
|
||
params = {
|
||
"page": request.args.get("page", 1, type=int),
|
||
"page_size": request.args.get("page_size", 20, type=int),
|
||
"company_id": request.args.get("company_id", ""),
|
||
"plan_type": request.args.get("plan_type", ""),
|
||
"status": request.args.get("status", type=int),
|
||
}
|
||
return jsonify(ppt_admin_service.list_products(params))
|
||
|
||
|
||
@ppt_admin_bp.route("/products", methods=["POST"])
|
||
@permission_required("config_manage")
|
||
def create_product():
|
||
return jsonify(ppt_admin_service.create_product(request.get_json(force=True)))
|
||
|
||
|
||
@ppt_admin_bp.route("/products/<product_id>", methods=["PUT"])
|
||
@permission_required("config_manage")
|
||
def update_product(product_id):
|
||
return jsonify(ppt_admin_service.update_product(product_id, request.get_json(force=True)))
|
||
|
||
|
||
@ppt_admin_bp.route("/products/<product_id>/status", methods=["PUT"])
|
||
@permission_required("config_manage")
|
||
def update_product_status(product_id):
|
||
return jsonify(ppt_admin_service.update_product_status(product_id, request.get_json(force=True)))
|
||
|
||
|
||
@ppt_admin_bp.route("/products/<product_id>/manual/upload", methods=["POST"])
|
||
@permission_required("config_manage")
|
||
def upload_manual(product_id):
|
||
file = request.files.get("file")
|
||
password = request.form.get("password", "")
|
||
if not file:
|
||
return jsonify({"code": 1001, "message": "请上传文件", "data": None})
|
||
return jsonify(ppt_admin_service.upload_manual(product_id, file, password))
|
||
|
||
|
||
@ppt_admin_bp.route("/products/<product_id>/manual/parse", methods=["POST"])
|
||
@permission_required("config_manage")
|
||
def parse_manual(product_id):
|
||
return jsonify(ppt_admin_service.parse_manual(product_id))
|
||
|
||
|
||
@ppt_admin_bp.route("/products/<product_id>/manual/review", methods=["PUT"])
|
||
@permission_required("config_manage")
|
||
def review_manual(product_id):
|
||
data = request.get_json(force=True)
|
||
if data.get("confirm") and not data.get("reviewedBy"):
|
||
data["reviewedBy"] = str(getattr(request, "user_id", ""))
|
||
return jsonify(ppt_admin_service.review_manual(product_id, data))
|
||
|
||
|
||
@ppt_admin_bp.route("/products/<product_id>/manual", methods=["GET"])
|
||
@jwt_required
|
||
def get_manual(product_id):
|
||
return jsonify(ppt_admin_service.get_manual(product_id))
|
||
|
||
|
||
@ppt_admin_bp.route("/products/<product_id>/manual/file", methods=["GET"])
|
||
def get_manual_file(product_id):
|
||
"""流式返回小册子 PDF 文件（支持 ?token= 以兼容 iframe/window.open）。"""
|
||
# 手动认证：支持 Authorization header 或 ?token= query param
|
||
from insurance.middleware.auth_middleware import jwt_required as _jwt_decorator
|
||
|
||
# 将 query param token 注入到 environ 中，供 jwt_required 读取
|
||
if not request.headers.get("Authorization") and request.args.get("token"):
|
||
request.environ["HTTP_AUTHORIZATION"] = f"Bearer {request.args.get('token')}"
|
||
|
||
# 用装饰器包装内联函数执行认证
|
||
@_jwt_decorator
|
||
def _check_and_serve():
|
||
from insurance.models.ppt_config import PptProduct
|
||
product = PptProduct.query.get(product_id)
|
||
if not product or not product.manual_file_url:
|
||
return jsonify({"code": 1002, "message": "小册子文件不存在", "data": None}), 404
|
||
if not os.path.exists(product.manual_file_url):
|
||
return jsonify({"code": 1002, "message": "小册子文件已被删除", "data": None}), 404
|
||
from insurance.config import get_storage_root
|
||
manual_root = os.path.abspath(os.path.join(get_storage_root(), "uploads", "product-manuals"))
|
||
manual_path = os.path.abspath(product.manual_file_url)
|
||
if not _is_within(manual_root, manual_path):
|
||
return jsonify({"code": 1002, "message": "小册子文件不存在", "data": None}), 404
|
||
return send_file(
|
||
manual_path,
|
||
mimetype="application/pdf",
|
||
as_attachment=False,
|
||
download_name=f"{product_id}_manual.pdf",
|
||
)
|
||
|
||
return _check_and_serve()
|
||
|
||
|
||
# ---- PPT 模板管理 ----
|
||
|
||
@ppt_admin_bp.route("/templates", methods=["GET"])
|
||
@jwt_required
|
||
def list_templates():
|
||
params = {
|
||
"page": request.args.get("page", 1, type=int),
|
||
"page_size": request.args.get("page_size", 20, type=int),
|
||
"plan_type": request.args.get("plan_type", ""),
|
||
"status": request.args.get("status", type=int),
|
||
}
|
||
return jsonify(ppt_admin_service.list_templates(params))
|
||
|
||
|
||
@ppt_admin_bp.route("/templates", methods=["POST"])
|
||
@permission_required("config_manage")
|
||
def create_template():
|
||
return jsonify(ppt_admin_service.create_template(request.get_json(force=True)))
|
||
|
||
|
||
@ppt_admin_bp.route("/templates/upload", methods=["POST"])
|
||
@permission_required("config_manage")
|
||
def upload_template():
|
||
"""上传并自动解析 PPTX 模板。"""
|
||
return jsonify(ppt_admin_service.upload_template(
|
||
request.files.get("file"),
|
||
request.form.to_dict(),
|
||
))
|
||
|
||
|
||
@ppt_admin_bp.route("/templates/<template_id>/asset", methods=["POST"])
|
||
@permission_required("config_manage")
|
||
def replace_template_asset(template_id):
|
||
"""替换模板源 PPTX 并重新解析页面结构。"""
|
||
return jsonify(ppt_admin_service.upload_template(
|
||
request.files.get("file"),
|
||
request.form.to_dict(),
|
||
template_id=template_id,
|
||
))
|
||
|
||
|
||
@ppt_admin_bp.route("/templates/<template_id>/file", methods=["GET"])
|
||
@jwt_required
|
||
def get_template_file(template_id):
|
||
"""查看或下载模板源 PPTX。"""
|
||
from insurance.models.ppt_config import PptTemplate
|
||
from insurance.ppt.template_asset_service import resolve_template_asset
|
||
|
||
template = PptTemplate.query.get(template_id)
|
||
if not template or not template.source_template_asset_id:
|
||
return jsonify({"code": 1002, "message": "模板文件不存在", "data": None}), 404
|
||
try:
|
||
file_path = resolve_template_asset(template.source_template_asset_id)
|
||
except ValueError:
|
||
file_path = None
|
||
if not file_path or not os.path.isfile(file_path):
|
||
return jsonify({"code": 1002, "message": "模板文件不存在", "data": None}), 404
|
||
return send_file(
|
||
file_path,
|
||
mimetype="application/vnd.openxmlformats-officedocument.presentationml.presentation",
|
||
as_attachment=False,
|
||
download_name=f"{template_id}.pptx",
|
||
)
|
||
|
||
|
||
@ppt_admin_bp.route("/templates/<template_id>", methods=["PUT"])
|
||
@permission_required("config_manage")
|
||
def update_template(template_id):
|
||
return jsonify(ppt_admin_service.update_template(template_id, request.get_json(force=True)))
|
||
|
||
|
||
@ppt_admin_bp.route("/templates/<template_id>/status", methods=["PUT"])
|
||
@permission_required("config_manage")
|
||
def update_template_status(template_id):
|
||
return jsonify(ppt_admin_service.update_template_status(template_id, request.get_json(force=True)))
|
||
|
||
|
||
# ---- 海报模板管理 ----
|
||
|
||
@ppt_admin_bp.route("/poster-templates", methods=["GET"])
|
||
@jwt_required
|
||
def list_poster_templates():
|
||
params = {
|
||
"page": request.args.get("page", 1, type=int),
|
||
"page_size": request.args.get("page_size", 20, type=int),
|
||
"status": request.args.get("status", type=int),
|
||
}
|
||
return jsonify(ppt_admin_service.list_poster_templates(params))
|
||
|
||
|
||
@ppt_admin_bp.route("/poster-templates", methods=["POST"])
|
||
@permission_required("config_manage")
|
||
def create_poster_template():
|
||
return jsonify(ppt_admin_service.create_poster_template(request.get_json(force=True)))
|
||
|
||
|
||
@ppt_admin_bp.route("/poster-templates/<int:template_id>", methods=["PUT"])
|
||
@permission_required("config_manage")
|
||
def update_poster_template(template_id):
|
||
return jsonify(ppt_admin_service.update_poster_template(template_id, request.get_json(force=True)))
|
||
|
||
|
||
@ppt_admin_bp.route("/poster-templates/<int:template_id>/status", methods=["PUT"])
|
||
@permission_required("config_manage")
|
||
def update_poster_template_status(template_id):
|
||
return jsonify(ppt_admin_service.update_poster_template_status(template_id, request.get_json(force=True)))
|
||
|
||
|
||
# ---- 文案模板管理 ----
|
||
|
||
@ppt_admin_bp.route("/copy-templates", methods=["GET"])
|
||
@jwt_required
|
||
def list_copy_templates():
|
||
params = {
|
||
"page": request.args.get("page", 1, type=int),
|
||
"page_size": request.args.get("page_size", 20, type=int),
|
||
"status": request.args.get("status", type=int),
|
||
}
|
||
return jsonify(ppt_admin_service.list_copy_templates(params))
|
||
|
||
|
||
@ppt_admin_bp.route("/copy-templates", methods=["POST"])
|
||
@permission_required("config_manage")
|
||
def create_copy_template():
|
||
return jsonify(ppt_admin_service.create_copy_template(request.get_json(force=True)))
|
||
|
||
|
||
@ppt_admin_bp.route("/copy-templates/<int:template_id>", methods=["PUT"])
|
||
@permission_required("config_manage")
|
||
def update_copy_template(template_id):
|
||
return jsonify(ppt_admin_service.update_copy_template(template_id, request.get_json(force=True)))
|
||
|
||
|
||
@ppt_admin_bp.route("/copy-templates/<int:template_id>", methods=["DELETE"])
|
||
@permission_required("config_manage")
|
||
def delete_copy_template(template_id):
|
||
return jsonify(ppt_admin_service.delete_copy_template(template_id))
|
||
|
||
|
||
# ---- 历史记录管理（需要审计权限） ----
|
||
|
||
@ppt_admin_bp.route("/history", methods=["GET"])
|
||
@permission_required("audit_view")
|
||
def list_history():
|
||
params = {
|
||
"page": request.args.get("page", 1, type=int),
|
||
"page_size": request.args.get("page_size", 20, type=int),
|
||
"user_id": request.args.get("user_id", ""),
|
||
"company_id": request.args.get("company_id", ""),
|
||
"action_type": request.args.get("action_type", ""),
|
||
}
|
||
return jsonify(ppt_admin_service.list_history(params))
|
||
|
||
|
||
@ppt_admin_bp.route("/history/<int:history_id>", methods=["DELETE"])
|
||
@permission_required("config_manage")
|
||
def delete_history(history_id):
|
||
return jsonify(ppt_admin_service.delete_history(history_id))
|
||
|
||
|
||
@ppt_admin_bp.route("/history/export", methods=["GET"])
|
||
@permission_required("audit_view")
|
||
def export_history():
|
||
from flask import Response
|
||
params = {
|
||
"user_id": request.args.get("user_id", ""),
|
||
"company_id": request.args.get("company_id", ""),
|
||
"action_type": request.args.get("action_type", ""),
|
||
}
|
||
csv_content = ppt_admin_service.export_history_csv(params)
|
||
return Response(
|
||
csv_content,
|
||
mimetype="text/csv",
|
||
headers={"Content-Disposition": "attachment; filename=ppt_history.csv"},
|
||
)
|
||
|
||
|
||
# ---- 系统配置 ----
|
||
|
||
@ppt_admin_bp.route("/available-models", methods=["GET"])
|
||
@permission_required("config_manage")
|
||
def get_available_models():
|
||
return jsonify(ppt_admin_service.get_available_models())
|
||
|
||
|
||
@ppt_admin_bp.route("/settings", methods=["GET"])
|
||
@permission_required("config_manage")
|
||
def get_settings():
|
||
return jsonify(ppt_admin_service.get_settings())
|
||
|
||
|
||
@ppt_admin_bp.route("/settings", methods=["PUT"])
|
||
@permission_required("config_manage")
|
||
def update_settings():
|
||
updated_by = getattr(request, "user_id", "")
|
||
return jsonify(ppt_admin_service.update_settings(request.get_json(force=True), updated_by))
|
||
|
||
|
||
@ppt_admin_bp.route("/sync-models", methods=["POST"])
|
||
@permission_required("config_manage")
|
||
def sync_models():
|
||
data = request.get_json(force=True)
|
||
return jsonify(ppt_admin_service.sync_models(
|
||
provider=data.get("provider", ""),
|
||
api_key=data.get("api_key", ""),
|
||
base_url=data.get("base_url", ""),
|
||
))
|
||
|
||
|
||
# ---- 数据清理 ----
|
||
|
||
@ppt_admin_bp.route("/cleanup", methods=["POST"])
|
||
@permission_required("config_manage")
|
||
def run_cleanup():
|
||
"""手动触发过期数据清理。"""
|
||
from insurance.utils.cleanup import cleanup_expired_data
|
||
dry_run = request.args.get("dry_run", "false").lower() == "true"
|
||
result = cleanup_expired_data(dry_run=dry_run)
|
||
return jsonify({"code": 0, "data": result})
|