#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
TextCraft Chat API 调用示例（requests + Basic Auth + 流式响应处理）
特性
- 使用 requests.Session 复用连接
- Basic 认证从环境变量读取，避免硬编码敏感信息
- 支持流式(text/event-stream)与非流式(application/json)两种返回
- 健壮的错误处理与超时设置
- 轻量的SSE解析器，能兼容常见的 data: 行式分片
"""
import os
import sys
import json
import contextlib
from typing import Generator, Iterable, Optional, Union, Dict, Any
import requests
from requests.auth import HTTPBasicAuth
from requests.exceptions import RequestException, Timeout, HTTPError, ConnectionError
DEFAULT_BASE_URL = "https://api.textcraft.example.com"
DEFAULT_TIMEOUT = (10, 300) # (connect_timeout, read_timeout)
class TextCraftClient:
"""
TextCraft API 客户端（Basic Auth）
"""
def __init__(
self,
base_url: str = DEFAULT_BASE_URL,
username: Optional[str] = None,
password: Optional[str] = None,
timeout: Union[float, tuple] = DEFAULT_TIMEOUT,
verify: Union[bool, str] = True,
session: Optional[requests.Session] = None,
) -> None:
"""
参数:
- base_url: API服务基础URL, 如 https://api.textcraft.example.com
- username: 基本认证用户名（建议从环境变量读取）
- password: 基本认证密码（建议从环境变量读取）
- timeout: 超时设置，float或(connect, read)元组
- verify: TLS证书校验，为True或CA证书路径
- session: 可注入自定义requests.Session
"""
self.base_url = base_url.rstrip("/")
self.timeout = timeout
self.verify = verify
self.session = session or requests.Session()
# 从环境变量兜底读取凭证，避免硬编码敏感信息
self.username = username or os.getenv("TEXTCRAFT_USERNAME")
self.password = password or os.getenv("TEXTCRAFT_PASSWORD")
if not self.username or not self.password:
raise ValueError(
"缺少Basic认证凭证。请通过参数或环境变量 "
"TEXTCRAFT_USERNAME / TEXTCRAFT_PASSWORD 提供。"
)
self.auth = HTTPBasicAuth(self.username, self.password)
self.default_headers = {
"User-Agent": "textcraft-python/1.0",
}
def _endpoint(self, path: str) -> str:
return f"{self.base_url.rstrip('/')}/{path.lstrip('/')}"
@staticmethod
def _extract_stream_piece(obj: Dict[str, Any]) -> Optional[str]:
"""
尝试从常见的SSE JSON数据结构中提取文本片段。
若结构未知，返回None让上层以原始字符串回退。
兼容多种常见字段命名。
"""
# 常见结构1：choices[].delta.content（OpenAI风格）
try:
return obj["choices"][0]["delta"].get("content")
except Exception:
pass
# 常见结构2：choices[].message.content（最终消息）
try:
return obj["choices"][0]["message"].get("content")
except Exception:
pass
# 常见结构3：直接 content 字段
try:
if isinstance(obj.get("content"), str):
return obj["content"]
except Exception:
pass
return None
@staticmethod
def _iter_sse_lines(resp: requests.Response) -> Iterable[str]:
"""
简单SSE解析：只处理 data: 行；忽略注释和空行。
以行作为边界，不做多行拼接；适合多数token级别增量场景。
"""
for raw in resp.iter_lines(decode_unicode=True):
if not raw:
continue
if raw.startswith(":"):
# SSE注释行
continue
if raw.startswith("data:"):
data = raw[5:].strip()
# 约定的结束信号
if data == "[DONE]":
return
yield data
def chat(
self,
messages: Iterable[Dict[str, str]],
model: str = "tc-1.2-chat",
temperature: float = 0.3,
stream: bool = True,
max_tokens: int = 600,
extra_headers: Optional[Dict[str, str]] = None,
) -> Union[Dict[str, Any], Generator[str, None, None]]:
"""
发起聊天请求。
- 若 stream=True，返回一个生成器，逐段产出文本片段(str)。
- 若 stream=False，返回一次性JSON结果(dict)。
注意：使用完毕后生成器会自动关闭底层响应，确保资源释放。
"""
url = self._endpoint("/v1/chat")
payload = {
"model": model,
"messages": list(messages),
"temperature": float(temperature),
"stream": bool(stream),
"max_tokens": int(max_tokens),
}
headers = dict(self.default_headers)
# 根据是否流式设置Accept
if stream:
headers["Accept"] = "text/event-stream, application/json;q=0.9"
else:
headers["Accept"] = "application/json"
headers["Content-Type"] = "application/json"
if extra_headers:
headers.update(extra_headers)
try:
if stream:
resp = self.session.post(
url,
headers=headers,
json=payload,
auth=self.auth,
timeout=self.timeout,
verify=self.verify,
stream=True,
)
resp.raise_for_status()
ctype = (resp.headers.get("Content-Type") or "").lower()
is_sse = "text/event-stream" in ctype
@contextlib.contextmanager
def _closer(r: requests.Response):
try:
yield
finally:
# 确保连接回收
r.close()
def _generator() -> Generator[str, None, None]:
with _closer(resp):
if is_sse:
for data in self._iter_sse_lines(resp):
# 优先尝试解析JSON提取文本片段
piece = None
if data and (data.startswith("{") or data.startswith("[")):
try:
obj = json.loads(data)
piece = self._extract_stream_piece(obj)
# 若无法提取但是JSON，回退为紧凑字符串输出
if piece is None:
piece = json.dumps(obj, ensure_ascii=False)
except json.JSONDecodeError:
# 非JSON片段，按原始字符串返回
piece = data
else:
piece = data
if piece:
yield piece
else:
# 一些服务即使指定了stream也返回完整JSON
chunk = resp.content.decode(resp.encoding or "utf-8", errors="replace")
try:
obj = json.loads(chunk)
# 直接输出完整消息体（非流模式的兼容分支）
text = self._extract_stream_piece(obj) or json.dumps(obj, ensure_ascii=False)
yield text
except json.JSONDecodeError:
# 回退为原始文本
yield chunk
return _generator()
else:
resp = self.session.post(
url,
headers=headers,
json=payload,
auth=self.auth,
timeout=self.timeout,
verify=self.verify,
)
resp.raise_for_status()
return resp.json()
except HTTPError as e:
# 服务端返回非2xx
detail = None
if e.response is not None:
try:
detail = e.response.json()
except Exception:
detail = e.response.text
raise RuntimeError(f"HTTP错误: {e} | 详情: {detail}") from e
except Timeout as e:
raise TimeoutError(f"请求超时: {e}") from e
except ConnectionError as e:
raise ConnectionError(f"连接失败: {e}") from e
except RequestException as e:
raise RuntimeError(f"请求异常: {e}") from e
def main() -> None:
"""
可执行示例：以流式打印返回内容到标准输出。
"""
# 示例消息（来自问题描述）
messages = [
{
"role": "system",
"content": "你是一名资深中文写作助手，擅长将多段素材整合为结构清晰的文章。输出使用中文，格式为标题、导语、正文小节、结尾。"
},
{
"role": "user",
"content": "素材1：本周我们上线了“离线草稿”与“自动保存”功能，解决了弱网环境内容丢失问题；素材2：新增草稿历史版本对比，并提供一键恢复；读者对象：写作爱好者与自媒体作者；风格：实用、清晰、适度幽默；长度：约600字。"
},
]
client = TextCraftClient(
base_url="https://api.textcraft.example.com",
# 为安全起见，用户名/密码请通过环境变量 TEXTCRAFT_USERNAME / TEXTCRAFT_PASSWORD 提供
# username=os.getenv("TEXTCRAFT_USERNAME"),
# password=os.getenv("TEXTCRAFT_PASSWORD"),
timeout=(10, 300),
verify=True,
)
# 流式打印
try:
stream = client.chat(
messages=messages,
model="tc-1.2-chat",
temperature=0.3,
stream=True,
max_tokens=600,
)
if isinstance(stream, dict):
# 极少数情况下服务端可能返回非流式JSON（即使指定了stream=True）
sys.stdout.write(json.dumps(stream, ensure_ascii=False))
sys.stdout.flush()
else:
for piece in stream:
sys.stdout.write(piece)
sys.stdout.flush()
except Exception as e:
# 统一错误输出
print(f"\n调用失败: {e}", file=sys.stderr)
sys.exit(1)
if __name__ == "__main__":
main()
-
参数说明：
- base_url：API基础地址。默认 https://api.textcraft.example.com。
- username/password：Basic认证用户名与密码。为安全起见请通过环境变量 TEXTCRAFT_USERNAME、TEXTCRAFT_PASSWORD 提供，避免硬编码。
- timeout：请求超时。可为单个浮点数或(connect_timeout, read_timeout)元组。示例中为(10, 300)，以支持较长流式会话。
- verify：TLS证书校验。True 为启用系统CA；也可传入自定义CA文件路径。不要关闭校验除非在受控环境下。
- messages：对话消息数组，每个元素包含 role 和 content。role 常见取值：system、user、assistant。
- model：模型名称，示例为 tc-1.2-chat。
- temperature：采样温度，0.0-1.0。
- stream：是否开启流式返回。True 时将返回生成器，逐段产出内容；False 时返回一次性JSON。
- max_tokens：生成上限token数。
- extra_headers：可选的自定义请求头（如业务追踪ID）。
-
使用示例：
# 1) 设置环境变量（Linux/macOS）
export TEXTCRAFT_USERNAME="test_analyst"
export TEXTCRAFT_PASSWORD="s3curePwd!42"
# Windows (PowerShell)
# $env:TEXTCRAFT_USERNAME="test_analyst"
# $env:TEXTCRAFT_PASSWORD="s3curePwd!42"
# 2) 运行脚本
python textcraft_client.py
或在代码中以库方式调用（非流式一次性结果示例）：
from textcraft_client import TextCraftClient
client = TextCraftClient()
resp = client.chat(
messages=[{"role":"user","content":"请用一句话介绍你自己"}],
model="tc-1.2-chat",
stream=False,
max_tokens=100
)
print(resp)
- 注意事项：
- 凭证安全：不要将用户名/密码硬编码进代码或提交到版本控制。请使用环境变量或安全的密钥管理方案。
- 超时与健壮性：流式响应可能较长，读超时(read timeout)请适当放宽；连接超时(connect timeout)保持较短以快速失败。
- 流式协议兼容：代码优先按 text/event-stream 解析SSE；若服务端返回application/json（即使请求了流式），也能兼容处理。
- 错误处理：已对HTTP错误、超时、连接错误等进行分类处理，并包含服务端错误体的安全输出。生产环境可对异常信息做更细粒度日志与观测。
- TLS 校验：保持 verify=True，除非在测试环境且已知信任的自签证书；禁用校验存在中间人攻击风险。
- 重试策略：示例未对POST做自动重试，以避免重复请求带来的副作用或计费。若确需重试，请在调用层根据幂等性和业务容忍度谨慎实现。
- 日志脱敏：在记录请求/响应日志时避免输出Authorization、密码以及完整响应内容（尤其包含用户隐私数据时）。
