from __future__ import annotations import json import zipfile from io import BytesIO from govcrawler.adapters import flk_npc class _Resp: status_code = 200 def __init__(self, payload: dict, content: bytes | None = None): self._payload = payload self.text = json.dumps(payload, ensure_ascii=False) self.content = content if content is not None else self.text.encode("utf-8") def json(self): return self._payload def raise_for_status(self): pass def _docx_bytes(paragraphs: list[str]) -> bytes: escaped = [ p.replace("&", "&").replace("<", "<").replace(">", ">") for p in paragraphs ] body = "".join(f"{p}" for p in escaped) xml = ( '' '' f"{body}" "" ) buf = BytesIO() with zipfile.ZipFile(buf, "w") as zf: zf.writestr("word/document.xml", xml) return buf.getvalue() def test_fetch_list_page_maps_law_fields(monkeypatch): seen = {} payload = { "total": 725, "rows": [ { "bbbs": "fb60d09eadb5483aa6e933e478b15a71", "title": "中华人民共和国监狱法", "gbrq": "2026-04-30", "sxrq": "2026-11-01", "sxx": 4, "zdjgName": "全国人民代表大会常务委员会", "flxz": "法律", "zdjgCodeId": 110, "flfgCodeId": 130, } ], "code": 200, "msg": "查询成功", } def _fake_post(url, *, json=None, headers=None, **_kw): seen["url"] = url seen["json"] = json seen["headers"] = headers return _Resp(payload) monkeypatch.setattr(flk_npc.httpx, "post", _fake_post) _list_url, items, fr = flk_npc.fetch_list_page( None, page_num=1, params={"page_size": 20, "flfg_code_ids": [110, 120, 130]}, ) assert fr.status == 200 assert seen["url"] == flk_npc.LIST_URL assert seen["json"]["flfgCodeId"] == [110, 120, 130] assert len(items) == 1 item = items[0] assert item.native_post_id == "fb60d09eadb5483aa6e933e478b15a71" assert item.title == "中华人民共和国监狱法" assert item.publisher == "全国人民代表大会常务委员会" assert item.publish_date.isoformat() == "2026-04-30" assert item.effective_date.isoformat() == "2026-11-01" assert item.is_effective is False assert item.content_category == "法律" assert item.content_subcategory == "130" assert item.metadata_json["flfgCodeId"] == 130 assert item.metadata_json["validity_state_code"] == 4 assert item.metadata_json["validity_state_label"] == "尚未生效" assert item.metadata_json["publish_date"] == "2026-04-30" assert item.metadata_json["effective_date"] == "2026-11-01" def test_modified_validity_state_is_still_effective(): assert flk_npc._validity_state_label(2) == "已修改" assert flk_npc._is_effective(2) is True def test_fetch_detail_maps_metadata_and_ofd_text(monkeypatch): detail_payload = { "msg": "操作成功", "code": 200, "data": { "bbbs": "fb60d09eadb5483aa6e933e478b15a71", "ossFile": { "ossWordPath": "prod/20260430/a.docx", "ossWordOfdPath": "prod/20260430/a.ofd", "ossPdfPath": "prod/20260430/a.pdf", "ossPdfOfdPath": None, }, "flxz": "法律", "zdjgName": "全国人民代表大会常务委员会", "gbrq": "2026-04-30", "sxrq": "2026-11-01", "title": "中华人民共和国监狱法", "sxx": 4, }, } preview_payload = { "msg": "获取预览链接成功！", "code": 200, "data": { "url": ( "https://flkofd.npc.gov.cn/reader?" "file=http://172.16.220.27:38080/law-search/amazonFile/" "ofdGenerateLink?filePath=prod/20260430/a.ofd" ) }, } info_payload = {"area": [[210, 297, 0], [210, 297, 0]]} page_payloads = [ {"areas": [{"lines": [{"chars": [{"char": "第"}, {"char": "一"}, {"char": "条"}]}]}]}, {"areas": [{"lines": [{"chars": [{"char": "监"}, {"char": "狱"}]}]}]}, ] calls = {"text": 0} def _fake_get(url, *, params=None, **_kw): if url == flk_npc.DETAIL_URL: return _Resp(detail_payload) if url == flk_npc.DOWNLOAD_PC_URL: fmt = params["format"] return _Resp( { "msg": "操作成功", "code": 200, "data": {"url": f"https://flkoss.example/{fmt}-signed-url"}, } ) if url == flk_npc.PREVIEW_URL: return _Resp(preview_payload) if url.endswith("/reader/info"): return _Resp(info_payload) if url.endswith("/reader/text"): idx = calls["text"] calls["text"] += 1 return _Resp(page_payloads[idx]) raise AssertionError(f"unexpected GET {url}") monkeypatch.setattr(flk_npc.httpx, "get", _fake_get) fr, fields = flk_npc.fetch_detail( None, url="https://flk.npc.gov.cn/detail?id=fb60d09eadb5483aa6e933e478b15a71", list_item=None, ) assert fr.status == 200 assert fields.title == "中华人民共和国监狱法" assert fields.publisher == "全国人民代表大会常务委员会" assert fields.publish_date.isoformat() == "2026-04-30" assert fields.effective_date.isoformat() == "2026-11-01" assert fields.is_effective is False assert "第一条" in fields.content_text assert "监狱" in fields.content_text assert "时效性：尚未生效" in fields.content_text assert fields.public_meta["ossWordOfdPath"] == "prod/20260430/a.ofd" assert fields.public_meta["validity_state_code"] == "4" assert fields.public_meta["validity_state_label"] == "尚未生效" assert fields.public_meta["publish_date"] == "2026-04-30" assert fields.public_meta["effective_date"] == "2026-11-01" assert fields.attachment_urls == [ "https://flkoss.example/docx-signed-url", "https://flkoss.example/pdf-signed-url", ] def test_fetch_detail_uses_docx_when_ofd_text_is_empty(monkeypatch): detail_payload = { "msg": "操作成功", "code": 200, "data": { "bbbs": "ff808181927f127601960f9a6e1305fb", "ossFile": { "ossWordPath": "prod/20250113/a.docx", "ossWordOfdPath": "prod/20250113/a.ofd", }, "flxz": "行政法规", "zdjgName": "国务院", "gbrq": "2025-01-13", "sxrq": "2025-04-01", "title": "公共安全视频图像信息系统管理条例", "sxx": 3, }, } docx_payload = _docx_bytes([ "公共安全视频图像信息系统管理条例", "第一条 为了规范公共安全视频图像信息系统管理，维护公共安全，保护个人隐私和个人信息权益，根据有关法律，制定本条例。", "第二条 本条例所称公共安全视频图像信息系统，是指通过在公共场所安装图像采集设备及相关设施，对涉及公共安全的区域进行视频图像信息收集、传输、显示、存储的系统。", ]) def _fake_get(url, *, params=None, **_kw): if url == flk_npc.DETAIL_URL: return _Resp(detail_payload) if url == flk_npc.DOWNLOAD_PC_URL: return _Resp( { "msg": "操作成功", "code": 200, "data": {"url": "https://flkoss.example/public-safety.docx"}, } ) if url == flk_npc.PREVIEW_URL: return _Resp({"msg": "获取预览链接成功！", "code": 500, "data": {}}) if url == "https://flkoss.example/public-safety.docx": return _Resp({}, content=docx_payload) raise AssertionError(f"unexpected GET {url}") monkeypatch.setattr(flk_npc.httpx, "get", _fake_get) _fr, fields = flk_npc.fetch_detail( None, url="https://flk.npc.gov.cn/detail?id=ff808181927f127601960f9a6e1305fb", list_item=None, ) assert "制定机关：国务院" in fields.content_text assert "第一条 为了规范公共安全视频图像信息系统管理" in fields.content_text assert "第二条 本条例所称公共安全视频图像信息系统" in fields.content_text assert len(fields.content_text) > flk_npc.MIN_USEFUL_BODY_CHARS def test_extract_docx_text_maps_paragraphs(): data = _docx_bytes(["第一条 正文", "第二条 继续"]) assert flk_npc._extract_docx_text(data) == "第一条 正文\n第二条 继续" def test_normalize_ofd_text_merges_visual_line_wraps(): raw = "\n".join( [ "第一条", "为了推动国家通用语言文字的规范化、标准化及其健", "康发展，发挥国家通用语言文字在社会主义现代化建设、", "各地区经济文化交流中的作用，铸牢中华民族共同体意识，坚定", "文化自信，根据宪法，制定本法。", "第二条", "本法所称的国家通用语言文字是普通话和规范汉", "—1—", "字，是国家法定全国通用的语言文字。", "第三条", "国家推广普及国家通用语言文字。", ] ) assert flk_npc.normalize_content_text(raw) == "\n".join( [ "第一条", "为了推动国家通用语言文字的规范化、标准化及其健康发展，发挥国家通用语言文字在社会主义现代化建设、各地区经济文化交流中的作用，铸牢中华民族共同体意识，坚定文化自信，根据宪法，制定本法。", "第二条", "本法所称的国家通用语言文字是普通话和规范汉字，是国家法定全国通用的语言文字。", "第三条", "国家推广普及国家通用语言文字。", ] ) def test_normalize_content_text_preserves_metadata_header(): raw = "\n".join( [ "中华人民共和国国家通用语言文字法", "制定机关：全国人民代表大会常务委员会", "法律法规分类：法律", "", "第一条", "为了推动国家通用语言文字的规范化、标准化及其健", "康发展，制定本法。", ] ) assert flk_npc.normalize_content_text(raw) == "\n".join( [ "中华人民共和国国家通用语言文字法", "制定机关：全国人民代表大会常务委员会", "法律法规分类：法律", "", "第一条", "为了推动国家通用语言文字的规范化、标准化及其健康发展，制定本法。", ] ) def test_normalize_content_text_drops_toc_and_merges_split_headings(): raw = "\n".join( [ "中华人民共和国民族团结进步促进法", "制定机关：全国人民代表大会常务委员会", "", "目", "录", "序", "言", "第一章", "总", "则", "第二章", "附", "则", "序", "言", "中国是世界上历史最悠久的国家之一，中华民族是有着五千", "多年文明史的伟大民族。", "—1—", "第一条", "为了铸牢中华民族共同体意识，制定本法。", ] ) assert flk_npc.normalize_content_text(raw) == "\n".join( [ "中华人民共和国民族团结进步促进法", "制定机关：全国人民代表大会常务委员会", "", "序言", "中国是世界上历史最悠久的国家之一，中华民族是有着五千多年文明史的伟大民族。", "第一条", "为了铸牢中华民族共同体意识，制定本法。", ] )