from __future__ import annotations import json import mimetypes import os import shutil import uuid import zipfile from datetime import datetime, timezone from pathlib import Path import asyncio from fastapi import APIRouter, Depends, File, HTTPException, UploadFile from fastapi.responses import Response, StreamingResponse from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.api.deps import get_admin_user, get_db from app.config import settings from app.core.models.brand import Brand from app.core.models.product import Product from app.core.models.user import User from app.plugins.opencart_import.importer import import_brands, import_categories, import_products_stream, sync_images_to_storage from app.plugins.opencart_import.schemas import ( AvailableFilesResponse, CustomerGroupMeta, ImageSyncRequest, LanguageMeta, MappingConfig, OcMeta, OptionTypeMeta, PreviewResponse, RunResponse, ServerFile, UploadResponse, UseFileRequest, ) router = APIRouter(prefix="/admin/opencart-import", tags=["OpenCart 迁移"]) # Separate router registered at /api level (no /admin prefix) for serving images static_router = APIRouter() TMP_BASE = Path("/tmp") _HERE = Path(__file__).parent # backend/ 根目录（plugins/opencart_import/ 上三级） _BACKEND_ROOT = _HERE.parent.parent.parent STATIC_OC_DIR = _BACKEND_ROOT / "static" / "uploads" / "opencart" STATIC_OC_DIR.mkdir(parents=True, exist_ok=True) # FTP / 手动上传投放目录：与 backend/ 同级的 storage/opencart_drop/ # 路径完全相对于项目，不依赖任何绝对路径 DROP_DIR = _BACKEND_ROOT / "storage" / "opencart_drop" DROP_DIR.mkdir(parents=True, exist_ok=True) def _image_base_url(tenant_domain: str | None = None) -> str: """ 构建图片访问 URL 的基础路径。 优先使用租户的公开域名（HTTPS），保证 Store 前端能正常加载图片。 回退到 API_BASE_URL（兼容旧配置）。 """ if tenant_domain and tenant_domain not in ("localhost", "127.0.0.1"): # 自动补协议：内网 IP / 本地开发回退 http，否则 https import re as _re if _re.match(r"^\d+\.\d+\.\d+\.\d+", tenant_domain): base = f"http://{tenant_domain}" else: base = f"https://{tenant_domain}" else: base = (getattr(settings, "API_BASE_URL", "") or "").rstrip("/") return f"{base}/api/static/uploads/opencart" def _tmp_dir(import_id: str) -> Path: return TMP_BASE / f"oc_import_{import_id}" def _load_data(import_id: str) -> dict: data_file = _tmp_dir(import_id) / "data.json" if not data_file.exists(): raise HTTPException(status_code=404, detail="Import session not found or expired") with open(data_file, "r", encoding="utf-8") as f: return json.load(f) # ── image serve endpoint (path-aware) ──────────────────────── @static_router.get("/static/uploads/opencart/{path:path}", include_in_schema=False) async def serve_oc_image(path: str): filepath = STATIC_OC_DIR / path if not filepath.exists() or not filepath.is_file(): raise HTTPException(status_code=404, detail="Image not found") mime, _ = mimetypes.guess_type(str(filepath)) mime = mime or "application/octet-stream" return Response(content=filepath.read_bytes(), media_type=mime) # ── shared helper ───────────────────────────────────────────── def _extract_zip_and_build_response(zip_path: Path, import_id: str) -> UploadResponse: tmp_dir = _tmp_dir(import_id) tmp_dir.mkdir(parents=True, exist_ok=True) try: with zipfile.ZipFile(zip_path, "r") as zf: zf.extractall(tmp_dir) except zipfile.BadZipFile: shutil.rmtree(tmp_dir, ignore_errors=True) raise HTTPException(status_code=400, detail="ZIP 文件损坏") data = _load_data(import_id) meta_raw = data.get("meta", {}) meta = OcMeta( exported_at=meta_raw.get("exported_at", ""), oc_prefix=meta_raw.get("oc_prefix", "oc_"), languages=[LanguageMeta(**la) for la in meta_raw.get("languages", [])], option_types=[OptionTypeMeta(**o) for o in meta_raw.get("option_types", [])], customer_groups=[CustomerGroupMeta(**c) for c in meta_raw.get("customer_groups", [])], counts=meta_raw.get("counts", {}), ) return UploadResponse(import_id=import_id, meta=meta) # ── server-side file listing & selection ───────────────────── @router.get("/available-files", response_model=AvailableFilesResponse) async def list_available_files( admin: User = Depends(get_admin_user), ): """列出通过 FTP 或其他方式上传到服务器 drop 目录的 zip 文件。""" files: list[ServerFile] = [] if DROP_DIR.exists(): for p in sorted(DROP_DIR.iterdir()): if p.is_file() and p.suffix.lower() == ".zip": stat = p.stat() files.append(ServerFile(filename=p.name, size=stat.st_size, mtime=stat.st_mtime)) return AvailableFilesResponse(drop_dir=str(DROP_DIR), files=files) @router.post("/use-file", response_model=UploadResponse) async def use_server_file( req: UseFileRequest, admin: User = Depends(get_admin_user), ): """使用 drop 目录中已有的 zip 文件创建导入 session，无需重新上传。""" zip_path = DROP_DIR / req.filename if not zip_path.exists() or not zip_path.is_file(): raise HTTPException(status_code=404, detail=f"文件不存在：{req.filename}") if zip_path.suffix.lower() != ".zip": raise HTTPException(status_code=400, detail="只支持 .zip 文件") # Prevent path traversal try: zip_path.relative_to(DROP_DIR) except ValueError: raise HTTPException(status_code=400, detail="非法文件路径") import_id = str(uuid.uuid4()) return _extract_zip_and_build_response(zip_path, import_id) # ── upload ──────────────────────────────────────────────────── @router.post("/upload", response_model=UploadResponse) async def upload_zip( file: UploadFile = File(...), admin: User = Depends(get_admin_user), ): if not (file.filename or "").endswith(".zip"): raise HTTPException(status_code=400, detail="请上传 .zip 文件") import_id = str(uuid.uuid4()) tmp_dir = _tmp_dir(import_id) tmp_dir.mkdir(parents=True, exist_ok=True) zip_path = tmp_dir / "upload.zip" content = await file.read() zip_path.write_bytes(content) response = _extract_zip_and_build_response(zip_path, import_id) zip_path.unlink(missing_ok=True) return response # ── preview ─────────────────────────────────────────────────── @router.post("/{import_id}/preview", response_model=PreviewResponse) async def preview( import_id: str, cfg: MappingConfig, db: AsyncSession = Depends(get_db), admin: User = Depends(get_admin_user), ): data = _load_data(import_id) tenant_id = admin.tenant_id existing_skus: set[str] = set() result = await db.execute(select(Product.sku).where(Product.tenant_id == tenant_id)) for row in result.scalars(): existing_skus.add(row) oc_skus = {p["model"] for p in data.get("products", [])} conflicts = len(oc_skus & existing_skus) return PreviewResponse( brands_total=len(data.get("brands", [])), categories_total=len(data.get("categories", [])), products_total=len(data.get("products", [])), sku_conflicts=conflicts, ) # ── run (SSE 流式推送) ───────────────────────────────────────── @router.post("/{import_id}/run") async def run_import( import_id: str, cfg: MappingConfig, db: AsyncSession = Depends(get_db), admin: User = Depends(get_admin_user), ): """ 流式导入：通过 SSE 实时推送进度。 遇到第一个失败立即推 error 事件并停止，调用方无需轮询。 """ data = _load_data(import_id) tenant_id = admin.tenant_id tmp_images_dir = _tmp_dir(import_id) / "images" from sqlalchemy import select as _select from app.core.models.tenant import Tenant as _Tenant _tenant_row = (await db.execute( _select(_Tenant.domain).where(_Tenant.id == tenant_id) )).scalar_one_or_none() image_base_url = _image_base_url(_tenant_row) oc_img_base = (cfg.oc_image_base_url or "").rstrip("/") async def sse_gen(): brands_created = brands_skipped = categories_created = 0 try: # 1. brands brands_created, brands_skipped = await import_brands( db, data.get("brands", []), tenant_id, image_base_url, cfg.conflict_strategy, oc_img_base=oc_img_base, ) # rebuild brand_id_map brand_id_map: dict[int, int] = {} for oc_b in data.get("brands", []): res = await db.execute(select(Brand).where(Brand.tenant_id == tenant_id, Brand.name == oc_b["name"])) b = res.scalars().first() if b: brand_id_map[oc_b["id"]] = b.id # 2. categories categories_created, cat_id_map = await import_categories( db, data.get("categories", []), cfg.language_id, tenant_id, image_base_url, cfg.conflict_strategy, oc_img_base=oc_img_base, english_language_id=cfg.english_language_id, ) # 3. products — 流式处理 final_stats = None had_error = False async for event in import_products_stream( db, data.get("products", []), cfg, tenant_id, brand_id_map, cat_id_map, image_base_url, tmp_images_dir, STATIC_OC_DIR, ): if event["type"] == "error": had_error = True await db.rollback() yield f"data: {json.dumps(event, ensure_ascii=False)}\n\n" return if event["type"] == "done": final_stats = event["stats"] await db.commit() else: yield f"data: {json.dumps(event, ensure_ascii=False)}\n\n" await asyncio.sleep(0) # 允许 uvicorn 立即发送此帧 if final_stats is None: return # 汇总结果 from app.plugins.opencart_import.schemas import RunResponse, ImportError as OcImportError product_result = RunResponse( brands_created=brands_created, brands_skipped=brands_skipped, categories_created=categories_created, products_created=final_stats["products_created"], products_skipped=final_stats["products_skipped"], products_overwritten=final_stats["products_overwritten"], variants_created=final_stats["variants_created"], tier_prices_created=final_stats["tier_prices_created"], images_copied=final_stats["images_copied"], images_missing=final_stats["images_missing"], errors=[OcImportError(**e) for e in final_stats["errors"]], skipped_skus=final_stats["skipped_skus"], ) # 保存报告文件 total_source = len(data.get("products", [])) report = { "timestamp": datetime.now(timezone.utc).isoformat(), "import_id": import_id, "summary": { "total_in_source": total_source, "products_created": product_result.products_created, "products_overwritten": product_result.products_overwritten, "products_skipped": product_result.products_skipped, "products_failed": len(product_result.errors), }, "skipped": [{"sku": s, "reason": "已存在，跳过（skip策略）"} for s in product_result.skipped_skus], "failed": [{"sku": e.sku, "name": e.name, "reason": e.reason} for e in product_result.errors], } report_file = STATIC_OC_DIR / f"import_report_{import_id}.json" report_file.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8") product_result.error_log_url = f"/admin/opencart-import/{import_id}/report" shutil.rmtree(_tmp_dir(import_id), ignore_errors=True) result_dict = product_result.model_dump() yield f"data: {json.dumps({'type': 'result', 'data': result_dict}, ensure_ascii=False)}\n\n" except Exception as exc: await db.rollback() yield f"data: {json.dumps({'type': 'error', 'sku': '', 'name': '', 'reason': str(exc)}, ensure_ascii=False)}\n\n" return StreamingResponse( sse_gen(), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "X-Accel-Buffering": "no", # 告知 Nginx 不要缓冲此响应 }, ) # ── 下载导入报告（需要管理员认证）──────────────────────────────── @router.get("/{import_id}/report") async def get_import_report( import_id: str, admin: User = Depends(get_admin_user), ): """下载指定导入批次的完整报告（含失败和跳过的商品列表）。""" report_file = STATIC_OC_DIR / f"import_report_{import_id}.json" if not report_file.exists(): raise HTTPException(status_code=404, detail="报告不存在或已被清理") return Response( content=report_file.read_bytes(), media_type="application/json", headers={"Content-Disposition": f"attachment; filename=import_report_{import_id[:8]}.json"}, ) # ── cleanup ─────────────────────────────────────────────────── @router.delete("/{import_id}") async def cleanup( import_id: str, admin: User = Depends(get_admin_user), ): shutil.rmtree(_tmp_dir(import_id), ignore_errors=True) return {"ok": True} # ── post-import: sync images to CDN storage ─────────────────── @router.post("/image-sync") async def image_sync( req: ImageSyncRequest, db: AsyncSession = Depends(get_db), admin: User = Depends(get_admin_user), ): """ 将 DB 中所有指向旧 OpenCart 站的图片 URL 并发下载并直传到 租户当前配置的存储后端（OSS / S3 / R2 / 本地），完成后更新 DB。 通过 SSE 实时推送进度。 """ from app.core.models.tenant import Tenant from app.core.models.storage_profile import StorageProfile from app.core.storage import get_storage_for_profile tenant_id = admin.tenant_id oc_img_base = (req.oc_image_base_url or "").rstrip("/") if not oc_img_base: raise HTTPException(status_code=400, detail="oc_image_base_url 不能为空") # 解析租户存储配置 tr = await db.execute(select(Tenant).where(Tenant.id == tenant_id)) tenant = tr.scalar_one_or_none() if not tenant: raise HTTPException(status_code=404, detail="租户不存在") profile = None if tenant.storage_profile_id: pr = await db.execute(select(StorageProfile).where(StorageProfile.id == tenant.storage_profile_id)) profile = pr.scalar_one_or_none() storage = get_storage_for_profile(profile) storage_prefix = f"tenant_{tenant_id}/" async def sse_gen(): try: async for event in sync_images_to_storage( db, tenant_id, oc_img_base, storage, storage_prefix ): yield f"data: {json.dumps(event, ensure_ascii=False)}\n\n" if event.get("type") in ("done",): return except Exception as exc: yield f"data: {json.dumps({'type': 'error', 'reason': str(exc)}, ensure_ascii=False)}\n\n" return StreamingResponse( sse_gen(), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "X-Accel-Buffering": "no", }, )