From bc1f19542fca524cb8b89b4a29ec20066248fb90 Mon Sep 17 00:00:00 2001 From: ninghongbin <2409766686@qq.com> Date: Tue, 2 Dec 2025 19:28:37 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B7=BB=E5=8A=A0ai=E6=99=BA=E8=83=BD=E4=BD=93?= =?UTF-8?q?=E6=A0=87=E4=B9=A6=E6=8E=A5=E5=8F=A3=EF=BC=8C=E4=BF=AE=E6=94=B9?= =?UTF-8?q?ai=5Fagent=E4=BB=85=E6=9A=B4=E9=9C=B2=E4=B8=80=E4=B8=AA?= =?UTF-8?q?=E6=96=B9=E6=B3=95=E4=BE=9B=E5=A4=96=E9=83=A8=E8=B0=83=E7=94=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- AI_Agent.py | 136 ++++++++++++++++++++++++---------------- config.py | 4 ++ main.py | 177 +++++++++++++++++++++++++++++++++++++++++++++++++++- 3 files changed, 259 insertions(+), 58 deletions(-) diff --git a/AI_Agent.py b/AI_Agent.py index 5863b70..0436191 100644 --- a/AI_Agent.py +++ b/AI_Agent.py @@ -8,11 +8,10 @@ from docx import Document from shutil import which import requests +from config import OLLAMA_BASE_URL, OLLAMA_MODEL + INPUT_WORD = r"C:\Users\YC\Desktop\1.docx" # 你的招标文件 OUTPUT_WORD = r"C:\Users\YC\Desktop\投标文件-最终版.docx" # 最终输出路径 -OLLAMA_MODEL = "alibayram/Qwen3-30B-A3B-Instruct-2507:latest" # 当前最强本地模型 -OLLAMA_BASE_URL = "http://192.168.110.5:11434" - # ==================== Ollama 本地调用（支持 128K 上下文 + 长输出）=================== @@ -82,7 +81,7 @@ def word_to_md(word_path: str) -> str: if not pandoc_cmd: common = [ os.path.expanduser(r"~\AppData\Local\Pandoc\pandoc.exe"), - r"C:\Program Files\Pandoc\pandoc.exe", + r"./util/pandoc.exe", ] for p in common: if os.path.exists(p): @@ -110,7 +109,7 @@ def word_to_md(word_path: str) -> str: # ==================== 新版：两步生成超级目录（永不超时）=================== def generate_full_outline(tender_md: str) -> str: tender_text = Path(tender_md).read_text(encoding="utf-8") - print(f"招标文件共 {len(tender_text)//2} 字，开始两阶段生成四级目录...") + print(f"招标文件共 {len(tender_text) // 2} 字，开始两阶段生成四级目录...") # 第一步：先让模型只看前 6 万字，生成一个【简洁但完整】的三级目录（超快，10秒内出） prompt1 = f"""请仔细阅读以下招标文件核心内容，只输出一个简洁但完整的三级目录（一级用“一、”，二级用“1、”，三级用“1.1、”）。 @@ -143,7 +142,7 @@ def generate_full_outline(tender_md: str) -> str: # 每 8个三级标题为一组，展开四级（稳到爆） full_outline = outline_skeleton + "\n" for i in range(0, len(level3_titles), 8): - batch = level3_titles[i:i+8] + batch = level3_titles[i:i + 8] batch_text = "\n".join(batch) prompt2 = f"""你是一位招投标专家，请把下面这几个三级标题分别展开成 10～18 个专业四级标题（格式必须是 1.1.1、1.1.2、……）。 @@ -157,7 +156,7 @@ def generate_full_outline(tender_md: str) -> str: 直接输出四级标题：""" - print(f" 正在展开第 {i//8 + 1} 组四级标题（{len(batch)}个）...") + print(f" 正在展开第 {i // 8 + 1} 组四级标题（{len(batch)}个）...") level4_text = call_llm([{"role": "user", "content": prompt2}], temperature=0.2, max_tokens=20000) full_outline += "\n" + level4_text + "\n" @@ -165,7 +164,7 @@ def generate_full_outline(tender_md: str) -> str: # 保存并返回 Path("output/四级目录.md").write_text(full_outline, encoding="utf-8") - print(f"超级四级目录生成成功！总计约 {len(full_outline)//2} 字（再也不怕超时了！）") + print(f"超级四级目录生成成功！总计约 {len(full_outline) // 2} 字（再也不怕超时了！）") return full_outline @@ -255,26 +254,37 @@ def update_word_toc(docx_path: str): print(f"Word目录自动更新失败（可手动右键更新）：{e}") -# ==================== 主流程 ==================== -def main(): - print("启动本地 Qwen3-30B 投标文件生成器（128K上下文版）\n") - os.makedirs("output", exist_ok=True) +# ==================== 统一对外暴露的方法（唯一入口）=================== +def generate_tender_from_input(input_word_path: str, output_word_path: str) -> bool: + """ + 统一对外暴露的投标文件生成方法 - # 1. 转换招标文件 - tender_md = word_to_md(INPUT_WORD) - tender_text = Path(tender_md).read_text(encoding="utf-8") + Args: + input_word_path: 输入招标文件路径（.docx/.doc） + output_word_path: 输出投标文件路径（.docx） - # 2. 生成超级详细目录 - outline = generate_full_outline(tender_md) + Returns: + bool: 生成成功返回True，失败返回False + """ + try: + print("启动投标文件生成器（统一入口）\n") + os.makedirs("output", exist_ok=True) - # 3. 分批生成正文（超长内容 - content = batch_fill_content(outline, tender_text) - content = expand_to_50000_words(content) + # 1. 转换招标文件 + tender_md = word_to_md(input_word_path) + tender_text = Path(tender_md).read_text(encoding="utf-8") - # 4. 合成最终 Markdown - final_md = f"""# 【投标单位全称】 + # 2. 生成超级详细目录 + outline = generate_full_outline(tender_md) -## {Path(INPUT_WORD).stem} - 投标文件 + # 3. 分批生成正文 + content = batch_fill_content(outline, tender_text) + content = expand_to_50000_words(content) + + # 4. 合成最终 Markdown + final_md = f"""# 【投标单位全称】 + +## {Path(input_word_path).stem} - 投标文件 {outline} @@ -287,41 +297,57 @@ def main(): - 类似业绩证明材料 - 偏离表 """ - final_md_path = "output/最终投标文件.md" - Path(final_md_path).write_text(final_md, encoding="utf-8") - print(f"\n最终 Markdown 生成成功！总计约 {len(final_md) // 2} 字") + final_md_path = "output/最终投标文件.md" + Path(final_md_path).write_text(final_md, encoding="utf-8") + print(f"\n最终 Markdown 生成成功！总计约 {len(final_md) // 2} 字") - # 5. 转 Word（三保险） - print("正在转换为 Word 文档...") - success = False - pandoc_cmd = which("pandoc") or which("pandoc.exe") - if pandoc_cmd and os.path.exists(pandoc_cmd): - cmd = [pandoc_cmd, final_md_path, "-o", OUTPUT_WORD, "--reference-doc=template.docx"] if os.path.exists( - "template.docx") else [pandoc_cmd, final_md_path, "-o", OUTPUT_WORD] - if subprocess.run(cmd, capture_output=True).returncode == 0: - success = True + # 5. 转 Word（三保险） + print("正在转换为 Word 文档...") + success = False + pandoc_cmd = which("pandoc") or which("pandoc.exe") + if pandoc_cmd and os.path.exists(pandoc_cmd): + cmd = [pandoc_cmd, final_md_path, "-o", output_word_path, + "--reference-doc=template.docx"] if os.path.exists( + "template.docx") else [pandoc_cmd, final_md_path, "-o", output_word_path] + if subprocess.run(cmd, capture_output=True).returncode == 0: + success = True - if not success: - print("Pandoc 失败，使用 python-docx 强制生成...") - doc = Document() - for line in final_md.split('\n'): - l = line.strip() - if l.startswith("# "): - doc.add_heading(l[2:], 0) - elif l.startswith("## "): - doc.add_heading(l[3:], 1) - elif l.startswith("### "): - doc.add_heading(l[4:], 2) - elif l.startswith("#### "): - doc.add_heading(l[5:], 3) - elif l: - doc.add_paragraph(l) - doc.save(OUTPUT_WORD) + if not success: + print("Pandoc 失败，使用 python-docx 强制生成...") + doc = Document() + for line in final_md.split('\n'): + l = line.strip() + if l.startswith("# "): + doc.add_heading(l[2:], 0) + elif l.startswith("## "): + doc.add_heading(l[3:], 1) + elif l.startswith("### "): + doc.add_heading(l[4:], 2) + elif l.startswith("#### "): + doc.add_heading(l[5:], 3) + elif l: + doc.add_paragraph(l) + doc.save(output_word_path) - update_word_toc(OUTPUT_WORD) - print(f"\n大功告成！投标文件已生成：") - print(f" {OUTPUT_WORD}") - print(f" 总字数约：{len(final_md) // 2} 字") + update_word_toc(output_word_path) + print(f"\n大功告成！投标文件已生成：") + print(f" {output_word_path}") + print(f" 总字数约：{len(final_md) // 2} 字") + + # 清理临时生成的md文件（可选，保留也可以） + if os.path.exists(tender_md): + os.remove(tender_md) + + return True + + except Exception as e: + print(f"投标文件生成失败：{str(e)}") + return False + + +# ==================== 原有主流程（保持不变，方便单独运行）=================== +def main(): + generate_tender_from_input(INPUT_WORD, OUTPUT_WORD) os.startfile(OUTPUT_WORD) diff --git a/config.py b/config.py index d96a5a2..3e08d28 100644 --- a/config.py +++ b/config.py @@ -64,3 +64,7 @@ class DetectionResponse(BaseModel): originalImgSize: List[int] targets: List[dict] processing_errors: List[str] = [] + +# 智能体相关 +OLLAMA_MODEL = "alibayram/Qwen3-30B-A3B-Instruct-2507:latest" # 当前最强本地模型 +OLLAMA_BASE_URL = "http://192.168.110.5:11434" \ No newline at end of file diff --git a/main.py b/main.py index f137d6c..bf18092 100644 --- a/main.py +++ b/main.py @@ -1,18 +1,30 @@ import io import os +import time import tempfile +import shutil from contextlib import asynccontextmanager +from pathlib import Path +from typing import Optional import requests import uvicorn from PIL import Image -from fastapi import FastAPI, HTTPException +from fastapi import FastAPI, HTTPException, UploadFile, File from fastapi.middleware.cors import CORSMiddleware +from fastapi.responses import FileResponse from pydantic import BaseModel, HttpUrl +from AI_Agent import generate_tender_from_input from config import DetectionResponse from process import detect_large_image_from_url +# 配置文件存储路径 +UPLOAD_DIR = Path("uploaded_files") +OUTPUT_DIR = Path("generated_tenders") +UPLOAD_DIR.mkdir(exist_ok=True) +OUTPUT_DIR.mkdir(exist_ok=True) + # 全局检测管理器 detector_manager = None @@ -28,9 +40,17 @@ async def lifespan(app: FastAPI): print(f"初始化失败：{str(e)}") raise yield + # 程序关闭时清理临时文件（可选） + print("清理临时文件...") + for file in UPLOAD_DIR.glob("*"): + try: + if file.is_file(): + file.unlink() + except Exception as e: + print(f"清理文件 {file} 失败：{e}") -app = FastAPI(lifespan=lifespan, title="目标检测API", version="1.0.0") +app = FastAPI(lifespan=lifespan, title="目标检测与投标文件生成API", version="1.0.0") # 配置跨域请求 app.add_middleware( @@ -51,6 +71,15 @@ class DetectionProcessRequest(BaseModel): url: HttpUrl +class TenderGenerateResponse(BaseModel): + """投标文件生成响应模型""" + status: str + message: str + relative_path: str + file_name: str + file_size: Optional[int] = None + + @app.post("/detect_image", response_model=DetectionResponse) async def run_detection_image(request: DetectionRequest): # 解析检测类型 @@ -114,12 +143,154 @@ async def get_supported_types(): return {"supported_types": []} +@app.post("/generate_tender", response_model=TenderGenerateResponse) +async def generate_tender_file(file: UploadFile = File(...)): + """ + 上传招标文件（Word格式），生成投标文件 + 支持的文件格式：.docx + 返回生成文件的相对路径，用于下载接口 + """ + upload_path = None + + # 验证文件类型 + if not file.filename.endswith((".docx", ".doc")): + raise HTTPException( + status_code=400, + detail="仅支持Word文件（.docx或.doc格式）" + ) + + try: + # 保存上传的文件 + timestamp = int(time.time()) + file_ext = Path(file.filename).suffix + upload_filename = f"tender_{timestamp}{file_ext}" + upload_path = UPLOAD_DIR / upload_filename + + # 保存文件内容 + with open(upload_path, "wb") as f: + shutil.copyfileobj(file.file, f) + + print(f"已接收上传文件：{upload_path}") + + # 生成输出文件名和路径 + output_filename = f"投标文件_生成版_{timestamp}.docx" + output_path = OUTPUT_DIR / output_filename + + # 调用AI_Agent的统一入口方法（仅这一个调用） + print("开始生成投标文件...") + generate_success = generate_tender_from_input( + input_word_path=str(upload_path), + output_word_path=str(output_path) + ) + + if not generate_success: + raise Exception("投标文件生成核心流程执行失败") + + # 验证生成的文件是否存在 + if not output_path.exists() or not output_path.is_file(): + raise Exception("投标文件生成后未找到目标文件") + + # 计算文件大小 + file_size = output_path.stat().st_size + + # 构建相对路径（相对于项目根目录） + relative_path = str(output_path.relative_to(Path.cwd())) + + print(f"投标文件生成成功：{output_path}") + + return { + "status": "success", + "message": "投标文件生成完成", + "relative_path": relative_path, + "file_name": output_filename, + "file_size": file_size + } + + except Exception as e: + error_msg = f"生成投标文件失败：{str(e)}" + print(error_msg) + raise HTTPException( + status_code=500, + detail=error_msg + ) + finally: + # 关闭上传文件流 + await file.close() + # 清理上传的原始文件，节省空间 + if upload_path and upload_path.exists(): + try: + upload_path.unlink() + print(f"已清理上传文件：{upload_path}") + except Exception as e: + print(f"清理上传文件失败：{e}") + + +@app.get("/download_file") +async def download_generated_file(relative_path: str): + """ + 根据相对路径下载生成的投标文件 + Args: + relative_path: 生成文件的相对路径（从/generate_tender接口获取） + """ + # 解析绝对路径，防止路径穿越攻击 + try: + abs_path = Path(relative_path).resolve() + + # 验证路径是否在允许的输出目录内 + if not abs_path.is_relative_to(OUTPUT_DIR.resolve()): + raise HTTPException( + status_code=403, + detail="访问禁止：文件路径不在允许范围内" + ) + + if not abs_path.exists() or not abs_path.is_file(): + raise HTTPException( + status_code=404, + detail="文件不存在或已被删除" + ) + + # 返回文件下载响应 + return FileResponse( + path=abs_path, + filename=abs_path.name, + media_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document" + ) + + except HTTPException: + raise + except Exception as e: + print(f"文件下载失败：{str(e)}") + raise HTTPException( + status_code=500, + detail=f"文件下载失败：{str(e)}" + ) + + +@app.get("/list_generated_files") +async def list_generated_files(): + """列出所有生成的投标文件（可选接口）""" + files = [] + for file in OUTPUT_DIR.glob("*.docx"): + files.append({ + "file_name": file.name, + "relative_path": str(file.relative_to(Path.cwd())), + "file_size": file.stat().st_size, + "created_time": file.stat().st_ctime + }) + return { + "total": len(files), + "files": files + } + + if __name__ == "__main__": import argparse + # 补充必要的导入（在主函数中导入，避免启动时依赖冲突） parser = argparse.ArgumentParser() parser.add_argument("--host", default="0.0.0.0") parser.add_argument("--port", type=int, default=8000) parser.add_argument("--reload", action="store_true") args = parser.parse_args() - uvicorn.run("main:app", host=args.host, port=args.port, reload=args.reload) + + uvicorn.run("main:app", host=args.host, port=args.port, reload=args.reload) \ No newline at end of file