RAG数据准备完整实战指南：从原始文档到高质量知识库

pip install presidio-analyzer presidio-anonymizer spacypython -m spacy download zh_core_web_sm

# sensitive_scanner.pyfrom presidio_analyzer import AnalyzerEnginefrom presidio_anonymizer import AnonymizerEngineimport reimport jsonfrom typing import List, Dictclass SensitiveDataScanner:    def __init__(self):        self.analyzer = AnalyzerEngine()        self.anonymizer = AnonymizerEngine()                # 自定义中文敏感信息识别规则        self.custom_patterns = {            'CHINA_ID': r'd{15}|d{18}',            'CHINA_PHONE': r'1[3-9]d{9}',            'BANK_CARD': r'd{16,19}',            'EMAIL': r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+.[a-zA-Z]{2,}'        }        def scan_document(self, text: str, language: str = "zh") -> List[Dict]:        """扫描文档中的敏感信息"""        results = []                # 使用Presidio进行基础扫描        presidio_results = self.analyzer.analyze(text=text, language=language)                for result in presidio_results:            results.append({                'type': result.entity_type,                'start': result.start,                'end': result.end,                'confidence': result.score,                'content': text[result.start:result.end]            })                # 补充自定义规则扫描        for pattern_name, pattern in self.custom_patterns.items():            matches = re.finditer(pattern, text)            for match in matches:                results.append({                    'type': pattern_name,                    'start': match.start(),                    'end': match.end(),                    'confidence': 0.9,                    'content': match.group()                })                return results        def generate_report(self, scan_results: List[Dict]) -> Dict:        """生成扫描报告"""        report = {            'total_sensitive_items': len(scan_results),            'types_distribution': {},            'high_risk_items': []        }                for item in scan_results:            # 统计敏感信息类型分布            item_type = item['type']            if item_type not in report['types_distribution']:                report['types_distribution'][item_type] = 0            report['types_distribution'][item_type] += 1                        # 标记高风险项            if item['confidence'] > 0.8:                report['high_risk_items'].append(item)                return report# 使用示例scanner = SensitiveDataScanner()test_text = "张三的电话是13800138000，身份证号码是110101199001011234，银行卡号是6222021234567890123"scan_results = scanner.scan_document(test_text)report = scanner.generate_report(scan_results)print("发现敏感信息：")for item in scan_results:    print(f"类型：{item['type']}, 内容：{item['content']}, 置信度：{item['confidence']}")

# data_freshness_checker.pyfrom datetime import datetime, timedeltaimport pandas as pdfrom typing import Dict, Listimport sqlite3class DataFreshnessChecker:    def __init__(self, db_path: str = "data_tracking.db"):        self.db_path = db_path        self.init_database()            def init_database(self):        """初始化数据跟踪数据库"""        conn = sqlite3.connect(self.db_path)        cursor = conn.cursor()                cursor.execute('''            CREATE TABLE IF NOT EXISTS document_tracking (                document_id TEXT PRIMARY KEY,                file_path TEXT,                last_modified DATE,                data_type TEXT,                expiry_threshold_days INTEGER,                status TEXT DEFAULT 'active'            )        ''')                conn.commit()        conn.close()        def register_document(self, doc_id: str, file_path: str,                          data_type: str, threshold_days: int = 365):        """注册文档到跟踪系统"""        conn = sqlite3.connect(self.db_path)        cursor = conn.cursor()                cursor.execute('''            INSERT OR REPLACE INTO document_tracking             (document_id, file_path, last_modified, data_type, expiry_threshold_days)            VALUES (?, ?, ?, ?, ?)        ''', (doc_id, file_path, datetime.now().date(), data_type, threshold_days))                conn.commit()        conn.close()        def check_outdated_data(self) -> List[Dict]:        """检查过时数据"""        conn = sqlite3.connect(self.db_path)        cursor = conn.cursor()                cursor.execute('''            SELECT document_id, file_path, last_modified, data_type, expiry_threshold_days            FROM document_tracking            WHERE status = 'active'        ''')                results = cursor.fetchall()        conn.close()                outdated_items = []        current_date = datetime.now().date()                for row in results:            doc_id, file_path, last_modified, data_type, threshold = row            last_mod_date = datetime.strptime(last_modified, '%Y-%m-%d').date()            days_since_update = (current_date - last_mod_date).days                        if days_since_update > threshold:                outdated_items.append({                    'document_id': doc_id,                    'file_path': file_path,                    'days_outdated': days_since_update,                    'data_type': data_type,                    'risk_level': self._calculate_risk_level(days_since_update, threshold)                })                return outdated_items        def _calculate_risk_level(self, days_outdated: int, threshold: int) -> str:        """计算风险等级"""        ratio = days_outdated / threshold        if ratio < 1.2:            return "低风险"        elif ratio < 2.0:            return "中风险"        else:            return "高风险"# 使用示例checker = DataFreshnessChecker()# 注册不同类型的文档checker.register_document("fin_001", "financial_report_2023.pdf", "财务数据", 90)checker.register_document("hr_001", "employee_handbook.pdf", "人事制度", 730)checker.register_document("cust_001", "customer_database.csv", "客户信息", 180)# 检查过时数据outdated_data = checker.check_outdated_data()for item in outdated_data:    print(f"文档：{item['document_id']}, 超期{item['days_outdated']}天, 风险等级：{item['risk_level']}")

# data_normalizer.pyimport reimport pandas as pdfrom typing import Dict, Listimport jiebafrom collections import defaultdictclass DataNormalizer:    def __init__(self):        self.address_mapping = self._load_address_mapping()        self.amount_patterns = [            (r'(d+(?:.d+)?)万元', lambda x: str(float(x.group(1)) * 10000) + '元'),            (r'(d+(?:.d+)?)千元', lambda x: str(float(x.group(1)) * 1000) + '元'),            (r'(d+(?:.d+)?)亿元', lambda x: str(float(x.group(1)) * 100000000) + '元'),        ]            def _load_address_mapping(self) -> Dict[str, str]:        """加载地址标准化映射表"""        return {            '北京海淀区': '北京市海淀区',            '上海浦东': '上海市浦东新区',            '深圳南山': '深圳市南山区',            # 可以从外部配置文件加载更多映射规则        }        def normalize_address(self, text: str) -> str:        """地址归一化"""        normalized_text = text        for source, target in self.address_mapping.items():            normalized_text = re.sub(source, target, normalized_text)                # 补全省市区层级        patterns = [            (r'([^省市区县]+)区', r'1区'),  # 保持区不变            (r'([^省市区县]+)市([^省市区县]+)区', r'1市2区'),  # 市区结构        ]                for pattern, replacement in patterns:            normalized_text = re.sub(pattern, replacement, normalized_text)                    return normalized_text        def normalize_amount(self, text: str) -> str:        """金额标准化"""        normalized_text = text        for pattern, converter in self.amount_patterns:            normalized_text = re.sub(pattern, converter, normalized_text)        return normalized_text        def normalize_company_names(self, text: str) -> str:        """公司名称标准化"""        # 移除常见的公司后缀变体        suffixes = ['有限公司', '股份有限公司', 'Limited', 'Ltd', 'Co.,Ltd']        normalized = text                for suffix in suffixes:            if normalized.endswith(suffix):                normalized = normalized[:-len(suffix)].strip()                break                # 统一添加标准后缀        if not any(normalized.endswith(s) for s in suffixes):            normalized += '有限公司'                    return normalized        def batch_normalize(self, texts: List[str]) -> List[str]:        """批量标准化处理"""        results = []        for text in texts:            normalized = text            normalized = self.normalize_address(normalized)            normalized = self.normalize_amount(normalized)            normalized = self.normalize_company_names(normalized)            results.append(normalized)        return results# 使用示例normalizer = DataNormalizer()test_texts = [    "在北京海淀区的某科技有限公司投资了100万元",    "上海浦东新区张江高科技园区的企业年收入达到5亿元",    "深圳南山区腾讯大厦的租金是每月50万元"]normalized_texts = normalizer.batch_normalize(test_texts)for original, normalized in zip(test_texts, normalized_texts):    print(f"原文：{original}")    print(f"标准化：{normalized}n")

# advanced_pdf_parser.pyimport PyMuPDF as fitzimport pandas as pdimport camelotfrom PIL import Imageimport pytesseractimport iofrom typing import Dict, List, Tupleimport jsonclass AdvancedPDFParser:    def __init__(self):        self.supported_strategies = ['text_extraction', 'table_extraction', 'ocr_fallback']            def parse_pdf(self, pdf_path: str, strategy: str = 'auto') -> Dict:        """解析PDF文档"""        doc = fitz.open(pdf_path)                if strategy == 'auto':            strategy = self._determine_best_strategy(doc)                result = {            'metadata': self._extract_metadata(doc),            'pages': [],            'tables': [],            'images': []        }                for page_num in range(len(doc)):            page = doc[page_num]            page_data = self._parse_page(page, page_num, strategy)            result['pages'].append(page_data)                # 提取表格        if strategy in ['table_extraction', 'auto']:            result['tables'] = self._extract_tables(pdf_path)                doc.close()        return result        def _determine_best_strategy(self, doc) -> str:        """自动确定最佳解析策略"""        total_text_length = 0        total_images = 0                for page in doc:            text = page.get_text()            total_text_length += len(text)            total_images += len(page.get_images())                # 如果文本很少但图片很多，可能是扫描件        if total_text_length < 1000 and total_images > 5:            return 'ocr_fallback'                # 检查是否包含大量表格        sample_text = doc[0].get_text() if len(doc) > 0 else ""        if self._contains_tables(sample_text):            return 'table_extraction'                return 'text_extraction'        def _parse_page(self, page, page_num: int, strategy: str) -> Dict:        """解析单个页面"""        page_data = {            'page_number': page_num,            'text': '',            'blocks': [],            'images': []        }                if strategy == 'ocr_fallback':            # OCR解析            pix = page.get_pixmap()            img_data = pix.tobytes("png")            img = Image.open(io.BytesIO(img_data))            page_data['text'] = pytesseract.image_to_string(img, lang='chi_sim+eng')        else:            # 标准文本提取            page_data['text'] = page.get_text()                        # 提取文档结构            blocks = page.get_text("dict")["blocks"]            for block in blocks:                if "lines" in block:                    block_info = {                        'bbox': block['bbox'],                        'text': '',                        'font_size': 0,                        'font_name': ''                    }                                        for line in block["lines"]:                        for span in line["spans"]:                            block_info['text'] += span['text']                            block_info['font_size'] = span['size']                            block_info['font_name'] = span['font']                                        page_data['blocks'].append(block_info)                return page_data        def _extract_tables(self, pdf_path: str) -> List[Dict]:        """提取表格数据"""        try:            tables = camelot.read_pdf(pdf_path, pages='all')            table_data = []                        for i, table in enumerate(tables):                table_info = {                    'table_id': i,                    'page': table.page,                    'accuracy': table.accuracy,                    'data': table.df.to_dict('records'),                    'shape': table.df.shape                }                table_data.append(table_info)                        return table_data        except Exception as e:            print(f"表格提取失败: {e}")            return []        def _extract_metadata(self, doc) -> Dict:        """提取文档元数据"""        metadata = doc.metadata        return {            'title': metadata.get('title', ''),            'author': metadata.get('author', ''),            'creator': metadata.get('creator', ''),            'creation_date': metadata.get('creationDate', ''),            'modification_date': metadata.get('modDate', ''),            'pages_count': len(doc)        }        def _contains_tables(self, text: str) -> bool:        """检测文本是否包含表格"""        table_indicators = ['表格', '序号', '项目', '金额', '数量', '单价']        line_count = len(text.split('n'))                # 如果文本行数多且包含表格关键词        if line_count > 20:            for indicator in table_indicators:                if indicator in text:                    return True        return False# 使用示例parser = AdvancedPDFParser()pdf_result = parser.parse_pdf("sample_document.pdf")print(f"文档包含 {pdf_result['metadata']['pages_count']} 页")print(f"提取到 {len(pdf_result['tables'])} 个表格")print(f"第一页文本预览：{pdf_result['pages'][0]['text'][:200]}...")

# data_anonymizer.pyfrom faker import Fakerimport reimport jsonfrom typing import Dict, List, Callableimport hashlibclass IntelligentAnonymizer:    def __init__(self, locale: str = 'zh_CN'):        self.fake = Faker(locale)        self.anonymization_rules = self._init_rules()        self.consistent_mapping = {}  # 保持一致性的映射表            def _init_rules(self) -> Dict[str, Callable]:        """初始化脱敏规则"""        return {            'CHINA_PHONE': self._anonymize_phone,            'CHINA_ID': self._anonymize_id_card,            'EMAIL': self._anonymize_email,            'PERSON_NAME': self._anonymize_name,            'BANK_CARD': self._anonymize_bank_card,            'ADDRESS': self._anonymize_address        }        def _generate_consistent_fake(self, original_value: str, fake_generator: Callable) -> str:        """生成一致性的虚假数据"""        # 使用原值的哈希作为种子，确保同一原值总是生成相同的虚假值        hash_seed = int(hashlib.md5(original_value.encode()).hexdigest()[:8], 16)                if original_value not in self.consistent_mapping:            # 临时设置种子            original_seed = self.fake.seed_instance(hash_seed)            fake_value = fake_generator()            self.fake.seed_instance(original_seed)  # 恢复原种子                        self.consistent_mapping[original_value] = fake_value                return self.consistent_mapping[original_value]        def _anonymize_phone(self, phone: str) -> str:        """脱敏手机号"""        return self._generate_consistent_fake(phone, self.fake.phone_number)        def _anonymize_id_card(self, id_card: str) -> str:        """脱敏身份证号"""        def generate_fake_id():            # 生成符合校验规则的假身份证号            area_code = "110101"  # 北京东城区            birth_date = self.fake.date_of_birth(minimum_age=18, maximum_age=80)            birth_str = birth_date.strftime("%Y%m%d")            sequence = f"{self.fake.random_int(min=100, max=999)}"                        # 简化的校验码计算            check_code = str(self.fake.random_int(min=0, max=9))            return area_code + birth_str + sequence + check_code                return self._generate_consistent_fake(id_card, generate_fake_id)        def _anonymize_email(self, email: str) -> str:        """脱敏邮箱"""        return self._generate_consistent_fake(email, self.fake.email)        def _anonymize_name(self, name: str) -> str:        """脱敏姓名"""        return self._generate_consistent_fake(name, self.fake.name)        def _anonymize_bank_card(self, card_number: str) -> str:        """脱敏银行卡号"""        def generate_fake_card():            # 生成16位假银行卡号            return ''.join([str(self.fake.random_int(min=0, max=9)) for _ in range(16)])                return self._generate_consistent_fake(card_number, generate_fake_card)        def _anonymize_address(self, address: str) -> str:        """脱敏地址"""        return self._generate_consistent_fake(address, self.fake.address)        def anonymize_text(self, text: str, sensitive_items: List[Dict]) -> str:        """对文本进行脱敏处理"""        anonymized_text = text                # 按位置倒序排序，避免替换后位置偏移        sorted_items = sorted(sensitive_items, key=lambda x: x['start'], reverse=True)                for item in sorted_items:            original_content = item['content']            item_type = item['type']                        if item_type in self.anonymization_rules:                fake_content = self.anonymization_rules[item_type](original_content)                anonymized_text = (anonymized_text[:item['start']] +                                  fake_content +                                  anonymized_text[item['end']:])                return anonymized_text        def create_anonymization_report(self, original_items: List[Dict]) -> Dict:        """创建脱敏处理报告"""        report = {            'total_items_processed': len(original_items),            'types_processed': {},            'consistency_maintained': True,            'processing_timestamp': self.fake.date_time().isoformat()        }                for item in original_items:            item_type = item['type']            if item_type not in report['types_processed']:                report['types_processed'][item_type] = 0            report['types_processed'][item_type] += 1                return report# 使用示例anonymizer = IntelligentAnonymizer()# 模拟敏感信息检测结果sensitive_data = [    {'type': 'CHINA_PHONE', 'start': 5, 'end': 16, 'content': '13800138000'},    {'type': 'PERSON_NAME', 'start': 0, 'end': 2, 'content': '张三'},    {'type': 'CHINA_ID', 'start': 20, 'end': 38, 'content': '110101199001011234'}]original_text = "张三的电话是13800138000，身份证号码是110101199001011234"anonymized_text = anonymizer.anonymize_text(original_text, sensitive_data)print(f"原文：{original_text}")print(f"脱敏后：{anonymized_text}")# 生成脱敏报告report = anonymizer.create_anonymization_report(sensitive_data)print(f"脱敏报告：{json.dumps(report, indent=2, ensure_ascii=False)}")

# semantic_chunker.pyimport refrom typing import List, Dict, Tuplefrom sentence_transformers import SentenceTransformerimport numpy as npfrom sklearn.metrics.pairwise import cosine_similarityclass SemanticChunker:    def __init__(self, model_name: str = 'all-MiniLM-L6-v2'):        self.model = SentenceTransformer(model_name)        self.max_chunk_size = 512        self.overlap_size = 50        self.similarity_threshold = 0.5            def chunk_by_structure(self, text: str) -> List[Dict]:        """基于文档结构进行分块"""        # 识别标题层级        heading_patterns = [            (r'^# (.+)$', 1),      # H1            (r'^## (.+)$', 2),     # H2            (r'^### (.+)$', 3),    # H3            (r'^#### (.+)$', 4),   # H4            (r'^第[一二三四五六七八九十d]+章s*(.+)', 1),  # 中文章节            (r'^第[一二三四五六七八九十d]+节s*(.+)', 2),  # 中文小节            (r'^d+.s+(.+)', 2), # 数字编号        ]                lines = text.split('n')        chunks = []        current_chunk = {            'content': '',            'level': 0,            'title': '',            'start_line': 0,            'end_line': 0        }                for i, line in enumerate(lines):            line = line.strip()            if not line:                continue                            # 检查是否为标题            is_heading = False            for pattern, level in heading_patterns:                match = re.match(pattern, line, re.MULTILINE)                if match:                    # 保存当前块                    if current_chunk['content']:                        current_chunk['end_line'] = i - 1                        chunks.append(current_chunk.copy())                                        # 开始新块                    current_chunk = {                        'content': line + 'n',                        'level': level,                        'title': match.group(1) if match.groups() else line,                        'start_line': i,                        'end_line': i                    }                    is_heading = True                    break                        if not is_heading:                current_chunk['content'] += line + 'n'                # 添加最后一个块        if current_chunk['content']:            current_chunk['end_line'] = len(lines) - 1            chunks.append(current_chunk)                return chunks        def chunk_by_semantics(self, text: str) -> List[Dict]:        """基于语义相似度进行分块"""        sentences = self._split_sentences(text)        if len(sentences) < 2:            return [{'content': text, 'semantic_score': 1.0}]                # 计算句子嵌入        embeddings = self.model.encode(sentences)                # 计算相邻句子的相似度        similarities = []        for i in range(len(embeddings) - 1):            sim = cosine_similarity([embeddings[i]], [embeddings[i + 1]])[0][0]            similarities.append(sim)                # 找到语义边界（相似度低的地方）        boundaries = [0]  # 起始边界        for i, sim in enumerate(similarities):            if sim < self.similarity_threshold:                boundaries.append(i + 1)        boundaries.append(len(sentences))  # 结束边界                # 生成语义块        chunks = []        for i in range(len(boundaries) - 1):            start_idx = boundaries[i]            end_idx = boundaries[i + 1]                        chunk_sentences = sentences[start_idx:end_idx]            chunk_content = ' '.join(chunk_sentences)                        # 计算块内语义一致性            if len(chunk_sentences) > 1:                chunk_embeddings = embeddings[start_idx:end_idx]                avg_similarity = np.mean([                    cosine_similarity([chunk_embeddings[j]], [chunk_embeddings[j + 1]])[0][0]                    for j in range(len(chunk_embeddings) - 1)                ])            else:                avg_similarity = 1.0                        chunks.append({                'content': chunk_content,                'semantic_score': avg_similarity,                'sentence_count': len(chunk_sentences),                'start_sentence': start_idx,                'end_sentence': end_idx - 1            })                return chunks        def adaptive_chunk(self, text: str) -> List[Dict]:        """自适应分块策略"""        # 首先尝试结构化分块        structure_chunks = self.chunk_by_structure(text)                final_chunks = []        for chunk in structure_chunks:            chunk_text = chunk['content']                        # 如果块太大，进行语义分块            if len(chunk_text) > self.max_chunk_size:                semantic_chunks = self.chunk_by_semantics(chunk_text)                                for sem_chunk in semantic_chunks:                    combined_chunk = {                        **chunk,  # 保留结构信息                        'content': sem_chunk['content'],                        'semantic_score': sem_chunk['semantic_score'],                        'chunk_method': 'structure_semantic'                    }                    final_chunks.append(combined_chunk)            else:                chunk['chunk_method'] = 'structure_only'                final_chunks.append(chunk)                return final_chunks        def _split_sentences(self, text: str) -> List[str]:        """智能句子分割"""        # 中英文句子分割规则        sentence_endings = r'[.!?。！？]+'        sentences = re.split(sentence_endings, text)                # 清理空句子和过短句子        cleaned_sentences = []        for sentence in sentences:            sentence = sentence.strip()            if len(sentence) > 10:  # 过滤过短的句子                cleaned_sentences.append(sentence)                return cleaned_sentences        def add_context_overlap(self, chunks: List[Dict]) -> List[Dict]:        """为分块添加上下文重叠"""        if len(chunks) <= 1:            return chunks                enhanced_chunks = []        for i, chunk in enumerate(chunks):            enhanced_chunk = chunk.copy()                        # 添加前文上下文            if i > 0:                prev_content = chunks[i-1]['content']                # 取前一块的后50个字符作为上下文                context_start = prev_content[-self.overlap_size:] if len(prev_content) > self.overlap_size else prev_content                enhanced_chunk['prev_context'] = context_start                        # 添加后文上下文            if i < len(chunks) - 1:                next_content = chunks[i+1]['content']                # 取后一块的前50个字符作为上下文                context_end = next_content[:self.overlap_size] if len(next_content) > self.overlap_size else next_content                enhanced_chunk['next_context'] = context_end                        # 生成完整的检索文本（包含上下文）            full_content = enhanced_chunk.get('prev_context', '') + chunk['content'] + enhanced_chunk.get('next_context', '')            enhanced_chunk['searchable_content'] = full_content                        enhanced_chunks.append(enhanced_chunk)                return enhanced_chunks# 使用示例chunker = SemanticChunker()sample_document = """# 企业知识管理系统设计方案## 1. 系统概述企业知识管理系统旨在整合企业内部的各类知识资源，提高知识的共享和利用效率。系统采用现代化的技术架构，支持多种知识形式的存储和检索。## 2. 技术架构### 2.1 前端架构前端采用React框架，提供直观的用户界面。支持响应式设计，适配多种设备。### 2.2 后端架构  后端基于Spring Boot框架，提供RESTful API接口。采用微服务架构，保证系统的可扩展性和稳定性。## 3. 数据管理系统支持结构化和非结构化数据的存储。采用向量数据库进行语义检索，提高检索的准确性。"""# 执行自适应分块chunks = chunker.adaptive_chunk(sample_document)enhanced_chunks = chunker.add_context_overlap(chunks)print("分块结果：")for i, chunk in enumerate(enhanced_chunks):    print(f"n块 {i+1}:")    print(f"标题：{chunk.get('title', '无标题')}")    print(f"层级：{chunk.get('level', 0)}")    print(f"方法：{chunk.get('chunk_method', '未知')}")    print(f"内容长度：{len(chunk['content'])}")    print(f"内容预览：{chunk['content'][:100]}...")

# qa_generator.pyimport openaiimport jsonfrom typing import List, Dict, Tupleimport refrom collections import defaultdictimport jieba.analyseclass QAGenerator:    def __init__(self, api_key: str, model: str = "gpt-3.5-turbo"):        openai.api_key = api_key        self.model = model        self.qa_templates = self._load_qa_templates()            def _load_qa_templates(self) -> List[Dict]:        """加载问答生成模板"""        return [            {                'type': 'factual',                'prompt': '基于以下文本，生成3-5个事实性问答对。问题应该具体明确，答案应该准确完整。',                'format': '问题：n答案：'            },            {                'type': 'conceptual',                 'prompt': '基于以下文本，生成2-3个概念性问答对。问题应该涉及定义、原理或概念解释。',                'format': '问题：n答案：'            },            {                'type': 'procedural',                'prompt': '基于以下文本，生成1-2个程序性问答对。问题应该涉及操作步骤或流程。',                'format': '问题：n答案：'            }        ]        def generate_qa_pairs(self, text_chunk: str, chunk_metadata: Dict = None) -> List[Dict]:        """为文本块生成问答对"""        all_qa_pairs = []                # 根据文本特征选择合适的模板        suitable_templates = self._select_templates(text_chunk)                for template in suitable_templates:            try:                qa_pairs = self._generate_with_template(text_chunk, template)                                # 为每个QA对添加元数据                for qa in qa_pairs:                    qa.update({                        'source_text': text_chunk[:200] + '...' if len(text_chunk) > 200 else text_chunk,                        'generation_type': template['type'],                        'chunk_metadata': chunk_metadata or {},                        'keywords': self._extract_keywords(text_chunk),                        'confidence_score': self._calculate_confidence(qa, text_chunk)                    })                                all_qa_pairs.extend(qa_pairs)                            except Exception as e:                print(f"生成QA对时出错: {e}")                continue                # 去重和质量过滤        filtered_qa_pairs = self._filter_qa_pairs(all_qa_pairs)                return filtered_qa_pairs        def _select_templates(self, text: str) -> List[Dict]:        """根据文本特征选择合适的模板"""        selected_templates = []                # 检测文本特征        has_definitions = bool(re.search(r'(是指|定义为|指的是|意思是)', text))        has_procedures = bool(re.search(r'(步骤|流程|方法|操作|执行)', text))        has_facts = bool(re.search(r'(数据|统计|结果|显示|表明)', text))                # 根据特征选择模板        if has_facts or len(text) > 100:            selected_templates.append(self.qa_templates[0])  # factual                if has_definitions:            selected_templates.append(self.qa_templates[1])  # conceptual                    if has_procedures:            selected_templates.append(self.qa_templates[2])  # procedural                # 如果没有明显特征，默认使用事实性模板        if not selected_templates:            selected_templates.append(self.qa_templates[0])                return selected_templates        def _generate_with_template(self, text: str, template: Dict) -> List[Dict]:        """使用指定模板生成QA对"""        prompt = f"""{template['prompt']}文本内容：{text}请按照以下格式输出：{template['format']}要求：1. 问题要具体、清晰，避免过于宽泛2. 答案要基于原文内容，不要添加额外信息3. 每个问答对之间用"---"分隔4. 确保问题的答案在原文中能找到依据"""                response = openai.ChatCompletion.create(            model=self.model,            messages=[                {"role": "system", "content": "你是一个专业的知识问答生成专家。"},                {"role": "user", "content": prompt}            ],            temperature=0.3,            max_tokens=1000        )                content = response.choices[0].message.content        return self._parse_qa_response(content)        def _parse_qa_response(self, response: str) -> List[Dict]:        """解析LLM生成的QA对"""        qa_pairs = []                # 按分隔符分割        sections = response.split('---')                for section in sections:            section = section.strip()            if not section:                continue                        # 提取问题和答案            lines = section.split('n')            question = ""            answer = ""                        current_part = None            for line in lines:                line = line.strip()                if line.startswith('问题：') or line.startswith('Q:'):                    current_part = 'question'                    question = line.replace('问题：', '').replace('Q:', '').strip()                elif line.startswith('答案：') or line.startswith('A:'):                    current_part = 'answer'                    answer = line.replace('答案：', '').replace('A:', '').strip()                elif current_part == 'question':                    question += ' ' + line                elif current_part == 'answer':                    answer += ' ' + line                        if question and answer:                qa_pairs.append({                    'question': question.strip(),                    'answer': answer.strip()                })                return qa_pairs        def _extract_keywords(self, text: str) -> List[str]:        """提取文本关键词"""        # 使用jieba提取关键词        keywords = jieba.analyse.extract_tags(text, topK=5, withWeight=False)        return keywords        def _calculate_confidence(self, qa_pair: Dict, source_text: str) -> float:        """计算QA对的置信度分数"""        question = qa_pair['question']        answer = qa_pair['answer']                # 基于多个因素计算置信度        score = 0.0                # 1. 答案在原文中的覆盖度        answer_words = set(jieba.lcut(answer))        source_words = set(jieba.lcut(source_text))        coverage = len(answer_words.intersection(source_words)) / len(answer_words) if answer_words else 0        score += coverage * 0.4                # 2. 问题的具体性（长度适中，包含关键词）        question_length = len(question)        if 10 <= question_length <= 50:            score += 0.3        elif question_length > 50:            score += 0.1                # 3. 答案的完整性（不要太短或太长）        answer_length = len(answer)        if 20 <= answer_length <= 200:            score += 0.3        elif answer_length > 200:            score += 0.1                return min(score, 1.0)        def _filter_qa_pairs(self, qa_pairs: List[Dict]) -> List[Dict]:        """过滤和去重QA对"""        # 按置信度排序        sorted_pairs = sorted(qa_pairs, key=lambda x: x['confidence_score'], reverse=True)                # 去重（基于问题相似度）        unique_pairs = []        seen_questions = set()                for qa in sorted_pairs:            question = qa['question']                        # 简单的去重策略：检查问题是否过于相似            is_duplicate = False            for seen_q in seen_questions:                if self._questions_similar(question, seen_q):                    is_duplicate = True                    break                        if not is_duplicate and qa['confidence_score'] > 0.5:                unique_pairs.append(qa)                seen_questions.add(question)                return unique_pairs[:5]  # 最多返回5个高质量QA对        def _questions_similar(self, q1: str, q2: str, threshold: float = 0.8) -> bool:        """检查两个问题是否相似"""        words1 = set(jieba.lcut(q1))        words2 = set(jieba.lcut(q2))                if not words1 or not words2:            return False                intersection = words1.intersection(words2)        union = words1.union(words2)                similarity = len(intersection) / len(union)        return similarity > threshold# 使用示例（需要OpenAI API密钥）# qa_generator = QAGenerator(api_key="your-openai-api-key")# sample_text = """# 企业知识管理系统是指通过信息技术手段，对企业内部的显性知识和隐性知识进行系统化管理的平台。# 该系统主要包括知识获取、知识存储、知识共享和知识应用四个核心模块。# 实施步骤包括：1）需求分析；2）系统设计；3）技术选型；4）开发实施；5）测试部署。# 据统计，实施知识管理系统的企业，其知识利用效率平均提升35%以上。# """# qa_pairs = qa_generator.generate_qa_pairs(sample_text)# print("生成的问答对：")# for i, qa in enumerate(qa_pairs, 1):#     print(f"n{i}. 问题：{qa['question']}")#     print(f"   答案：{qa['answer']}")#     print(f"   置信度：{qa['confidence_score']:.2f}")#     print(f"   类型：{qa['generation_type']}")

# metadata_manager.pyimport sqlite3import jsonfrom datetime import datetimefrom typing import Dict, List, Optionalfrom enum import Enumimport uuidclass SensitiveLevel(Enum):    PUBLIC = "公开"    INTERNAL = "内部"    CONFIDENTIAL = "机密"    TOP_SECRET = "绝密"class DataType(Enum):    DOCUMENT = "文档"    DATABASE = "数据库"    API = "接口"    FILE = "文件"class MetadataManager:    def __init__(self, db_path: str = "metadata.db"):        self.db_path = db_path        self.init_database()        def init_database(self):        """初始化元数据数据库"""        conn = sqlite3.connect(self.db_path)        cursor = conn.cursor()                # 创建元数据表        cursor.execute('''            CREATE TABLE IF NOT EXISTS metadata (                id TEXT PRIMARY KEY,                data_source TEXT NOT NULL,                file_path TEXT,                data_type TEXT NOT NULL,                sensitive_level TEXT NOT NULL,                owner TEXT,                department TEXT,                created_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,                updated_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,                file_size INTEGER,                file_format TEXT,                encoding TEXT,                language TEXT,                keywords TEXT,                description TEXT,                tags TEXT,                access_count INTEGER DEFAULT 0,                last_access_time TIMESTAMP,                checksum TEXT,                version TEXT DEFAULT '1.0',                parent_id TEXT,                is_active BOOLEAN DEFAULT 1            )        ''')                # 创建权限控制表        cursor.execute('''            CREATE TABLE IF NOT EXISTS permissions (                id TEXT PRIMARY KEY,                metadata_id TEXT,                role TEXT NOT NULL,                permission_type TEXT NOT NULL,                granted_by TEXT,                granted_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,                expires_time TIMESTAMP,                is_active BOOLEAN DEFAULT 1,                FOREIGN KEY (metadata_id) REFERENCES metadata (id)            )        ''')                # 创建数据血缘表        cursor.execute('''            CREATE TABLE IF NOT EXISTS data_lineage (                id TEXT PRIMARY KEY,                source_id TEXT,                target_id TEXT,                relationship_type TEXT,                created_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,                FOREIGN KEY (source_id) REFERENCES metadata (id),                FOREIGN KEY (target_id) REFERENCES metadata (id)            )        ''')                conn.commit()        conn.close()        def register_data_source(self,                            data_source: str,                           file_path: str,                           data_type: DataType,                           sensitive_level: SensitiveLevel,                           **kwargs) -> str:        """注册数据源"""                metadata_id = str(uuid.uuid4())                conn = sqlite3.connect(self.db_path)        cursor = conn.cursor()                # 插入元数据        cursor.execute('''            INSERT INTO metadata (                id, data_source, file_path, data_type, sensitive_level,                owner, department, file_size, file_format, encoding,                language, keywords, description, tags, checksum, version            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)        ''', (            metadata_id,            data_source,            file_path,            data_type.value,            sensitive_level.value,            kwargs.get('owner', ''),            kwargs.get('department', ''),            kwargs.get('file_size', 0),            kwargs.get('file_format', ''),            kwargs.get('encoding', 'utf-8'),            kwargs.get('language', 'zh'),            json.dumps(kwargs.get('keywords', []), ensure_ascii=False),            kwargs.get('description', ''),            json.dumps(kwargs.get('tags', []), ensure_ascii=False),            kwargs.get('checksum', ''),            kwargs.get('version', '1.0')        ))                conn.commit()        conn.close()                return metadata_id        def set_permissions(self,                        metadata_id: str,                       role: str,                        permission_type: str,                       granted_by: str,                       expires_time: Optional[datetime] = None):        """设置数据权限"""                permission_id = str(uuid.uuid4())                conn = sqlite3.connect(self.db_path)        cursor = conn.cursor()                cursor.execute('''            INSERT INTO permissions (                id, metadata_id, role, permission_type, granted_by, expires_time            ) VALUES (?, ?, ?, ?, ?, ?)        ''', (            permission_id,            metadata_id,            role,            permission_type,            granted_by,            expires_time.isoformat() if expires_time else None        ))                conn.commit()        conn.close()                return permission_id        def check_permission(self, metadata_id: str, role: str, permission_type: str) -> bool:        """检查数据访问权限"""                conn = sqlite3.connect(self.db_path)        cursor = conn.cursor()                cursor.execute('''            SELECT COUNT(*) FROM permissions p            JOIN metadata m ON p.metadata_id = m.id            WHERE p.metadata_id = ?             AND p.role = ?             AND p.permission_type = ?            AND p.is_active = 1            AND m.is_active = 1            AND (p.expires_time IS NULL OR p.expires_time > ?)        ''', (metadata_id, role, permission_type, datetime.now().isoformat()))                result = cursor.fetchone()        conn.close()                return result[0] > 0        def update_access_log(self, metadata_id: str):        """更新访问日志"""                conn = sqlite3.connect(self.db_path)        cursor = conn.cursor()                cursor.execute('''            UPDATE metadata             SET access_count = access_count + 1,                last_access_time = ?            WHERE id = ?        ''', (datetime.now().isoformat(), metadata_id))                conn.commit()        conn.close()        def create_data_lineage(self, source_id: str, target_id: str, relationship_type: str):        """创建数据血缘关系"""                lineage_id = str(uuid.uuid4())                conn = sqlite3.connect(self.db_path)        cursor = conn.cursor()                cursor.execute('''            INSERT INTO data_lineage (id, source_id, target_id, relationship_type)            VALUES (?, ?, ?, ?)        ''', (lineage_id, source_id, target_id, relationship_type))                conn.commit()        conn.close()                return lineage_id        def get_metadata(self, metadata_id: str) -> Optional[Dict]:        """获取元数据信息"""                conn = sqlite3.connect(self.db_path)        cursor = conn.cursor()                cursor.execute('''            SELECT * FROM metadata WHERE id = ? AND is_active = 1        ''', (metadata_id,))                result = cursor.fetchone()        conn.close()                if result:            columns = [desc[0] for desc in cursor.description]            metadata = dict(zip(columns, result))                        # 解析JSON字段            if metadata['keywords']:                metadata['keywords'] = json.loads(metadata['keywords'])            if metadata['tags']:                metadata['tags'] = json.loads(metadata['tags'])                        return metadata                return None        def search_metadata(self,                        data_type: Optional[str] = None,                       sensitive_level: Optional[str] = None,                       department: Optional[str] = None,                       keywords: Optional[List[str]] = None) -> List[Dict]:        """搜索元数据"""                conn = sqlite3.connect(self.db_path)        cursor = conn.cursor()                # 构建查询条件        conditions = ["is_active = 1"]        params = []                if data_type:            conditions.append("data_type = ?")            params.append(data_type)                if sensitive_level:            conditions.append("sensitive_level = ?")            params.append(sensitive_level)                if department:            conditions.append("department = ?")            params.append(department)                if keywords:            for keyword in keywords:                conditions.append("(keywords LIKE ? OR description LIKE ? OR tags LIKE ?)")                keyword_pattern = f"%{keyword}%"                params.extend([keyword_pattern, keyword_pattern, keyword_pattern])                where_clause = " AND ".join(conditions)        query = f"SELECT * FROM metadata WHERE {where_clause} ORDER BY updated_time DESC"                cursor.execute(query, params)        results = cursor.fetchall()        conn.close()                # 转换为字典列表        metadata_list = []        if results:            columns = [desc[0] for desc in cursor.description]            for result in results:                metadata = dict(zip(columns, result))                                # 解析JSON字段                if metadata['keywords']:                    metadata['keywords'] = json.loads(metadata['keywords'])                if metadata['tags']:                    metadata['tags'] = json.loads(metadata['tags'])                                metadata_list.append(metadata)                return metadata_list        def generate_governance_report(self) -> Dict:        """生成数据治理报告"""                conn = sqlite3.connect(self.db_path)        cursor = conn.cursor()                # 统计基本信息        cursor.execute("SELECT COUNT(*) FROM metadata WHERE is_active = 1")        total_sources = cursor.fetchone()[0]                cursor.execute("""            SELECT data_type, COUNT(*)             FROM metadata WHERE is_active = 1             GROUP BY data_type        """)        type_distribution = dict(cursor.fetchall())                cursor.execute("""            SELECT sensitive_level, COUNT(*)             FROM metadata WHERE is_active = 1             GROUP BY sensitive_level        """)        sensitivity_distribution = dict(cursor.fetchall())                cursor.execute("""            SELECT department, COUNT(*)             FROM metadata WHERE is_active = 1             GROUP BY department        """)        department_distribution = dict(cursor.fetchall())                # 访问统计        cursor.execute("""            SELECT AVG(access_count), MAX(access_count), MIN(access_count)            FROM metadata WHERE is_active = 1        """)        access_stats = cursor.fetchone()                # 权限统计        cursor.execute("""            SELECT permission_type, COUNT(*)             FROM permissions WHERE is_active = 1             GROUP BY permission_type        """)        permission_stats = dict(cursor.fetchall())                conn.close()                return {            'total_data_sources': total_sources,            'type_distribution': type_distribution,            'sensitivity_distribution': sensitivity_distribution,            'department_distribution': department_distribution,            'access_statistics': {                'average_access': round(access_stats[0] or 0, 2),                'max_access': access_stats[1] or 0,                'min_access': access_stats[2] or 0            },            'permission_statistics': permission_stats,            'report_generated_at': datetime.now().isoformat()        }# 使用示例metadata_manager = MetadataManager()# 注册数据源doc_id = metadata_manager.register_data_source(    data_source="HR_handbook_v2.pdf",    file_path="/documents/hr/handbook.pdf",    data_type=DataType.DOCUMENT,    sensitive_level=SensitiveLevel.INTERNAL,    owner="张三",    department="人力资源部",    file_size=2048000,    file_format="PDF",    keywords=["人事制度", "员工手册", "薪酬体系"],    description="公司员工手册第二版",    tags=["HR", "制度", "内部"])# 设置权限metadata_manager.set_permissions(    metadata_id=doc_id,    role="HR专员",    permission_type="读取",    granted_by="HR主管")# 检查权限has_permission = metadata_manager.check_permission(doc_id, "HR专员", "读取")print(f"HR专员是否有读取权限：{has_permission}")# 生成治理报告report = metadata_manager.generate_governance_report()print(f"数据治理报告：{json.dumps(report, indent=2, ensure_ascii=False)}")

# quality_monitor.pyimport timeimport jsonfrom datetime import datetime, timedeltafrom typing import Dict, List, Tupleimport threadingfrom dataclasses import dataclassimport logging@dataclassclass QualityMetric:    name: str    value: float    threshold: float    status: str    last_updated: datetimeclass QualityMonitor:    def __init__(self, config_path: str = "quality_config.json"):        self.config = self._load_config(config_path)        self.metrics = {}        self.alerts = []        self.monitoring_active = False                # 设置日志        logging.basicConfig(            level=logging.INFO,            format='%(asctime)s - %(levelname)s - %(message)s',            handlers=[                logging.FileHandler('quality_monitor.log'),                logging.StreamHandler()            ]        )        self.logger = logging.getLogger(__name__)        def _load_config(self, config_path: str) -> Dict:        """加载监控配置"""        default_config = {            "monitoring_interval": 300,  # 5分钟            "alert_threshold": {                "document_parse_failure_rate": 0.05,  # 5%                "sensitive_data_leak_rate": 0.0,      # 0%                "qa_recall_rate": 0.85,               # 85%                "system_response_time": 2.0,          # 2秒                "storage_usage_rate": 0.80             # 80%            },            "notification": {                "email_enabled": True,                "webhook_enabled": False,                "email_recipients": ["admin@company.com"],                "webhook_url": ""            }        }                try:            with open(config_path, 'r', encoding='utf-8') as f:                loaded_config = json.load(f)                default_config.update(loaded_config)        except FileNotFoundError:            self.logger.warning(f"配置文件 {config_path} 不存在，使用默认配置")            # 创建默认配置文件            with open(config_path, 'w', encoding='utf-8') as f:                json.dump(default_config, f, indent=2, ensure_ascii=False)                return default_config        def register_metric(self, name: str, threshold: float):        """注册监控指标"""        self.metrics[name] = QualityMetric(            name=name,            value=0.0,            threshold=threshold,            status="UNKNOWN",            last_updated=datetime.now()        )        self.logger.info(f"注册监控指标: {name}, 阈值: {threshold}")        def update_metric(self, name: str, value: float):        """更新指标值"""        if name not in self.metrics:            self.logger.warning(f"指标 {name} 未注册")            return                metric = self.metrics[name]        metric.value = value        metric.last_updated = datetime.now()                # 判断状态        if name in ["document_parse_failure_rate", "sensitive_data_leak_rate", "system_response_time", "storage_usage_rate"]:            # 这些指标越低越好            metric.status = "NORMAL" if value <= metric.threshold else "ALERT"        else:            # qa_recall_rate等指标越高越好            metric.status = "NORMAL" if value >= metric.threshold else "ALERT"                # 如果状态异常，生成告警        if metric.status == "ALERT":            self._generate_alert(metric)                self.logger.info(f"更新指标 {name}: {value} ({metric.status})")        def _generate_alert(self, metric: QualityMetric):        """生成告警"""        alert = {            'metric_name': metric.name,            'current_value': metric.value,            'threshold': metric.threshold,            'status': metric.status,            'timestamp': datetime.now().isoformat(),            'message': f"指标 {metric.name} 异常: 当前值 {metric.value}, 阈值 {metric.threshold}"        }                self.alerts.append(alert)        self.logger.error(alert['message'])                # 发送通知        if self.config['notification']['email_enabled']:            self._send_email_alert(alert)        def _send_email_alert(self, alert: Dict):        """发送邮件告警（模拟实现）"""        self.logger.info(f"发送邮件告警: {alert['message']}")        # 实际项目中这里会调用邮件服务API        def run_quality_checks(self) -> Dict:        """执行质量检查"""        check_results = {            'check_time': datetime.now().isoformat(),            'results': {}        }                # 1. 文档解析失效率检查        parse_failure_rate = self._check_document_parse_rate()        self.update_metric('document_parse_failure_rate', parse_failure_rate)        check_results['results']['document_parse_failure_rate'] = parse_failure_rate                # 2. 敏感信息泄露检查        sensitive_leak_rate = self._check_sensitive_data_leak()        self.update_metric('sensitive_data_leak_rate', sensitive_leak_rate)        check_results['results']['sensitive_data_leak_rate'] = sensitive_leak_rate                # 3. QA召回率检查        qa_recall = self._check_qa_recall_rate()        self.update_metric('qa_recall_rate', qa_recall)        check_results['results']['qa_recall_rate'] = qa_recall                # 4. 系统响应时间检查        response_time = self._check_system_response_time()        self.update_metric('system_response_time', response_time)        check_results['results']['system_response_time'] = response_time                # 5. 存储使用率检查        storage_usage = self._check_storage_usage()        self.update_metric('storage_usage_rate', storage_usage)        check_results['results']['storage_usage_rate'] = storage_usage                return check_results        def _check_document_parse_rate(self) -> float:        """检查文档解析成功率"""        # 模拟检查逻辑        # 实际项目中会查询解析日志或数据库        import random        return random.uniform(0.02, 0.08)  # 模拟2%-8%的失效率        def _check_sensitive_data_leak(self) -> float:        """检查敏感信息泄露率"""        # 实际项目中会扫描处理后的文档        import random        return random.uniform(0.0, 0.01)  # 模拟0%-1%的泄露率        def _check_qa_recall_rate(self) -> float:        """检查QA召回率"""        # 实际项目中会运行测试集        import random        return random.uniform(0.75, 0.95)  # 模拟75%-95%的召回率        def _check_system_response_time(self) -> float:        """检查系统响应时间"""        # 模拟API响应时间检查        import random        return random.uniform(0.5, 3.0)  # 模拟0.5-3秒的响应时间        def _check_storage_usage(self) -> float:        """检查存储使用率"""        # 实际项目中会检查磁盘使用情况        import random        return random.uniform(0.60, 0.90)  # 模拟60%-90%的使用率        def start_monitoring(self):        """启动持续监控"""        if self.monitoring_active:            self.logger.warning("监控已在运行中")            return                self.monitoring_active = True                # 注册默认指标        for metric_name, threshold in self.config['alert_threshold'].items():            self.register_metric(metric_name, threshold)                # 启动监控线程        monitor_thread = threading.Thread(target=self._monitoring_loop)        monitor_thread.daemon = True        monitor_thread.start()                self.logger.info("质量监控已启动")        def _monitoring_loop(self):        """监控循环"""        while self.monitoring_active:            try:                self.run_quality_checks()                time.sleep(self.config['monitoring_interval'])            except Exception as e:                self.logger.error(f"监控过程中出错: {e}")                time.sleep(60)  # 出错后等待1分钟再重试        def stop_monitoring(self):        """停止监控"""        self.monitoring_active = False        self.logger.info("质量监控已停止")        def get_dashboard_data(self) -> Dict:        """获取监控面板数据"""        dashboard_data = {            'metrics': {},            'alerts': self.alerts[-10:],  # 最近10条告警            'summary': {                'total_metrics': len(self.metrics),                'normal_metrics': 0,                'alert_metrics': 0,                'last_check': None            }        }                for name, metric in self.metrics.items():            dashboard_data['metrics'][name] = {                'value': metric.value,                'threshold': metric.threshold,                'status': metric.status,                'last_updated': metric.last_updated.isoformat()            }                        if metric.status == 'NORMAL':                dashboard_data['summary']['normal_metrics'] += 1            elif metric.status == 'ALERT':                dashboard_data['summary']['alert_metrics'] += 1                        # 更新最后检查时间            if (dashboard_data['summary']['last_check'] is None or                 metric.last_updated > datetime.fromisoformat(dashboard_data['summary']['last_check'])):                dashboard_data['summary']['last_check'] = metric.last_updated.isoformat()                return dashboard_data        def generate_quality_report(self, days: int = 7) -> Dict:        """生成质量报告"""        end_time = datetime.now()        start_time = end_time - timedelta(days=days)                # 过滤时间范围内的告警        period_alerts = [            alert for alert in self.alerts            if start_time <= datetime.fromisoformat(alert['timestamp']) <= end_time        ]                # 按指标分组统计告警        alert_stats = {}        for alert in period_alerts:            metric_name = alert['metric_name']            if metric_name not in alert_stats:                alert_stats[metric_name] = 0            alert_stats[metric_name] += 1                # 计算指标健康度        health_score = 0        if self.metrics:            normal_count = sum(1 for m in self.metrics.values() if m.status == 'NORMAL')            health_score = (normal_count / len(self.metrics)) * 100                report = {            'report_period': {                'start_time': start_time.isoformat(),                'end_time': end_time.isoformat(),                'days': days            },            'overall_health_score': round(health_score, 2),            'metrics_summary': {                'total_metrics': len(self.metrics),                'normal_metrics': sum(1 for m in self.metrics.values() if m.status == 'NORMAL'),                'alert_metrics': sum(1 for m in self.metrics.values() if m.status == 'ALERT')            },            'alert_summary': {                'total_alerts': len(period_alerts),                'alerts_by_metric': alert_stats,                'most_problematic_metric': max(alert_stats.items(), key=lambda x: x[1])[0] if alert_stats else None            },            'current_metrics': {                name: {                    'value': metric.value,                    'threshold': metric.threshold,                    'status': metric.status                }                for name, metric in self.metrics.items()            },            'recommendations': self._generate_recommendations()        }                return report        def _generate_recommendations(self) -> List[str]:        """生成改进建议"""        recommendations = []                for name, metric in self.metrics.items():            if metric.status == 'ALERT':                if name == 'document_parse_failure_rate':                    recommendations.append("建议检查PDF解析器配置，可能需要升级OCR引擎或优化文档预处理流程")                elif name == 'sensitive_data_leak_rate':                    recommendations.append("发现敏感信息泄露，建议立即检查脱敏规则和二次校验机制")                elif name == 'qa_recall_rate':                    recommendations.append("QA召回率偏低，建议优化文档分块策略和检索算法参数")                elif name == 'system_response_time':                    recommendations.append("系统响应时间过长，建议检查数据库查询效率和服务器资源使用情况")                elif name == 'storage_usage_rate':                    recommendations.append("存储空间不足，建议清理过期数据或扩容存储设备")                if not recommendations:            recommendations.append("所有指标正常，建议继续保持当前的数据处理质量")                return recommendations# 使用示例monitor = QualityMonitor()# 启动监控monitor.start_monitoring()# 等待一段时间让监控运行time.sleep(2)# 获取监控面板数据dashboard = monitor.get_dashboard_data()print("监控面板数据：")print(json.dumps(dashboard, indent=2, ensure_ascii=False))# 生成质量报告quality_report = monitor.generate_quality_report(days=1)print("n质量报告：")print(json.dumps(quality_report, indent=2, ensure_ascii=False))# 停止监控monitor.stop_monitoring()

# robust_pdf_processor.pyimport PyMuPDF as fitzimport pytesseractfrom PIL import Imageimport camelotimport iofrom typing import Dict, Listclass RobustPDFProcessor:    def __init__(self):        self.ocr_languages = 'chi_sim+eng'  # 中英文OCR        self.fallback_strategies = ['text_extraction', 'ocr', 'hybrid']        def process_with_fallback(self, pdf_path: str) -> Dict:        """多策略PDF处理"""        results = {}                for strategy in self.fallback_strategies:            try:                if strategy == 'text_extraction':                    result = self._extract_text_native(pdf_path)                elif strategy == 'ocr':                    result = self._extract_text_ocr(pdf_path)                elif strategy == 'hybrid':                    result = self._extract_text_hybrid(pdf_path)                                # 验证提取质量                if self._validate_extraction_quality(result):                    results[strategy] = result                    break                                except Exception as e:                print(f"策略 {strategy} 失败: {e}")                continue                # 特殊处理：表格提取        try:            tables = self._extract_tables_robust(pdf_path)            results['tables'] = tables        except:            results['tables'] = []                return results        def _validate_extraction_quality(self, text: str) -> bool:        """验证提取质量"""        if len(text) < 50:  # 文本太短            return False                # 检查乱码比例        total_chars = len(text)        weird_chars = sum(1 for c in text if ord(c) > 65535 or c in '□■●◆')        weird_ratio = weird_chars / total_chars                return weird_ratio < 0.1  # 乱码比例小于10%        def _extract_tables_robust(self, pdf_path: str) -> List[Dict]:        """鲁棒的表格提取"""        tables = []                # 策略1：使用camelot        try:            camelot_tables = camelot.read_pdf(pdf_path, pages='all')            for table in camelot_tables:                if table.accuracy > 0.7:  # 只保留准确度高的表格                    tables.append({                        'method': 'camelot',                        'accuracy': table.accuracy,                        'data': table.df.to_dict('records')                    })        except:            pass                # 策略2：基于文本模式识别        if not tables:            try:                doc = fitz.open(pdf_path)                for page in doc:                    text = page.get_text()                    potential_tables = self._detect_text_tables(text)                    tables.extend(potential_tables)                doc.close()            except:                pass                return tables