import base64 import hashlib import hmac import http.client import os import re from email.utils import formatdate from uuid import uuid4 from urllib.parse import quote, urlsplit, urlunsplit from fastapi import APIRouter, Depends, File, Form, HTTPException, Query, Request, UploadFile, status from sqlalchemy import delete, func, or_, select from sqlalchemy.orm import Session, selectinload from ..auth import create_token, get_actor_id, require_admin, verify_password from ..config import get_settings from ..database import get_db from ..models import ( AdminUser, AuditLog, Campaign, CtaBanner, Destination, HeroSlide, HotelGroup, Lead, MapImage, MediaAsset, Product, ProductImage, RouteSection, RouteSectionProduct, SiteVersion, ThemeCard, VehicleOption, utc_now, ) from ..schemas import AdminProductQuery, LeadQuery, LeadStatus, LeadStatusIn, LoginIn, ProductCreateIn, ProductStatus, ProductUpdateIn, SiteConfigPatchIn, SiteConfigReorderIn from ..seed import create_media, reset_guizhou_content from ..route_sections import replace_route_section_products, route_section_dict, route_section_query from ..serializers import admin_product_dict, destination_dict, encode_value, hotel_group_dict, lead_dict, model_dict from .shared import site_config router = APIRouter(prefix="/api/admin") MEDIA_UPLOAD_MAX_BYTES = 5 * 1024 * 1024 MEDIA_ALLOWED_MIME_TYPES = { "image/jpeg": "jpg", "image/png": "png", "image/webp": "webp", "image/gif": "gif", } SITE_CONFIG_MODULES = { "heroSlides": { "model": HeroSlide, "entity": "hero_slide", "primary": "title", "fields": {"title", "kicker", "image", "isActive", "sortOrder"}, "none_to_empty": {"image"}, "create_defaults": {"kicker": "", "image": ""}, }, "destinations": { "model": Destination, "entity": "destination", "primary": "name", "fields": {"name", "slug", "region", "image", "isHot", "isActive", "sortOrder"}, "none_to_empty": set(), "create_defaults": {"isHot": False}, }, "map": { "model": MapImage, "entity": "map_image", "primary": "image", "fields": {"image", "isActive"}, "none_to_empty": set(), "create_defaults": {}, "ordered": False, "singleton": True, }, "themes": { "model": ThemeCard, "entity": "theme_card", "primary": "label", "fields": {"label", "image", "targetType", "targetValue", "isActive", "sortOrder"}, "none_to_empty": {"image"}, "create_defaults": {"image": ""}, }, "campaigns": { "model": Campaign, "entity": "campaign", "primary": "title", "fields": {"title", "slug", "description", "coverImage", "priceAmount", "priceUnit", "tags", "status", "startsAt", "endsAt"}, "none_to_empty": set(), "empty_to_none": {"description", "coverImage", "priceUnit", "startsAt", "endsAt"}, "create_defaults": {"description": None, "coverImage": None, "priceAmount": None, "priceUnit": "起/人", "tags": [], "status": "draft", "startsAt": None, "endsAt": None}, "ordered": False, }, "routeSections": { "model": RouteSection, "entity": "route_section", "primary": "title", "fields": {"title", "subtitle", "isActive", "sortOrder", "productIds"}, "none_to_empty": set(), "empty_to_none": {"subtitle"}, "create_defaults": {}, }, "hotelGroups": { "model": HotelGroup, "entity": "hotel_group", "primary": "title", "fields": {"title", "description", "image", "coverImage", "priceAmount", "priceUnit", "tags", "status", "isActive", "sortOrder"}, "none_to_empty": set(), "empty_to_none": {"description", "image", "coverImage", "priceUnit"}, "create_defaults": {"description": None, "image": None, "coverImage": None, "priceAmount": None, "priceUnit": "起/晚", "tags": [], "status": "published"}, }, "vehicleOptions": { "model": VehicleOption, "entity": "vehicle_option", "primary": "title", "fields": {"title", "description", "image", "isActive", "sortOrder"}, "none_to_empty": set(), "empty_to_none": {"description", "image"}, "create_defaults": {"description": None, "image": None}, }, "ctaBanners": { "model": CtaBanner, "entity": "cta_banner", "primary": "alt", "fields": {"alt", "image", "targetType", "targetValue", "isActive", "sortOrder"}, "none_to_empty": {"image", "targetType"}, "create_defaults": {"image": "", "targetType": ""}, }, } PUBLISH_STATUSES = {"draft", "published"} def media_error(status_code: int, message: str, code: str, details: dict | None = None) -> None: raise HTTPException(status_code=status_code, detail={"message": message, "code": code, "details": details or {}}) def normalize_media_group(group: str | None) -> str: value = (group or "general").strip() or "general" if len(value) > 32 or not re.fullmatch(r"[A-Za-z0-9_-]+", value): media_error(400, "素材分组只能包含字母、数字、下划线和中划线", "MEDIA_UPLOAD_INVALID_GROUP", {"group": group}) return value def safe_media_name(filename: str | None) -> str: name = os.path.basename((filename or "upload").replace("\\", "/")).strip() return name[:120] or "upload" def detect_image_mime(head: bytes) -> str | None: if head.startswith(b"\xff\xd8\xff"): return "image/jpeg" if head.startswith(b"\x89PNG\r\n\x1a\n"): return "image/png" if head.startswith((b"GIF87a", b"GIF89a")): return "image/gif" if len(head) >= 12 and head[:4] == b"RIFF" and head[8:12] == b"WEBP": return "image/webp" return None def inspect_upload_image(file: UploadFile) -> tuple[str, int]: content_type = (file.content_type or "").split(";", 1)[0].strip().lower() if content_type == "image/jpg": content_type = "image/jpeg" if content_type and content_type not in MEDIA_ALLOWED_MIME_TYPES and content_type != "application/octet-stream": media_error(400, "仅支持上传 JPG、PNG、WebP 或 GIF 图片", "MEDIA_UPLOAD_INVALID_TYPE", {"mimeType": content_type}) file.file.seek(0, os.SEEK_END) size_bytes = file.file.tell() file.file.seek(0) if size_bytes <= 0: media_error(400, "上传文件不能为空", "MEDIA_UPLOAD_EMPTY_FILE") if size_bytes > MEDIA_UPLOAD_MAX_BYTES: media_error( 413, "图片大小不能超过 5MB", "MEDIA_UPLOAD_TOO_LARGE", {"maxBytes": MEDIA_UPLOAD_MAX_BYTES, "sizeBytes": size_bytes}, ) head = file.file.read(512) file.file.seek(0) detected = detect_image_mime(head) if not detected: media_error(400, "仅支持上传 JPG、PNG、WebP 或 GIF 图片", "MEDIA_UPLOAD_INVALID_TYPE") if content_type in MEDIA_ALLOWED_MIME_TYPES and content_type != detected: media_error( 400, "图片内容与文件类型不一致", "MEDIA_UPLOAD_TYPE_MISMATCH", {"mimeType": content_type, "detectedMimeType": detected}, ) return detected, size_bytes def media_object_key(group: str, mime_type: str) -> str: extension = MEDIA_ALLOWED_MIME_TYPES[mime_type] return f"admin/{group}/{utc_now().strftime('%Y/%m/%d')}/{uuid4().hex}.{extension}" def oss_settings(): settings = get_settings() values = { "OSS_ACCESS_KEY_ID": settings.oss_access_key_id, "OSS_ACCESS_KEY_SECRET": settings.oss_access_key_secret, "OSS_ENDPOINT": settings.oss_endpoint, "OSS_BUCKET_NAME": settings.oss_bucket_name, } missing = [key for key, value in values.items() if not (value and value.strip())] if missing: media_error(503, "OSS 存储配置不完整", "MEDIA_STORAGE_NOT_CONFIGURED", {"missing": missing}) return settings def normalized_oss_host(endpoint: str, bucket: str) -> tuple[str, str]: raw = endpoint.strip().rstrip("/") if "://" not in raw: raw = f"https://{raw}" parsed = urlsplit(raw) if parsed.scheme not in {"http", "https"} or not parsed.netloc: media_error(503, "OSS Endpoint 配置不正确", "MEDIA_STORAGE_INVALID_ENDPOINT") host = parsed.netloc if not host.startswith(f"{bucket}."): host = f"{bucket}.{host}" return parsed.scheme, host def upload_image_to_oss(file_obj, key: str, mime_type: str, size_bytes: int) -> str: settings = oss_settings() bucket = settings.oss_bucket_name.strip() scheme, host = normalized_oss_host(settings.oss_endpoint, bucket) date = formatdate(usegmt=True) canonical_resource = f"/{bucket}/{key}" string_to_sign = f"PUT\n\n{mime_type}\n{date}\n{canonical_resource}" signature = base64.b64encode( hmac.new(settings.oss_access_key_secret.strip().encode("utf-8"), string_to_sign.encode("utf-8"), hashlib.sha1).digest() ).decode("ascii") quoted_key = quote(key, safe="/-_.~") headers = { "Authorization": f"OSS {settings.oss_access_key_id.strip()}:{signature}", "Content-Type": mime_type, "Content-Length": str(size_bytes), "Date": date, } connection_class = http.client.HTTPSConnection if scheme == "https" else http.client.HTTPConnection connection = connection_class(host, timeout=20) try: file_obj.seek(0) connection.request("PUT", f"/{quoted_key}", body=file_obj, headers=headers) response = connection.getresponse() response.read(2048) if response.status < 200 or response.status >= 300: media_error(502, "OSS 上传失败", "MEDIA_STORAGE_UPLOAD_FAILED", {"status": response.status}) except OSError: media_error(502, "OSS 上传失败", "MEDIA_STORAGE_UPLOAD_FAILED") finally: connection.close() return urlunsplit((scheme, host, f"/{quoted_key}", "", "")) def normalize_images(images): return [ {"url": image.url.strip(), "alt": image.alt.strip() if image.alt else None, "sortOrder": index} for index, image in enumerate(images or []) if image.url.strip() ] def normalize_detail_sections(sections): normalized = [] for section in sections or []: blocks = [] for block in section.blocks: if block.type == "image": url = block.url.strip() if url: blocks.append({"type": "image", "url": url, "alt": block.alt.strip() if block.alt else None}) else: text = block.text.strip() if text: blocks.append({"type": "text", "text": text}) key = section.key.strip() label = section.label.strip() if key and label and blocks: normalized.append({"key": key, "label": label, "title": section.title.strip() if section.title else None, "blocks": blocks}) return normalized def audit(db: Session, actor_id: str | None, action: str, entity: str, entity_id: str | None = None, after=None, before=None) -> None: db.add( AuditLog( actorId=actor_id, action=action, entity=entity, entityId=entity_id, before=encode_value(before) if before is not None else None, after=encode_value(after) if after is not None else None, ) ) def site_config_error(status_code: int, message: str, code: str, details: dict | None = None) -> None: raise HTTPException(status_code=status_code, detail={"message": message, "code": code, "details": details or {}}) def site_module(module: str) -> dict: config = SITE_CONFIG_MODULES.get(module) if not config: site_config_error(400, "模块不存在或无权限操作", "MODULE_CONFIG_FORBIDDEN", {"module": module}) return config def site_slugify(value: str) -> str: return quote(value, safe="").replace("%", "").lower() def clean_site_value(value): return value.strip() if isinstance(value, str) else value def normalize_campaign_tags(value) -> list[str]: if value is None: return [] tags = [item.strip() for item in value if isinstance(item, str) and item.strip()] if len(tags) > 3: site_config_error(422, "tags cannot exceed 3", "MODULE_CONFIG_VALIDATION_ERROR", {"field": "tags", "max": 3}) return tags def site_field_value(config: dict, field: str, value): if field == "tags": return normalize_campaign_tags(value) value = clean_site_value(value) if value == "" and field in config.get("empty_to_none", set()): return None if value is None and field in config["none_to_empty"]: return "" return value def validate_publish_status(value): status_value = clean_site_value(value) if status_value not in PUBLISH_STATUSES: site_config_error(422, "status must be draft or published", "MODULE_CONFIG_VALIDATION_ERROR", {"field": "status"}) return status_value def validate_site_primary(config: dict, body: SiteConfigPatchIn) -> str: primary = config["primary"] value = clean_site_value(getattr(body, primary)) if not value: site_config_error(422, "必填字段不能为空", "MODULE_CONFIG_VALIDATION_ERROR", {"field": primary}) return value def next_site_sort_order(db: Session, model) -> int: highest = db.scalar(select(func.max(model.sortOrder))) return (highest + 1) if highest is not None else 0 def hero_slide_admin_dict(item: HeroSlide) -> dict: return { "id": item.id, "title": item.title, "kicker": item.kicker or None, "image": item.image or None, "isActive": item.isActive, "sortOrder": item.sortOrder, "createdAt": encode_value(item.createdAt), "updatedAt": encode_value(item.updatedAt), } def map_image_admin_dict(item: MapImage) -> dict: return { "id": item.id, "image": item.image or None, "isActive": item.isActive, "createdAt": encode_value(item.createdAt), "updatedAt": encode_value(item.updatedAt), } def site_item_dict(module: str, item) -> dict: if module == "heroSlides": return hero_slide_admin_dict(item) if module == "map": return map_image_admin_dict(item) if module == "routeSections": return route_section_dict(item) if module == "hotelGroups": return hotel_group_dict(item) return destination_dict(item) if module == "destinations" else model_dict(item) def module_items(db: Session, config: dict) -> list: model = config["model"] stmt = select(model) if model is RouteSection: stmt = stmt.options(selectinload(RouteSection.products).selectinload(RouteSectionProduct.product)) if config.get("ordered", True): stmt = stmt.order_by(model.sortOrder.asc()) else: stmt = stmt.order_by(model.createdAt.asc()) return db.scalars(stmt).all() def normalize_site_sort_orders(items: list) -> None: for index, item in enumerate(items): item.sortOrder = index def site_create_payload(module: str, config: dict, body: SiteConfigPatchIn, db: Session) -> dict: fields = body.model_fields_set payload = { field: site_field_value(config, field, getattr(body, field)) for field in config["fields"] if field in fields and field != "productIds" } payload[config["primary"]] = validate_site_primary(config, body) for field, value in config["create_defaults"].items(): payload.setdefault(field, value) if "isActive" in config["fields"]: payload["isActive"] = payload.get("isActive", True) if config.get("ordered", True) and payload.get("sortOrder") is None: payload["sortOrder"] = next_site_sort_order(db, config["model"]) if module == "destinations": slug = clean_site_value(payload.get("slug")) payload["slug"] = slug or site_slugify(payload["name"]) if module == "campaigns": slug = clean_site_value(payload.get("slug")) if not slug: site_config_error(422, "required field is empty", "MODULE_CONFIG_VALIDATION_ERROR", {"field": "slug"}) payload["slug"] = slug if not payload.get("priceUnit"): payload["priceUnit"] = "起/人" payload["status"] = validate_publish_status(payload.get("status", "draft")) if module == "hotelGroups": if not payload.get("priceUnit"): payload["priceUnit"] = "起/晚" payload["status"] = validate_publish_status(payload.get("status", "published")) if not payload.get("coverImage") and payload.get("image"): payload["coverImage"] = payload["image"] if not payload.get("image") and payload.get("coverImage"): payload["image"] = payload["coverImage"] if "status" in fields and "isActive" not in fields: payload["isActive"] = payload["status"] == "published" if "isActive" in fields and "status" not in fields: payload["status"] = "published" if payload.get("isActive") else "draft" return payload def apply_site_patch(module: str, config: dict, item, body: SiteConfigPatchIn) -> None: fields = body.model_fields_set for field in config["fields"]: if field not in fields: continue if field == "productIds": continue value = site_field_value(config, field, getattr(body, field)) if field == config["primary"] and not value: site_config_error(422, "必填字段不能为空", "MODULE_CONFIG_VALIDATION_ERROR", {"field": field}) if module == "destinations" and field == "slug" and not value: site_config_error(422, "必填字段不能为空", "MODULE_CONFIG_VALIDATION_ERROR", {"field": field}) if module == "campaigns" and field == "slug" and not value: site_config_error(422, "required field is empty", "MODULE_CONFIG_VALIDATION_ERROR", {"field": field}) if module in {"campaigns", "hotelGroups"} and field == "status": value = validate_publish_status(value) setattr(item, field, value) if module == "hotelGroups": if "coverImage" in fields and "image" not in fields: item.image = item.coverImage if "image" in fields and "coverImage" not in fields: item.coverImage = item.image if "status" in fields and "isActive" not in fields: item.isActive = item.status == "published" if "isActive" in fields and "status" not in fields: item.status = "published" if item.isActive else "draft" def create_route_section_config(body: SiteConfigPatchIn, request: Request, db: Session) -> dict: config = site_module("routeSections") items = module_items(db, config) fields = body.model_fields_set now = utc_now() section = RouteSection( title=validate_site_primary(config, body), subtitle=site_field_value(config, "subtitle", body.subtitle) if "subtitle" in fields else None, sortOrder=( site_field_value(config, "sortOrder", body.sortOrder) if "sortOrder" in fields and body.sortOrder is not None else max([item.sortOrder for item in items], default=-1) + 1 ), isActive=site_field_value(config, "isActive", body.isActive) if "isActive" in fields else True, createdAt=now, updatedAt=now, ) section.products = [] db.add(section) db.flush() if "productIds" in fields: replace_route_section_products(db, section, body.productIds or []) db.flush() after = route_section_dict(section) audit(db, get_actor_id(request), "create", config["entity"], section.id, after) db.commit() return after def load_product(db: Session, product_id: str) -> Product: return db.scalars( select(Product) .options(selectinload(Product.destination).selectinload(Destination.aliases), selectinload(Product.images)) .where(Product.id == product_id) ).one() @router.post("/auth/login") def login(body: LoginIn, db: Session = Depends(get_db)): user = db.scalar(select(AdminUser).where(AdminUser.email == body.email)) if not user or not user.isActive or not verify_password(body.password, user.passwordHash): raise HTTPException(status_code=401, detail="账号或密码错误") return { "token": create_token(user), "user": {"id": user.id, "email": user.email, "name": user.name, "role": user.role}, } @router.get("/me") def me(user: AdminUser = Depends(require_admin)): return {"id": user.id, "email": user.email, "name": user.name, "role": user.role} @router.get("/dashboard") def dashboard(_user: AdminUser = Depends(require_admin), db: Session = Depends(get_db)): stats = { "productCount": db.scalar(select(func.count()).select_from(Product)), "publishedProductCount": db.scalar(select(func.count()).select_from(Product).where(Product.status == "published")), "destinationCount": db.scalar(select(func.count()).select_from(Destination).where(Destination.isActive.is_(True))), "newLeadCount": db.scalar(select(func.count()).select_from(Lead).where(Lead.status == "new")), "leadCount": db.scalar(select(func.count()).select_from(Lead)), "campaignCount": db.scalar(select(func.count()).select_from(Campaign)), } recent = db.scalars( select(Lead).options(selectinload(Lead.sourceProduct), selectinload(Lead.assignedUser)).order_by(Lead.createdAt.desc()).limit(5) ).all() return {"stats": stats, "recentLeads": [lead_dict(lead) for lead in recent]} @router.get("/products") def list_products( keyword: str | None = None, status_value: ProductStatus | None = Query(default=None, alias="status"), take: int = Query(default=100, ge=1, le=200), _user: AdminUser = Depends(require_admin), db: Session = Depends(get_db), ): query = AdminProductQuery(keyword=keyword, status=status_value, take=take) stmt = ( select(Product) .options(selectinload(Product.destination).selectinload(Destination.aliases), selectinload(Product.images)) .order_by(Product.sortWeight.asc(), Product.updatedAt.desc()) .limit(query.take) ) if query.status: stmt = stmt.where(Product.status == query.status) if query.keyword: pattern = f"%{query.keyword}%" stmt = stmt.where(or_(Product.title.ilike(pattern), Product.subtitle.ilike(pattern), Product.tags.any(query.keyword))) products = db.scalars(stmt).unique().all() return {"items": [admin_product_dict(product) for product in products]} @router.post("/products", status_code=status.HTTP_201_CREATED) def create_product(body: ProductCreateIn, request: Request, _user: AdminUser = Depends(require_admin), db: Session = Depends(get_db)): images = normalize_images(body.images) sections = normalize_detail_sections(body.detailSections) payload = body.model_dump(exclude={"images", "detailSections"}) payload["priceUnit"] = body.priceUnit or "起/人" payload["detailSections"] = sections or None payload["publishedAt"] = utc_now() if body.status == "published" else None product = Product(**payload) db.add(product) db.flush() for image in images: db.add(ProductImage(productId=product.id, **image)) create_media(db, image["url"], "product-detail", image["alt"] or product.title) db.flush() product = load_product(db, product.id) audit(db, get_actor_id(request), "create", "product", product.id, admin_product_dict(product)) db.commit() return admin_product_dict(product) @router.patch("/products/{product_id}") def update_product(product_id: str, body: ProductUpdateIn, request: Request, _user: AdminUser = Depends(require_admin), db: Session = Depends(get_db)): product = db.scalars( select(Product) .options(selectinload(Product.images), selectinload(Product.destination).selectinload(Destination.aliases)) .where(Product.id == product_id) ).first() if not product: raise HTTPException(status_code=404, detail="线路不存在") before = admin_product_dict(product) fields = body.model_fields_set payload = body.model_dump(exclude={"images", "detailSections"}, exclude_unset=True) for key, value in payload.items(): setattr(product, key, value) if "detailSections" in fields: product.detailSections = normalize_detail_sections(body.detailSections) if "status" in fields and body.status == "published" and before.get("status") != "published": product.publishedAt = utc_now() if "images" in fields: db.execute(delete(ProductImage).where(ProductImage.productId == product.id)) for image in normalize_images(body.images): db.add(ProductImage(productId=product.id, **image)) create_media(db, image["url"], "product-detail", image["alt"] or product.title) db.flush() product = load_product(db, product.id) audit(db, get_actor_id(request), "update", "product", product.id, admin_product_dict(product), before) db.commit() return admin_product_dict(product) @router.get("/destinations") def admin_destinations(_user: AdminUser = Depends(require_admin), db: Session = Depends(get_db)): destinations = db.scalars( select(Destination).options(selectinload(Destination.aliases)).order_by(Destination.sortOrder.asc()) ).all() counts = {} if destinations: rows = db.execute( select(Product.destinationId, func.count(Product.id)) .where(Product.destinationId.in_([destination.id for destination in destinations])) .group_by(Product.destinationId) ).all() counts = {destination_id: count for destination_id, count in rows if destination_id} return {"items": [destination_dict(destination, include_count=True, product_count=counts.get(destination.id, 0)) for destination in destinations]} @router.get("/site-config") def admin_site_config(_user: AdminUser = Depends(require_admin), db: Session = Depends(get_db)): destinations = db.scalars( select(Destination).options(selectinload(Destination.aliases)).order_by(Destination.sortOrder.asc()) ).all() return { "heroSlides": [ hero_slide_admin_dict(item) for item in db.scalars(select(HeroSlide).order_by(HeroSlide.sortOrder.asc())).all() ], "destinations": [destination_dict(item) for item in destinations], "map": [ map_image_admin_dict(item) for item in db.scalars(select(MapImage).order_by(MapImage.createdAt.asc())).all() ], "themes": [ model_dict(item) for item in db.scalars(select(ThemeCard).order_by(ThemeCard.sortOrder.asc())).all() ], "ctaBanners": [ model_dict(item) for item in db.scalars(select(CtaBanner).order_by(CtaBanner.sortOrder.asc())).all() ], "campaigns": [ model_dict(item) for item in db.scalars(select(Campaign).order_by(Campaign.updatedAt.desc())).all() ], "routeSections": [route_section_dict(item) for item in db.scalars(route_section_query()).all()], "hotelGroups": [ hotel_group_dict(item) for item in db.scalars(select(HotelGroup).order_by(HotelGroup.sortOrder.asc())).all() ], "vehicleOptions": [ model_dict(item) for item in db.scalars(select(VehicleOption).order_by(VehicleOption.sortOrder.asc())).all() ], } @router.post("/site-config/{module}", status_code=status.HTTP_201_CREATED) def create_site_config( module: str, body: SiteConfigPatchIn, request: Request, _user: AdminUser = Depends(require_admin), db: Session = Depends(get_db), ): config = site_module(module) if module == "routeSections": return create_route_section_config(body, request, db) if config.get("fixed"): site_config_error(405, "固定模块不支持新增", "MODULE_CONFIG_CREATE_UNSUPPORTED", {"module": module}) if config.get("singleton") and module_items(db, config): site_config_error(409, "地图图片已存在", "MAP_IMAGE_ALREADY_EXISTS", {"module": module}) item = config["model"](**site_create_payload(module, config, body, db)) db.add(item) db.flush() after = site_item_dict(module, item) audit(db, get_actor_id(request), "create", config["entity"], item.id, after) db.commit() return after @router.patch("/site-config/{module}/reorder") def reorder_site_config( module: str, body: SiteConfigReorderIn, request: Request, _user: AdminUser = Depends(require_admin), db: Session = Depends(get_db), ): config = site_module(module) if not config.get("ordered", True): site_config_error(400, "模块不支持排序", "MODULE_CONFIG_REORDER_UNSUPPORTED", {"module": module}) items = module_items(db, config) current_ids = [item.id for item in items] requested_ids = body.itemIds if len(set(requested_ids)) != len(requested_ids) or set(requested_ids) != set(current_ids): site_config_error( 400, "排序配置项必须完整且不能重复", "MODULE_CONFIG_REORDER_INVALID", {"module": module, "itemIds": requested_ids}, ) items_by_id = {item.id: item for item in items} ordered_items = [items_by_id[item_id] for item_id in requested_ids] before = [site_item_dict(module, item) for item in items] normalize_site_sort_orders(ordered_items) after = [site_item_dict(module, item) for item in ordered_items] audit(db, get_actor_id(request), "reorder", config["entity"], module, {"items": after}, {"items": before}) db.commit() return {"items": after} @router.patch("/site-config/{module}/{item_id}") def update_site_config( module: str, item_id: str, body: SiteConfigPatchIn, request: Request, _user: AdminUser = Depends(require_admin), db: Session = Depends(get_db), ): config = site_module(module) item = db.get(config["model"], item_id) if not item: site_config_error(404, "维护项不存在", "MODULE_CONFIG_NOT_FOUND", {"module": module, "id": item_id}) before = site_item_dict(module, item) apply_site_patch(module, config, item, body) if module == "routeSections" and "productIds" in body.model_fields_set: replace_route_section_products(db, item, body.productIds or []) db.flush() after = site_item_dict(module, item) audit(db, get_actor_id(request), "update", config["entity"], item.id, after, before) db.commit() return after @router.delete("/site-config/{module}/{item_id}") def delete_site_config( module: str, item_id: str, request: Request, _user: AdminUser = Depends(require_admin), db: Session = Depends(get_db), ): config = site_module(module) if config.get("fixed"): site_config_error(405, "固定模块不支持删除", "MODULE_CONFIG_DELETE_UNSUPPORTED", {"module": module}) item = db.get(config["model"], item_id) if not item: site_config_error(404, "维护项不存在", "MODULE_CONFIG_NOT_FOUND", {"module": module, "id": item_id}) if module == "destinations": product_count = db.scalar(select(func.count()).select_from(Product).where(Product.destinationId == item.id)) if product_count: site_config_error(409, "配置项仍被商品引用，不能删除", "MODULE_CONFIG_CONFLICT", {"module": module, "id": item_id}) before = site_item_dict(module, item) db.delete(item) db.flush() if config.get("ordered", True): normalize_site_sort_orders(module_items(db, config)) result = {"id": item_id} audit(db, get_actor_id(request), "delete", config["entity"], item_id, result, before) db.commit() return result @router.get("/leads") def list_leads( status_value: LeadStatus | None = Query(default=None, alias="status"), take: int = Query(default=100, ge=1, le=200), _user: AdminUser = Depends(require_admin), db: Session = Depends(get_db), ): query = LeadQuery(status=status_value, take=take) stmt = ( select(Lead) .options(selectinload(Lead.sourceProduct), selectinload(Lead.assignedUser)) .order_by(Lead.createdAt.desc()) .limit(query.take) ) if query.status: stmt = stmt.where(Lead.status == query.status) leads = db.scalars(stmt).all() return {"items": [lead_dict(lead) for lead in leads]} @router.patch("/leads/{lead_id}/status") def update_lead_status(lead_id: str, body: LeadStatusIn, request: Request, _user: AdminUser = Depends(require_admin), db: Session = Depends(get_db)): lead = db.get(Lead, lead_id) if not lead: raise HTTPException(status_code=404, detail="线索不存在") before = model_dict(lead) lead.status = body.status db.flush() audit(db, get_actor_id(request), "update_status", "lead", lead.id, model_dict(lead), before) db.commit() return model_dict(lead) @router.get("/media-assets") def list_media_assets(_user: AdminUser = Depends(require_admin), db: Session = Depends(get_db)): assets = db.scalars(select(MediaAsset).order_by(MediaAsset.createdAt.desc()).limit(200)).all() return {"items": [model_dict(asset) for asset in assets]} @router.post("/media-assets/upload", status_code=status.HTTP_201_CREATED) def upload_media_asset( request: Request, file: UploadFile = File(...), group: str | None = Form(default="general"), _user: AdminUser = Depends(require_admin), db: Session = Depends(get_db), ): safe_group = normalize_media_group(group) mime_type, size_bytes = inspect_upload_image(file) key = media_object_key(safe_group, mime_type) url = upload_image_to_oss(file.file, key, mime_type, size_bytes) asset = MediaAsset( url=url, name=safe_media_name(file.filename), mimeType=mime_type, sizeBytes=size_bytes, group=safe_group, ) db.add(asset) db.flush() after = model_dict(asset) audit(db, get_actor_id(request), "upload", "media_asset", asset.id, after) db.commit() return after @router.post("/reset-guizhou-content") def reset_content(request: Request, _user: AdminUser = Depends(require_admin), db: Session = Depends(get_db)): result = reset_guizhou_content(db) audit(db, get_actor_id(request), "reset_guizhou_content", "site_content", after=result) db.commit() return result @router.post("/publish", status_code=status.HTTP_201_CREATED) def publish(request: Request, _user: AdminUser = Depends(require_admin), db: Session = Depends(get_db)): snapshot = site_config(db, active_only=True) version = SiteVersion(title=f"manual-{utc_now().isoformat()}", status="published", snapshot=snapshot, publishedAt=utc_now()) db.add(version) db.flush() audit(db, get_actor_id(request), "publish", "site_version", version.id, model_dict(version)) db.commit() return model_dict(version)