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>
148 lines
4.5 KiB
Python
148 lines
4.5 KiB
Python
"""法规原文读取服务 — 从 law-pack 目录读取 markdown,从 SQLite 取列表"""
|
|
import sqlite3
|
|
import logging
|
|
from typing import Optional, Dict, Any, List
|
|
from pathlib import Path
|
|
|
|
from app.config import SQLITE_PATH, LAW_PACK_DIR
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class LawReader:
|
|
"""法规原文读取单例"""
|
|
|
|
_instance: Optional["LawReader"] = None
|
|
|
|
@classmethod
|
|
def get_instance(cls) -> "LawReader":
|
|
if cls._instance is None:
|
|
cls._instance = cls()
|
|
return cls._instance
|
|
|
|
def __init__(self):
|
|
self._conn: Optional[sqlite3.Connection] = None
|
|
|
|
def _ensure_conn(self):
|
|
if self._conn is None and SQLITE_PATH.exists():
|
|
self._conn = sqlite3.connect(str(SQLITE_PATH), check_same_thread=False)
|
|
self._conn.row_factory = sqlite3.Row
|
|
|
|
def list_laws(
|
|
self,
|
|
category: Optional[str] = None,
|
|
province: Optional[str] = None,
|
|
keyword: Optional[str] = None,
|
|
page: int = 1,
|
|
page_size: int = 50,
|
|
) -> Dict[str, Any]:
|
|
"""法规列表(分页 + 过滤)"""
|
|
self._ensure_conn()
|
|
if self._conn is None:
|
|
return {"results": [], "total": 0, "page": page, "page_size": page_size}
|
|
|
|
where = []
|
|
params: List[Any] = []
|
|
if category:
|
|
where.append("category = ?")
|
|
params.append(category)
|
|
if province:
|
|
where.append("province = ?")
|
|
params.append(province)
|
|
if keyword:
|
|
where.append("name LIKE ?")
|
|
params.append(f"%{keyword}%")
|
|
|
|
where_clause = f"WHERE {' AND '.join(where)}" if where else ""
|
|
|
|
# 总数
|
|
total = self._conn.execute(
|
|
f"SELECT COUNT(*) FROM laws {where_clause}", params
|
|
).fetchone()[0]
|
|
|
|
# 分页
|
|
offset = (page - 1) * page_size
|
|
rows = self._conn.execute(
|
|
f"""
|
|
SELECT id, name, category, publish_date, province, city, clause_count, file_path
|
|
FROM laws {where_clause}
|
|
ORDER BY category, name
|
|
LIMIT ? OFFSET ?
|
|
""",
|
|
params + [page_size, offset],
|
|
).fetchall()
|
|
|
|
results = [
|
|
{
|
|
"law_id": row["id"],
|
|
"name": row["name"],
|
|
"category": row["category"],
|
|
"publish_date": row["publish_date"],
|
|
"province": row["province"],
|
|
"city": row["city"],
|
|
"clause_count": row["clause_count"],
|
|
"file_path": row["file_path"],
|
|
}
|
|
for row in rows
|
|
]
|
|
|
|
return {
|
|
"results": results,
|
|
"total": total,
|
|
"page": page,
|
|
"page_size": page_size,
|
|
}
|
|
|
|
def get_law(self, law_id: int) -> Optional[Dict[str, Any]]:
|
|
"""法规详情(含原文 + 条文列表)"""
|
|
self._ensure_conn()
|
|
if self._conn is None:
|
|
return None
|
|
|
|
row = self._conn.execute(
|
|
"SELECT * FROM laws WHERE id = ?", (law_id,)
|
|
).fetchone()
|
|
if row is None:
|
|
return None
|
|
|
|
# 读取原文(路径校验防穿越)
|
|
file_path = Path(row["file_path"])
|
|
try:
|
|
# 确保在 law-pack 目录内
|
|
abs_path = file_path.resolve()
|
|
pack_root = LAW_PACK_DIR.resolve()
|
|
if not str(abs_path).startswith(str(pack_root)):
|
|
logger.error(f"路径穿越尝试: {file_path}")
|
|
return None
|
|
content = abs_path.read_text(encoding="utf-8")
|
|
except Exception as e:
|
|
logger.error(f"读取原文失败 {file_path}: {e}")
|
|
content = f"[读取失败: {e}]"
|
|
|
|
# 条文列表
|
|
clauses = self._conn.execute(
|
|
"SELECT id, chapter, clause_no, content FROM clauses WHERE law_id = ? ORDER BY id",
|
|
(law_id,),
|
|
).fetchall()
|
|
clause_list = [
|
|
{
|
|
"clause_id": c["id"],
|
|
"chapter": c["chapter"],
|
|
"clause_no": c["clause_no"],
|
|
"content": c["content"],
|
|
}
|
|
for c in clauses
|
|
]
|
|
|
|
return {
|
|
"law_id": row["id"],
|
|
"name": row["name"],
|
|
"category": row["category"],
|
|
"publish_date": row["publish_date"],
|
|
"province": row["province"],
|
|
"city": row["city"],
|
|
"content": content,
|
|
"file_path": row["file_path"],
|
|
"clauses": clause_list,
|
|
}
|