New file |
| | |
| | | # -*- coding: utf-8 -*- |
| | | # |
| | | # @author: lyg |
| | | # @date: 2025-5-7 |
| | | # @version: 1 |
| | | # @description:è§è§è¯å«ææ¡£å
容 |
| | | |
| | | from langchain_openai.chat_models import ChatOpenAI |
| | | from langchain_core.prompts import HumanMessagePromptTemplate, ChatPromptTemplate |
| | | from langchain_core.messages import HumanMessage, SystemMessage |
| | | from langchain_core.output_parsers import JsonOutputParser |
| | | from docx import Document |
| | | from PIL import Image |
| | | from io import BytesIO |
| | | import re |
| | | import json |
| | | import base64 |
| | | |
| | | |
| | | class VisionTest: |
| | | def __init__(self, file): |
| | | self.llm = ChatOpenAI(temperature=0, model="qwen2.5-72b-instruct", |
| | | base_url="https://dashscope.aliyuncs.com/compatible-mode/v1", api_key="sk-15ecf7e273ad4b729c7f7f42b542749e") |
| | | image = base64.b64encode(open(file, 'rb').read()).decode() |
| | | self.prompt = ChatPromptTemplate.from_messages([ |
| | | SystemMessage("ä½ æ¯ä¸ä¸ªèµæ·±è½¯ä»¶å·¥ç¨å¸ï¼è¯·åæå¾çåçé®é¢ã"), |
| | | HumanMessage(content=[ |
| | | {"type": "text", "text": "{msg}"}, |
| | | { |
| | | "type": "image_url", |
| | | "image_url": {"url": f"data:image/jpeg;base64,{image}"}, |
| | | } |
| | | ]) |
| | | ]) |
| | | |
| | | def run(self, msg): |
| | | chain = self.prompt | self.llm |
| | | resp = chain.invoke({"msg": msg}) |
| | | print(resp.content) |
| | | |
| | | # def get_document_chapters(doc_path): |
| | | # doc = Document(doc_path) |
| | | # chapters = [] |
| | | # current_chapter = None |
| | | |
| | | # for para in doc.paragraphs: |
| | | # if para.style.name.startswith('Heading'): # æ£æ¥æ¯å¦ä¸ºæ 颿 ·å¼ |
| | | # level = int(para.style.name.replace('Heading', '')) # è·åæ é¢çº§å« |
| | | # current_chapter = {'level': level, 'title': para.text, 'content': []} |
| | | # chapters.append(current_chapter) |
| | | # elif current_chapter is not None: |
| | | # current_chapter['content'].append(para.text) # æ·»å å
容å°å½åç« è |
| | | |
| | | # return chapters |
| | | |
| | | def has_image(self, paragraph): |
| | | # éè¿æ£æ¥XMLä¸çåµå
¥å¼å¯¹è±¡æ¥å¤ææ¯å¦æå¾ç |
| | | xml = paragraph._element.xml |
| | | return 'w:object' in xml or 'w:drawing' in xml |
| | | |
| | | def convert_blob_to_png_base64(self, image_blob): |
| | | try: |
| | | # æå¼å¾ç |
| | | image = Image.open(BytesIO(image_blob)) |
| | | # å建å
åç¼å²åº |
| | | buffer = BytesIO() |
| | | # ä¿å为PNGæ ¼å¼ |
| | | image.save(buffer, format="PNG") |
| | | # è·åPNGæ ¼å¼çäºè¿å¶æ°æ® |
| | | png_data = buffer.getvalue() |
| | | # 转æ¢ä¸ºBase64ç¼ç |
| | | base64_data = base64.b64encode(png_data).decode('utf-8') |
| | | return base64_data |
| | | except Exception as e: |
| | | print(f"Error: {e}") |
| | | return None |
| | | |
| | | def get_image_blob(self, paragraph): |
| | | # éåæ®µè½ä¸çææRun对象ï¼å¾çé常å¨åç¬çRunä¸ï¼ |
| | | for run in paragraph.runs: |
| | | xml = run._element.xml |
| | | if xml.find('v:imagedata') != -1: |
| | | # ä½¿ç¨æ£åè¡¨è¾¾å¼æ¥æ¾r:id屿§ |
| | | match = re.search(r'r:id="([^"]+)"', xml) |
| | | if match: |
| | | r_id = match.group(1) |
| | | if r_id: |
| | | # è·åå¾çä¿¡æ¯ |
| | | image_part = paragraph.part.rels[r_id].target_part |
| | | return image_part.blob |
| | | if xml.find('wp:inline') != -1 or xml.find('wp:anchor') != -1: |
| | | # ä½¿ç¨æ£åè¡¨è¾¾å¼æ¥æ¾r:embed屿§ |
| | | match = re.search(r'r:embed="([^"]+)"', xml) |
| | | if match: |
| | | r_id = match.group(1) |
| | | if r_id: |
| | | # è·åå¾çä¿¡æ¯ |
| | | image_part = paragraph.part.rels[r_id].target_part |
| | | return image_part.blob |
| | | return None |
| | | |
| | | def loadDoc(self): |
| | | doc = Document('./static/doc/ZLæ ¼å¼(å
¬å¼).docx') |
| | | # æç
§æ é¢è·å段è½çå±çº§ç»æ |
| | | titles = [] |
| | | for paragraph in doc.paragraphs: |
| | | if paragraph.text != "": |
| | | # æåä¸ä¸ºç©º |
| | | if paragraph.style.base_style is not None: |
| | | # æbase_style |
| | | if paragraph.style.base_style.name.startswith('Heading'): |
| | | # æ¯æ é¢ |
| | | level = int(paragraph.style.base_style.name.split(' ')[-1]) |
| | | obj = {} |
| | | obj["level"] = level |
| | | obj["text"] = paragraph.text |
| | | obj["child"] = [] |
| | | titles.append(obj) |
| | | else: |
| | | length = len(titles) |
| | | if "child" in titles[length -1]: |
| | | obj = {} |
| | | obj["text"] = paragraph.text |
| | | titles[length -1]['child'].append(obj) |
| | | else: |
| | | # 没æbase_style |
| | | length = len(titles) |
| | | obj = {} |
| | | obj["text"] = paragraph.text |
| | | if length > 0 and "child" in titles[length -1]: |
| | | # å¦ææ¯æ é¢å
çappendè¿æ é¢çchild |
| | | titles[length -1]['child'].append(obj) |
| | | else: |
| | | # éæ é¢å
çç´æ¥æ¾å¨ç¬¬ä¸å± |
| | | titles.append(obj) |
| | | else: |
| | | # æå为空æ¶ï¼å¯è½æ¯å¾çæè
è¡¨æ ¼ |
| | | if self.has_image(paragraph): |
| | | # å½å段è½ä¸ºå¾ç |
| | | obj = {} |
| | | # è·åå¾ççblob |
| | | img = self.get_image_blob(paragraph) |
| | | if img is not None: |
| | | imgBase64 = self.convert_blob_to_png_base64(img) |
| | | if imgBase64 is not None: |
| | | obj["imgBase64"] = imgBase64 |
| | | titles[length -1]['child'].append(obj) |
| | | # å¨è¿éæ©å±å¤æè¡¨æ ¼ |
| | | print(titles) |
| | | # for para in doc.paragraphs: |
| | | # print(para.text) |
| | | # print('------------------------') |
| | | |
| | | if __name__ == '__main__': |
| | | vision = VisionTest("./images/test.png") |
| | | # vision.run("é®é¢") |
| | | vision.loadDoc() |