134 lines
3.8 KiB
Python
134 lines
3.8 KiB
Python
"""
|
||
工具函数模块：语言检测、HTML清洗、内容提取、时间解析、AI API调用、文本换行
|
||
"""
|
||
|
||
import re
|
||
import requests
|
||
from datetime import datetime, timedelta, timezone
|
||
from email.utils import parsedate_to_datetime
|
||
|
||
|
||
def is_chinese(text):
|
||
"""判断文本是否为中文（汉字占比 > 20%）"""
|
||
if not text:
|
||
return True
|
||
chinese_chars = sum(1 for c in text if '\u4e00' <= c <= '\u9fff')
|
||
return chinese_chars > len(text) * 0.2
|
||
|
||
|
||
def clean_html(html_text):
|
||
"""清理HTML标签，合并空白字符"""
|
||
if not html_text:
|
||
return ""
|
||
text = re.sub(r'<[^>]+>', ' ', html_text)
|
||
text = re.sub(r'\s+', ' ', text)
|
||
return text.strip()
|
||
|
||
|
||
def extract_content(entry):
|
||
"""从RSS条目提取完整正文内容"""
|
||
content = entry.get('summary', '') or entry.get('description', '')
|
||
if not content:
|
||
content_list = entry.get('content', [{}])
|
||
if content_list and isinstance(content_list, list):
|
||
content = content_list[0].get('value', '')
|
||
return clean_html(content)
|
||
|
||
|
||
def parse_pub_time(entry):
|
||
"""解析发布时间，统一转换为北京时间（UTC+8）"""
|
||
beijing_tz = timezone(timedelta(hours=8))
|
||
|
||
time_fields = ['published_parsed', 'updated_parsed', 'created_parsed']
|
||
for field in time_fields:
|
||
if entry.get(field):
|
||
try:
|
||
utc_time = datetime(*entry[field][:6], tzinfo=timezone.utc)
|
||
return utc_time.astimezone(beijing_tz).replace(tzinfo=None)
|
||
except Exception:
|
||
continue
|
||
|
||
for field in ['published', 'updated', 'created']:
|
||
if entry.get(field):
|
||
try:
|
||
dt = parsedate_to_datetime(entry[field])
|
||
if dt.tzinfo is not None:
|
||
return dt.astimezone(beijing_tz).replace(tzinfo=None)
|
||
return dt
|
||
except Exception:
|
||
continue
|
||
|
||
return datetime.now()
|
||
|
||
|
||
def call_ai(messages, api_key, base_url, model, temperature=0.3, max_tokens=800, purpose=None):
|
||
"""调用 OpenAI 兼容的 Chat Completion API"""
|
||
if not api_key:
|
||
raise ValueError('OPENAI_API_KEY 未配置')
|
||
|
||
headers = {
|
||
'Content-Type': 'application/json',
|
||
'Authorization': f'Bearer {api_key}'
|
||
}
|
||
|
||
data = {
|
||
'model': model,
|
||
'messages': messages,
|
||
'temperature': temperature,
|
||
'max_tokens': max_tokens,
|
||
}
|
||
|
||
try:
|
||
response = requests.post(
|
||
f'{base_url}/chat/completions',
|
||
headers=headers,
|
||
json=data,
|
||
timeout=120
|
||
)
|
||
response.raise_for_status()
|
||
result = response.json()['choices'][0]['message']['content']
|
||
_record_api_success(model, purpose, max_tokens)
|
||
return result
|
||
except Exception:
|
||
_record_api_failure(model, purpose, max_tokens)
|
||
raise
|
||
|
||
|
||
def wrap_text(text, font, max_width, draw):
|
||
"""将长文本按像素宽度自动换行，返回行列表"""
|
||
lines = []
|
||
current_line = ""
|
||
|
||
for char in text:
|
||
test_line = current_line + char
|
||
bbox = draw.textbbox((0, 0), test_line, font=font)
|
||
line_width = bbox[2] - bbox[0]
|
||
|
||
if line_width <= max_width:
|
||
current_line = test_line
|
||
else:
|
||
if current_line:
|
||
lines.append(current_line)
|
||
current_line = char
|
||
|
||
if current_line:
|
||
lines.append(current_line)
|
||
|
||
return lines
|
||
|
||
|
||
def _record_api_success(model, purpose, max_tokens):
|
||
try:
|
||
from .monitor import get_monitor
|
||
get_monitor().record_api_call(model or "unknown", purpose or "unknown", max_tokens, True)
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def _record_api_failure(model, purpose, max_tokens):
|
||
try:
|
||
from .monitor import get_monitor
|
||
get_monitor().record_api_call(model or "unknown", purpose or "unknown", max_tokens, False)
|
||
except Exception:
|
||
pass
|