tw2/qingjian_cs/app/auto_draw.py

from __future__ import annotations

import asyncio
import os
import sys
from pathlib import Path
from typing import Any

import requests

from .config import AUTO_DRAW_ENDPOINT, AUTO_DRAW_TIMEOUT_SECONDS


def _add_legacy_tw_path() -> None:
    root = os.getenv("LEGACY_TW_ROOT", r"D:\main\sandbox\tw_terminator").strip()
    if not root:
        return
    p = Path(root)
    if p.exists() and str(p) not in sys.path:
        sys.path.insert(0, str(p))


async def _draw_via_legacy_tw(
    image_url: str,
    customer_id: str,
    requirement: str = "",
) -> dict[str, Any]:
    _add_legacy_tw_path()
    from image.image_processor import image_processor  # type: ignore
    from services.service_tuhui_upload import upload_to_tuhui  # type: ignore

    prompt = requirement.strip() or "按原图做高清处理，保留主体细节，输出清晰可用版本"
    process_res = await image_processor.process_image(
        image_url=image_url,
        operation="enhance",
        requirements="complexity:normal",
        gemini_prompt=prompt,
        aspect_ratio="1:1",
        perspective="no",
        proc_type="",
        subject="",
        quality="",
    )
    if not process_res.get("success"):
        return {"ok": False, "error": str(process_res.get("message", "process_failed"))}

    ok, link, _ = await upload_to_tuhui(
        process_res["result_path"],
        title=f"客户{customer_id[-4:]}-预览图" if customer_id else "预览图",
        description="AI自动作图预览",
        price=1,
    )
    if not ok:
        return {"ok": False, "error": str(link)}
    return {"ok": True, "url": str(link)}


def _draw_via_http_endpoint(image_url: str, customer_id: str, requirement: str = "") -> dict[str, Any]:
    if not AUTO_DRAW_ENDPOINT:
        return {"ok": False, "error": "AUTO_DRAW_ENDPOINT not configured"}
    payload = {
        "image_url": image_url,
        "customer_id": customer_id,
        "requirement": requirement,
    }
    resp = requests.post(AUTO_DRAW_ENDPOINT, json=payload, timeout=AUTO_DRAW_TIMEOUT_SECONDS)
    if resp.status_code != 200:
        return {"ok": False, "error": f"http_{resp.status_code}:{resp.text[:200]}"}
    data = resp.json() if resp.text else {}
    url = str(data.get("url", "") or data.get("preview_url", "") or "")
    if not url:
        return {"ok": False, "error": "missing_preview_url"}
    return {"ok": True, "url": url}


async def auto_draw_preview(
    image_url: str,
    customer_id: str,
    requirement: str = "",
) -> dict[str, Any]:
    """
    统一自动作图入口：
    1) 优先走 tw_terminator 的 Gemini 作图链路
    2) 失败时回退 AUTO_DRAW_ENDPOINT
    """
    try:
        return await _draw_via_legacy_tw(image_url=image_url, customer_id=customer_id, requirement=requirement)
    except Exception as e:
        legacy_error = str(e)

    try:
        data = await asyncio.to_thread(
            _draw_via_http_endpoint,
            image_url,
            customer_id,
            requirement,
        )
        if data.get("ok"):
            return data
        return {"ok": False, "error": f"legacy:{legacy_error}; endpoint:{data.get('error','unknown')}"}
    except Exception as e:
        return {"ok": False, "error": f"legacy:{legacy_error}; endpoint:{e}"}