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, Lead, MediaAsset, Product, ProductImage, SiteVersion, ThemeCard, utc_now, ) from ..schemas import AdminProductQuery, LeadQuery, LeadStatus, LeadStatusIn, LoginIn, ProductCreateIn, ProductStatus, ProductUpdateIn, SiteConfigPatchIn, SiteConfigReorderIn from ..seed import create_media, reset_guizhou_content from ..serializers import admin_product_dict, destination_dict, encode_value, 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}, }, "themes": { "model": ThemeCard, "entity": "theme_card", "primary": "label", "fields": {"label", "image", "targetType", "targetValue", "isActive", "sortOrder"}, "none_to_empty": {"image"}, "create_defaults": {"image": ""}, }, "ctaBanners": { "model": CtaBanner, "entity": "cta_banner", "primary": "alt", "fields": {"alt", "image", "targetType", "targetValue", "isActive", "sortOrder"}, "none_to_empty": {"image", "targetType"}, "create_defaults": {"image": "", "targetType": ""}, }, } 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 site_field_value(config: dict, field: str, value): value = clean_site_value(value) if value is None and field in config["none_to_empty"]: return "" return 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 site_item_dict(module: str, item) -> dict: if module == "heroSlides": return hero_slide_admin_dict(item) return destination_dict(item) if module == "destinations" else model_dict(item) def module_items(db: Session, model) -> list: return db.scalars(select(model).order_by(model.sortOrder.asc())).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 } payload[config["primary"]] = validate_site_primary(config, body) for field, value in config["create_defaults"].items(): payload.setdefault(field, value) payload["isActive"] = payload.get("isActive", True) if 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"]) 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 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}) setattr(item, field, value) 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], "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() ], } @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) 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) items = module_items(db, config["model"]) 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) 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) 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() normalize_site_sort_orders(module_items(db, config["model"])) 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)