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 '" 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