refactor(server): 移除鉴权并归档遗留路由至 legacy/
- 删除 auth.py 与 deps.py,各路由去除 get_current_user 依赖 - collection.py 更名为 materials.py,冻结链路(ozon/publish/shops/categories)移入 legacy/ - 扩展默认生图服务端口并入 8800 并自动迁移旧配置,水印默认文案改为 Panda Store - 新增 docs/v2.1/backend-structure.md 后端结构盘点文档
This commit is contained in:
@@ -0,0 +1,86 @@
|
||||
"""套图业务逻辑(从 api/suite.py 下沉):文本素材转换 / 模型校验 / 生成图回写商品素材。
|
||||
|
||||
api 层只留参数校验与调用;本模块可独立测试。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
|
||||
from schemas.suite import TextMaterial
|
||||
from services.tasks import TaskImage
|
||||
|
||||
|
||||
def texts_to_raw(texts: list[TextMaterial]) -> dict:
|
||||
"""前端组装的文本素材 → prompt 上下文用的 raw dict(后写的覆盖先写的)。"""
|
||||
raw: dict = {}
|
||||
for t in texts:
|
||||
if t.kind == "title" and t.content:
|
||||
raw["title"] = t.content
|
||||
elif t.kind == "price" and t.content:
|
||||
raw["price"] = t.content
|
||||
elif t.kind == "brand" and t.content:
|
||||
raw["brand"] = t.content
|
||||
elif t.kind == "params" and t.pairs:
|
||||
merged = {p["key"]: p["value"] for p in (raw.get("params") or [])}
|
||||
for p in t.pairs:
|
||||
merged.setdefault(p["key"], p["value"])
|
||||
raw["params"] = [{"key": k, "value": v} for k, v in merged.items()]
|
||||
elif t.kind == "selling_point" and t.content:
|
||||
raw["sellingPoints"] = t.content
|
||||
elif t.kind == "desc" and t.content:
|
||||
raw["desc"] = t.content
|
||||
elif t.kind == "sales" and t.content:
|
||||
raw["sales"] = t.content
|
||||
elif t.kind == "shop" and t.content:
|
||||
raw["shop"] = t.content
|
||||
return raw
|
||||
|
||||
|
||||
def validate_model(provider_name: str, model: str | None) -> None:
|
||||
"""按 provider 校验模型名白名单,不合法抛 ValueError。"""
|
||||
from schemas.suite import RIGHTAPI_MODELS, TONGYI_MODELS
|
||||
|
||||
if provider_name == "tongyi" and model and model not in TONGYI_MODELS:
|
||||
raise ValueError(f"不支持的模型: {model}(tongyi 支持: {TONGYI_MODELS})")
|
||||
if provider_name == "rightapi" and model and model not in RIGHTAPI_MODELS:
|
||||
raise ValueError(f"不支持的模型: {model}(rightapi 支持: {RIGHTAPI_MODELS})")
|
||||
|
||||
|
||||
async def append_generated_asset(product_id: str, image: TaskImage) -> str | None:
|
||||
"""把一张生成完成的图追加为 product_assets(generated),并累加 asset_counts。返回 asset_id。
|
||||
|
||||
作为 run_suite 的逐张回调使用(签名须为 (image)),调用方用 partial 绑定 product_id。
|
||||
"""
|
||||
from sqlalchemy import func, select
|
||||
|
||||
from db import get_session_factory
|
||||
from models import Product, ProductAsset
|
||||
|
||||
pid = uuid.UUID(product_id)
|
||||
async with get_session_factory()() as db:
|
||||
count = await db.scalar(
|
||||
select(func.count(ProductAsset.id)).where(
|
||||
ProductAsset.product_id == pid,
|
||||
ProductAsset.group_key == "generated",
|
||||
)
|
||||
)
|
||||
asset = ProductAsset(
|
||||
product_id=pid,
|
||||
group_key="generated",
|
||||
variant_name=None,
|
||||
sort_order=count or 0,
|
||||
type="img",
|
||||
source_url="",
|
||||
stored_url=image.url,
|
||||
status="uploaded",
|
||||
)
|
||||
db.add(asset)
|
||||
await db.flush()
|
||||
|
||||
product = await db.get(Product, pid)
|
||||
if product is not None:
|
||||
counts = dict(product.asset_counts or {})
|
||||
counts["generated"] = int(counts.get("generated") or 0) + 1
|
||||
product.asset_counts = counts
|
||||
await db.commit()
|
||||
return str(asset.id)
|
||||
Reference in New Issue
Block a user