feat: QYLAW 法律法规知识库
- 语义检索(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>
This commit is contained in:
@@ -0,0 +1,266 @@
|
||||
"""法规切片解析器 — 按"第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 <法规文件路径>")
|
||||
Reference in New Issue
Block a user