图片完整代码与下载
Python 3.10+ / Node.js 22+，无需第三方SDK。读取服务端环境变量 MXB_API_KEY，提交一次、保存任务号、轮询、保存全部返回图片。MJ一次4/5张也会完整保存。
下载：Python · Node.js · AS Gemini工具/视频理解Python · 完整Markdown。Key设置见快速上手，视频生成另见视频代码。
可下载的请求文件
| 目的 | JSON | 接口与准备 |
|---|---|---|
| MJ文生图 | mj-text.json | MJ专用组；比例size、数字quality |
| MJ双参考 | mj-reference.json | 替换两个素材URL |
| Seedream Lite 4K | seedream-lite-4k.json | resolution=4k |
| Seedream Pro 2K | seedream-pro-2k.json | resolution=2k |
| Seedream双参考 | seedream-reference.json | 替换两个素材URL |
| Sunburst 4K/max | gpt-sunburst-4k.json | 需要相应型号与规格权限 |
| Adobe中质量 | adobe-medium-2k.json | 固定medium，默认2K；4K写法见GPT图片章节 |
| GPT双参考 | gpt-reference.json | 替换URL，编辑入口 |
| 香蕉4K | banana-4k.json | Flash图片完整preview后缀 |
| Grok图片 | grok-image.json | 图片型号，非视频 |
示例JSON不含Key。文件中 media.example.com 是素材占位，不能直接用于生成；换成自己的可读取URL，或删掉占位images后通过下面的本地参考参数传图。
Python：MJ完整生成与全部下载
把Python与对应JSON下载到同一工作目录，在已设置MXB_API_KEY的终端运行：
bash
python mxb-image.py --request mj-text.json --output-dir mj-run-01
结果包括 submission.json（提交状态/任务号）、result.json（最终响应）以及 image-01.png 等实际图片文件。扩展名依据文件头选择，不把JPEG仅改名为PNG。每次新生成使用一个新的输出目录，避免误重复提交。
继续查询已有任务、恢复下载（不会重新生成）：
bash
python mxb-image.py --task "$MXB_TASK_ID" --output-dir mj-run-01
本地多参考图片，使用原文生图JSON加两个素材，不需要先手工上传：
bash
python mxb-image.py --request seedream-lite-4k.json \
--reference product.png --reference palette.png --output-dir seedream-run-01
--reference 按顺序追加。MJ/Seedream编码为字符串数组并走生成入口；GPT/Gemini编码为image_url对象并自动走编辑入口。需要明确编辑路径时用 --edit，但目标模型必须支持。
不读取文件时也可用基本参数：
bash
python mxb-image.py --model gemini-3.1-flash-image-preview \
--size 3840x2160 --prompt "横版中文茶饮海报，标题茶香入夏" --output-dir banana-run-01
使用 --request 时，以JSON文件中的model/prompt/size/quality等为准；不要再用其他基本参数期待覆盖文件。等待时限由 --timeout 控制，默认1800秒，超时不取消服务端任务。
Node.js：请求文件与续查
bash
# Bash；已设置MXB_API_KEY
unset MXB_TASK_ID
export MXB_REQUEST='mj-text.json'
export MXB_OUTPUT_DIR='mj-node-run-01'
node mxb-image.mjs
# 用真实任务号续查，不发起新生成
export MXB_TASK_ID='替换为真实任务号'
node mxb-image.mjs
可选变量：MXB_MODEL / MXB_PROMPT / MXB_SIZE（没有MXB_REQUEST时使用）；MXB_REFERENCE单图；MXB_REFERENCES为JSON字符串数组（多图时使用，例如 ["product.png","palette.png"]）；MXB_EDIT=true强制编辑。输出同样保存全部图片。
Windows PowerShell 7
powershell
$secureKey = Read-Host 'API Key' -AsSecureString
$env:MXB_API_KEY = [System.Net.NetworkCredential]::new('', $secureKey).Password
python .\mxb-image.py --request .\mj-text.json --output-dir .\mj-run-01
Node.js对应写法：
powershell
Remove-Item Env:MXB_TASK_ID -ErrorAction SilentlyContinue
$env:MXB_REQUEST = '.\mj-text.json'
$env:MXB_OUTPUT_DIR = '.\mj-node-run-01'
node .\mxb-image.mjs
以上两种方式任选一种；两者都执行就会创建两次任务并分别计费。不要把Bash的反斜杠续行直接贴入PowerShell。
失败、恢复与示例边界
提交前先创建submission.json。如果该文件已存在，脚本拒绝新POST，避免把一次提交失败当成未受理重复收费。先检查其中的task_id和调用记录；需要新作品时用新的输出目录。提交超时且没有任务号应人工核对，脚本不擅自重试。
查询可对临时GET错误做有界退避；状态失败、身份不符、未知状态或结果为空会停止。下载失败时已经保存的任务号与result.json还在，使用原任务续查/重新下载，不重新生成。有效期见图片任务。
示例每张参考最多25MiB、下载图片最多64MiB，这些是客户端保守上限，不是所有模型的能力上限。代码检查Content-Type和常见文件头，生产系统仍应使用图像解码库验证完整像素，按业务配置存储和大小限制。结果URL下载不携带API Key，API鉴权请求不跟随重定向。
代码与模型参数按当前实现和既有实际出图记录核对，发布时进行离线流程测试。本次文档更新没有新发起付费生成，不把离线测试当成所有模型/参数组合的新实测。
Python完整源码
python
#!/usr/bin/env python3
"""Python 3.10+, standard library. Submit once, resume, save ALL returned images."""
import argparse
import base64
import json
import mimetypes
import os
import re
import time
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path
ORIGIN = 'https://openpix.cc'
LIMIT = 64 * 1024 * 1024 # Client download bound, not the API specification.
class NoRedirect(urllib.request.HTTPRedirectHandler):
def redirect_request(self, req, fp, code, msg, headers, newurl):
return None
class APIError(RuntimeError):
def __init__(self, code, message, retry_after=None):
super().__init__(f'HTTP {code}: {message}')
self.code, self.retry_after = code, retry_after
def request(key, path, payload=None, timeout=30):
if not path.startswith('/v1/'):
raise ValueError('Expected a fixed /v1/ path')
req = urllib.request.Request(ORIGIN + path,
data=None if payload is None else json.dumps(payload, ensure_ascii=False).encode('utf8'),
headers={'Authorization': 'Bearer ' + key, 'Content-Type': 'application/json'},
method='GET' if payload is None else 'POST')
try:
with urllib.request.build_opener(NoRedirect()).open(req, timeout=timeout) as response:
result = json.load(response)
except urllib.error.HTTPError as error:
raise APIError(error.code, error.read(2048).decode('utf8', 'replace'), error.headers.get('Retry-After')) from error
if not isinstance(result, dict) or (result.get('error') and not (result.get('status') or result.get('state'))):
raise RuntimeError('API returned an error or invalid JSON object')
return result
def prepare(body, references=(), force_edit=False):
if not isinstance(body, dict) or not isinstance(body.get('model'), str):
raise ValueError('Request must contain model')
if not isinstance(body.get('prompt'), str) or not body['prompt'].strip():
raise ValueError('prompt must be a nonempty string')
body = dict(body)
string_refs = body['model'].startswith(('mj-', 'seedream-'))
if references:
images = list(body.get('images', []))
for name in references:
file = Path(name)
mime = mimetypes.guess_type(file.name)[0]
if mime not in ('image/png', 'image/jpeg', 'image/webp'):
raise ValueError('Local references must be PNG/JPEG/WebP')
if not 0 < file.stat().st_size <= 25 * 1024 * 1024:
raise ValueError('Example accepts each reference up to 25 MiB; model limits may be lower')
url = 'data:' + mime + ';base64,' + base64.b64encode(file.read_bytes()).decode('ascii')
images.append(url if string_refs else {'image_url': url})
body['images'] = images
edit = force_edit or (bool(body.get('images') or body.get('image')) and not string_refs)
return body, '/v1/images/' + ('edits' if edit else 'generations') + '/async'
def task_id(value):
if not isinstance(value, str) or not re.fullmatch(r'[A-Za-z0-9_-]+', value):
raise ValueError('Invalid task ID; check logs before another POST')
return value
def wait(key, identifier, seconds=1800):
identifier = task_id(identifier)
deadline, delay = time.monotonic() + seconds, 3
while time.monotonic() < deadline:
try:
task = request(key, '/v1/images/tasks/' + identifier)
except APIError as error:
if error.code not in (429, 500, 502, 503, 504):
raise
try:
delay = min(30, max(1, float(error.retry_after)))
except (ValueError, TypeError):
delay = min(30, delay * 1.5)
time.sleep(delay)
continue
except (urllib.error.URLError, TimeoutError):
time.sleep(delay)
delay = min(30, delay * 1.5)
continue
returned = task.get('task_id') or task.get('id')
if returned and returned != identifier:
raise RuntimeError('Task ID mismatch')
if task.get('submission_uncertain'):
raise RuntimeError('Submission uncertain; inspect original task and billing, do not resubmit')
status = str(task.get('status') or task.get('state') or '').lower()
if status in ('completed', 'succeeded', 'success'):
nested = task.get('result')
data = task.get('data') or (nested.get('data') if isinstance(nested, dict) else None)
if not isinstance(data, list) or not data or not all(isinstance(x, dict) for x in data):
raise RuntimeError('Completed task has no valid images')
return data, task
if status in ('failed', 'failure', 'error', 'cancelled', 'canceled', 'expired'):
raise RuntimeError('Task failed: ' + json.dumps(task.get('error'), ensure_ascii=False))
if status not in ('processing', 'queued', 'running', 'pending', 'in_progress'):
raise RuntimeError('Unknown task status: ' + status)
value = task.get('poll_after_ms', 3000)
delay = min(30, max(1, value / 1000)) if isinstance(value, (int, float)) else 3
time.sleep(delay)
raise TimeoutError('Resume using --task ' + identifier + '. Server task is not cancelled.')
def image_extension(data):
if data.startswith(b'\x89PNG\r\n\x1a\n'): return '.png'
if data.startswith(b'\xff\xd8\xff'): return '.jpg'
if data[:4] == b'RIFF' and data[8:12] == b'WEBP': return '.webp'
if data[:6] in (b'GIF87a', b'GIF89a'): return '.gif'
raise RuntimeError('Invalid or unsupported image signature; no output file replaced')
def save_images(items, directory):
directory = Path(directory)
directory.mkdir(parents=True, exist_ok=True)
saved = []
for i, item in enumerate(items, 1):
if item.get('url'):
url = urllib.parse.urlsplit(item['url'])
if url.scheme not in ('https', 'http') or url.username or url.password or not url.hostname:
raise ValueError('Invalid image result URL')
# Separate request: no API Key sent to result URLs or redirects.
with urllib.request.urlopen(item['url'], timeout=120) as response:
if not response.headers.get('Content-Type', '').lower().startswith('image/'):
raise RuntimeError('Result URL did not return an image')
data = response.read(LIMIT + 1)
elif item.get('b64_json'):
if len(item['b64_json']) > (LIMIT + 2) // 3 * 4:
raise RuntimeError('Image exceeds client limit')
data = base64.b64decode(item['b64_json'], validate=True)
else:
raise RuntimeError('Missing image URL/Base64')
if not data or len(data) > LIMIT:
raise RuntimeError('Image empty or exceeds client limit')
target = directory / f'image-{i:02d}{image_extension(data)}'
temp = target.with_suffix(target.suffix + '.part')
temp.write_bytes(data)
os.replace(temp, target)
saved.append(str(target))
return saved
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--request', help='UTF-8 JSON request; preserves model-specific fields')
parser.add_argument('--model', default='gpt-image-2.5')
parser.add_argument('--prompt')
parser.add_argument('--size', default='1024x1024')
parser.add_argument('--reference', action='append', default=[])
parser.add_argument('--edit', action='store_true')
parser.add_argument('--task', help='Resume; sends no generation POST')
parser.add_argument('--timeout', type=int, default=1800)
parser.add_argument('--output-dir', default='images')
args = parser.parse_args()
key = os.environ.get('MXB_API_KEY')
if not key: parser.error('Set MXB_API_KEY')
if args.timeout <= 0: parser.error('--timeout must be positive')
directory = Path(args.output_dir)
directory.mkdir(parents=True, exist_ok=True)
identifier = args.task
if not identifier:
body = json.loads(Path(args.request).read_text(encoding='utf-8-sig')) if args.request else {
'model': args.model, 'prompt': args.prompt, 'size': args.size, 'n': 1}
body, path = prepare(body, args.reference, args.edit)
state_file = directory / 'submission.json'
state = {'state': 'submitting', 'created_at': int(time.time()), 'model': body['model'], 'path': path}
# Exclusive state file prevents accidentally repeating a billable POST in the same directory.
with state_file.open('x', encoding='utf8') as stream:
json.dump(state, stream, ensure_ascii=False, indent=2)
try:
created = request(key, path, body, timeout=60) # Exactly one POST.
identifier = task_id(created.get('task_id') or created.get('id'))
except Exception:
state['state'] = 'submission_uncertain'
state_file.write_text(json.dumps(state, indent=2), encoding='utf8')
raise RuntimeError('Submission not retried. Inspect calling logs before creating another task.')
state.update(state='accepted', task_id=identifier)
state_file.write_text(json.dumps(state, indent=2), encoding='utf8')
print('Task:', identifier, flush=True)
data, final = wait(key, identifier, args.timeout)
(directory / 'result.json').write_text(json.dumps(final, ensure_ascii=False, indent=2), encoding='utf8')
for name in save_images(data, directory): print('Saved:', name)
if __name__ == '__main__': main()
Node.js完整源码
javascript
// Node.js 22+. Submit once; resume with MXB_TASK_ID; save ALL returned images.
import {readFile,writeFile,mkdir,rename,stat} from 'node:fs/promises';
import {extname,join} from 'node:path';
import {pathToFileURL} from 'node:url';
const origin='https://openpix.cc',limit=64*1024*1024;
const pause=ms=>new Promise(resolve=>setTimeout(resolve,ms));
export class APIError extends Error {
constructor(status,body,retryAfter){super(`HTTP ${status}: ${body.slice(0,2048)}`);this.status=status;this.retryAfter=retryAfter;}
}
export async function request(key,path,body,timeout=30000){
if(!path.startsWith('/v1/'))throw new Error('Expected fixed /v1/ path');
const r=await fetch(origin+path,{method:body===undefined?'GET':'POST',redirect:'error',
headers:{Authorization:`Bearer ${key}`,'Content-Type':'application/json'},
body:body===undefined?undefined:JSON.stringify(body),signal:AbortSignal.timeout(timeout)});
if(!r.ok)throw new APIError(r.status,await r.text(),r.headers.get('retry-after'));
const value=await r.json();
if(!value||typeof value!=='object'||Array.isArray(value)||(value.error&&!(value.status||value.state)))throw new Error('Invalid API response');
return value;
}
export async function prepare(input,references=[],forceEdit=false){
if(!input||typeof input.model!=='string'||typeof input.prompt!=='string'||!input.prompt.trim())throw new Error('model and nonempty prompt required');
const body={...input},stringRefs=/^(mj-|seedream-)/.test(body.model);
if(references.length){
body.images=[...(body.images||[])];
for(const file of references){
const mime={'.png':'image/png','.jpg':'image/jpeg','.jpeg':'image/jpeg','.webp':'image/webp'}[extname(file).toLowerCase()];
const {size}=await stat(file);
if(!mime||size<=0||size>25*1024*1024)throw new Error('Local reference must be PNG/JPEG/WebP up to 25 MiB; model limits may be lower');
const url=`data:${mime};base64,${(await readFile(file)).toString('base64')}`;
body.images.push(stringRefs?url:{image_url:url});
}
}
const edit=forceEdit||(!stringRefs&&Boolean(body.image||body.images?.length));
return {body,path:`/v1/images/${edit?'edits':'generations'}/async`};
}
export function taskId(id){if(typeof id!=='string'||!/^[A-Za-z0-9_-]+$/.test(id))throw new Error('Invalid task ID');return id;}
export async function wait(key,id,timeout=1800000){
taskId(id);const deadline=Date.now()+timeout;let delay=3000;
while(Date.now()<deadline){
let t;
try{t=await request(key,'/v1/images/tasks/'+id);}
catch(e){
if(e instanceof APIError&&![429,500,502,503,504].includes(e.status))throw e;
if(!(e instanceof APIError)&&!['TypeError','TimeoutError','AbortError'].includes(e.name))throw e;
const retry=e.retryAfter==null?NaN:Number(e.retryAfter)*1000;
delay=Number.isFinite(retry)?Math.max(1000,Math.min(30000,retry)):Math.min(30000,delay*1.5);
await pause(delay);continue;
}
if((t.task_id||t.id)&&(t.task_id||t.id)!==id)throw new Error('Task ID mismatch');
if(t.submission_uncertain)throw new Error('Submission uncertain; inspect original task before another POST');
const status=String(t.status||t.state||'').toLowerCase();
if(['completed','succeeded','success'].includes(status)){
const data=t.data?.length?t.data:t.result?.data;
if(!Array.isArray(data)||!data.length||!data.every(x=>x&&typeof x==='object'))throw new Error('No valid images');
return {data,task:t};
}
if(['failed','failure','error','cancelled','canceled','expired'].includes(status))throw new Error(`Task failed: ${JSON.stringify(t.error)}`);
if(!['processing','queued','running','pending','in_progress'].includes(status))throw new Error('Unknown task status: '+status);
delay=typeof t.poll_after_ms==='number'?Math.max(1000,Math.min(30000,t.poll_after_ms)):3000;await pause(delay);
}
throw new Error(`Wait ended; resume with MXB_TASK_ID=${id}. Server task is not cancelled.`);
}
export function imageExtension(b){
if(b.subarray(0,8).equals(Buffer.from([137,80,78,71,13,10,26,10])))return '.png';
if(b.subarray(0,3).equals(Buffer.from([255,216,255])))return '.jpg';
if(b.toString('ascii',0,4)==='RIFF'&&b.toString('ascii',8,12)==='WEBP')return '.webp';
if(['GIF87a','GIF89a'].includes(b.toString('ascii',0,6)))return '.gif';
throw new Error('Invalid image signature');
}
export async function saveImages(items,directory){
await mkdir(directory,{recursive:true});const names=[];
for(const [i,item] of items.entries()){
let bytes;
if(item.url){
const url=new URL(item.url);
if(!['https:','http:'].includes(url.protocol)||url.username||url.password)throw new Error('Invalid image URL');
const r=await fetch(url,{signal:AbortSignal.timeout(120000)}); // No API credentials.
if(!r.ok||!r.headers.get('content-type')?.startsWith('image/'))throw new Error('Download is not an image');
const chunks=[];let length=0;
for await(const chunk of r.body){length+=chunk.length;if(length>limit)throw new Error('Image exceeds client limit');chunks.push(chunk);}
bytes=Buffer.concat(chunks);
}else if(item.b64_json){
if(item.b64_json.length>Math.ceil(limit/3)*4)throw new Error('Image exceeds client limit');
bytes=Buffer.from(item.b64_json,'base64');
}else throw new Error('Missing URL/Base64');
if(!bytes.length||bytes.length>limit)throw new Error('Empty or oversized image');
const name=join(directory,`image-${String(i+1).padStart(2,'0')}${imageExtension(bytes)}`);
await writeFile(name+'.part',bytes);await rename(name+'.part',name);names.push(name);
}
return names;
}
async function main(){
const key=process.env.MXB_API_KEY;if(!key)throw new Error('Set MXB_API_KEY');
const directory=process.env.MXB_OUTPUT_DIR||'images';await mkdir(directory,{recursive:true});
let id=process.env.MXB_TASK_ID;
if(!id){
const input=process.env.MXB_REQUEST?JSON.parse((await readFile(process.env.MXB_REQUEST,'utf8')).replace(/^\uFEFF/,'')):{
model:process.env.MXB_MODEL||'gpt-image-2.5',prompt:process.env.MXB_PROMPT,size:process.env.MXB_SIZE||'1024x1024',n:1};
const refs=process.env.MXB_REFERENCES?JSON.parse(process.env.MXB_REFERENCES):(process.env.MXB_REFERENCE?[process.env.MXB_REFERENCE]:[]);
const {body,path}=await prepare(input,refs,process.env.MXB_EDIT==='true');
const stateFile=join(directory,'submission.json'),state={state:'submitting',model:body.model,path,created_at:Date.now()};
await writeFile(stateFile,JSON.stringify(state,null,2),{flag:'wx'});
try{const created=await request(key,path,body,60000);id=taskId(created.task_id||created.id);}
catch(e){await writeFile(stateFile,JSON.stringify({...state,state:'submission_uncertain'},null,2));throw new Error('Submission not retried; inspect calling logs before another POST. '+e.message);}
await writeFile(stateFile,JSON.stringify({...state,state:'accepted',task_id:id},null,2));console.log('Task:',id);
}
const {data,task}=await wait(key,id);await writeFile(join(directory,'result.json'),JSON.stringify(task,null,2));
for(const name of await saveImages(data,directory))console.log('Saved:',name);
}
if(process.argv[1]&&import.meta.url===pathToFileURL(process.argv[1]).href)main().catch(e=>{console.error(e.message);process.exitCode=1;});
AS Gemini完整源码
使用方法与计费见AS Gemini专章。tools模式通常产生两次模型调用，video模式上传本地MP4进行理解。
python
#!/usr/bin/env python3
"""Python 3.10+, standard library. AS Gemini tool roundtrip or native MP4 understanding."""
import argparse
import base64
import json
import os
import re
import urllib.request
from pathlib import Path
class NoRedirect(urllib.request.HTTPRedirectHandler):
def redirect_request(self, req, fp, code, msg, headers, newurl): return None
def post(key, path, body):
req = urllib.request.Request('https://openpix.cc' + path,
data=json.dumps(body, ensure_ascii=False).encode('utf8'),
headers={'Authorization': 'Bearer ' + key, 'Content-Type': 'application/json'}, method='POST')
# Billable POST, deliberately no automatic retry.
with urllib.request.build_opener(NoRedirect()).open(req, timeout=180) as response:
value = json.load(response)
if value.get('error'): raise RuntimeError(json.dumps(value['error'], ensure_ascii=False))
return value
def run_tools(key, model):
tools = [{'type': 'function', 'function': {
'name': 'lookup_stock', 'description': '查询指定商品和仓库的库存',
'parameters': {'type': 'object', 'properties': {'sku': {'type': 'string'}, 'warehouse': {'type': 'string'}},
'required': ['sku', 'warehouse']}}}]
messages = [{'role': 'user', 'content': '请查询 TEA-01 在上海仓的库存，告诉我能否发货。'}]
first = post(key, '/v1/chat/completions', {'model': model, 'messages': messages, 'tools': tools, 'tool_choice': 'auto', 'stream': False})
if not first.get('choices'): raise RuntimeError('No assistant choice')
assistant = first['choices'][0]['message']
calls = assistant.get('tool_calls') or []
if not calls: raise RuntimeError('No tool call; inspect answer instead of assuming tool success')
messages.append(assistant) # Keep COMPLETE assistant message and tool metadata.
for call in calls:
function = call['function']
if function['name'] != 'lookup_stock' or not call.get('id'): raise RuntimeError('Unexpected tool')
args = json.loads(function['arguments'])
if not isinstance(args, dict) or not all(isinstance(args.get(k), str) for k in ('sku', 'warehouse')):
raise RuntimeError('Invalid tool arguments')
# Fixture only: replace this explicitly allowed function with your application lookup.
result = {'sku': args['sku'], 'warehouse': args['warehouse'],
'stock': 17 if args['sku'] == 'TEA-01' and args['warehouse'] in ('上海', '上海仓') else 0,
'source': 'demo_fixture'}
messages.append({'role': 'tool', 'tool_call_id': call['id'], 'content': json.dumps(result, ensure_ascii=False)})
second = post(key, '/v1/chat/completions', {'model': model, 'messages': messages, 'tools': tools, 'tool_choice': 'auto', 'stream': False})
print(json.dumps({'first_usage': first.get('usage'), 'second_response': second}, ensure_ascii=False, indent=2))
def video_body(file, prompt):
path = Path(file)
# Conservative sample limit, not the service's maximum.
if path.suffix.lower() != '.mp4' or not 0 < path.stat().st_size <= 16 * 1024 * 1024:
raise ValueError('Example accepts MP4 up to 16 MiB; start with a short clip')
raw = path.read_bytes()
if len(raw) < 12 or raw[4:8] != b'ftyp': raise ValueError('Expected MP4 ftyp signature')
return {'contents': [{'role': 'user', 'parts': [
{'inlineData': {'mimeType': 'video/mp4', 'data': base64.b64encode(raw).decode('ascii')}}, {'text': prompt}]}]}
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('mode', choices=['tools', 'video'])
parser.add_argument('--model', default='gemini-3.8-flash')
parser.add_argument('--file')
parser.add_argument('--prompt', default='按时间顺序描述视频中的主体、动作和文字，不猜测不可见内容。')
args = parser.parse_args();key = os.environ.get('MXB_API_KEY')
if not key: parser.error('Set MXB_API_KEY')
if not re.fullmatch(r'[A-Za-z0-9._-]+', args.model): parser.error('Invalid model ID')
if args.mode == 'tools': run_tools(key, args.model)
else:
if not args.file: parser.error('--file is required for video')
response = post(key, '/v1beta/models/' + args.model + ':generateContent', video_body(args.file, args.prompt))
texts = [p['text'] for c in response.get('candidates', []) for p in c.get('content', {}).get('parts', []) if isinstance(p.get('text'), str)]
if not texts: raise RuntimeError('No text; inspect finishReason/promptFeedback before another POST')
print('\n'.join(texts));print(json.dumps({'usageMetadata': response.get('usageMetadata')}, ensure_ascii=False, indent=2))
if __name__ == '__main__': main()