add llm2docx

jpai_llm2docx/Dockerfile

+# 使用 Python 官方镜像作为基础镜像
+FROM docker.m.daocloud.io/python:3.9.20
+# FROM python:3.9
+
+# 设置工作目录
+WORKDIR /app
+
+# 安装 Python 依赖
+COPY requirements.txt .
+# RUN python3 -m pip install --no-cache-dir -r requirements.txt
+# RUN pip install -i https://pypi.mirrors.ustc.edu.cn/simple/ --no-cache-dir -r requirements.txt
+RUN python3 -m pip install "fastapi[all]" && pip3 install python-docx && pip3 install requests && pip3 install python-multipart
+
+# 复制应用代码
+COPY . .
+
+# 暴露容器的 2286s 端口
+EXPOSE 2286
+
+# 映射卷
+VOLUME /app/config
+
+# 运行 FastAPI 应用
+CMD ["python", "main.py"]
\ No newline at end of file

jpai_llm2docx/auth.py

+import json
+import requests
+import redis
+
+# 配置 Redis
+redis_client = redis.StrictRedis(host='192.168.252.71', port=6379, db=0, decode_responses=True)
+
+# 获取有效 Token 的函数
+def auth(login_url, username, password, refresh_url, vaild_url=None):
+    cache_key = f"token:{username}"
+    refresh_cache_key = f"refresh_token:{username}"
+
+    # 尝试从 Redis 获取 Token
+    token = redis_client.get(cache_key)
+    if token and vaild_url is None:
+        if validate_token(vaild_url, token):  # 确保 Token 有效
+            return token
+
+    # 尝试从 Redis 获取 Refresh Token 并刷新 Token
+    refresh_token = redis_client.get(refresh_cache_key)
+    if refresh_token:
+        new_token = refresh_token_method(refresh_url, refresh_token)
+        if new_token:
+            redis_client.set(cache_key, new_token, ex=3600)  # 缓存新 Token，有效期 1 小时
+            return new_token
+
+    # 通过用户名和密码登录获取 Token
+    payload = json.dumps({
+        "username": username,
+        "password": password
+    })
+    headers = {
+        'Content-Type': 'application/json'
+    }
+
+    response = requests.post(login_url, headers=headers, data=payload)
+    if response.status_code == 200:
+        resp_data = response.json()
+        access_token = resp_data.get("data", {}).get("accessToken")
+        refresh_token = resp_data.get("data", {}).get("refreshToken")
+
+        if access_token:
+            redis_client.set(cache_key, access_token, ex=3600)  # 缓存 Token，有效期 1 小时
+        if refresh_token:
+            redis_client.set(refresh_cache_key, refresh_token, ex=86400)  # 缓存 Refresh Token，有效期 1 天
+
+        return access_token
+    else:
+        return ""
+
+# 验证 Token 的有效性（具体逻辑需根据 API 定义实现）
+def validate_token(vaild_url, token):
+    headers = {
+        'Authorization': f'Bearer {token}'
+    }
+
+    response = requests.post(vaild_url, headers=headers)
+
+    if response.status_code == 200:
+        resp_data = response.json()
+        code = resp_data.get("code")
+        if code == 401:
+            return False
+        return True
+    else:
+        return False
+
+# 使用 Refresh Token 刷新 Token
+def refresh_token_method(refresh_url, refresh_token):
+    headers = {
+        'Content-Type': 'application/json'
+    }
+
+    # 将 Refresh Token 作为 URL 参数
+    url_with_params = f"{refresh_url}?refreshToken={refresh_token}"
+    response = requests.post(url_with_params, headers=headers)
+
+    if response.status_code == 200:
+        resp_data = response.json()
+        return resp_data.get("data", {}).get("accessToken")
+    else:
+        return ""
+
+if __name__ == "__main__":
+    upload_url = "http://192.168.252.71:48080/admin-api/infra/file/upload"
+    login_url = "http://192.168.252.71:48080/admin-api/system/auth/login"
+    username = "admin"
+    password = "admin@jpai.com"
+    refresh_url = "http://192.168.252.71:48080/admin-api/system/auth/refresh-token"
+    vaild_url = "http://192.168.252.71:48080/admin-api/system/auth/get-permission-info"
+    print(auth(login_url, username, password, refresh_url, vaild_url))

jpai_llm2docx/build.sh

+docker build -t llm2docx:lastest .
\ No newline at end of file

jpai_llm2docx/config/config.json

+{
+    "username": "admin",
+    "password": "admin@jpai.com",
+    "upload_url": "http://192.168.252.71:48080/admin-api/infra/file/upload",
+    "login_url": "http://192.168.252.71:48080/admin-api/system/auth/login",
+    "refresh_url": "http://192.168.252.71:48080/admin-api/system/auth/refresh-token",
+    "vaild_url": "http://192.168.252.71:48080/admin-api/system/auth/get-permission-info"
+}
\ No newline at end of file

jpai_llm2docx/core.py

+import json
+from docx import Document
+from docx.enum.text import WD_BREAK
+from docx.oxml.ns import qn
+from pathlib import Path
+import requests
+import copy
+
+from processer import clone_paragraph_style, delete_paragraph
+from auth import auth
+
+def replace_text_in_docx(json_data, temp_dir=""):
+    template_name = json_data["文件ID"]
+    if "文件ID" not in json_data.keys() or Path(f"./templates/{template_name}.docx").exists() is False:
+        return None
+    # 加载 Word 文档
+    doc = Document(f"./templates/{template_name}.docx")
+    # 设置默认字体为宋体
+    doc.styles['Normal'].font.name = u'宋体'
+    doc.styles['Normal']._element.rPr.rFonts.set(qn('w:eastAsia'), u'宋体')
+    
+    # 遍历 json_data 中的每个键值对
+    for key, value in json_data.items():
+        # 跳过 "文件ID" 键，因为它用于标识文件
+        if key == "文件ID":
+            continue
+        
+        # 如果值是列表，我们需要逐项插入
+        if isinstance(value, list):
+            if "表" in key and len(json_data[key]) != 0:
+                insert_table(doc, value)
+            elif "项" in key and len(json_data[key]) != 0:
+                if "项页" in key:
+                    insert_secton(doc, value, True)
+                else:
+                    insert_secton(doc, value)
+            else:
+                insert_list(doc, key, value)
+        else:
+            # 替换单个值
+            insert_one(doc, key, value)
+    
+    # 保存修改后的文档
+    new_doc_path = Path(temp_dir) / f"{template_name}_modified.docx"
+    # new_doc_path = f"./cache/{template_name}_modified.docx"
+    doc.save(new_doc_path)
+    return new_doc_path
+
+def insert_table(doc, list_value):
+    placeholders = list(list_value[0].keys())
+    # 遍历文档中的所有表格
+    for table in doc.tables:
+        early_stop = False
+        # 遍历表格中的所有行
+        for i, row in enumerate(table.rows):
+            start_idx = -1
+            # 检查行中的单元格是否包含占位符
+            for j, cell in enumerate(row.cells):
+                if j >= len(placeholders):
+                    break
+                placeholder = placeholders[j]
+                if f'{{{{{placeholder}}}}}' in cell.text:
+                    if start_idx < 0:
+                        start_idx = i
+                        early_stop = True
+                    for i in range(len(list_value) - 1):
+                        new_row = table.add_row().cells
+                        for j, cell in enumerate(row.cells):
+                            new_row[j].text = cell.text
+                    break
+                
+        if early_stop:
+            # 填充
+            for i, row in enumerate(table.rows):
+                if i >= start_idx and (i - start_idx) < len(list_value):
+                    fill_values = list_value[i - start_idx]
+                    for key, val in fill_values.items():
+                        for j, cell in enumerate(row.cells):
+                            if key in cell.text:
+                                cell.text = val
+
+            break
+
+def insert_secton(doc, list_value, switch_page=False):
+    # 定位
+    paragraphs_to_repeat = []
+    start = False
+    start_flag = "start"
+    end_flag = "end"
+    start_Paragraph = None
+    repeat_count = len(list_value) - 1
+    start_placeholder = f'{{{{{start_flag}}}}}'
+    end_placeholder = f'{{{{{end_flag}}}}}'
+    for i, paragraph in enumerate(doc.paragraphs):
+        if start_placeholder in paragraph.text:
+            start = True
+            continue
+        if start:
+            if start_Paragraph is None:
+                start_Paragraph = paragraph
+            paragraphs_to_repeat.append(paragraph)
+        if end_placeholder in paragraph.text:
+            paragraphs_to_repeat.pop(-1)
+            # start = False
+            break
+
+    # 重复复制段落
+    for _ in range(repeat_count):
+        # 在段落之前插入新段落
+        for paragraph_i in paragraphs_to_repeat:
+            new_paragraph = start_Paragraph.insert_paragraph_before(paragraph_i.text)
+            clone_paragraph_style(paragraph_i, new_paragraph)
+        if switch_page and start_Paragraph is not None:
+            new_paragraph = start_Paragraph.insert_paragraph_before('')
+            new_paragraph.add_run().add_break(WD_BREAK.PAGE)
+    # 重复替换段落
+    for i in range(repeat_count + 1):
+        for key, val in list_value[i].items():
+            insert_one(doc, key, val, True)
+    # 清楚其他占位符
+    for parag in doc.paragraphs:
+        if start_placeholder in parag.text or end_placeholder in parag.text:
+            delete_paragraph(parag)
+
+def insert_list(doc, key, list_value):
+    # 定位
+    start_Paragraph = None
+    repeat_count = len(list_value) - 1
+    for paragraph in doc.paragraphs:
+        if f'{{{{{key}}}}}' in paragraph.text:
+            start_Paragraph = paragraph
+    if start_Paragraph is not None:
+        for _ in range(repeat_count):
+            new_paragraph = start_Paragraph.insert_paragraph_before(f'{{{{{key}}}}}', style=paragraph.style)
+            clone_paragraph_style(start_Paragraph, new_paragraph)
+    for val in list_value:
+        insert_one(doc, key, val, True)
+
+def insert_one(doc, key, value, onlyOne=False):
+    if "\n" in value:
+        value = value.split("\n")
+    for paragraph in doc.paragraphs:
+        if f'{{{{{key}}}}}' in paragraph.text:
+            runs = paragraph.runs
+            find = False
+            for i, run in enumerate(runs):
+                if f'{{{{{key}}}}}' in run.text:
+                    find = True
+                    # 替换文本，保持格式不变
+                    if isinstance(value, list):
+                        for i, val in enumerate(value):
+                            if i < len(value) - 1:
+                                new_paragraph = paragraph.insert_paragraph_before(val, style=paragraph.style)
+                                clone_paragraph_style(paragraph, new_paragraph, True)
+                                continue
+                            run.text = run.text.replace(f'{{{{{key}}}}}', value[-1])
+                    else:
+                        run.text = run.text.replace(f'{{{{{key}}}}}', value)
+            if find is False:
+                origin_parag = copy.deepcopy(paragraph)
+                paragraph.text = paragraph.text.replace(f'{{{{{key}}}}}', value)
+                clone_paragraph_style(origin_parag, paragraph, True)
+            if onlyOne:
+                break
+
+def upload_docx_to_server(new_doc_path, data):
+    upload_url = data["upload_url"]
+    login_url = data["login_url"]
+    username = data["username"]
+    password = data["password"]
+    refresh_url = data["refresh_url"]
+    vaild_url = data["vaild_url"]
+
+    # 确保文件存在
+    if not Path(new_doc_path).is_file():
+        raise FileNotFoundError(f"The file {new_doc_path} does not exist.")
+
+    with open(new_doc_path, 'rb') as file:
+        # 构建 multipart/form-data 请求体
+        files = {'file': (Path(new_doc_path).name, file)}
+
+        # 发送 POST 请求上传文档
+        auth_token = auth(login_url, username, password, refresh_url, vaild_url)
+        if auth_token == "":
+            return {"msg": "auth error", "data": ""}
+        headers = {
+            'Authorization': f'Bearer {auth_token}',  # 使用 Bearer 令牌
+        }
+        response = requests.post(upload_url, headers=headers, files=files)
+        if response.status_code == 200:
+            resp_data = response.json()
+            # 如果需要删除本地文件
+            return {"msg": resp_data.get('msg'), "data": resp_data.get('data')}
+        else:
+            return {"msg": "upload fail", "data": ""}

jpai_llm2docx/main.py

+import uvicorn
+from fastapi import FastAPI, Form
+import json
+import re
+from pathlib import Path
+import shutil
+import zipfile
+import tempfile
+
+from core import replace_text_in_docx, upload_docx_to_server
+
+app = FastAPI()
+# 定义一个正则表达式来匹配 JSON 数据
+json_pattern = re.compile(r'```json\s+(.*?)\s+```', re.DOTALL)
+with open("./config/config.json", 'r', encoding='utf-8') as file:
+    data = json.load(file)
+
+@app.post("/llm2docx/")
+async def create_upload_file(keyData: str = Form(...)):
+    # 使用正则表达式查找所有 JSON 数据
+    matches = json_pattern.findall(keyData)
+    if matches:
+        results = []
+        temp_dir = tempfile.mkdtemp()  # 创建一个临时目录来存储生成的 .docx 文件
+        try:
+            for json_str in matches:
+                try:
+                    # 尝试解析 JSON 字符串
+                    json_data = json.loads(json_str.strip())
+                except:
+                    # 如果 JSON 字符串无效，返回错误信息
+                    continue
+                
+                new_doc_path = replace_text_in_docx(json_data, temp_dir)
+                if new_doc_path is None:
+                    return {"msg": "新docx文件建立失败", "data": ""}
+                results.append(new_doc_path)
+            if len(results) == 0:
+                return {"msg": "JSON 字符串无效", "data": ""}
+            # 将所有 .docx 文件打包成压缩文件
+            zip_path = Path(temp_dir) / "word_files.zip"
+            with zipfile.ZipFile(zip_path, 'w') as zipf:
+                for doc_path in results:
+                    zipf.write(doc_path, arcname=doc_path.name)
+            
+            # 上传压缩文件到服务器
+            res = upload_docx_to_server(zip_path, data)
+            
+            # 清理临时文件
+            shutil.rmtree(temp_dir)
+            
+            return res
+        
+        except Exception as e:
+            # 如果发生异常，清理临时文件并返回错误信息
+            shutil.rmtree(temp_dir)
+            return {"msg": f"处理过程中发生错误: {str(e)}", "data": ""}
+    else:
+        # 如果没有找到 JSON 数据，返回错误信息
+        return {"msg": "没有找到JSON 数据", "data": ""}
+
+if __name__ == "__main__":
+    uvicorn.run("main:app", host="0.0.0.0", port=2286, reload=True)
\ No newline at end of file

jpai_llm2docx/main_v1.py

+import uvicorn
+from fastapi import FastAPI, Form
+import json
+import re
+from pathlib import Path
+
+from core import replace_text_in_docx, upload_docx_to_server
+
+app = FastAPI()
+# 定义一个正则表达式来匹配 JSON 数据
+json_pattern = re.compile(r'```json\s+(.*?)\s+```', re.DOTALL)
+with open("./config/config.json", 'r', encoding='utf-8') as file:
+    data = json.load(file)
+
+@app.post("/llm2docx/")
+async def create_upload_file(keyData: str = Form(...)):
+    # 使用正则表达式查找 JSON 数据
+    match = json_pattern.search(keyData)
+    if match:
+        # 提取 JSON 字符串
+        json_str = match.group(1).strip()
+        try:
+            # 尝试解析 JSON 字符串
+            json_data = json.loads(json_str)
+        except:
+            # 如果 JSON 字符串无效，返回错误信息
+            return {"msg": "AI关键信息抽取错误", "data": ""}
+        
+        new_doc_path = replace_text_in_docx(json_data)
+        if new_doc_path is None:
+            return {"msg": "AI关键信息抽取错误", "data": ""}
+
+        res = upload_docx_to_server(new_doc_path, data)
+        if Path(new_doc_path).exists():
+            Path(new_doc_path).unlink()
+        return res
+        
+    else:
+        # 如果没有找到 JSON 数据，返回错误信息
+        return {"msg": "AI关键信息抽取错误", "data": ""}
+
+if __name__ == "__main__":
+    uvicorn.run("main:app", host="0.0.0.0", port=2286, reload=True)

jpai_llm2docx/processer.py

+
+def delete_paragraph(paragraph):
+    """
+    删除指定的段落
+    :param paragraph: 要删除的段落对象。
+    """
+    # 获取段落的 XML 元素
+    p_element = paragraph._element
+    # 获取段落的父元素（即文档的 body）
+    parent_element = p_element.getparent()
+    # 从父元素中移除段落的 XML 元素
+    parent_element.remove(p_element)
+
+def clone_paragraph_style(paragraph, new_paragraph, noBold=False):
+    # 复制段落属性
+    new_paragraph.paragraph_format.alignment = paragraph.paragraph_format.alignment
+    new_paragraph.paragraph_format.keep_together = paragraph.paragraph_format.keep_together
+    new_paragraph.paragraph_format.keep_with_next = paragraph.paragraph_format.keep_with_next
+    new_paragraph.paragraph_format.left_indent = paragraph.paragraph_format.left_indent
+    new_paragraph.paragraph_format.right_indent = paragraph.paragraph_format.right_indent
+    new_paragraph.paragraph_format.first_line_indent = paragraph.paragraph_format.first_line_indent
+    new_paragraph.paragraph_format.space_after = paragraph.paragraph_format.space_after
+    new_paragraph.paragraph_format.space_before = paragraph.paragraph_format.space_before
+    new_paragraph.paragraph_format.widow_control = paragraph.paragraph_format.widow_control
+
+    # 复制每个运行的样式和文本
+    if len(paragraph.runs) == 0:
+        return
+    original_run = paragraph.runs[0]
+    for new_run  in new_paragraph.runs:
+        if noBold is False:
+            new_run.font.bold = original_run.font.bold
+        new_run.font.italic = original_run.font.italic
+        new_run.font.underline = original_run.font.underline
+        new_run.font.color.rgb = original_run.font.color.rgb
+        new_run.font.size = original_run.font.size
+        new_run.font.name = original_run.font.name

jpai_llm2docx/run.sh

+docker run -d -p 2286:2286 --name llm2docx_api -v ./config:/app/config llm2docx:lastest
\ No newline at end of file

jpai_llm2docx/test.py

+import shutil
+import tempfile
+
+print(shutil.__file__)  # 打印 shutil 模块的路径
+print(tempfile.__file__)  # 打印 tempfile 模块的路径
\ No newline at end of file