import requests import re import os import json from dotenv import load_dotenv load_dotenv() DOC_STORAGE_PATH = os.getenv("DOC_STORAGE_PATH") # XWiki 基础 URL 和认证信息 base_url = "http://localhost:8081" username = "ZHU" password = "p@ssw0rd" def wiki_search_documents(search_query): # 搜索文档 search_url = f"{base_url}/bin/view/Main/Search?text={search_query}&f_type=DOCUMENT&f_locale=zh_CN&f_locale=&r=1" #search_url = f"{base_url}/xwiki/rest/wikis/query?q={search_query}" #search_url = f"{base_url}/wikis/XWiki/spaces/Main/documents?query={search_query}" # 发起请求 response = requests.get(search_url, auth=(username, password)) ret = [] if response.status_code == 200: search_results = response.text match_result = re.compile(r'\n(.*?)').findall(search_results) if match_result: index = 1000 for k,v in match_result: doc = { "id": f"{index}", "title": v, "url": k, "pub_date":""} ret.append(doc) index = index + 1 else: print(f"搜索失败,状态码:{response.status_code}") if len(ret) > 0: with open(f"{DOC_STORAGE_PATH}\wiki_cache.json", "w", encoding="utf-8") as f: f.write(json.dumps({"data":ret}, ensure_ascii=False, indent=4)) return ret def get_wiki_document(title: str, url: str): get_url = f"{base_url}{url}?xpage=plain" print(f"get_wiki_document {get_url}") response = requests.get(get_url, auth=(username, password)) if response.status_code == 200: search_results = response.text return {"url": url,"title":title, "text": search_results} return None def wiki_get_document(title: str): print(">>>>>> wiki_get_document ", title) with open(f"{DOC_STORAGE_PATH}\wiki_cache.json", "r", encoding="utf-8") as f: data = json.load(f) for v in data["data"]: if v["title"] == title or v["id"] == title: return get_wiki_document(v["title"] , v["url"]) return None if __name__ == "__main__": ret = wiki_search_documents("保险公司 大模型") print(ret) content = wiki_get_document("在保险公司实施大模型项目的规定") print(content)