"""法规切片解析器 — 按"第X条"切片,提取 metadata 解析 markdown 法规文件: - 法规名:文件名去 .md 后缀,去末尾 _YYYYMMDD - 发布日期:文件名末尾 _YYYYMMDD - 章节:跟踪"第X章"上下文 - 条文:按"第X条"切片,内容聚合到下一个"第X条"/"第X章"/"第X节" - 地方性法规:从 地方性法规区域映射.json 附加省份/市级 用法: from scripts.parse_clause import parse_law_file clauses = parse_law_file("/data/law-pack/法律/中华人民共和国民法典_20200528.md", "法律") """ import json import re import logging from dataclasses import dataclass, field from pathlib import Path from typing import List, Optional, Dict, Any logger = logging.getLogger(__name__) # 中文数字正则(支持"第一""第二十二""第一百二十六"等) CN_NUM = r"[一二三四五六七八九十百千零〇两]+" # 条文开头:第X条 + 全角空格/普通空格 CLAUSE_RE = re.compile(rf"^第({CN_NUM})条[\s\u3000]+(.*)") # 章节开头:第X章 + 标题 CHAPTER_RE = re.compile(rf"^第{CN_NUM}章[\s\u3000]+(.+)") # 节开头:第X节 SECTION_RE = re.compile(rf"^第{CN_NUM}节[\s\u3000]+(.+)") # 编开头:第X编 PART_RE = re.compile(rf"^第{CN_NUM}编[\s\u3000]+(.+)") # 文件名日期后缀 DATE_SUFFIX_RE = re.compile(r"_(\d{8})$") @dataclass class Clause: """法规条文""" law_name: str category: str chapter: Optional[str] = None # 当前章节,如"第一章 基本规定" clause_no: str = "" # 条号,如"第一条" / "第143条" content: str = "" # 条文内容 publish_date: Optional[str] = None # 发布日期 YYYY-MM-DD file_path: str = "" # 原文路径 province: Optional[str] = None # 省份(地方性法规) city: Optional[str] = None # 市级(地方性法规) region_level: Optional[str] = None # 区域级别(省级/市级) def to_dict(self) -> Dict[str, Any]: return { "law_name": self.law_name, "category": self.category, "chapter": self.chapter, "clause_no": self.clause_no, "content": self.content, "publish_date": self.publish_date, "file_path": self.file_path, "province": self.province, "city": self.city, "region_level": self.region_level, } def extract_name_and_date(filename: str) -> tuple[str, Optional[str]]: """从文件名提取法规名和发布日期 Args: filename: 文件名(含或不含 .md 后缀) Returns: (法规名, 发布日期 YYYY-MM-DD 或 None) """ stem = Path(filename).stem match = DATE_SUFFIX_RE.search(stem) if match: date_str = match.group(1) name = stem[: match.start()] # 格式化日期 YYYY-MM-DD date_formatted = f"{date_str[:4]}-{date_str[4:6]}-{date_str[6:8]}" return name, date_formatted return stem, None def load_region_mapping(mapping_path: Path) -> Dict[str, Optional[Dict]]: """加载地方性法规区域映射 Returns: {文件名: {region_name, province, level, region_id, province_id} | None} """ if not mapping_path.exists(): logger.warning(f"区域映射文件不存在: {mapping_path}") return {} with open(mapping_path, "r", encoding="utf-8") as f: return json.load(f) def parse_law_file( filepath: str | Path, category: str, region_mapping: Optional[Dict] = None, ) -> List[Clause]: """解析单个法规文件,按"第X条"切片 Args: filepath: 法规 markdown 文件路径 category: 法规类别(法律/行政法规/监察法规/司法解释/地方性法规) region_mapping: 地方性法规区域映射(文件名 -> {province, city, ...}) Returns: 条文列表;若文件无法识别"第X条"则返回空列表 """ filepath = Path(filepath) filename = filepath.name law_name, publish_date = extract_name_and_date(filename) # 地方性法规附加区域信息 province = None city = None region_level = None if category == "地方性法规" and region_mapping is not None: info = region_mapping.get(filename) if info is None: # 映射为 null,跳过(调用方负责记录) return [] province = info.get("province") region_level = info.get("level") # 市级:region_name 如果不是省份名,则视为市级 region_name = info.get("region_name") if region_name and province and region_name != province: city = region_name # 读取文件 try: text = filepath.read_text(encoding="utf-8") except Exception as e: logger.error(f"读取失败 {filepath}: {e}") return [] lines = text.split("\n") clauses: List[Clause] = [] current_chapter: Optional[str] = None current_clause: Optional[Clause] = None current_content_lines: List[str] = [] def _flush_current(): """将当前条文写入列表""" nonlocal current_clause, current_content_lines if current_clause is not None: content = "\n".join(current_content_lines).strip() if content: current_clause.content = content current_clause.chapter = current_chapter clauses.append(current_clause) current_clause = None current_content_lines = [] for line in lines: stripped = line.strip() # 跳过空行(但保留在内容中,后续 strip 处理) if not stripped: if current_clause is not None: current_content_lines.append("") continue # 检测章节 chap_match = CHAPTER_RE.match(stripped) if chap_match: _flush_current() current_chapter = stripped continue # 检测节(不切片,只更新上下文) if SECTION_RE.match(stripped): _flush_current() continue # 检测编(不切片) if PART_RE.match(stripped): _flush_current() continue # 检测条文 clause_match = CLAUSE_RE.match(stripped) if clause_match: _flush_current() clause_no = f"第{clause_match.group(1)}条" first_line = clause_match.group(2).strip() current_clause = Clause( law_name=law_name, category=category, clause_no=clause_no, publish_date=publish_date, file_path=str(filepath), province=province, city=city, region_level=region_level, ) current_content_lines = [first_line] if first_line else [] continue # 普通行:若在条文中,追加到内容 if current_clause is not None: current_content_lines.append(stripped) _flush_current() return clauses def scan_category_dir( law_pack_dir: Path, category: str, region_mapping: Optional[Dict] = None, ) -> tuple[List[Clause], List[str]]: """扫描某类别目录下所有法规文件 Args: law_pack_dir: 法规根目录 category: 类别(法律/行政法规/...) region_mapping: 区域映射(仅地方性法规需要) Returns: (条文列表, 跳过文件列表[null 映射或解析失败]) """ cat_dir = law_pack_dir / category if not cat_dir.exists(): logger.warning(f"类别目录不存在: {cat_dir}") return [], [] all_clauses: List[Clause] = [] skipped: List[str] = [] md_files = sorted(cat_dir.glob("*.md")) for md_file in md_files: clauses = parse_law_file(md_file, category, region_mapping) if not clauses: # 区分 null 映射跳过 vs 解析失败 if category == "地方性法规" and region_mapping is not None: info = region_mapping.get(md_file.name) if info is None: skipped.append(f"[null映射] {md_file.name}") else: skipped.append(f"[无条文] {md_file.name}") else: skipped.append(f"[无条文] {md_file.name}") else: all_clauses.extend(clauses) return all_clauses, skipped if __name__ == "__main__": # 自测:解析民法典 import sys logging.basicConfig(level=logging.INFO) test_file = sys.argv[1] if len(sys.argv) > 1 else None if test_file: clauses = parse_law_file(test_file, "法律") print(f"解析 {test_file}: {len(clauses)} 条") for c in clauses[:3]: print(f" {c.clause_no} [{c.chapter}] {c.content[:50]}...") else: print("用法: python parse_clause.py <法规文件路径>")