#!/usr/bin/env python3 """qianwen 千问 AI CLI 工具 用法: qianwen chat "你好" # 对话 qianwen image "画一只猫" # 生图 qianwen vision <图片路径> # 看图分析 qianwen tts "文字" # 语音合成(如果有) """ import sys, os, json, base64, argparse from pathlib import Path API_KEY = os.environ.get("DASHSCOPE_API_KEY", "") NATIVE_BASE = "https://dashscope.aliyuncs.com/api/v1" COMPAT_BASE = "https://dashscope.aliyuncs.com/compatible-mode/v1" # 确保API Key存在 def ensure_key(): if not API_KEY: print("❌ 请设置 DASHSCOPE_API_KEY 环境变量", file=sys.stderr) sys.exit(1) def cmd_chat(args): """对话""" ensure_key() import requests resp = requests.post( f"{COMPAT_BASE}/chat/completions", headers={"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}, json={ "model": args.model or "qwen-plus", "messages": [{"role": "user", "content": args.text}], "max_tokens": args.max_tokens }, timeout=60 ) result = resp.json() print(result["choices"][0]["message"]["content"]) def cmd_image(args): """生成图片""" ensure_key() prompt = args.text if args.file: with open(args.file) as f: prompt = f.read() import requests # wan系列使用原生端点，qwen-image系列使用compatible-mode is_wan = args.model and "wan" in args.model.lower() if is_wan: # wan模型: DashScope原生API — input.messages 结构 url = f"{NATIVE_BASE}/services/aigc/multimodal-generation/generation" payload = { "model": args.model, "input": { "messages": [ { "role": "user", "content": [{"text": prompt}] } ] }, "parameters": { "size": args.size, "n": args.n, "prompt_extend": True } } else: # qwen-image模型: compatible-mode url = f"{COMPAT_BASE}/images/generations" payload = { "model": args.model, "prompt": prompt, "n": args.n, "size": args.size } resp = requests.post( url, headers={"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}, json=payload, timeout=180 ) result = resp.json() if "output" in result: output_dir = Path(args.output) output_dir.mkdir(parents=True, exist_ok=True) results_list = [] out = result["output"] if isinstance(out, dict) and "results" in out: results_list = out["results"] elif isinstance(out, dict) and "image_url" in out: results_list = [out] elif isinstance(out, dict) and "choices" in out: # wan模型: output.choices[].message.content[].image for choice in out["choices"]: msg = choice.get("message", {}) content = msg.get("content", []) for c in content: if isinstance(c, dict) and "image" in c: results_list.append({"url": c["image"]}) for i, img in enumerate(results_list): url = img.get("url", "") or img.get("image_url", "") if url: r = requests.get(url, timeout=30) ext = url.split(".")[-1].split("?")[0] if ext not in ("png", "jpg", "jpeg", "webp", "gif"): ext = "png" fname = f"qwen-img-{i+1}.{ext}" if len(results_list) > 1 else f"qwen-img.{ext}" fpath = output_dir / fname fpath.write_bytes(r.content) print(f"✅ 已保存: {fpath}") else: print(f" ⚠️ 结果无URL: {json.dumps(img)[:200]}") else: print(f"❌ 生成失败: {json.dumps(result, indent=2, ensure_ascii=False)[:500]}") def cmd_vision(args): """分析图片""" ensure_key() import requests img_path = args.image ext = Path(img_path).suffix.lower() mime_map = {".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".png": "image/png", ".gif": "image/gif", ".webp": "image/webp"} with open(img_path, "rb") as f: img_b64 = base64.b64encode(f.read()).decode() prompt_text = args.text or "请描述这张图片的内容" resp = requests.post( f"{COMPAT_BASE}/chat/completions", headers={"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}, json={ "model": args.model or "qwen-vl-plus", "messages": [{ "role": "user", "content": [ {"type": "text", "text": prompt_text}, {"type": "image_url", "image_url": {"url": f"data:{mime_map.get(ext, 'image/jpeg')};base64,{img_b64}"}} ] }], "max_tokens": args.max_tokens }, timeout=30 ) result = resp.json() print(result["choices"][0]["message"]["content"]) def cmd_usage(args): """查看用量""" ensure_key() import requests resp = requests.get( f"{BASE_URL}/dashboard/usage", headers={"Authorization": f"Bearer {API_KEY}"}, timeout=30 ) print(json.dumps(resp.json(), indent=2, ensure_ascii=False)) def main(): parser = argparse.ArgumentParser(description="千问 AI CLI 工具") parser.add_argument("-m", "--model", help="模型名") parser.add_argument("--max-tokens", type=int, default=1000, help="最大token数") parser.add_argument("-n", type=int, default=1, help="生成数量(生图)") parser.add_argument("--size", default="1024*1024", help="图片尺寸(生图)") parser.add_argument("-o", "--output", default="./output", help="输出目录") parser.add_argument("-f", "--file", help="从文件读取prompt") parser.add_argument("text", nargs="?", help="输入文本") parser.add_argument("image", nargs="?", help="图片路径(vision)") # 解析子命令 if len(sys.argv) < 2: parser.print_help() print("\n子命令:") print(" chat 对话") print(" image 生成图片") print(" vision 分析图片") print(" usage 查看用量") return # 映射子命令 subcmd = sys.argv[1] if subcmd in ("chat", "image", "vision", "usage"): known, unknown = parser.parse_known_args(sys.argv[2:]) known.model = known.model or None # 自动设置默认模型 if subcmd == "chat" and not known.model: known.model = "qwen-plus" elif subcmd == "image" and not known.model: known.model = "wan2.6-t2i" elif subcmd == "vision" and not known.model: known.model = "qwen-vl-plus" if subcmd == "chat": known.text = sys.argv[2] if len(sys.argv) > 2 else input("prompt> ") cmd_chat(known) elif subcmd == "image": if not known.text and not known.file: print("❌ 需要提供 prompt (text 或 -f file)", file=sys.stderr) sys.exit(1) cmd_image(known) elif subcmd == "vision": known.image = sys.argv[2] if len(sys.argv) > 2 else None if not known.image: print("❌ 需要提供图片路径", file=sys.stderr) sys.exit(1) known.text = sys.argv[3] if len(sys.argv) > 3 else "请描述这张图片" cmd_vision(known) elif subcmd == "usage": cmd_usage(known) else: parser.print_help() if __name__ == "__main__": main()