freedak
641e33b834
- 语义检索(FAISS + embedding)+ 精确查找(法规名+条号) - RAG 问答(SSE 流式,支持 thinking 折叠显示) - 法规浏览(原文阅读) - 历史记录(检索+对话持久化到 SQLite) - 设置页(系统提示词/模板/LLM 参数可配置) - 检索质量评估脚本 Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
267 lines
8.7 KiB
Python
267 lines
8.7 KiB
Python
"""法规切片解析器 — 按"第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 <法规文件路径>")
|