fachuan / apps /client /services /text_parser.py
jun2024's picture
Upload 2241 files
f8b03c1 verified
Raw History Blame Contribute Delete
19.9 kB
"""当事人文本解析器。"""
from __future__ import annotations
import re
from typing import Any
# 关键字列表,用于智能分割无换行文本
_FIELD_KEYWORDS: list[str] = [
"名称",
"企业名称",
"公司名称",
"单位名称",
"姓名",
"类型",
"当事人",
"法定代表人",
"法人代表",
"负责人",
"统一社会信用代码",
"社会信用代码",
"信用代码",
"身份证号码",
"身份证号",
"身份证",
"证件号码",
"地址",
"注册地址",
"经营地址",
"联系地址",
"住址",
"住所地",
"住所",
"联系电话",
"电话",
"联系方式",
"手机",
]
# 角色标签模式(用于分割多当事人文本)
# 统一支持:
# - 带序号:被告一 / 被告1 / 甲方二
# - 带括号注释:原告(反诉被告)
# - 带冒号或仅空格分隔:原告:张三 / 原告 张三
_ROLE_LABELS: tuple[str, ...] = (
"再审被申请人",
"再审申请人",
"申请再审人",
"被申请复议人",
"申请复议人",
"原审被告",
"原审原告",
"被申请执行人",
"申请执行人",
"被上诉人",
"被申请人",
"被执行人",
"被答辩人",
"申请人",
"上诉人",
"答辩人",
"第三人",
"委托方",
"受托方",
"委托人",
"当事人",
"原告",
"被告",
"甲方",
"乙方",
"丙方",
"丁方",
)
_ROLE_LABELS_ALT = "|".join(re.escape(label) for label in _ROLE_LABELS)
_ROLE_SPLIT_STRS: list[str] = [
rf"(?:{re.escape(label)})\s*(?:[一二三四五六七八九十\d]+)?\s*(?:([^)]*)|\([^)]*\))?\s*(?:[::]\s*|\s+)"
for label in _ROLE_LABELS
]
_ROLE_SPLIT_PATTERNS: list[re.Pattern[str]] = [re.compile(p, re.IGNORECASE) for p in _ROLE_SPLIT_STRS]
# 角色标签 + 名称捕获模式(用于提取名称)
_ROLE_NAME_PATTERNS: list[re.Pattern[str]] = [re.compile(rf"{p}([^\n]+)", re.IGNORECASE) for p in _ROLE_SPLIT_STRS]
_ETHNICITY_PATTERN = re.compile(
r"[,,]\s*(?:男|女|汉族|回族|满族|蒙古族|维吾尔族|藏族|壮族|朝鲜族|苗族|瑶族|"
r"土家族|布依族|侗族|白族|哈尼族|哈萨克族|黎族|傣族|畲族|傈僳族|仡佬族|东乡族|"
r"高山族|拉祜族|水族|佤族|纳西族|羌族|土族|仫佬族|锡伯族|柯尔克孜族|达斡尔族|"
r"景颇族|毛南族|撒拉族|布朗族|塔吉克族|阿昌族|普米族|鄂温克族|怒族|京族|基诺族|"
r"德昂族|保安族|俄罗斯族|裕固族|乌孜别克族|门巴族|鄂伦春族|独龙族|塔塔尔族|"
r"赫哲族|珞巴族).*"
)
_BIRTH_DATE_PATTERN = re.compile(r"[,,]\s*\d{4}年\d{1,2}月\d{1,2}日.*")
_CREDIT_CODE_PATTERN = re.compile(
r"(?:统一社会信用代码|信用代码|社会信用代码)\s*(?:[::]|为|是)?\s*([A-Z0-9]{18})",
re.IGNORECASE,
)
_CREDIT_CODE_FALLBACK_PATTERN = re.compile(r"\b([A-Z0-9]{18})\b", re.IGNORECASE)
_ID_NUMBER_PATTERN = re.compile(
r"(?:身份证号码|身份证号|身份证|证件号码)\s*(?:[::]|为|是)?\s*([0-9Xx]{15,18})",
re.IGNORECASE,
)
_ID_NUMBER_FALLBACK_PATTERN = re.compile(r"\b([1-9]\d{16}[0-9Xx]|[1-9]\d{14})\b")
_ADDRESS_PATTERN = re.compile(
r"(?:注册地址|经营地址|联系地址|地址|住址|住所地|住所)\s*(?:[::]|为|是)?\s*([^\n;;。]*?)"
r"(?=\n|$|;|;|。|(?:联系电话|电话|联系方式|手机|法定代表人|法人代表|法定负责人|负责人)\s*(?:[::]|为|是)?)",
re.IGNORECASE,
)
_PHONE_PATTERN = re.compile(
r"(?:联系电话|电话|联系方式|手机|联系电话号码|联系人电话)\s*(?:[::]|为|是)?\s*([0-9\-\+\s]{7,20})",
re.IGNORECASE,
)
_PHONE_FALLBACK_PATTERN = re.compile(r"(?<!\d)(1[3-9]\d{9}|(?:0\d{2,3}-?)?\d{7,8})(?!\d)")
_LEGAL_REP_PATTERN = re.compile(
r"(?:法定代表人|法人代表|法定负责人|负责人)\s*(?:[::]|为|是)?\s*([^\n;;。]*?)"
r"(?=\n|$|;|;|。|(?:联系电话|电话|联系方式|手机|注册地址|经营地址|联系地址|地址|住址|住所地|住所)\s*(?:[::]|为|是)?)",
re.IGNORECASE,
)
_PAREN_CLEANUP_PATTERN = re.compile(r"([^)]*)|\([^)]*\)")
_WHITESPACE_PATTERN = re.compile(r"\s+")
_ROLE_NAME_FALLBACK_PATTERN = re.compile(
r"(?:^|\n)\s*"
rf"(?:{_ROLE_LABELS_ALT})"
r"\s*(?:[一二三四五六七八九十\d]+)?\s*(?:([^)]*)|\([^)]*\))?\s*(?:[::]\s*|\s+)([^\n,,;;。]+)",
re.IGNORECASE,
)
_ROLE_PREFIX_CLEANUP_PATTERN = re.compile(
rf"^(?:{_ROLE_LABELS_ALT})\s*(?:[一二三四五六七八九十\d]+)?\s*(?:([^)]*)|\([^)]*\))?\s*(?:[::]\s*|\s+)",
re.IGNORECASE,
)
_LEGAL_ENTITY_NAME_PATTERN = re.compile(
r"([\u4e00-\u9fa5A-Za-z0-9()()·]{4,}"
r"(?:有限责任公司|有限公司|股份有限公司|股份公司|集团有限公司|集团|合伙企业|律师事务所|研究院|银行|医院|学校|中心|店|厂))"
)
_NATURAL_PERSON_NAME_PATTERN = re.compile(r"([\u4e00-\u9fa5·]{2,10})\s*[,,]\s*(?:男|女)")
_LEADING_NAME_BEFORE_FIELD_PATTERN = re.compile(
r"^\s*([^\n;;。]{2,80}?)"
r"(?=\s*(?:统一社会信用代码|社会信用代码|信用代码|身份证号码|身份证号|身份证|证件号码|"
r"法定代表人|法人代表|法定负责人|负责人|联系电话|电话|联系方式|手机|联系电话号码|联系人电话|"
r"地址|住址|住所地|住所|注册地址|经营地址|联系地址))"
)
_LEADING_PERSON_NAME_PATTERN = re.compile(
r"^\s*([\u4e00-\u9fa5·]{2,10})"
r"(?=\s*(?:[,,;;]?\s*(?:男|女|出生|身份证号码|身份证号|身份证|证件号码|联系电话|电话|联系方式|手机|地址|住址|住所)))"
)
_ADDRESS_LINE_FALLBACK_PATTERN = re.compile(
r"^(?:中国)?(?:北京市|天津市|上海市|重庆市|[\u4e00-\u9fa5]{2,}(?:省|自治区|特别行政区))"
r"[\u4e00-\u9fa5A-Za-z0-9\-()()#号室栋单元路街道区县乡镇村]{4,}$"
)
_TRAILING_GENDER_PATTERN = re.compile(r"(?:\s|[,,])(?:男|女)\s*$")
_TRAILING_BIRTH_INFO_PATTERN = re.compile(r"(?:\s|[,,])\d{4}年\d{1,2}月\d{1,2}日(?:出生)?\s*$")
_ENUMERATION_PREFIX_PATTERN = re.compile(r"(?m)^\s*(?:[((]?[一二三四五六七八九十\d]+[))\.、]|[-*•])\s*")
_LEGAL_KEYWORDS: tuple[str, ...] = (
"有限公司",
"股份公司",
"集团",
"企业",
"厂",
"店",
"中心",
"协会",
"基金会",
"研究院",
"学校",
"医院",
"银行",
)
def parse_client_text(text: str) -> dict[str, Any]:
"""
解析当事人文本信息,支持有换行和无换行格式
"""
if not text or not text.strip():
return _empty_result()
# 预处理:在关键字前插入换行,方便后续解析
text = _normalize_text(text.strip())
# 尝试解析
parties = _extract_parties(text)
if not parties:
# 如果角色标签匹配失败,尝试直接提取字段
return _parse_fields_directly(text)
return parties[0]
def parse_multiple_clients_text(text: str) -> list[dict[str, Any]]:
"""解析包含多个当事人的文本"""
if not text or not text.strip():
return []
text = _normalize_text(text.strip())
return _extract_parties(text)
_FIELD_KEYWORDS_PATTERN = re.compile(r"(?<!\n)(" + "|".join(re.escape(kw) for kw in _FIELD_KEYWORDS) + r")\s*[::]")
_INLINE_BREAK_KEYWORDS_ALT = "|".join(
re.escape(keyword) for keyword in dict.fromkeys([*_FIELD_KEYWORDS, *_ROLE_LABELS])
)
_INLINE_BREAK_PATTERN = re.compile(rf"(?<!\n)(?<=\S)\s+(?=(?:{_INLINE_BREAK_KEYWORDS_ALT})\s*(?:[::]|为|是|\S))")
_FULL_STOP_BREAK_PATTERN = re.compile(r"[。]+(?=\s*\S)")
def _normalize_text(text: str) -> str:
"""预处理文本:统一换行、拆分内联字段、清理列表序号"""
text = text.replace("\r\n", "\n").replace("\r", "\n")
text = text.replace(";", "\n").replace(";", "\n")
text = _FULL_STOP_BREAK_PATTERN.sub("\n", text)
text = _ENUMERATION_PREFIX_PATTERN.sub("", text)
text = _INLINE_BREAK_PATTERN.sub("\n", text)
text = _FIELD_KEYWORDS_PATTERN.sub(r"\n\g<0>", text)
text = re.sub(r"\n{2,}", "\n", text)
return text.strip()
def _parse_fields_directly(text: str) -> dict[str, Any]:
"""直接从文本提取字段(无角色标签时使用)"""
return _parse_single_party(text, use_smart_name=True)
_SMART_NAME_PATTERN = re.compile(
r"^[甲乙丙丁]方\s*(?:([^)]*))?\s*[::]\s*(.+?)(?=法定代表人|统一社会信用代码|地址|电话|$)",
re.DOTALL,
)
# 支持字段式名称(保留冒号约束,避免过度吞并整段文本)
_NAME_FIELD_PATTERN = re.compile(
r"(?:^|\n)\s*(?:名称|企业名称|公司名称|单位名称|姓名|当事人名称)\s*(?:[::]|为|是)?\s*([^\n]+)",
re.IGNORECASE,
)
_NON_NAME_KEYWORDS: tuple[str, ...] = (
"统一社会信用代码",
"社会信用代码",
"信用代码",
"身份证",
"证件号码",
"法定代表人",
"法人代表",
"负责人",
"联系电话",
"联系方式",
"电话",
"手机",
"地址",
"住址",
"住所",
"注册地址",
"经营地址",
"联系地址",
)
def _extract_name_smart(text: str) -> str | None:
"""智能提取名称"""
name = _extract_name(text)
if name:
return name
# 支持 "名称: xxx" 格式
match = _NAME_FIELD_PATTERN.search(text)
if match:
name = _clean_name_candidate(match.group(1))
if _is_valid_name_candidate(name):
return name
match2 = _SMART_NAME_PATTERN.search(text)
if match2:
name = _clean_name_candidate(_WHITESPACE_PATTERN.sub("", match2.group(1).strip()))
if _is_valid_name_candidate(name):
return name
leading_name_match = _LEADING_NAME_BEFORE_FIELD_PATTERN.search(text)
if leading_name_match:
name = _clean_name_candidate(leading_name_match.group(1))
if _is_valid_name_candidate(name):
return name
# 兼容无冒号的角色写法:如 "被告 张三,男..."
role_match = _ROLE_NAME_FALLBACK_PATTERN.search(text)
if role_match:
name = _clean_name_candidate(role_match.group(1))
if _is_valid_name_candidate(name):
return name
leading_person_match = _LEADING_PERSON_NAME_PATTERN.search(text)
if leading_person_match:
name = _clean_name_candidate(leading_person_match.group(1))
if _is_valid_name_candidate(name):
return name
# 兼容仅出现公司全称的文本
legal_name_match = _LEGAL_ENTITY_NAME_PATTERN.search(text)
if legal_name_match:
name = _clean_name_candidate(legal_name_match.group(1))
if _is_valid_name_candidate(name):
return name
# 兼容自然人常见写法:如 "李四,男..."
person_match = _NATURAL_PERSON_NAME_PATTERN.search(text)
if person_match:
name = _clean_name_candidate(person_match.group(1))
if _is_valid_name_candidate(name):
return name
# 兜底:第一行即名称
first_line_name = _extract_name_from_first_meaningful_line(text)
if first_line_name:
return first_line_name
return None
def _clean_name_candidate(name_part: str) -> str:
"""清洗名称候选值(保留公司括号地名,不做激进删减)"""
name = _ROLE_PREFIX_CLEANUP_PATTERN.sub("", name_part.strip())
name = _ETHNICITY_PATTERN.sub("", name)
name = _BIRTH_DATE_PATTERN.sub("", name)
name = _TRAILING_BIRTH_INFO_PATTERN.sub("", name)
name = _TRAILING_GENDER_PATTERN.sub("", name)
return name.strip().strip(",,;;。")
def _is_valid_name_candidate(name: str) -> bool:
"""判断名称候选值是否有效,避免把字段值误判为名称"""
if not name or len(name) < 2 or len(name) > 120:
return False
if any(keyword in name for keyword in _NON_NAME_KEYWORDS):
return False
if any(keyword.startswith(name) for keyword in _NON_NAME_KEYWORDS):
return False
if any(role.startswith(name) for role in _ROLE_LABELS):
return False
compact = _WHITESPACE_PATTERN.sub("", name)
if not compact:
return False
if compact.isdigit():
return False
if _ID_NUMBER_FALLBACK_PATTERN.fullmatch(compact):
return False
if _CREDIT_CODE_FALLBACK_PATTERN.fullmatch(compact) and any(ch.isalpha() for ch in compact):
return False
return True
def _extract_name_from_first_meaningful_line(text: str) -> str | None:
"""从第一条有效信息行提取名称"""
for raw_line in text.splitlines():
line = raw_line.strip()
if not line:
continue
line = _clean_name_candidate(line)
if _is_valid_name_candidate(line):
return line
return None
def _extract_parties(text: str) -> list[dict[str, Any]]:
"""提取所有当事人信息"""
parties = []
# 定义角色标签模式(支持 甲方(原告)、乙方(被告)等格式)
role_patterns = _ROLE_SPLIT_PATTERNS
# 找到所有角色标签的位置
all_matches: list[re.Match[str]] = []
for compiled in role_patterns:
all_matches.extend(compiled.finditer(text))
# 按位置排序,去除同一起始位置的重复匹配(保留最长匹配)
all_matches.sort(key=lambda x: (x.start(), -(x.end() - x.start())))
seen_starts: set[int] = set()
deduped: list[re.Match[str]] = []
for m in all_matches:
if m.start() not in seen_starts:
seen_starts.add(m.start())
deduped.append(m)
all_matches = deduped
# 提取每个当事人的信息
for i, match in enumerate(all_matches):
start_pos = match.end()
# 确定当事人信息的结束位置
if i + 1 < len(all_matches):
end_pos = all_matches[i + 1].start()
else:
end_pos = len(text)
# 提取当事人信息段落
party_text = text[start_pos:end_pos].strip()
if party_text:
# 重新构造完整的当事人信息(包含角色标签)
role_label = text[match.start() : match.end()]
full_party_text = role_label + party_text
party_info = _parse_single_party(full_party_text, use_smart_name=True)
if party_info["name"]: # 只有名称不为空才添加
parties.append(party_info)
# 如果没有找到角色标签,尝试直接解析
if not parties:
party_info = _parse_single_party(text, use_smart_name=True)
if party_info["name"]:
parties.append(party_info)
return parties
def _parse_single_party(text: str, *, use_smart_name: bool = False) -> dict[str, Any]:
"""解析单个当事人信息"""
result = _empty_result()
name = _extract_name_smart(text) if use_smart_name else _extract_name(text)
if name:
result["name"] = name
result["client_type"] = _determine_client_type(name, text)
# 提取统一社会信用代码
credit_code = _extract_credit_code(text)
if credit_code:
result["id_number"] = credit_code
result["client_type"] = "legal" # 有统一社会信用代码的是法人
# 提取身份证号码
if not result["id_number"]:
id_number = _extract_id_number(text)
if id_number:
result["id_number"] = id_number
result["client_type"] = "natural" # 有身份证号的是自然人
# 提取地址
address = _extract_address(text)
if address:
result["address"] = address
# 提取电话
phone = _extract_phone(text)
if phone:
result["phone"] = phone
# 提取法定代表人
legal_rep = _extract_legal_representative(text)
if legal_rep:
result["legal_representative"] = legal_rep
if result["client_type"] == "natural":
result["client_type"] = "legal" # 有法定代表人的是法人
return result
def _extract_name(text: str) -> str | None:
"""提取名称"""
# 定义角色标签模式(支持 甲方(原告)、乙方(被告)等格式)
role_patterns = _ROLE_NAME_PATTERNS
for compiled in role_patterns:
match = compiled.search(text)
if match:
name = _clean_name_candidate(match.group(1))
if _is_valid_name_candidate(name):
return name.strip()
return None
def _extract_credit_code(text: str) -> str | None:
"""提取统一社会信用代码"""
match = _CREDIT_CODE_PATTERN.search(text)
if match:
return match.group(1).strip().upper()
# 无标签兜底:仅接受含字母的18位编码,避免误判身份证号
# 但需要排除明显是身份证号的情况(文本中有"身份证"关键词)
for fallback_match in _CREDIT_CODE_FALLBACK_PATTERN.finditer(text):
code = fallback_match.group(1).strip().upper()
if not any(ch.isalpha() for ch in code):
continue
# 排除身份证号:如果编码附近有"身份证"关键词,跳过
# 计算编码在文本中的位置
start_pos = fallback_match.start()
end_pos = fallback_match.end()
# 检查前面 20 个字符内是否有"身份证"
context_before = text[max(0, start_pos - 20) : start_pos].lower()
if "身份证" in context_before:
continue
return code
return None
def _extract_id_number(text: str) -> str | None:
"""提取身份证号码"""
match = _ID_NUMBER_PATTERN.search(text)
if match:
return match.group(1).strip().upper()
fallback = _ID_NUMBER_FALLBACK_PATTERN.search(text)
return fallback.group(1).strip().upper() if fallback else None
def _extract_address(text: str) -> str | None:
"""提取地址"""
match = _ADDRESS_PATTERN.search(text)
if match:
address = match.group(1).strip()
if address:
return address
# 无标签兜底:从明显的省市开头行提取
for raw_line in text.splitlines():
line = raw_line.strip()
if line and _ADDRESS_LINE_FALLBACK_PATTERN.fullmatch(line):
return line
return None
def _extract_phone(text: str) -> str | None:
"""提取电话号码"""
match = _PHONE_PATTERN.search(text)
if match:
phone = _WHITESPACE_PATTERN.sub("", match.group(1).strip())
if phone:
return phone
# 无标签兜底:优先手机号,其次座机
fallback_phone: str | None = None
for fallback in _PHONE_FALLBACK_PATTERN.finditer(text):
candidate = _WHITESPACE_PATTERN.sub("", fallback.group(1).strip())
digits_only = re.sub(r"\D", "", candidate)
if len(digits_only) == 11 and digits_only.startswith("1"):
return candidate
if fallback_phone is None:
fallback_phone = candidate
return fallback_phone
def _extract_legal_representative(text: str) -> str | None:
"""提取法定代表人"""
match = _LEGAL_REP_PATTERN.search(text)
if not match:
return None
legal_rep = _clean_name_candidate(match.group(1))
return legal_rep or None
def _determine_client_type(name: str, text: str) -> str:
"""根据名称和文本内容判断客户类型"""
if _extract_credit_code(text):
return "legal"
if _extract_legal_representative(text):
return "legal"
if any(kw in name for kw in _LEGAL_KEYWORDS):
return "legal"
if _extract_id_number(text):
return "natural"
return "natural"
def _empty_result() -> dict[str, Any]:
"""返回空的解析结果"""
return {
"name": "",
"phone": "",
"address": "",
"client_type": "natural",
"id_number": "",
"legal_representative": "",
}