#!/usr/bin/env python3 """Small standalone client: own-account API token, bounded polling, no dependencies.""" import argparse import json import math import os from pathlib import Path import re import stat import sys import time import urllib.error import urllib.request from urllib.parse import urlsplit DEFAULT_BASE_URL = 'https://aiwenti.cn' class ClientError(Exception): pass class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self,*args,**kwargs): raise ClientError('服务地址发生跳转，请核对地址；令牌没有转发给其他网站。') def read_config(path): """Read an owner-held configuration, tightening its permissions on the open file.""" try: if path.is_symlink():raise ClientError('请选择自己保存的接入配置文件，不能使用文件链接。') descriptor=os.open(path,os.O_RDONLY|getattr(os,'O_NOFOLLOW',0)) with os.fdopen(descriptor,'r',encoding='utf-8') as handle: details=os.fstat(handle.fileno()) if not stat.S_ISREG(details.st_mode) or details.st_size>16384: raise ClientError('配置文件无效，请从 API 页面重新复制并保存。') if hasattr(os,'getuid') and details.st_uid!=os.getuid(): raise ClientError('请使用自己保存的接入配置文件。') if hasattr(os,'fchmod'):os.fchmod(handle.fileno(),0o600) value=json.load(handle) except (OSError,ValueError,UnicodeError): raise ClientError('无法读取接入配置，请从 API 页面重新复制并保存。') from None if isinstance(value, dict) and set(value) == {'mcpServers'}: try: servers = value['mcpServers'] if not isinstance(servers, dict) or set(servers) != {'wenti'}:raise ValueError() server = servers['wenti'] if (not isinstance(server, dict) or set(server) != {'type', 'url', 'headers'} or server['type'] != 'streamableHttp' or not isinstance(server['url'], str) or not isinstance(server['headers'], dict) or set(server['headers']) != {'Authorization'}):raise ValueError() address = urlsplit(server['url']) authorization = server['headers']['Authorization'] if (address.path != '/api/mcp' or address.query or address.fragment or not isinstance(authorization, str) or not authorization.startswith('Bearer ')):raise ValueError() value = {'base_url': address._replace(path='').geturl(), 'api_key': authorization[7:]} except (TypeError, KeyError, ValueError): raise ClientError('配置文件无效，请复制 API 页面的文提配置。') from None if (not isinstance(value,dict) or set(value)!={'base_url','api_key'} or not all(isinstance(value[name],str) for name in value)): raise ClientError('配置文件无效，请从 API 页面重新复制并保存。') return value class Client: def __init__(self,base,key,*,opener=None): try: if not isinstance(base,str) or any(char.isspace() for char in base):raise ValueError() parsed=urlsplit(base) if parsed.port==0:raise ValueError() except ValueError: raise ClientError('服务地址无效，请复制 API 页面显示的完整地址。') from None if (parsed.scheme not in {'http','https'} or not parsed.hostname or parsed.username or parsed.password or parsed.path not in ('','/') or parsed.query or parsed.fragment): raise ClientError('服务地址需要是网站根地址。') if parsed.scheme=='http' and parsed.hostname not in {'localhost','127.0.0.1','::1'}: raise ClientError('远程服务请使用 HTTPS 地址，本机可使用 http://127.0.0.1:8794。') if not isinstance(key,str) or not re.fullmatch(r'VTK-[A-Za-z0-9_-]{43}',key): raise ClientError('请在工作区 API 页面生成令牌并复制配置，或提供私密令牌文件。') self.base,self.key=base.rstrip('/'),key handlers=[NoRedirect()] if parsed.hostname in {'localhost','127.0.0.1','::1'}:handlers.append(urllib.request.ProxyHandler({})) self.opener=opener or urllib.request.build_opener(*handlers) def call(self,path,payload=None,*,raw=False): data=json.dumps(payload,ensure_ascii=False).encode() if payload is not None else None request=urllib.request.Request(self.base+path,data=data,headers={ 'Authorization':'Bearer '+self.key,'Content-Type':'application/json', 'ngrok-skip-browser-warning':'1'}) try: with self.opener.open(request,timeout=20) as response: body=response.read(2*1024*1024+1) if len(body)>2*1024*1024:raise ClientError('接口返回内容超过读取上限。') return body if raw else json.loads(body) except urllib.error.HTTPError as error: try:message=json.loads(error.read(8192)).get('error',{}).get('message') except (ValueError,OSError):message=None raise ClientError(message or f'接口请求失败：HTTP {error.code}') from None except (urllib.error.URLError,OSError):raise ClientError('无法连接服务，请检查网站地址和 Mac 是否在线。') from None except ValueError:raise ClientError('接口没有返回有效 JSON。') from None def extract(self,text,*,format='text',timeout=300): record=self.call('/api/jobs',{'text':text,'duplicate_policy':'reuse'}) reused=record.get('reused') if record.get('reused') in {'active','result'} else None identity=record.get('id') if not isinstance(identity,str) or not re.fullmatch(r'[a-f0-9]{32}',identity):raise ClientError('接口没有返回有效任务编号。') deadline=time.monotonic()+timeout while record.get('status') in {'queued','running'}: if time.monotonic()>=deadline:raise ClientError('等待超时，任务仍在服务端运行。可用 status 命令继续查询 '+identity) print(record.get('stage_label','正在处理'),file=sys.stderr) time.sleep(1.5) record=self.call('/api/jobs/'+identity) if record.get('status')!='complete':raise ClientError(record.get('message') or '任务未完整成功，请在工作区查看。') if format=='json': return (json.dumps(self.report(record,reused=reused),ensure_ascii=False,indent=2)+'\n').encode('utf-8') endpoint='/text' if format=='text' else '/subtitles/'+format return self.call('/api/jobs/'+identity+endpoint,raw=True) @staticmethod def report(record,*,reused=None): result=record.get('result') or {} text=result.get('original_text') if 'original_text' in result else result.get('text') if not isinstance(text,str) or not text.strip(): raise ClientError('这条任务没有可核验的文案正文，请查看任务详情。') billing=record.get('billing') or {} points=billing.get('charged_points') charged=(round(points/10,6) if type(points) in (int,float) and math.isfinite(points) and points>=0 else None) return {'task_id':record['id'],'status':'complete','source_url':result.get('source'), 'title':result.get('title'),'text':text,'duration_seconds':result.get('duration'), 'reused':reused,'quota':{'consumed_minutes':0 if reused else charged, 'task_charged_minutes':charged,'billing_state':billing.get('state')}, 'warnings':result.get('warnings') or [],'assessment':None} def main(argv=None): parser=argparse.ArgumentParser(description=__doc__) parser.add_argument('--base-url',help='服务根地址，默认使用下载此工具的网站') parser.add_argument('--config',type=Path,help='API 页面下载的文提接入.json') parser.add_argument('--key-file',type=Path) sub=parser.add_subparsers(dest='command',required=True) extract=sub.add_parser('extract',help='提取单条分享链接，自动等待并复用已有结果') extract.add_argument('text');extract.add_argument('--format',choices=['text','srt','vtt','json'],default='text') extract.add_argument('--output',type=Path);extract.add_argument('--timeout',type=int,default=300) status=sub.add_parser('status');status.add_argument('job_id') sub.add_parser('me') options=parser.parse_args(argv) try: config=read_config(options.config) if options.config else {} if options.key_file: if options.key_file.stat().st_mode & 0o077:raise ClientError('令牌文件应仅自己可读，请先执行 chmod 600。') key=options.key_file.read_text().strip() else:key=config.get('api_key',os.environ.get('VIDEO_TEXT_API_KEY','')).strip() client=Client(options.base_url or config.get('base_url') or DEFAULT_BASE_URL,key) if options.command=='extract': if not 5<=options.timeout<=1800:raise ClientError('等待时间需要是 5～1800 秒。') body=client.extract(options.text,format=options.format,timeout=options.timeout) if options.output:options.output.write_bytes(body) else:sys.stdout.buffer.write(body) elif options.command=='status': if not re.fullmatch(r'[a-f0-9]{32}',options.job_id):raise ClientError('任务编号无效。') print(json.dumps(client.call('/api/jobs/'+options.job_id),ensure_ascii=False)) else:print(json.dumps(client.call('/api/auth/me'),ensure_ascii=False)) except (ClientError,OSError) as error: print(str(error) if isinstance(error,ClientError) else '无法读取或写入本地文件。',file=sys.stderr) return 1 return 0 if __name__=='__main__':raise SystemExit(main())