|
|
@@ -0,0 +1,563 @@
|
|
|
+"""
|
|
|
+数学题目查重助手 - 使用向量相似度比对题目是否重复
|
|
|
+"""
|
|
|
+import json
|
|
|
+import os
|
|
|
+import pickle
|
|
|
+import re
|
|
|
+import numpy as np
|
|
|
+import faiss
|
|
|
+import pymysql
|
|
|
+import concurrent.futures
|
|
|
+from openai import OpenAI
|
|
|
+from typing import List, Dict, Tuple, Optional
|
|
|
+from config import DB_HOST, DB_PORT, DB_DATABASE, DB_USERNAME, DB_PASSWORD, OPENAI_API_KEY, OPENAI_BASE_URL
|
|
|
+
|
|
|
+class QuestionDuplicateChecker:
|
|
|
+ """题目查重器"""
|
|
|
+
|
|
|
+ def __init__(self, index_path="questions_tem.index", metadata_path="questions_tem_metadata.pkl"):
|
|
|
+ """
|
|
|
+ 初始化查重器
|
|
|
+
|
|
|
+ Args:
|
|
|
+ index_path: FAISS索引保存路径
|
|
|
+ metadata_path: 元数据保存路径
|
|
|
+ """
|
|
|
+ self.client = OpenAI(
|
|
|
+ api_key=OPENAI_API_KEY,
|
|
|
+ base_url=OPENAI_BASE_URL
|
|
|
+ )
|
|
|
+ self.index_path = index_path
|
|
|
+ self.metadata_path = metadata_path
|
|
|
+ self.index = None
|
|
|
+ self.metadata = [] # 存储题目ID和文本，以便回显
|
|
|
+
|
|
|
+ # 权重设置
|
|
|
+ self.weights = {
|
|
|
+ 'stem': 0.8,
|
|
|
+ 'options': 0.05,
|
|
|
+ 'answer': 0.05,
|
|
|
+ 'solution': 0.1
|
|
|
+ }
|
|
|
+ # 维度调整：4部分拼接后的总维度为 3072 * 4 = 12288
|
|
|
+ self.dimension = 3072 * 4
|
|
|
+
|
|
|
+ self._load_index()
|
|
|
+
|
|
|
+ def clean_text(self, text: str) -> str:
|
|
|
+ """清理文本：移除图片标签、归一化空白字符"""
|
|
|
+ if not text:
|
|
|
+ return ""
|
|
|
+ # 移除 <image .../> 标签
|
|
|
+ text = re.sub(r'<image\s+[^>]*/>', '', text)
|
|
|
+ # 归一化空白：将多个空格、换行替换为单个空格
|
|
|
+ text = re.sub(r'\s+', ' ', text)
|
|
|
+ return text.strip()
|
|
|
+
|
|
|
+ def normalize_options(self, options) -> str:
|
|
|
+ """标准化选项内容"""
|
|
|
+ if not options:
|
|
|
+ return ""
|
|
|
+
|
|
|
+ data = None
|
|
|
+ if isinstance(options, str):
|
|
|
+ # 如果是字符串，尝试解析 JSON
|
|
|
+ text = options.strip()
|
|
|
+ if (text.startswith('{') and text.endswith('}')) or (text.startswith('[') and text.endswith(']')):
|
|
|
+ try:
|
|
|
+ data = json.loads(text)
|
|
|
+ except:
|
|
|
+ pass
|
|
|
+ if data is None:
|
|
|
+ return self.clean_text(options)
|
|
|
+ else:
|
|
|
+ data = options
|
|
|
+
|
|
|
+ if isinstance(data, dict):
|
|
|
+ # 排序并清理字典内容
|
|
|
+ sorted_data = {str(k).strip(): self.clean_text(str(v)) for k, v in sorted(data.items())}
|
|
|
+ return self.clean_text(json.dumps(sorted_data, ensure_ascii=False))
|
|
|
+ elif isinstance(data, list):
|
|
|
+ # 清理列表内容
|
|
|
+ cleaned_data = [self.clean_text(str(v)) for v in data]
|
|
|
+ return self.clean_text(json.dumps(cleaned_data, ensure_ascii=False))
|
|
|
+
|
|
|
+ return self.clean_text(str(data))
|
|
|
+
|
|
|
+ def _load_index(self):
|
|
|
+ """加载或初始化索引"""
|
|
|
+ if os.path.exists(self.index_path) and os.path.exists(self.metadata_path):
|
|
|
+ try:
|
|
|
+ self.index = faiss.read_index(self.index_path)
|
|
|
+ # 检查索引维度是否匹配，如果不匹配则重新初始化（针对权重升级）
|
|
|
+ if self.index.d != self.dimension:
|
|
|
+ print(f"⚠️ 索引维度不匹配 ({self.index.d} != {self.dimension})，正在重新初始化...")
|
|
|
+ self._init_new_index()
|
|
|
+ else:
|
|
|
+ with open(self.metadata_path, 'rb') as f:
|
|
|
+ self.metadata = pickle.load(f)
|
|
|
+ print(f"✓ 已加载现有索引，包含 {len(self.metadata)} 道题目")
|
|
|
+ except Exception as e:
|
|
|
+ print(f"⚠️ 加载索引失败: {e}，将初始化新索引")
|
|
|
+ self._init_new_index()
|
|
|
+ else:
|
|
|
+ self._init_new_index()
|
|
|
+
|
|
|
+ def _init_new_index(self):
|
|
|
+ """初始化新的FAISS索引"""
|
|
|
+ # 使用内积索引，配合权重拼接向量
|
|
|
+ self.index = faiss.IndexFlatIP(self.dimension)
|
|
|
+ self.metadata = []
|
|
|
+ print("✓ 已初始化新的FAISS索引")
|
|
|
+
|
|
|
+ def save_index(self):
|
|
|
+ """保存索引和元数据"""
|
|
|
+ faiss.write_index(self.index, self.index_path)
|
|
|
+ with open(self.metadata_path, 'wb') as f:
|
|
|
+ pickle.dump(self.metadata, f)
|
|
|
+ print(f"✓ 索引和元数据已保存到 {self.index_path}")
|
|
|
+
|
|
|
+ def get_weighted_embedding(self, question: Dict) -> np.ndarray:
|
|
|
+ """
|
|
|
+ 获取加权拼接向量
|
|
|
+ 计算 4 个部分的 embedding，并乘以各自权重的平方根后拼接
|
|
|
+ """
|
|
|
+ parts = {
|
|
|
+ 'stem': self.clean_text(question.get('stem', '') or ''),
|
|
|
+ 'options': self.normalize_options(question.get('options', '') or ''),
|
|
|
+ 'answer': self.clean_text(question.get('answer', '') or ''),
|
|
|
+ 'solution': self.clean_text(question.get('solution', '') or '')
|
|
|
+ }
|
|
|
+
|
|
|
+ embeddings = []
|
|
|
+ for key, text in parts.items():
|
|
|
+ # 获取单部分 embedding
|
|
|
+ emb = self.get_embedding(text if text.strip() else "无") # 防止空文本
|
|
|
+ if emb is None:
|
|
|
+ return None
|
|
|
+
|
|
|
+ # 归一化该部分的向量
|
|
|
+ norm = np.linalg.norm(emb)
|
|
|
+ if norm > 0:
|
|
|
+ emb = emb / norm
|
|
|
+
|
|
|
+ # 乘以权重平方根：这样点积结果就是权重加权和
|
|
|
+ weight_factor = np.sqrt(self.weights[key])
|
|
|
+ embeddings.append(emb * weight_factor)
|
|
|
+
|
|
|
+ # 拼接成一个大向量
|
|
|
+ mega_vector = np.concatenate(embeddings).astype('float32')
|
|
|
+ return mega_vector
|
|
|
+
|
|
|
+ def get_embedding(self, text: str) -> np.ndarray:
|
|
|
+ """获取文本的embedding向量"""
|
|
|
+ try:
|
|
|
+ # 统一清理文本
|
|
|
+ cleaned_text = self.clean_text(text)
|
|
|
+ if not cleaned_text.strip():
|
|
|
+ cleaned_text = "无"
|
|
|
+
|
|
|
+ response = self.client.embeddings.create(
|
|
|
+ model="text-embedding-3-large",
|
|
|
+ input=cleaned_text
|
|
|
+ )
|
|
|
+ embedding = np.array(response.data[0].embedding).astype('float32')
|
|
|
+ return embedding
|
|
|
+ except Exception as e:
|
|
|
+ print(f"❌ 获取embedding失败: {e}")
|
|
|
+ return None
|
|
|
+
|
|
|
+ def get_embeddings_batch(self, texts: List[str]) -> List[np.ndarray]:
|
|
|
+ """批量获取向量化结果"""
|
|
|
+ try:
|
|
|
+ if not texts:
|
|
|
+ return []
|
|
|
+
|
|
|
+ # 统一清理文本
|
|
|
+ cleaned_texts = []
|
|
|
+ for t in texts:
|
|
|
+ c = self.clean_text(t)
|
|
|
+ cleaned_texts.append(c if c.strip() else "无")
|
|
|
+
|
|
|
+ response = self.client.embeddings.create(
|
|
|
+ model="text-embedding-3-large",
|
|
|
+ input=cleaned_texts
|
|
|
+ )
|
|
|
+ return [np.array(data.embedding).astype('float32') for data in response.data]
|
|
|
+ except Exception as e:
|
|
|
+ print(f"❌ 批量获取 embedding 失败: {e}")
|
|
|
+ return []
|
|
|
+
|
|
|
+ def get_weighted_embeddings_batch(self, questions: List[Dict]) -> List[np.ndarray]:
|
|
|
+ """
|
|
|
+ 批量获取加权拼接向量
|
|
|
+ """
|
|
|
+ all_texts = []
|
|
|
+ for q in questions:
|
|
|
+ # 获取 4 个部分并处理空值占位
|
|
|
+ stem = self.clean_text(q.get('stem', '') or '')
|
|
|
+ opts = self.normalize_options(q.get('options', '') or '')
|
|
|
+ ans = self.clean_text(q.get('answer', '') or '')
|
|
|
+ sol = self.clean_text(q.get('solution', '') or '')
|
|
|
+
|
|
|
+ # OpenAI API 不接受空字符串，强制填充 "无"
|
|
|
+ all_texts.append(stem if stem.strip() else "无")
|
|
|
+ all_texts.append(opts if opts.strip() else "无")
|
|
|
+ all_texts.append(ans if ans.strip() else "无")
|
|
|
+ all_texts.append(sol if sol.strip() else "无")
|
|
|
+
|
|
|
+ # 批量请求 (每题 4 个部分)
|
|
|
+ all_embeddings = self.get_embeddings_batch(all_texts)
|
|
|
+ if not all_embeddings:
|
|
|
+ return []
|
|
|
+
|
|
|
+ results = []
|
|
|
+ for i in range(len(questions)):
|
|
|
+ q_embs = all_embeddings[i*4 : (i+1)*4]
|
|
|
+ if len(q_embs) < 4:
|
|
|
+ results.append(None)
|
|
|
+ continue
|
|
|
+
|
|
|
+ weighted_parts = []
|
|
|
+ keys = ['stem', 'options', 'answer', 'solution']
|
|
|
+ for j, emb in enumerate(q_embs):
|
|
|
+ # 归一化
|
|
|
+ norm = np.linalg.norm(emb)
|
|
|
+ if norm > 0:
|
|
|
+ emb = emb / norm
|
|
|
+ # 加权
|
|
|
+ weight_factor = np.sqrt(self.weights[keys[j]])
|
|
|
+ weighted_parts.append(emb * weight_factor)
|
|
|
+
|
|
|
+ mega_vector = np.concatenate(weighted_parts).astype('float32')
|
|
|
+ results.append(mega_vector)
|
|
|
+
|
|
|
+ return results
|
|
|
+
|
|
|
+ def combine_content(self, q: Dict) -> str:
|
|
|
+ """组合题目内容用于预览显示"""
|
|
|
+ stem = q.get('stem', '') or ''
|
|
|
+ return self.clean_text(stem) # 预览以清理后的题干为主
|
|
|
+
|
|
|
+ def fetch_question_from_db(self, question_id: int) -> Optional[Dict]:
|
|
|
+ """从数据库获取题目信息"""
|
|
|
+ try:
|
|
|
+ conn = pymysql.connect(
|
|
|
+ host=DB_HOST,
|
|
|
+ port=DB_PORT,
|
|
|
+ user=DB_USERNAME,
|
|
|
+ password=DB_PASSWORD,
|
|
|
+ database=DB_DATABASE,
|
|
|
+ charset='utf8mb4',
|
|
|
+ cursorclass=pymysql.cursors.DictCursor
|
|
|
+ )
|
|
|
+ with conn.cursor() as cursor:
|
|
|
+ sql = "SELECT id, stem, options, answer, solution, is_repeat FROM questions_tem WHERE id = %s"
|
|
|
+ cursor.execute(sql, (question_id,))
|
|
|
+ result = cursor.fetchone()
|
|
|
+ return result
|
|
|
+ except Exception as e:
|
|
|
+ print(f"❌ 数据库查询失败: {e}")
|
|
|
+ return None
|
|
|
+ finally:
|
|
|
+ if 'conn' in locals() and conn:
|
|
|
+ conn.close()
|
|
|
+
|
|
|
+ def confirm_repeat(self, question_id: int, is_repeat: int) -> bool:
|
|
|
+ """
|
|
|
+ 人工确认结果后更新数据库和向量库
|
|
|
+ is_repeat: 0 代表无相似题，1 代表有重复
|
|
|
+ """
|
|
|
+ # 1. 更新数据库状态
|
|
|
+ self.update_is_repeat(question_id, is_repeat)
|
|
|
+
|
|
|
+ # 2. 如果标记为非重复，将其加入向量库
|
|
|
+ if is_repeat == 0:
|
|
|
+ question = self.fetch_question_from_db(question_id)
|
|
|
+ if question:
|
|
|
+ mega_vec = self.get_weighted_embedding(question)
|
|
|
+ if mega_vec is not None:
|
|
|
+ # 转换维度以便 FAISS 接收
|
|
|
+ mega_vec = mega_vec.reshape(1, -1)
|
|
|
+ self.add_to_index(question_id, mega_vec, self.clean_text(question.get('stem', ''))[:100])
|
|
|
+ return True
|
|
|
+ return True
|
|
|
+
|
|
|
+ def update_is_repeat(self, question_id: int, is_repeat: int):
|
|
|
+ """更新数据库中的 is_repeat 字段"""
|
|
|
+ try:
|
|
|
+ conn = pymysql.connect(
|
|
|
+ host=DB_HOST,
|
|
|
+ port=DB_PORT,
|
|
|
+ user=DB_USERNAME,
|
|
|
+ password=DB_PASSWORD,
|
|
|
+ database=DB_DATABASE,
|
|
|
+ charset='utf8mb4'
|
|
|
+ )
|
|
|
+ with conn.cursor() as cursor:
|
|
|
+ sql = "UPDATE questions_tem SET is_repeat = %s WHERE id = %s"
|
|
|
+ cursor.execute(sql, (is_repeat, question_id))
|
|
|
+ conn.commit()
|
|
|
+ print(f"✓ 题目 ID {question_id} 的 is_repeat 已更新为 {is_repeat}")
|
|
|
+ except Exception as e:
|
|
|
+ print(f"❌ 更新 is_repeat 失败: {e}")
|
|
|
+ finally:
|
|
|
+ if 'conn' in locals() and conn:
|
|
|
+ conn.close()
|
|
|
+
|
|
|
+ def call_openai_for_duplicate_check(self, q1: Dict, q2: Dict) -> bool:
|
|
|
+ """
|
|
|
+ 调用 GPT-4o 进行深度对比，判断两道题是否为重复题
|
|
|
+ """
|
|
|
+ try:
|
|
|
+ prompt = f"""你是一个专业的数学题目查重专家。请对比以下两道题目，判断它们是否为“重复题”。
|
|
|
+“重复题”的定义：题目背景、考查知识点、核心逻辑以及数值完全一致，或者仅有极其微小的表述差异但不影响题目本质。
|
|
|
+
|
|
|
+题目 1:
|
|
|
+题干: {q1.get('stem', '')}
|
|
|
+选项: {q1.get('options', '')}
|
|
|
+答案: {q1.get('answer', '')}
|
|
|
+解析: {q1.get('solution', '')}
|
|
|
+
|
|
|
+题目 2:
|
|
|
+题干: {q2.get('stem', '')}
|
|
|
+选项: {q2.get('options', '')}
|
|
|
+答案: {q2.get('answer', '')}
|
|
|
+解析: {q2.get('solution', '')}
|
|
|
+
|
|
|
+请分析两道题的相似性，并给出最终结论。
|
|
|
+结论必须以 JSON 格式输出，包含以下字段：
|
|
|
+- is_duplicate: 布尔值，True 表示是重复题，False 表示不是
|
|
|
+- reason: 简短的分析理由
|
|
|
+
|
|
|
+只输出 JSON 即可，不要有其他解释说明。"""
|
|
|
+
|
|
|
+ response = self.client.chat.completions.create(
|
|
|
+ model="gpt-4o",
|
|
|
+ messages=[
|
|
|
+ {"role": "system", "content": "你是一个专业的数学题目查重助手。"},
|
|
|
+ {"role": "user", "content": prompt}
|
|
|
+ ],
|
|
|
+ response_format={"type": "json_object"}
|
|
|
+ )
|
|
|
+
|
|
|
+ result = json.loads(response.choices[0].message.content)
|
|
|
+ is_dup = result.get('is_duplicate', False)
|
|
|
+ reason = result.get('reason', '')
|
|
|
+ print(f"🤖 GPT-4o 判定结果: {'重复' if is_dup else '不重复'} | 理由: {reason}")
|
|
|
+ return is_dup
|
|
|
+ except Exception as e:
|
|
|
+ print(f"❌ GPT-4o 兜底计算失败: {e}")
|
|
|
+ return False
|
|
|
+
|
|
|
+ def check_duplicate(self, question_id: int, threshold: float = 0.85) -> Dict:
|
|
|
+ """
|
|
|
+ 查重主逻辑 (未入库模式：不更新数据库，不自动入库)
|
|
|
+ """
|
|
|
+ # 1. 获取题目信息
|
|
|
+ question = self.fetch_question_from_db(question_id)
|
|
|
+ if not question:
|
|
|
+ return {"status": "error", "message": f"未找到ID为 {question_id} 的题目"}
|
|
|
+
|
|
|
+ # 2. 获取加权拼接向量
|
|
|
+ mega_vector = self.get_weighted_embedding(question)
|
|
|
+ if mega_vector is None:
|
|
|
+ return {"status": "error", "message": "无法生成题目向量"}
|
|
|
+
|
|
|
+ # 3. 搜索索引
|
|
|
+ if self.index.ntotal == 0:
|
|
|
+ return {
|
|
|
+ "status": "success",
|
|
|
+ "message": "向量库为空",
|
|
|
+ "is_duplicate": False,
|
|
|
+ "top_similar": []
|
|
|
+ }
|
|
|
+
|
|
|
+ # 搜索最相似的题目
|
|
|
+ k = min(3, self.index.ntotal)
|
|
|
+ query_vec = mega_vector.reshape(1, -1)
|
|
|
+ similarities, indices = self.index.search(query_vec, k)
|
|
|
+
|
|
|
+ best_similar = None
|
|
|
+ is_duplicate = False
|
|
|
+ gpt_checked = False
|
|
|
+
|
|
|
+ for i in range(k):
|
|
|
+ idx = indices[0][i]
|
|
|
+ score = float(similarities[0][i])
|
|
|
+ if idx != -1:
|
|
|
+ similar_info = self.metadata[idx]
|
|
|
+ similar_id = similar_info['id']
|
|
|
+
|
|
|
+ if similar_id == question_id:
|
|
|
+ continue
|
|
|
+
|
|
|
+ best_similar = {
|
|
|
+ "id": similar_id,
|
|
|
+ "similarity": round(score, 4),
|
|
|
+ "similar_point": similar_info.get('text_preview', '')
|
|
|
+ }
|
|
|
+
|
|
|
+ if score >= threshold:
|
|
|
+ is_duplicate = True
|
|
|
+ elif score > 0.5:
|
|
|
+ # 相似度在 0.5 ~ 0.85 之间，调用 GPT-4o 兜底
|
|
|
+ similar_question = self.fetch_question_from_db(similar_id)
|
|
|
+ if similar_question:
|
|
|
+ is_duplicate = self.call_openai_for_duplicate_check(question, similar_question)
|
|
|
+ gpt_checked = True
|
|
|
+ break
|
|
|
+
|
|
|
+ if is_duplicate:
|
|
|
+ return {
|
|
|
+ "status": "warning",
|
|
|
+ "message": "该题目与已存在题目相似，请人工核验" + (" (经 GPT-4o 兜底确认)" if gpt_checked else ""),
|
|
|
+ "is_duplicate": True,
|
|
|
+ "gpt_checked": gpt_checked,
|
|
|
+ "top_similar": [best_similar]
|
|
|
+ }
|
|
|
+ else:
|
|
|
+ return {
|
|
|
+ "status": "success",
|
|
|
+ "message": "该题目通过查重",
|
|
|
+ "is_duplicate": False,
|
|
|
+ "top_similar": [best_similar] if best_similar else []
|
|
|
+ }
|
|
|
+
|
|
|
+ def check_duplicate_by_content(self, question_data: Dict, threshold: float = 0.85) -> Dict:
|
|
|
+ """
|
|
|
+ 基于原始文本内容进行查重 (预检模式)
|
|
|
+ """
|
|
|
+ # 1. 获取加权拼接向量
|
|
|
+ mega_vector = self.get_weighted_embedding(question_data)
|
|
|
+ if mega_vector is None:
|
|
|
+ return {"status": "error", "message": "无法生成题目向量"}
|
|
|
+
|
|
|
+ # 2. 检索
|
|
|
+ if self.index.ntotal == 0:
|
|
|
+ return {"status": "success", "is_duplicate": False, "top_similar": []}
|
|
|
+
|
|
|
+ k = 1 # 预检通常只需返回最相似的一个
|
|
|
+ query_vec = mega_vector.reshape(1, -1)
|
|
|
+ similarities, indices = self.index.search(query_vec, k)
|
|
|
+
|
|
|
+ if indices[0][0] != -1:
|
|
|
+ score = float(similarities[0][0])
|
|
|
+ similar_info = self.metadata[indices[0][0]]
|
|
|
+ similar_id = similar_info['id']
|
|
|
+ best_similar = {
|
|
|
+ "id": similar_id,
|
|
|
+ "similarity": round(score, 4),
|
|
|
+ "similar_point": similar_info.get('text_preview', '')
|
|
|
+ }
|
|
|
+
|
|
|
+ is_duplicate = False
|
|
|
+ gpt_checked = False
|
|
|
+
|
|
|
+ if score >= threshold:
|
|
|
+ is_duplicate = True
|
|
|
+ elif score > 0.5:
|
|
|
+ # 相似度在 0.5 ~ 0.85 之间，调用 GPT-4o 兜底
|
|
|
+ similar_question = self.fetch_question_from_db(similar_id)
|
|
|
+ if similar_question:
|
|
|
+ is_duplicate = self.call_openai_for_duplicate_check(question_data, similar_question)
|
|
|
+ gpt_checked = True
|
|
|
+
|
|
|
+ if is_duplicate:
|
|
|
+ return {
|
|
|
+ "status": "warning",
|
|
|
+ "is_duplicate": True,
|
|
|
+ "gpt_checked": gpt_checked,
|
|
|
+ "top_similar": [best_similar]
|
|
|
+ }
|
|
|
+ return {"status": "success", "is_duplicate": False, "top_similar": [best_similar]}
|
|
|
+
|
|
|
+ return {"status": "success", "is_duplicate": False, "top_similar": []}
|
|
|
+
|
|
|
+ def add_to_index(self, question_id: int, vector: np.ndarray, text: str):
|
|
|
+ """将题目加入索引"""
|
|
|
+ if any(m['id'] == question_id for m in self.metadata):
|
|
|
+ return
|
|
|
+
|
|
|
+ self.index.add(vector)
|
|
|
+ self.metadata.append({
|
|
|
+ 'id': question_id,
|
|
|
+ 'text_preview': self.clean_text(text)[:100]
|
|
|
+ })
|
|
|
+ self.save_index()
|
|
|
+
|
|
|
+ def sync_all_from_db(self, batch_size=50, max_workers=5):
|
|
|
+ """同步数据库中所有题目到索引 (支持加权模式 + 批量 + 多线程)"""
|
|
|
+ print("正在进行全量同步 (优化版 - 加权模式)...")
|
|
|
+ existing_ids = {m['id'] for m in self.metadata}
|
|
|
+ try:
|
|
|
+ conn = pymysql.connect(
|
|
|
+ host=DB_HOST,
|
|
|
+ port=DB_PORT,
|
|
|
+ user=DB_USERNAME,
|
|
|
+ password=DB_PASSWORD,
|
|
|
+ database=DB_DATABASE,
|
|
|
+ charset='utf8mb4',
|
|
|
+ cursorclass=pymysql.cursors.DictCursor
|
|
|
+ )
|
|
|
+ with conn.cursor() as cursor:
|
|
|
+ sql = "SELECT id, stem, options, answer, solution FROM questions_tem"
|
|
|
+ cursor.execute(sql)
|
|
|
+ all_questions = cursor.fetchall()
|
|
|
+
|
|
|
+ new_questions = [q for q in all_questions if q['id'] not in existing_ids]
|
|
|
+ total_new = len(new_questions)
|
|
|
+
|
|
|
+ if total_new == 0:
|
|
|
+ print("✅ 已经是最新状态，无需同步。")
|
|
|
+ return
|
|
|
+
|
|
|
+ print(f"📊 数据库总计: {len(all_questions)}, 需同步新增: {total_new}")
|
|
|
+
|
|
|
+ # 分块处理
|
|
|
+ chunks = [new_questions[i:i + batch_size] for i in range(0, total_new, batch_size)]
|
|
|
+
|
|
|
+ def process_chunk(chunk):
|
|
|
+ mega_vectors = self.get_weighted_embeddings_batch(chunk)
|
|
|
+ return chunk, mega_vectors
|
|
|
+
|
|
|
+ # 使用线程池并发
|
|
|
+ with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
|
|
|
+ future_to_chunk = {executor.submit(process_chunk, chunk): chunk for chunk in chunks}
|
|
|
+
|
|
|
+ count = 0
|
|
|
+ for future in concurrent.futures.as_completed(future_to_chunk):
|
|
|
+ chunk, mega_vectors = future.result()
|
|
|
+ if mega_vectors:
|
|
|
+ for i, mega_vec in enumerate(mega_vectors):
|
|
|
+ if mega_vec is not None:
|
|
|
+ self.index.add(mega_vec.reshape(1, -1))
|
|
|
+ self.metadata.append({
|
|
|
+ 'id': chunk[i]['id'],
|
|
|
+ 'text_preview': self.clean_text(chunk[i].get('stem', ''))[:100]
|
|
|
+ })
|
|
|
+
|
|
|
+ count += len(chunk)
|
|
|
+ print(f"✅ 已同步进度: {count}/{total_new}")
|
|
|
+
|
|
|
+ # 每 500 条保存一次
|
|
|
+ if count % 500 == 0:
|
|
|
+ self.save_index()
|
|
|
+
|
|
|
+ self.save_index()
|
|
|
+ print(f"🎉 同步完成！当前索引总数: {len(self.metadata)}")
|
|
|
+ except Exception as e:
|
|
|
+ print(f"❌ 同步失败: {e}")
|
|
|
+ import traceback
|
|
|
+ traceback.print_exc()
|
|
|
+ finally:
|
|
|
+ if 'conn' in locals() and conn:
|
|
|
+ conn.close()
|
|
|
+
|
|
|
+if __name__ == "__main__":
|
|
|
+ # 测试代码
|
|
|
+ checker = QuestionDuplicateChecker()
|
|
|
+ # checker.sync_all_from_db() # 首次运行时同步
|
|
|
+ # result = checker.check_duplicate(10)
|
|
|
+ # print(json.dumps(result, ensure_ascii=False, indent=2))
|
Hai Lin
3 周之前