#!/usr/bin/env python3 """ Web 壳 —— 让 agent.py 变成可通过浏览器访问的在线服务 参与者不需要改这个文件，只改 agent.py 即可。 启动方式（本地调试）： DASHSCOPE_API_KEY=sk-xxx python3 server.py Deploy Skill 会自动用 gunicorn 启动： gunicorn -b 0.0.0.0:8080 server:app """ import json import os import threading import time import uuid from pathlib import Path from flask import Flask, request, jsonify, send_from_directory def _load_dotenv(): """读取同目录下的 .env（KEY=VALUE 每行一条），不覆盖已有环境变量。 部署到云端后 .env 会随代码一起上传，这样服务就能拿到 DASHSCOPE_API_KEY。""" env_file = Path(__file__).resolve().parent / ".env" if not env_file.exists(): return for line in env_file.read_text(encoding="utf-8").splitlines(): line = line.strip() if not line or line.startswith("#") or "=" not in line: continue key, _, value = line.partition("=") key, value = key.strip(), value.strip().strip('"').strip("'") if key and key not in os.environ: os.environ[key] = value _load_dotenv() # 必须在 import agent 之前执行，agent.py 在导入时就会读环境变量 # 导入 agent.py 里的核心函数 from agent import ( # noqa: E402 read_input_data, get_product_title, step_descriptions, step_images, step_video, step_strategy, API_KEY, TEXT_MODEL, IMAGE_MODEL, VIDEO_MODEL ) app = Flask(__name__) # 任务状态存储（内存，重启丢失，demo 够用） tasks = {} # task_id -> {status, progress, output_dir, input_dir, error} WORK_DIR = Path("/tmp/qianwenai-agent") WORK_DIR.mkdir(parents=True, exist_ok=True) # ============================================================ # 前端页面 # ============================================================ INDEX_HTML = """ 一键出海 · 商品素材 Agent

🌏 一键出海 · 商品素材 Agent

输入商品信息，自动生成美国🇺🇸 韩国🇰🇷 巴西🇧🇷 三国上架素材

💡 也可以粘贴你自己商品的完整 JSON 数据

正在准备...

📝 三语商品文案

🇺🇸 English 🇰🇷 한국어 🇧🇷 Português

🖼️ 商品图片

📋 策略文档

文本模型: MODEL_TEXT | 图像模型: MODEL_IMAGE | 视频模型: MODEL_VIDEO
""".replace('MODEL_TEXT', TEXT_MODEL).replace('MODEL_IMAGE', IMAGE_MODEL).replace('MODEL_VIDEO', VIDEO_MODEL) # ============================================================ # API 路由 # ============================================================ @app.route('/') def index(): return INDEX_HTML @app.route('/api/generate', methods=['POST']) def generate(): if not API_KEY: return jsonify(error="DASHSCOPE_API_KEY 未设置"), 500 data = request.get_json(force=True, silent=True) if not data: return jsonify(error="请提供商品 JSON 数据"), 400 task_id = uuid.uuid4().hex[:8] input_dir = str(WORK_DIR / f"input_{task_id}") output_dir = str(WORK_DIR / f"output_{task_id}") os.makedirs(f"{input_dir}/product_info", exist_ok=True) os.makedirs(output_dir, exist_ok=True) # 保存输入 with open(f"{input_dir}/product_info/product_basic.json", "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) tasks[task_id] = { "status": "running", "percent": 10, "message": "正在生成三语文案...", "output_dir": output_dir, "input_dir": input_dir, "error": None, } def run(): t = tasks[task_id] try: product_data = read_input_data(input_dir) if not product_data: raise RuntimeError("未读取到商品数据") t.update(percent=15, message=f"正在生成三语文案（模型 {TEXT_MODEL}）...") step_descriptions(product_data, output_dir) t.update(percent=40, message=f"正在生成 6 张商品图片（模型 {IMAGE_MODEL}）...") step_images(product_data, output_dir) t.update(percent=75, message=f"正在生成商品视频（模型 {VIDEO_MODEL}）...") step_video(product_data, output_dir) t.update(percent=90, message="正在生成策略文档...") step_strategy(product_data, output_dir) t.update(status="done", percent=100, message="全部完成！") except Exception as e: t.update(status="error", error=str(e)) threading.Thread(target=run, daemon=True).start() return jsonify(task_id=task_id, status="running") @app.route('/api/status/') def status(task_id): t = tasks.get(task_id) if not t: return jsonify(status="not_found"), 404 return jsonify( status=t["status"], percent=t.get("percent", 0), message=t.get("message", ""), error=t.get("error"), ) @app.route('/files//') def serve_file(task_id, filename): t = tasks.get(task_id) if not t: return "Not found", 404 return send_from_directory(t["output_dir"], filename) # ============================================================ # 启动 # ============================================================ if __name__ == "__main__": port = int(os.environ.get("PORT", 8080)) print(f"🌏 Agent Web 服务启动: http://localhost:{port}") print(f" 文本模型: {TEXT_MODEL} | 图像: {IMAGE_MODEL} | 视频: {VIDEO_MODEL}") if not API_KEY: print(" ⚠️ DASHSCOPE_API_KEY 未设置！请设置后重启") app.run(host="0.0.0.0", port=port, debug=False)