generated from dellevin/template
301 lines
9.5 KiB
Python
301 lines
9.5 KiB
Python
from flask import Flask, request, jsonify
|
||
import re
|
||
import jieba
|
||
import json
|
||
import os
|
||
import logging
|
||
from datetime import datetime
|
||
|
||
app = Flask(__name__)
|
||
|
||
# --- 日志配置 ---
|
||
# 创建一个logger
|
||
logger = logging.getLogger('user_activity')
|
||
logger.setLevel(logging.INFO)
|
||
|
||
# 创建一个handler，用于写入日志文件
|
||
file_handler = logging.FileHandler('useruse.log', mode='a', encoding='utf-8')
|
||
file_handler.setLevel(logging.INFO)
|
||
|
||
# 创建一个handler，用于输出到控制台（可选）
|
||
console_handler = logging.StreamHandler()
|
||
console_handler.setLevel(logging.INFO)
|
||
|
||
# 定义日志格式
|
||
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
|
||
file_handler.setFormatter(formatter)
|
||
console_handler.setFormatter(formatter)
|
||
|
||
# 给logger添加handler
|
||
logger.addHandler(file_handler)
|
||
logger.addHandler(console_handler)
|
||
|
||
# --- 从文件加载自定义词汇 ---
|
||
JIEBA_DICT_FILE = 'jieba.txt'
|
||
|
||
|
||
def load_custom_words_from_file(file_path):
|
||
"""从指定文件加载自定义词汇"""
|
||
custom_words = set() # 使用set避免重复
|
||
if not os.path.exists(file_path):
|
||
print(f"警告: 找不到自定义词汇文件 '{file_path}'，将使用空列表。")
|
||
return list(custom_words), set()
|
||
|
||
try:
|
||
with open(file_path, 'r', encoding='utf-8') as f:
|
||
for line in f:
|
||
word = line.strip() # 去除行首行尾的空白字符和换行符
|
||
if word: # 确保不是空行
|
||
custom_words.add(word)
|
||
except Exception as e:
|
||
print(f"错误: 读取自定义词汇文件 '{file_path}' 时发生异常: {e}")
|
||
|
||
# 从加载的词汇中筛选出需要特殊处理的复合词（包含大写字母的驼峰命名）
|
||
compound_words = {word for word in custom_words if any(c.isupper() for c in word)}
|
||
simple_custom_words = custom_words - compound_words
|
||
|
||
return list(simple_custom_words), compound_words
|
||
|
||
|
||
# 从文件加载自定义词汇
|
||
SIMPLE_CUSTOM_WORDS, COMPOUND_WORDS_SET = load_custom_words_from_file(JIEBA_DICT_FILE)
|
||
|
||
# 为简单词汇添加到jieba词典
|
||
for word in SIMPLE_CUSTOM_WORDS:
|
||
jieba.add_word(word)
|
||
# print(f"成功从 '{JIEBA_DICT_FILE}' 加载了 {len(SIMPLE_CUSTOM_WORDS)} 个简单自定义词汇。")
|
||
# print(f"识别出 {len(COMPOUND_WORDS_SET)} 个复合词汇待处理: {list(COMPOUND_WORDS_SET)[:10]}...") # 打印前10个作为示例
|
||
|
||
# 预编译复合词的正则表达式模式，以提高性能
|
||
# 例如: (MemoFlow|MookNote|VoceChat)
|
||
COMPOUND_PATTERN = re.compile('|'.join(re.escape(w) for w in COMPOUND_WORDS_SET), re.IGNORECASE)
|
||
|
||
|
||
def tokenize_text(text):
|
||
"""
|
||
对中英文及数字进行分词
|
||
- 复合词：优先匹配自定义的驼峰命名复合词
|
||
- 中文：使用 jieba 分词
|
||
- 英文：按空格和标点分词
|
||
- 数字：单独提取
|
||
"""
|
||
if not text or not text.strip():
|
||
return []
|
||
|
||
tokens = []
|
||
# 定义正则表达式模式
|
||
# 中文字符范围
|
||
chinese_pattern = re.compile(r'[\u4e00-\u9fff]+')
|
||
# 英文单词（包含连字符）
|
||
english_pattern = re.compile(r'[a-zA-Z]+(?:-[a-zA-Z]+)*')
|
||
# 数字（包含小数、负数）
|
||
number_pattern = re.compile(r'-?\d+\.?\d*')
|
||
|
||
# 当前处理位置
|
||
pos = 0
|
||
text_length = len(text)
|
||
|
||
while pos < text_length:
|
||
char = text[pos]
|
||
|
||
# 跳过空白字符
|
||
if char.isspace():
|
||
pos += 1
|
||
continue
|
||
|
||
# --- 新增逻辑：优先匹配复合词 ---
|
||
compound_match = COMPOUND_PATTERN.match(text, pos)
|
||
if compound_match:
|
||
matched_word = compound_match.group()
|
||
tokens.append({
|
||
'text': matched_word,
|
||
'type': 'compound_eng',
|
||
'length': len(matched_word)
|
||
})
|
||
pos += len(matched_word)
|
||
continue # 匹配到复合词后，跳过后续的其他匹配
|
||
|
||
# 匹配中文字符串
|
||
if chinese_pattern.match(char):
|
||
# 找到连续的中文字符
|
||
chinese_str = ''
|
||
while pos < text_length and chinese_pattern.match(text[pos]):
|
||
chinese_str += text[pos]
|
||
pos += 1
|
||
|
||
# 使用 jieba 对中文字符串进行分词
|
||
# 注意：这里的 jieba.cut() 会使用我们之前 add_word 添加的自定义词汇
|
||
chinese_tokens = list(jieba.cut(chinese_str))
|
||
for token in chinese_tokens:
|
||
if token.strip():
|
||
tokens.append({
|
||
'text': token,
|
||
'type': 'chinese',
|
||
'length': len(token)
|
||
})
|
||
|
||
# 匹配英文单词
|
||
elif english_pattern.match(char):
|
||
match = english_pattern.match(text, pos)
|
||
if match:
|
||
word = match.group()
|
||
tokens.append({
|
||
'text': word,
|
||
'type': 'english',
|
||
'length': len(word)
|
||
})
|
||
pos += len(word)
|
||
|
||
# 匹配数字
|
||
elif number_pattern.match(char) or (char == '-' and pos + 1 < text_length and text[pos + 1].isdigit()):
|
||
match = number_pattern.match(text, pos)
|
||
if match:
|
||
number = match.group()
|
||
tokens.append({
|
||
'text': number,
|
||
'type': 'number',
|
||
'length': len(number)
|
||
})
|
||
pos += len(number)
|
||
|
||
# 其他字符（标点符号等），单独处理
|
||
else:
|
||
tokens.append({
|
||
'text': char,
|
||
'type': 'punctuation',
|
||
'length': 1
|
||
})
|
||
pos += 1
|
||
|
||
return tokens
|
||
|
||
|
||
def get_token_stats(tokens):
|
||
"""获取分词统计信息"""
|
||
stats = {
|
||
'total': len(tokens),
|
||
'chinese': 0,
|
||
'english': 0,
|
||
'compound_eng': 0, # 为复合词类型添加计数
|
||
'number': 0,
|
||
'punctuation': 0
|
||
}
|
||
|
||
for token in tokens:
|
||
token_type = token['type']
|
||
if token_type in stats:
|
||
stats[token_type] += 1
|
||
|
||
# 重新计算总数，确保准确性
|
||
stats['total'] = sum(v for k, v in stats.items() if k != 'total')
|
||
|
||
return stats
|
||
|
||
|
||
# 定义用户Token文件路径
|
||
TOKEN_FILE_PATH = 'usertoken.json'
|
||
|
||
|
||
# 启动时加载Token文件
|
||
def load_valid_tokens():
|
||
"""从文件加载有效的Token列表"""
|
||
if not os.path.exists(TOKEN_FILE_PATH):
|
||
print(f"错误: 找不到Token文件 '{TOKEN_FILE_PATH}'")
|
||
return {}
|
||
|
||
try:
|
||
with open(TOKEN_FILE_PATH, 'r', encoding='utf-8') as f:
|
||
tokens_data = json.load(f)
|
||
|
||
# 将列表转换为以token为键的字典，便于快速查找
|
||
token_map = {}
|
||
for item in tokens_data:
|
||
# 确保数据格式正确
|
||
if 'user' in item and 'token' in item:
|
||
token_map[item['token']] = item['user']
|
||
else:
|
||
print(f"警告: Token文件中发现格式错误的条目: {item}")
|
||
|
||
# print(f"成功从 '{TOKEN_FILE_PATH}' 加载了 {len(token_map)} 个有效Token。")
|
||
return token_map
|
||
|
||
except json.JSONDecodeError as e:
|
||
print(f"错误: Token文件 '{TOKEN_FILE_PATH}' 不是有效的JSON格式。{e}")
|
||
return {}
|
||
except Exception as e:
|
||
print(f"错误: 读取Token文件时发生异常: {e}")
|
||
return {}
|
||
|
||
|
||
# 程序启动时加载一次
|
||
VALID_TOKENS_MAP = load_valid_tokens()
|
||
|
||
|
||
def get_user_by_token(token):
|
||
"""根据token获取用户名，如果无效则返回None"""
|
||
return VALID_TOKENS_MAP.get(token)
|
||
|
||
|
||
def authenticate_request():
|
||
"""检查请求头中的Authorization token，并返回用户信息"""
|
||
auth_header = request.headers.get('Authorization')
|
||
if not auth_header or not auth_header.startswith('Bearer '):
|
||
return None, "缺少或格式错误的Authorization头。请使用 'Authorization: Bearer <your_token>'"
|
||
|
||
token = auth_header.split(" ", 1)[1]
|
||
user = get_user_by_token(token)
|
||
|
||
if not user:
|
||
return None, "无效的Token"
|
||
|
||
return user, None
|
||
|
||
|
||
@app.route('/tokenize', methods=['POST'])
|
||
def tokenize_endpoint():
|
||
"""处理分词请求的 API 端点"""
|
||
user, error_msg = authenticate_request()
|
||
if not user:
|
||
return jsonify({'success': False, 'error': error_msg}), 401
|
||
|
||
data = request.get_json()
|
||
input_text = data.get('input_text', '')
|
||
|
||
# 记录日志：谁在什么时候输入了什么内容
|
||
log_message = f"User '{user}' submitted text: '{input_text}'"
|
||
logger.info(log_message)
|
||
print(f"Log Recorded: {log_message}") # 控制台也打印一下，方便即时查看
|
||
|
||
try:
|
||
# 执行分词
|
||
tokens = tokenize_text(input_text)
|
||
|
||
# 获取统计信息
|
||
stats = get_token_stats(tokens)
|
||
|
||
# 返回成功的 JSON 响应，包含请求的用户信息
|
||
return jsonify({
|
||
'success': True,
|
||
'tokens': tokens,
|
||
'stats': stats,
|
||
'original_text': input_text,
|
||
'requested_by': user # 添加请求者信息
|
||
})
|
||
|
||
except Exception as e:
|
||
# 如果处理出错，也记录一条错误日志
|
||
error_log_message = f"Error processing request for user '{user}': {str(e)}"
|
||
logger.error(error_log_message)
|
||
|
||
return jsonify({
|
||
'success': False,
|
||
'error': f'分词处理失败: {str(e)}'
|
||
}), 500
|
||
|
||
|
||
# --- 以下为原有的分词和统计逻辑，保持不变 ---
|
||
# (tokenize_text 和 get_token_stats 已经在上面定义过了)
|
||
|
||
|
||
if __name__ == '__main__':
|
||
app.run(debug=False, host='0.0.0.0', port=27056) # 生产环境建议关闭debug |