feat(gui): replace AI studio with product suite

app/image_studio_images.py

@@ -25,6 +25,7 @@ DEFAULT_CONNECT_TIMEOUT_SECONDS = 5
 DEFAULT_READ_TIMEOUT_SECONDS = 30
 DEFAULT_THUMBNAIL_SIZE = 220
 DEFAULT_THUMBNAIL_WORKERS = 4
+MAX_ORIGINAL_ASSETS = 16
 
 
 class ImageStudioImageError(RuntimeError):
@@ -311,6 +312,203 @@ def _existing_local_asset(asset):
         return None
 
 
+def import_original_files(
+    project_id,
+    file_paths,
+    *,
+    path=None,
+    config=None,
+    image_root=None,
+    max_assets=MAX_ORIGINAL_ASSETS,
+):
+    """Validate and atomically copy local product images into one studio project."""
+
+    imported = []
+    errors = []
+    for file_path in file_paths or []:
+        try:
+            with open(os.path.abspath(str(file_path)), "rb") as fh:
+                content = fh.read(int(ORIGINAL_MAX_BYTES) + 1)
+            imported.append(
+                import_original_bytes(
+                    project_id,
+                    content,
+                    filename_hint=os.path.basename(str(file_path)),
+                    path=path,
+                    config=config,
+                    image_root=image_root,
+                    max_assets=max_assets,
+                )
+            )
+        except Exception as exc:
+            errors.append({"path": os.path.abspath(str(file_path)), "error": str(exc)})
+    return {"assets": imported, "errors": errors, "limit": int(max_assets)}
+
+
+def import_original_bytes(
+    project_id,
+    content,
+    *,
+    filename_hint="clipboard.png",
+    path=None,
+    config=None,
+    image_root=None,
+    max_assets=MAX_ORIGINAL_ASSETS,
+):
+    cfg = appconfig.load_config() if config is None else config
+    database_path = path or appconfig.db_path(cfg)
+    project = image_studio.get_project(project_id, path=database_path)
+    if project is None:
+        raise ImageStudioImageError("商品套图任务不存在")
+    image_bytes = bytes(content or b"")
+    if not image_bytes:
+        raise ImageStudioImageError("商品原图内容为空")
+    if len(image_bytes) > int(ORIGINAL_MAX_BYTES):
+        raise ImageStudioImageError("商品原图超过大小上限")
+    info = _image_info(image_bytes)
+    originals = image_studio.list_assets(
+        project.id,
+        kind=image_studio.ASSET_KIND_ORIGINAL,
+        path=database_path,
+    )
+    active_originals = [
+        asset for asset in originals if asset.status != image_studio.ASSET_STATUS_MISSING
+    ]
+    digest = hashlib.sha256(image_bytes).hexdigest()[:16]
+    duplicate = next(
+        (
+            asset
+            for asset in originals
+            if digest in os.path.basename(str(asset.local_path or ""))
+            and _existing_local_asset(asset) is not None
+        ),
+        None,
+    )
+    if duplicate is not None:
+        return duplicate
+    if len(active_originals) >= int(max_assets):
+        raise ImageStudioImageError("商品原图最多只能添加%d张" % int(max_assets))
+
+    directory = image_studio.project_image_dirs(
+        image_root or appconfig.image_dir(cfg),
+        project,
+    )["originals"]
+    os.makedirs(directory, exist_ok=True)
+    order = max([int(asset.source_order or 0) for asset in originals] + [0]) + 1
+    extension = _extension_for_format(info["format"])
+    safe_hint = os.path.splitext(os.path.basename(str(filename_hint or "image")))[0]
+    safe_hint = "".join(ch if ch.isalnum() or ch in "_-" else "_" for ch in safe_hint)
+    safe_hint = safe_hint.strip("_")[:32] or "image"
+    final_path = os.path.join(
+        directory,
+        "original_%02d_%s_%s%s" % (order, safe_hint, digest, extension),
+    )
+    temp_path = final_path + ".tmp-" + uuid.uuid4().hex
+    created = False
+    try:
+        with open(temp_path, "wb") as fh:
+            fh.write(image_bytes)
+        with open(temp_path, "rb") as fh:
+            _image_info(fh.read())
+        os.replace(temp_path, final_path)
+        created = True
+        return image_studio.add_asset(
+            project.id,
+            image_studio.ASSET_KIND_ORIGINAL,
+            local_path=final_path,
+            aspect_ratio="%d:%d" % (int(info["width"]), int(info["height"])),
+            source_order=order,
+            path=database_path,
+        )
+    except Exception as exc:
+        for candidate in (temp_path, final_path if created else None):
+            if candidate and os.path.exists(candidate):
+                try:
+                    os.remove(candidate)
+                except OSError:
+                    pass
+        if isinstance(exc, ImageStudioImageError):
+            raise
+        raise ImageStudioImageError(f"保存商品原图失败：{exc}") from exc
+
+
+def trash_generated_asset(asset_id, *, path=None, config=None, image_root=None):
+    """Move one generated image to the app-managed trash without losing DB history."""
+
+    cfg = appconfig.load_config() if config is None else config
+    database_path = path or appconfig.db_path(cfg)
+    asset = image_studio.get_asset(asset_id, path=database_path)
+    if asset is None or not str(asset.kind or "").startswith("generated_"):
+        raise ImageStudioImageError("只能删除商品套图生成结果")
+    source_path = os.path.abspath(str(asset.local_path or ""))
+    if not source_path or not os.path.isfile(source_path):
+        raise ImageStudioImageError("生成图片文件不存在")
+    project = image_studio.get_project(asset.project_id, path=database_path)
+    if project is None:
+        raise ImageStudioImageError("商品套图任务不存在")
+    root = image_studio.project_image_dirs(
+        image_root or appconfig.image_dir(cfg),
+        project,
+    )["root"]
+    trash_dir = os.path.join(root, ".trash")
+    os.makedirs(trash_dir, exist_ok=True)
+    trash_path = os.path.join(
+        trash_dir,
+        "%s_%s" % (uuid.uuid4().hex, os.path.basename(source_path)),
+    )
+    try:
+        os.replace(source_path, trash_path)
+        image_studio.update_asset_local_path(
+            asset.id,
+            trash_path,
+            status=image_studio.ASSET_STATUS_MISSING,
+            path=database_path,
+        )
+    except Exception as exc:
+        if os.path.isfile(trash_path) and not os.path.exists(source_path):
+            try:
+                os.replace(trash_path, source_path)
+            except OSError:
+                pass
+        raise ImageStudioImageError(f"删除生成图片失败：{exc}") from exc
+    return {
+        "asset_id": asset.id,
+        "original_path": source_path,
+        "trash_path": trash_path,
+    }
+
+
+def restore_trashed_asset(record, *, path=None, config=None):
+    cfg = appconfig.load_config() if config is None else config
+    database_path = path or appconfig.db_path(cfg)
+    payload = dict(record or {})
+    asset = image_studio.get_asset(payload.get("asset_id"), path=database_path)
+    if asset is None:
+        raise ImageStudioImageError("待撤销的生成图片记录不存在")
+    trash_path = os.path.abspath(str(payload.get("trash_path") or ""))
+    original_path = os.path.abspath(str(payload.get("original_path") or ""))
+    if not os.path.isfile(trash_path):
+        raise ImageStudioImageError("废纸篓中的生成图片不存在")
+    if os.path.exists(original_path):
+        stem, extension = os.path.splitext(original_path)
+        original_path = "%s_restored_%s%s" % (stem, uuid.uuid4().hex[:8], extension)
+    os.makedirs(os.path.dirname(original_path), exist_ok=True)
+    os.replace(trash_path, original_path)
+    try:
+        return image_studio.update_asset_local_path(
+            asset.id,
+            original_path,
+            status=image_studio.ASSET_STATUS_AVAILABLE,
+            path=database_path,
+        )
+    except Exception as exc:
+        try:
+            os.replace(original_path, trash_path)
+        except OSError:
+            pass
+        raise ImageStudioImageError(f"撤销删除生成图片失败：{exc}") from exc
+
+
 def download_original_asset(asset_id, *, path=None, config=None, image_root=None, session=None):
     """Download one Shopee original image into originals/ and mark its asset available."""