acm / api /image_tasks.py
xiaolongmr
daoru lishi jilu shi zidong shibie wai ceng mulu
c4b43c4
Raw History Blame Contribute Delete
29 kB
from __future__ import annotations
import json
import os
import re
import shutil
import tempfile
import zipfile
from datetime import datetime
from decimal import Decimal, ROUND_HALF_UP
from pathlib import Path, PurePosixPath
from typing import Any
from fastapi import APIRouter, BackgroundTasks, File, Header, HTTPException, Query, Request, UploadFile
from fastapi.concurrency import run_in_threadpool
from fastapi.responses import FileResponse
from pydantic import BaseModel, Field
from api.image_inputs import parse_image_edit_request, read_image_sources
from api.support import require_identity, resolve_image_base_url
from services.content_filter import check_request
from services.config import config
from services.auth_service import DAILY_GIFT_ACCOUNT_POOL_MODEL_ID, auth_service
from services.image_task_service import image_task_service
from services.log_service import LoggedCall
from services.pool_pricing_service import pool_pricing_service
from services.share_evidence_service import share_evidence_service
from services.third_party_image_service import third_party_image_service
def _credit_amount(value: object) -> int | float:
try:
amount = max(Decimal("0"), Decimal(str(value or 0))).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
except Exception:
amount = Decimal("0")
return int(amount) if amount == amount.to_integral_value() else float(amount)
def _third_party_activity_quote(identity: dict[str, object], *, group_id: str, base_unit_price: object) -> dict[str, object]:
pricing_config = config.normal_user_third_party_image_pricing
if str(identity.get("role") or "").strip().lower() != "normal" or not bool(pricing_config.get("enabled")):
base = _credit_amount(base_unit_price)
return {
"applies": False,
"target_type": "third_party_group",
"third_party_group_id": group_id,
"base_unit_price": base,
"pay_percentage": 100,
"unit_price": base,
"quantity": 1,
"estimated_total": base,
"rule_id": "",
"rule_name": "",
}
return pool_pricing_service.quote(
identity,
quantity=1,
pricing_config=pricing_config,
target_type="third_party_group",
third_party_group_id=group_id,
base_unit_price=base_unit_price,
)
class ImageGenerationTaskRequest(BaseModel):
client_task_id: str = Field(..., min_length=1)
prompt: str = Field(..., min_length=1)
model: str = "gpt-image-2"
size: str | None = None
quality: str | None = None
resolution: str | None = None
image_resolution: str | None = None
third_party_image_group_id: str | None = None
account_pool_strategy: str = "own_first"
memory: bool = False
memory_conversation_id: str = ""
memory_reset: bool = False
class ResumePollRequest(BaseModel):
extra_timeout_secs: float | None = Field(default=None, ge=5.0, le=120.0)
class ShareEvidenceRequest(BaseModel):
conversation_id: str = Field(..., min_length=1)
class ImageHistoryExportRequest(BaseModel):
"""导出请求体。
conversations 由前端带上(本地会话结构:标题、分组、轮次),服务端原样
写进 manifest;include_images 决定是否把原图一起打进 zip。
"""
conversations: list[Any] = Field(default_factory=list)
include_images: bool = True
def _parse_task_ids(value: str) -> list[str]:
return [item.strip() for item in value.split(",") if item.strip()]
def _require_owned_task(identity: dict[str, object], conversation_id: str) -> dict[str, object]:
"""校验这条上游会话确实属于当前用户,返回内部任务字典。
没有这层校验的话,任何人拿到一个别人的 conversation_id 就能把它变成公开分享链接。
⚠️ 生图任务记录可能已被清理,但官网取证台账里还有这条会话 → 也算「属于你」。
否则用户点「查看官网」时只会看到一句像越权警告的 404,而真实原因是
「这条会话已过保留期,程序已按策略把官网对话删了」。
"""
task = image_task_service.find_owned_conversation(identity, conversation_id)
if task is None:
raise HTTPException(status_code=404, detail={
"error": "这条会话的官网记录已不存在:可能已过保留期被自动清理,或不属于当前账号",
"reason": "conversation_not_found",
})
return task
def _internal_model_limit_kwargs(identity: dict[str, object], model: object) -> dict[str, object]:
if str(identity.get("role") or "").strip().lower() != "user":
return {}
model_id = str(model or "gpt-image-2").strip() or "gpt-image-2"
return {
"internal_model_limit_user_id": str(identity.get("id") or "").strip(),
"internal_model_limit_model_id": model_id,
"internal_model_limit_amount": 1,
}
def _credit_kwargs(identity: dict[str, object], model: object) -> dict[str, object]:
if str(identity.get("role") or "").strip().lower() != "normal":
return _internal_model_limit_kwargs(identity, model)
pricing = pool_pricing_service.quote(identity, 1)
return {
"credit_user_id": str(identity.get("id") or "").strip(),
"credit_amount": pricing["estimated_total"],
"credit_model_id": DAILY_GIFT_ACCOUNT_POOL_MODEL_ID,
"credit_pricing": pricing,
}
def _third_party_wallet_remaining(identity: dict[str, object], group_id: str, model_id: str) -> int | float:
try:
summary = auth_service.credit_summary(identity)
except Exception:
return 0
wallets = summary.get("third_party_image_wallets") if isinstance(summary, dict) else []
wallet_key = f"{group_id}:{model_id}"
wallet_items = wallets if isinstance(wallets, list) else []
balances: list[int | float] = []
for wallet in wallet_items:
if not isinstance(wallet, dict):
continue
if str(wallet.get("id") or "") == wallet_key or (
str(wallet.get("group_id") or "") == group_id
and str(wallet.get("model_id") or "").strip() == model_id
):
try:
balances.append(_credit_amount(wallet.get("remaining")))
except (TypeError, ValueError):
balances.append(0)
return max(balances) if balances else 0
def _image_api_billing_kwargs(
identity: dict[str, object],
*,
model: object,
quality: object = "",
resolution: object = "",
image_resolution: object = "",
third_party_image_group_id: object = "",
) -> dict[str, object]:
resolution = str(resolution or image_resolution or "").strip().lower()
group_id = str(third_party_image_group_id or "").strip()
explicit_third_party = resolution in {"1k", "2k", "4k"} or bool(group_id)
route = None
if not explicit_third_party:
find_named_route = getattr(third_party_image_service, "find_named_route", None)
route = find_named_route(model, advance=False) if callable(find_named_route) else None
if not explicit_third_party and route is None:
return _credit_kwargs(identity, model)
require_4k = resolution == "4k"
if route is None and str(identity.get("role") or "").strip().lower() == "normal":
find_wallet_route = getattr(third_party_image_service, "find_wallet_route_for_identity", None)
if callable(find_wallet_route):
route = find_wallet_route(
identity,
model,
group_id=group_id,
require_4k=require_4k,
resolution=resolution,
advance=False,
)
if route is None:
route = third_party_image_service.find_route(
model,
group_id=group_id,
require_4k=require_4k,
resolution=resolution or "1k",
advance=False,
)
if route is None:
raise ValueError("No enabled image API model route")
try:
third_party_image_service.assert_access(identity, route)
except PermissionError as exc:
raise PermissionError(str(exc)) from exc
cost = third_party_image_service.cost_for_route(route, quality, 1)
common = {
"resolution": str(route.model.get("resolution") or resolution or "1k"),
"third_party_image_group_id": str(route.group.get("id") or ""),
}
route_model_id = str(route.model.get("id") or "")
if str(identity.get("role") or "").strip().lower() != "normal":
return {
**common,
**_internal_model_limit_kwargs(identity, route_model_id),
}
billing_mode = str(route.group.get("billing_mode") or "")
route_group_id = str(route.group.get("id") or "")
if billing_mode == "general_multiplier" and _third_party_wallet_remaining(identity, route_group_id, route_model_id) > 0:
# Keep existing model-specific dedicated balances usable even if an
# older/stale group setting still says general-only.
billing_mode = "both"
wallet_pricing = _third_party_activity_quote(identity, group_id=route_group_id, base_unit_price=1)
wallet_cost = _credit_amount(wallet_pricing.get("estimated_total", 1))
if billing_mode == "dedicated_wallet" or (
billing_mode == "both" and cost > 0 and _third_party_wallet_remaining(identity, route_group_id, route_model_id) >= wallet_cost
):
wallet_kwargs: dict[str, object] = {
**common,
"third_party_wallet_user_id": str(identity.get("id") or "").strip(),
"third_party_wallet_group_id": route_group_id,
"third_party_wallet_model_id": route_model_id,
"third_party_wallet_amount": wallet_cost,
"third_party_wallet_cost_per_image": _credit_amount(wallet_pricing.get("unit_price", 1)),
"credit_pricing": wallet_pricing,
}
if billing_mode == "both" and cost > 0:
fallback_pricing = _third_party_activity_quote(identity, group_id=route_group_id, base_unit_price=cost)
wallet_kwargs["third_party_wallet_fallback_credit_user_id"] = str(identity.get("id") or "").strip()
wallet_kwargs["third_party_wallet_fallback_credit_amount"] = fallback_pricing["estimated_total"]
wallet_kwargs["third_party_wallet_fallback_credit_model_id"] = f"{route_group_id}:{route_model_id}"
wallet_kwargs["third_party_wallet_fallback_credit_pricing"] = fallback_pricing
return wallet_kwargs
pricing = _third_party_activity_quote(identity, group_id=route_group_id, base_unit_price=cost)
return {
**common,
"credit_user_id": str(identity.get("id") or "").strip(),
"credit_amount": pricing["estimated_total"],
"credit_model_id": f"{route_group_id}:{route_model_id}",
"credit_pricing": pricing,
}
def _generation_billing_kwargs(identity: dict[str, object], body: ImageGenerationTaskRequest) -> dict[str, object]:
return _image_api_billing_kwargs(
identity,
model=body.model,
quality=body.quality,
resolution=body.resolution,
image_resolution=body.image_resolution,
third_party_image_group_id=body.third_party_image_group_id,
)
def _effective_account_pool_strategy(identity: dict[str, object], value: object) -> str:
if identity.get("role") == "admin" or bool(identity.get("account_pool_enabled")):
return str(value or "own_first")
return "admin_only"
async def filter_or_log(call: LoggedCall, text: str) -> None:
try:
await run_in_threadpool(check_request, text)
except HTTPException as exc:
call.log("调用失败", status="failed", error=str(exc.detail))
raise
# ---------------------------------------------------------------------------
# 历史记录导出 / 导入
#
# 包结构(zip):
# manifest.json 记录本体(tasks 为原始任务结构,不做 _public_task 裁剪)
# images/<相对路径> 任务里引用到的原图,路径与 /images/<rel> 完全一致
#
# 不写 image_index.json:/images/{path} 直接读文件,历史记录靠 task.data[].url
# 就能显示;索引只影响「图片管理」页的列表。刻意不碰索引,避免与
# image_storage_service 的写锁互相干扰。
# ---------------------------------------------------------------------------
HISTORY_FORMAT = "chatgpt2api-image-history"
HISTORY_VERSION = 1
_IMAGE_URL_RE = re.compile(r"/images/(.+)$")
_ZIP_MAX_UNCOMPRESSED = 8 * 1024 * 1024 * 1024 # 防 zip bomb:解压后上限 8GB
def _rel_from_image_url(url: object) -> str:
"""从 `<任意域名>/images/<rel>` 里取出安全的相对路径,非法则返回空串。"""
text = str(url or "").strip()
if not text:
return ""
match = _IMAGE_URL_RE.search(text)
if not match:
return ""
rel = match.group(1).split("?", 1)[0].split("#", 1)[0].strip()
if not rel:
return ""
pure = PurePosixPath(rel)
if pure.is_absolute() or ".." in pure.parts:
return ""
return pure.as_posix()
def _collect_task_image_rels(tasks: list[object]) -> list[str]:
rels: set[str] = set()
for task in tasks:
if not isinstance(task, dict):
continue
data = task.get("data")
if not isinstance(data, list):
continue
for item in data:
if not isinstance(item, dict):
continue
rel = _rel_from_image_url(item.get("url"))
if rel:
rels.add(rel)
return sorted(rels)
def _rewrite_task_image_urls(task: dict, base_url: str) -> None:
"""把记录里的图片地址改写到当前部署,否则换环境后图片全部指向旧域名。"""
data = task.get("data")
if not isinstance(data, list):
return
for item in data:
if not isinstance(item, dict):
continue
rel = _rel_from_image_url(item.get("url"))
if rel:
item["url"] = f"{base_url}/images/{rel}" if base_url else f"/images/{rel}"
def _build_history_archive(
identity: dict[str, object],
include_images: bool,
conversations: list[object],
) -> tuple[str, str, dict[str, object]]:
tasks = image_task_service.export_owned_tasks(identity)
rels = _collect_task_image_rels(tasks) if include_images else []
manifest = {
"format": HISTORY_FORMAT,
"version": HISTORY_VERSION,
"exported_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"owner_id": str(identity.get("id") or ""),
"count": len(tasks),
"conversation_count": len(conversations),
"includes_images": bool(include_images),
# 历史记录的「会话」结构(标题/分组/轮次)存在浏览器本地,服务端不知道,
# 由前端在导出时一并带上来,导入时原样还回去,否则导入后标题和分组会丢。
"conversations": conversations,
"tasks": tasks,
}
images_dir = config.images_dir
handle, archive_path = tempfile.mkstemp(prefix="image-history-", suffix=".zip")
os.close(handle)
written = 0
missing = 0
# 图片本身已是压缩格式,用 STORED 省 CPU(线上只有 2 个共享 vCPU)
with zipfile.ZipFile(archive_path, "w", zipfile.ZIP_STORED) as archive:
archive.writestr("manifest.json", json.dumps(manifest, ensure_ascii=False, indent=2))
for rel in rels:
source = images_dir / rel
try:
if not source.is_file():
missing += 1
continue
archive.write(source, f"images/{rel}")
written += 1
except OSError:
missing += 1
filename = f"image-history-{datetime.now().strftime('%Y%m%d-%H%M%S')}.zip"
stats = {"tasks": len(tasks), "images": written, "missing_images": missing}
return archive_path, filename, stats
def _history_prefix(archive: zipfile.ZipFile) -> str:
"""返回 manifest.json 在包内的目录前缀(含结尾斜杠,根目录时为空串)。
正常导出的包 manifest.json 就在根目录。但用解压工具或部分下载器重新打包时,
会给整包套一层目录,变成 `image-history-xxx/manifest.json`。这时按根目录找
会直接报「缺少 manifest.json」,用户会以为包坏了,实际只是路径多了一层。
只接受正好一层目录,避免误命中 images/ 内部碰巧叫 manifest.json 的文件。
"""
names = set(archive.namelist())
if "manifest.json" in names:
return ""
candidates = [name[: -len("manifest.json")] for name in names if name.endswith("/manifest.json")]
candidates = [item for item in candidates if item.count("/") == 1]
return candidates[0] if len(candidates) == 1 else ""
def _apply_history_archive(identity: dict[str, object], file_obj) -> dict[str, object]:
try:
archive = zipfile.ZipFile(file_obj)
except zipfile.BadZipFile as exc:
raise HTTPException(status_code=400, detail={"error": "文件不是有效的 zip 包"}) from exc
with archive:
prefix = _history_prefix(archive)
try:
raw_manifest = archive.read(f"{prefix}manifest.json")
except KeyError as exc:
raise HTTPException(status_code=400, detail={"error": "包里缺少 manifest.json,不是本系统导出的历史记录"}) from exc
try:
manifest = json.loads(raw_manifest.decode("utf-8"))
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
raise HTTPException(status_code=400, detail={"error": "manifest.json 无法解析"}) from exc
if not isinstance(manifest, dict) or manifest.get("format") != HISTORY_FORMAT:
raise HTTPException(status_code=400, detail={"error": "包格式不匹配,不是图片历史记录包"})
tasks = manifest.get("tasks")
if not isinstance(tasks, list):
tasks = []
images_dir = config.images_dir
total_uncompressed = sum(info.file_size for info in archive.infolist())
if total_uncompressed > _ZIP_MAX_UNCOMPRESSED:
raise HTTPException(status_code=400, detail={"error": "包解压后过大,已拒绝导入"})
images_written = 0
images_skipped = 0
images_prefix = f"{prefix}images/"
for info in archive.infolist():
if info.is_dir() or not info.filename.startswith(images_prefix):
continue
rel = info.filename[len(images_prefix):]
pure = PurePosixPath(rel)
if not rel or pure.is_absolute() or ".." in pure.parts:
continue # 防 zip slip
target = images_dir / pure
try:
if target.exists():
images_skipped += 1
continue
target.parent.mkdir(parents=True, exist_ok=True)
with archive.open(info) as source, target.open("wb") as destination:
shutil.copyfileobj(source, destination, length=1024 * 256)
images_written += 1
except OSError:
images_skipped += 1
base_url = config.base_url
for task in tasks:
if isinstance(task, dict):
_rewrite_task_image_urls(task, base_url)
result = image_task_service.import_tasks(identity, tasks)
result["images"] = images_written
result["images_skipped"] = images_skipped
conversations = manifest.get("conversations")
if isinstance(conversations, list):
# 会话结构还给前端,由它写回本地存储(服务端不持有这些数据)
result["conversations"] = conversations
return result
def create_router() -> APIRouter:
router = APIRouter()
@router.get("/api/image-tasks")
async def list_image_tasks(
ids: str = Query(default=""),
authorization: str | None = Header(default=None),
):
identity = require_identity(authorization)
return await run_in_threadpool(image_task_service.list_tasks, identity, _parse_task_ids(ids))
@router.post("/api/image-tasks/export")
async def export_image_history(
body: ImageHistoryExportRequest,
background_tasks: BackgroundTasks,
authorization: str | None = Header(default=None),
):
"""把当前用户的全部图片历史导出成 zip(记录 + 本地会话结构 + 可选原图)。"""
identity = require_identity(authorization)
archive_path, filename, _stats = await run_in_threadpool(
_build_history_archive,
identity,
bool(body.include_images),
list(body.conversations or []),
)
def _drop_archive(path: str = archive_path) -> None:
try:
os.unlink(path)
except OSError:
pass
background_tasks.add_task(_drop_archive)
return FileResponse(
archive_path,
media_type="application/zip",
filename=filename,
headers={"Cache-Control": "no-store"},
)
@router.post("/api/image-tasks/import")
async def import_image_history(
file: UploadFile = File(...),
authorization: str | None = Header(default=None),
):
"""导入历史记录 zip:同 ID 跳过、其余合并,图片落到本机 images 目录。"""
identity = require_identity(authorization)
name = str(file.filename or "")
if not name.lower().endswith(".zip"):
raise HTTPException(status_code=400, detail={"error": "请选择 .zip 格式的历史记录包"})
try:
return await run_in_threadpool(_apply_history_archive, identity, file.file)
finally:
await file.close()
@router.post("/api/image-tasks/generations")
async def create_generation_task(
body: ImageGenerationTaskRequest,
request: Request,
authorization: str | None = Header(default=None),
):
identity = require_identity(authorization)
await filter_or_log(LoggedCall(identity, "/api/image-tasks/generations", body.model, "文生图任务", request_text=body.prompt), body.prompt)
try:
return await run_in_threadpool(
image_task_service.submit_generation,
identity,
client_task_id=body.client_task_id,
prompt=body.prompt,
model=body.model,
size=body.size,
quality=str(body.quality or ""),
base_url=resolve_image_base_url(request),
memory_enabled=body.memory,
memory_conversation_id=body.memory_conversation_id,
memory_reset=body.memory_reset,
account_pool_strategy=_effective_account_pool_strategy(identity, body.account_pool_strategy),
**_generation_billing_kwargs(identity, body),
)
except PermissionError as exc:
raise HTTPException(status_code=403, detail={"error": str(exc)}) from exc
except ValueError as exc:
error_text = str(exc)
status_code = 402 if "次数不足" in error_text or "not enough" in error_text.lower() or "balance" in error_text.lower() else 400
raise HTTPException(status_code=status_code, detail={"error": str(exc)}) from exc
@router.post("/api/image-tasks/edits")
async def create_edit_task(
request: Request,
authorization: str | None = Header(default=None),
):
identity = require_identity(authorization)
payload, image_sources, mask_sources = await parse_image_edit_request(request)
client_task_id = str(payload.get("client_task_id") or "").strip()
if not client_task_id:
raise HTTPException(status_code=400, detail={"error": "client_task_id is required"})
prompt = str(payload["prompt"])
model = str(payload["model"])
await filter_or_log(LoggedCall(identity, "/api/image-tasks/edits", model, "图生图任务", request_text=prompt), prompt)
images = await read_image_sources(image_sources)
masks = await read_image_sources(mask_sources) if mask_sources else None
try:
return await run_in_threadpool(
image_task_service.submit_edit,
identity,
client_task_id=client_task_id,
prompt=prompt,
model=model,
size=payload["size"],
base_url=resolve_image_base_url(request),
images=images,
masks=masks,
memory_enabled=bool(payload.get("memory")),
memory_conversation_id=str(payload.get("memory_conversation_id") or ""),
memory_reset=bool(payload.get("memory_reset")),
account_pool_strategy=_effective_account_pool_strategy(identity, payload.get("account_pool_strategy") or "own_first"),
quality=str(payload.get("quality") or ""),
**_image_api_billing_kwargs(
identity,
model=model,
quality=payload.get("quality"),
resolution=payload.get("resolution"),
image_resolution=payload.get("image_resolution"),
third_party_image_group_id=payload.get("third_party_image_group_id"),
),
)
except PermissionError as exc:
raise HTTPException(status_code=403, detail={"error": str(exc)}) from exc
except ValueError as exc:
status_code = 402 if "次数不足" in str(exc) else 400
raise HTTPException(status_code=status_code, detail={"error": str(exc)}) from exc
@router.post("/api/image-tasks/{task_id}/resume-poll")
async def resume_image_poll(
task_id: str,
body: ResumePollRequest,
authorization: str | None = Header(default=None),
):
identity = require_identity(authorization)
try:
return await run_in_threadpool(
image_task_service.resume_poll,
identity,
task_id,
body.extra_timeout_secs or config.image_timeout_retry_secs,
)
except ValueError as exc:
raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc
@router.get("/api/image-tasks/share-evidence")
async def share_evidence_status(
conversation_id: str = Query(..., min_length=1),
authorization: str | None = Header(default=None),
):
"""查询某条会话的「官网来源」取证状态(链接、剩余有效秒数)。"""
identity = require_identity(authorization)
_require_owned_task(identity, conversation_id)
return share_evidence_service.describe(conversation_id)
@router.post("/api/image-tasks/share-evidence")
async def create_share_evidence(
body: ShareEvidenceRequest,
authorization: str | None = Header(default=None),
):
"""把一条会话取成为 ChatGPT 官方匿名分享链接(幂等)。"""
identity = require_identity(authorization)
if not config.image_share_evidence_enabled:
raise HTTPException(status_code=403, detail={"error": "后台未开启「官网来源取证」,无法生成分享链接"})
task = _require_owned_task(identity, body.conversation_id)
try:
await run_in_threadpool(
share_evidence_service.create_share,
conversation_id=body.conversation_id,
owner_id=str(identity.get("id") or ""),
account_email=str(task.get("account_email") or ""),
)
except ValueError as exc:
raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc
except RuntimeError as exc:
raise HTTPException(status_code=502, detail={"error": str(exc)}) from exc
return share_evidence_service.describe(body.conversation_id)
@router.delete("/api/image-tasks/share-evidence")
async def revoke_share_evidence(
conversation_id: str = Query(..., min_length=1),
authorization: str | None = Header(default=None),
):
"""提前撤销分享链接并回收会话。"""
identity = require_identity(authorization)
_require_owned_task(identity, conversation_id)
try:
await run_in_threadpool(share_evidence_service.revoke, conversation_id)
except ValueError as exc:
raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc
except RuntimeError as exc:
raise HTTPException(status_code=502, detail={"error": str(exc)}) from exc
return share_evidence_service.describe(conversation_id)
return router