From 2f3830bd5d8590e147947d2af434dbac94934327 Mon Sep 17 00:00:00 2001 From: sansen Date: Fri, 14 Aug 2026 17:06:10 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E6=89=B9=E9=87=8F=E6=8E=A5?= =?UTF-8?q?=E5=8F=A3=EF=BC=8C=E5=AE=8C=E5=96=84=E6=97=A5=E6=9C=9F=E8=A7=A3?= =?UTF-8?q?=E6=9E=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- client_app.py | 75 +++++++++++++++++- era_data.py | 10 +++ parser.py | 205 ++++++++++++++++++++++++++++++-------------------- 3 files changed, 203 insertions(+), 87 deletions(-) diff --git a/client_app.py b/client_app.py index 9ea334f..c81363d 100644 --- a/client_app.py +++ b/client_app.py @@ -1,5 +1,6 @@ import json -from fastapi import FastAPI, HTTPException, Header, Form, Request +from time import perf_counter +from fastapi import FastAPI, HTTPException, Header, Request from fastapi.responses import JSONResponse from fastapi.middleware.cors import CORSMiddleware from loguru import logger @@ -16,21 +17,36 @@ app.add_middleware( allow_headers=["*"], ) +# batch 接口单次请求的条目上限（防滥用） +BATCH_MAX_ITEMS = 1000 + + +def _check_auth(Authorization: str): + if Authorization != "3FFA6A4B073CF065969630692331A873": + raise HTTPException(status_code=403, detail="未授权的操作") + + +def _parse_urlencoded_content(body: bytes) -> str: + """手动解析 x-www-form-urlencoded 的 content 字段：绕过 starlette 1MB 表单限制""" + from urllib.parse import parse_qs + params = parse_qs(body.decode("utf-8"), keep_blank_values=True) + return params.get("content", [""])[0] + @app.post("/convert/start") async def convert_str( request: Request, - content: str = Form(...), # ✅ 改为接收 x-www-form-urlencoded 参数 Authorization: str = Header(None) ): """ 带调试信息的转换接口 (支持 x-www-form-urlencoded) + 手动解析表单体：绕过 starlette 1MB 表单限制（族谱全文可达数 MB） """ - if Authorization != "3FFA6A4B073CF065969630692331A873": - raise HTTPException(status_code=403, detail="未授权的操作") + _check_auth(Authorization) try: + content = _parse_urlencoded_content(await request.body()) res = convert_text(content) logger.info(f"转换结果: {json.dumps(res, ensure_ascii=False, indent=2)}") return JSONResponse(content={"code": 0, "result": res}) @@ -42,6 +58,57 @@ async def convert_str( ) +@app.post("/convert/batch") +async def convert_batch( + request: Request, + Authorization: str = Header(None) +): + """ + 批量转换接口：短文本多任务场景，一次请求处理 N 条 content。 + 请求: {"items": [{"id": "row1", "content": "..."}, ...]}（application/json） + 响应: {"code": 0, "result": [{"id": "row1", "result": [...]}, ...]} + - id 透传用于前端回填，缺省按数组顺序编号 + - 单条解析失败不影响其他条目（该条返回 error 字段） + - 每条 content 独立解析，位置语义与单接口一致（从 0 起） + """ + _check_auth(Authorization) + + # 请求体解析与入参校验失败 → 400（客户端错误） + try: + payload = json.loads((await request.body()).decode("utf-8")) + except Exception as e: + logger.warning(f"batch 请求体解析失败: {str(e)}") + raise HTTPException(status_code=400, detail="请求体必须是合法 JSON") + items = payload.get("items") if isinstance(payload, dict) else None + if not isinstance(items, list) or not items: + raise HTTPException(status_code=400, detail="items 必须是非空数组") + if len(items) > BATCH_MAX_ITEMS: + raise HTTPException(status_code=400, detail=f"items 超过上限 {BATCH_MAX_ITEMS} 条") + + try: + t0 = perf_counter() + results = [] + for i, item in enumerate(items): + item_id = item.get("id") if isinstance(item, dict) else None + content = item.get("content", "") if isinstance(item, dict) else "" + if item_id is None: + item_id = str(i) + try: + results.append({"id": item_id, "result": convert_text(content)}) + except Exception as e: + logger.error(f"batch 条目 {item_id} 解析异常: {str(e)}") + results.append({"id": item_id, "error": str(e)}) + dt = perf_counter() - t0 + logger.info(f"批量转换: {len(results)} 条, 耗时 {dt * 1000:.0f}ms") + return JSONResponse(content={"code": 0, "result": results}) + except Exception as e: + logger.error(f"批量转换异常: {str(e)}") + return JSONResponse( + content={"code": 500, "result": f"批量转换服务异常: {str(e)}"}, + status_code=500 + ) + + if __name__ == "__main__": import uvicorn diff --git a/era_data.py b/era_data.py index 8c874fa..328206e 100644 --- a/era_data.py +++ b/era_data.py @@ -47,6 +47,16 @@ TEMPLE_ALIASES = { "始皇帝": "始皇", # CSV 用"始皇帝"，正文常用"秦始皇" } +# === 年号别名（正文异写/避讳/简称 → CSV 规范写法） === +# 匹配后统一归一为规范年号再查表；别名本身同时加入白名单（KNOWN_ERAS_ALL） +ERA_ALIASES = { + "崇正": "崇祯", # 避"祯"字讳（清代文献常见） + "宏治": "弘治", # 避"弘"字讳（避乾隆帝弘历名） + "宣得": "宣德", # 同音误写（明宣德，族谱常见） + "大得": "大德", # 同音误写（元大德） + "靖国": "建中靖国", # 简称（北宋建中靖国，元年即 1101） +} + # === 常见年号白名单（优先匹配，避免正则贪心误匹配） === KNOWN_ERAS = { # 秦汉 diff --git a/parser.py b/parser.py index b6fe700..efc9f17 100644 --- a/parser.py +++ b/parser.py @@ -3,7 +3,7 @@ import re import pandas as pd from opencc import OpenCC -from era_data import dynasty_alias, KNOWN_ERAS, VARIANT_MAP, TEMPLE_ALIASES +from era_data import dynasty_alias, KNOWN_ERAS, VARIANT_MAP, TEMPLE_ALIASES, ERA_ALIASES cc = OpenCC('t2s') @@ -119,13 +119,18 @@ for _, row in era_df.iterrows(): if span is not None: era_span.setdefault(dynasty, {}).setdefault(era, span) -# 有效年号白名单 = 手工精选 ∪ CSV 年号（自动补全：白名单不应小于数据表）。 +# 有效年号白名单 = 手工精选 ∪ CSV 年号 ∪ 年号别名键（自动补全：白名单不应小于数据表）。 # 排除：民国（有专门匹配段）、带括号的年号（如"(后元)"，防止与东汉"中元"等歧义） _csv_eras = { e for e in era_df["年号"] if e and e != '民国' and '(' not in e and '（' not in e and 2 <= len(e) <= 8 } -KNOWN_ERAS_ALL = KNOWN_ERAS | _csv_eras +KNOWN_ERAS_ALL = KNOWN_ERAS | _csv_eras | set(ERA_ALIASES) + +def _norm_era(era): + """年号归一：OpenCC 繁转简后套年号别名（崇正→崇祯），返回规范写法""" + era = cc.convert(era) + return ERA_ALIASES.get(era, era) # === 庙号表 === @@ -256,70 +261,40 @@ _patterns_no_dynasty = _build_pattern(with_dynasty=False) _patterns_dynasty = _build_pattern(with_dynasty=True) -# === 农历月日（干支纪年后的"十一月十四""闰六月廿一日"） === -_month_special = {"正": 1, "冬": 11, "腊": 12} - -def _parse_month(text): - """月文本转数字：正月=1、冬月=11、腊月=12，其余走中文数字（十一月→11）""" - return _month_special.get(text, chinese_to_num(text)) - -def _parse_day(text): - """日文本转数字：剥"初"前缀（初八→8、初十→10），廿三→23""" - if text.startswith("初"): - text = text[1:] - return chinese_to_num(text) - -_month_day_re = re.compile( - r'(?P闰)?(?P[正冬腊〇零一二三四五六七八九十廿卄卅卌]{1,3})月' - r'(?:(?P[初〇零一二三四五六七八九十廿卄卅卌]{1,3})(?:日)?)?' -) -# 嵌入干支正则的可选月日片段（整体捕获，事后用 _month_day_re 解析） -_md_body = r'(?:闰)?[正冬腊〇零一二三四五六七八九十廿卄卅卌]{1,3}月' \ - r'(?:[初〇零一二三四五六七八九十廿卄卅卌]{1,3}(?:日)?)?' -# 族谱常见的生死动词：月日与（公元对照）之间可能隔着"生/殁"等单字 -_md_verb = r'[生死殁卒薨殓]' - -def _attach_month_day(result, m): - """把匹配中的月日组（md1/md2/md，括号前/后位置）并入结果字段""" - md = None - for name in ("md1", "md2", "md"): - if name in m.groupdict() and m.group(name): - md = m.group(name) - break - if not md: - return - mm = _month_day_re.match(md) - if not mm: - return - result["月"] = _parse_month(mm.group("month")) - if mm.group("day"): - result["日"] = _parse_day(mm.group("day")) - result["月日"] = md - - # === 2c. 干支+（公元对照）（如"清同治壬申（1872）十一月十四生"） === -# 干支纪年旁直接给出公元年份：合并为一条结果、对照校验干支、紧跟的月日并入字段。 -# 置于公元段之前（见 extract_era_years），避免"（1872）"被单独摘出成公元纪年。 +# 干支纪年旁直接给出公元年份：合并为一条结果、对照校验干支。 +# 除括号外还支持："公元"标记（清同治壬申公元1872年）、横线（壬申-1872）、 +# 无分隔连写（壬申1872年）。其余内容（如"十一月十四"）不并入原文。 +# 置于公元段之前（见 extract_era_years），避免对照年份被单独摘出成公元纪年。 _dynasty_opt = (r'(?P(?:(?[〇零○O一二三四五六七八九\d０-９]{3,4})\s*年?[)）]' +# 公元对照年份：阿拉伯/全角/中文数字 3-4 位，可带"年"字 +_gz_gy_year = r'[〇零○O一二三四五六七八九\d０-９]{3,4}' +# 干支与对照年份之间的连接方式（按优先级）： +# 括号（1872）/ 公元标记（，公元1872年）/ 横线（-1872）/ 无分隔连写（壬申1872年） +# 空白只允许空格/制表符（[ \t]*），不允许换行——行尾的 \n 不得被吞进原文 +# 公元标记支持 公元/西元/公历 三种写法 +_gz_gy_sep = ( + r'(?:' + r'(?:[（(][ \t]*(?P' + _gz_gy_year + r')[ \t]*年?[)）])' + r'|(?:[，,、]?[ \t]*(?:公元|西元|公历)[ \t]*(?P' + _gz_gy_year + r')[ \t]*年?)' + r'|(?:[-—–－~][ \t]*(?P' + _gz_gy_year + r')[ \t]*年?)' + r'|(?P' + _gz_gy_year + r')[ \t]*年?' + r')' +) +# 干支前的修饰词（如"光绪岁次辛丑""岁在壬戌"）：碑刻/谱牒常见 +_gz_mod = r'(?:岁[次在])?' _ganzhi_gongyuan_re = re.compile( _dynasty_opt + _gz_era_group + + _gz_mod + r'(?P' + _gz_pair + r')(?:年)?' - + r'(?P' + _md_body + r')?' - + r'(?:' + _md_verb + r')?' - + _gz_gy_paren - + r'(?P' + _md_body + r')?' + + _gz_gy_sep ) # 裸干支+（对照）：无年号可锚定（如"壬申（1872）"），公元取对照值 _ganzhi_bare_gongyuan_re = re.compile( r'(?P' + _gz_pair + r')(?:年)?' - + r'(?P' + _md_body + r')?' - + r'(?:' + _md_verb + r')?' - + _gz_gy_paren - + r'(?P' + _md_body + r')?' + + _gz_gy_sep ) @@ -342,28 +317,66 @@ def extract_era_years(text): # 覆盖集：先匹配的段落占据位置，后续段落跳过重叠（防止同一文本被多段重复摘出） covered = set() + # 民国纪年的公元对照仅支持括号形式（"民国三十八年（1949）"）： + # 民国X年已是确定年份，逗号/横线/连写分隔多为并列句（如"民国三十八年，公元二〇二四年"是两个日期），不合并 + _mg_gy_paren = r'(?:[（(][ \t]*(?P' + _gz_gy_year + r')[ \t]*年?[)）])?' + # === 1. 民国 === - minguo_pattern = re.compile(r'民[国國]\s*([〇零一二三四五六七八九十百廿卄卅卌元\d０-９壹貳贰叁肆伍陸陆柒捌玖拾]+)\s*年') + minguo_pattern = re.compile( + r'民[国國]\s*([〇零一二三四五六七八九十百廿卄卅卌元\d０-９壹貳贰叁肆伍陸陆柒捌玖拾]+)\s*年' + + _mg_gy_paren + ) for m in minguo_pattern.finditer(simplified_text): year_text = m.group(1) year_num = chinese_to_num(year_text) + # 公元对照（如"民国三十八年（1949）"）：取对照值为公元，不一致时标注 + gy_year = None + for name in ("gy_year", "gy_year2", "gy_year3", "gy_year4"): + if name in m.groupdict() and m.group(name): + gy_year = chinese_to_num(m.group(name)) + break + if gy_year is not None and not (1000 <= gy_year <= 2100): + continue start, end = m.start(), m.end() + # 尾随干支校验（如"民国三十八年己丑"）：并入原文同一单位（同公元段逻辑） + ganzhi = simplified_text[end:end + 2] + if ganzhi in _ganzhi_to_offset: + end += 2 + else: + j = end + while j < len(simplified_text) and simplified_text[j] in '[]【】': + j += 1 + ganzhi = simplified_text[j:j + 2] + if ganzhi not in _ganzhi_to_offset: + ganzhi = None covered.update(range(start, end)) raw_fan = original_text[start:end] - results.append({ + gregorian = gy_year if gy_year is not None else 1911 + year_num + result = { "原文": raw_fan, - "原文(简体)": m.group(0), + "原文(简体)": simplified_text[start:end], "朝代": "民国", "年号": "民国纪年", "年数": year_num, - "公元": 1911 + year_num, - "公元中文": f"{arabic_to_chinese_year(1911 + year_num)}年", + "公元": gregorian, + "公元中文": f"{arabic_to_chinese_year(gregorian)}年", "位置": {"起始": start, "结束": end}, "类型": "民国纪年", - }) + } + if ganzhi: + result["干支"] = ganzhi + if gregorian_to_ganzhi(gregorian) != ganzhi: + result["干支不符"] = True + if gy_year is not None and gy_year != 1911 + year_num: + result["干支不符"] = True + result["干支推算"] = 1911 + year_num + results.append(result) - # 1b. 民国+干支纪年（如"民国丁卯""民国甲戌"） - minguo_gz = re.compile(r'民[国國]\s*(?P' + _gz_pair + r')') + # 1b. 民国+干支纪年（如"民国丁卯""民国甲戌"），支持（公元对照）合并 + minguo_gz = re.compile( + r'民[国國]\s*(?P' + _gz_pair + r')(?:年)?' + + _mg_gy_paren + ) for m in minguo_gz.finditer(simplified_text): ganzhi = m.group("ganzhi") gz_offset = _ganzhi_to_offset.get(ganzhi, -1) @@ -372,22 +385,34 @@ def extract_era_years(text): # 民国元年=1912, base_gz = (1912-4)%60 = 8 (己酉) base_gz = (1912 - 4) % 60 year_in_era = ((gz_offset - base_gz) % 60) + 1 - gregorian = 1912 + year_in_era - 1 + computed = 1912 + year_in_era - 1 + # 公元对照（如"民国壬申（1932）"）：取对照值为公元，不一致时标注 + gy_year = None + for name in ("gy_year", "gy_year2", "gy_year3", "gy_year4"): + if name in m.groupdict() and m.group(name): + gy_year = chinese_to_num(m.group(name)) + break + if gy_year is not None and not (1000 <= gy_year <= 2100): + continue start, end = m.start(), m.end() covered.update(range(start, end)) raw_fan = original_text[start:end] - results.append({ + result = { "原文": raw_fan, "原文(简体)": m.group(0), "朝代": "民国", "年号": "民国纪年", "年数": year_in_era, "干支": ganzhi, - "公元": gregorian, - "公元中文": f"{arabic_to_chinese_year(gregorian)}年", + "公元": gy_year if gy_year is not None else computed, + "公元中文": f"{arabic_to_chinese_year(gy_year if gy_year is not None else computed)}年", "位置": {"起始": start, "结束": end}, "类型": "民国纪年（干支）", - }) + } + if gy_year is not None and gy_year != computed: + result["干支不符"] = True + result["干支推算"] = computed + results.append(result) # === 2b. 年号+公元（如"清康熙公元一六八七年丁卯"） === # 年号后直接给出公元年份：年数=公元-年号起始+1，与"年号+年数"公式自洽； @@ -395,14 +420,14 @@ def extract_era_years(text): _era_gongyuan = re.compile( r'(?P(?:(?' + _known_era_alt + r')' - + r'公元\s*(?P[〇零○O一二三四五六七八九\d０-９]{3,4})\s*年?' + + r'(?:公元|西元|公历)\s*(?P[〇零○O一二三四五六七八九\d０-９]{3,4})\s*年?' + r'(?P' + _gz_pair + r')?' ) for m in _era_gongyuan.finditer(simplified_text): start, end = m.start(), m.end() if any(pos in covered for pos in range(start, end)): continue - era = cc.convert(m.group("era")) + era = _norm_era(m.group("era")) year_num = chinese_to_num(m.group("gy_year")) if not (1000 <= year_num <= 2100): continue @@ -458,9 +483,14 @@ def extract_era_years(text): def _process_ganzhi_gongyuan(m): raw_dynasty = m.group("dynasty") if "dynasty" in m.groupdict() else "" dynasty = normalize_dynasty_name(raw_dynasty or "") - era = cc.convert(m.group("era")) if "era" in m.groupdict() and m.group("era") else None + era = _norm_era(m.group("era")) if "era" in m.groupdict() and m.group("era") else None ganzhi = m.group("ganzhi") - gy_year = chinese_to_num(m.group("gy_year")) + # 对照年份可能来自四种连接方式之一（gy_year/gy_year2/gy_year3/gy_year4） + gy_year = next( + chinese_to_num(m.group(name)) + for name in ("gy_year", "gy_year2", "gy_year3", "gy_year4") + if m.group(name) + ) if ganzhi not in _ganzhi_to_offset or not (1000 <= gy_year <= 2100): return start, end = m.start(), m.end() @@ -512,7 +542,6 @@ def extract_era_years(text): elif gregorian_to_ganzhi(gy_year) != ganzhi: # 裸干支+对照：无年号可锚定，直接用公元年份的干支校验（同公元段约定） result["干支不符"] = True - _attach_month_day(result, m) results.append(result) for m in _ganzhi_gongyuan_re.finditer(simplified_text): @@ -522,13 +551,13 @@ def extract_era_years(text): # === 2. 公元（允许无"年"字）=== gongyuan_pattern = re.compile( - r'(公元)?\s*([〇零○O一二三四五六七八九\d０-９]{3,4})\s*(?:（.*?）)?\s*(年)?' + r'(?:公元|西元|公历)?\s*([〇零○O一二三四五六七八九\d０-９]{3,4})\s*(?:（.*?）)?\s*(年)?' ) for m in gongyuan_pattern.finditer(simplified_text): start, end = m.start(), m.end() if any(pos in covered for pos in range(start, end)): continue - year_text = m.group(2) + year_text = m.group(1) year_num = chinese_to_num(year_text) if not (1000 <= year_num <= 2100): continue @@ -567,7 +596,7 @@ def extract_era_years(text): """处理干支纪年匹配""" raw_dynasty = m.group("dynasty") if "dynasty" in m.groupdict() else "" dynasty = normalize_dynasty_name(raw_dynasty or "") - era = cc.convert(m.group("era")) + era = _norm_era(m.group("era")) ganzhi = m.group("ganzhi") if ganzhi not in _ganzhi_to_offset: return @@ -622,16 +651,14 @@ def extract_era_years(text): span = era_span.get(dynasty, {}).get(era) if span is not None and gregorian > base + span - 1: result["超出使用期"] = True - _attach_month_day(result, m) results.append(result) # 模式A：有朝代前缀（如"清雍正乙巳"）；年号组 _gz_era_group 见模块级定义 ganzhi_dynasty = re.compile( r'(?P(?:(?' + _gz_pair + r')(?:年)?' - # 紧跟的农历月日并入同一单位（如"清雍正乙巳十一月十二午时生"） - + r'(?P' + _md_body + r')?' ) for m in ganzhi_dynasty.finditer(simplified_text): _process_ganzhi_match(m) @@ -642,8 +669,8 @@ def extract_era_years(text): ganzhi_no_dynasty = re.compile( r'(?P)' + r'(?P' + _long_era_alt + r')' + + _gz_mod + r'(?P' + _gz_pair + r')(?:年)?' - + r'(?P' + _md_body + r')?' ) for m in ganzhi_no_dynasty.finditer(simplified_text): _process_ganzhi_match(m) @@ -697,7 +724,7 @@ def extract_era_years(text): """处理一个年号正则匹配结果""" raw_dynasty = m.group("dynasty") if "dynasty" in m.groupdict() else "" dynasty = normalize_dynasty_name(raw_dynasty or "") - era = cc.convert(m.group("era")) + era = _norm_era(m.group("era")) # 过滤误匹配：年号是单个数字字符（如"民国二十六年"中的"二"） if len(era) == 1 and era in '零一二三四五六七八九十廿卄卅卌': @@ -738,13 +765,21 @@ def extract_era_years(text): break start, end = m.start(), m.end() + # 尾随干支校验（如"光绪二十六年庚子"）：并入原文同一单位（同公元段逻辑） + ganzhi = None + if base: + ganzhi = simplified_text[end:end + 2] + if ganzhi in _ganzhi_to_offset: + end += 2 + else: + ganzhi = None raw_fan = original_text[start:end] covered.update(range(start, end)) if base: gregorian = base + year_num - 1 result = { "原文": raw_fan, - "原文(简体)": m.group(0), + "原文(简体)": m.group(0) + (ganzhi or ''), "朝代": dynasty, "年号": era, "年数": year_num, @@ -753,6 +788,10 @@ def extract_era_years(text): "位置": {"起始": start, "结束": end}, "类型": "古代纪年", } + if ganzhi: + result["干支"] = ganzhi + if gregorian_to_ganzhi(gregorian) != ganzhi: + result["干支不符"] = True if len(multi) > 1: result["多候选"] = True result["候选"] = [list(c) for c in multi]