Python操作Word文档（python-docx）

╰+哭是因爲堅強的太久メ 2023-02-24 13:25 8阅读 0赞

**需求：生成word文档并返回url地址:**

# -*- coding:utf-8 -*-
    # 消费者
    import base64
    import json
    import os
    import smtplib
    import time
    from email.mime.multipart import MIMEMultipart
    from email.mime.text import MIMEText
    
    import docx
    import pika
    import pytz
    import requests
    
    
    address = 'localhost:8901'
    
    auth = pika.PlainCredentials('guest', 'guest')  # auth info
    connection = pika.BlockingConnection(pika.ConnectionParameters(
        '127.0.0.1', 5672, '/', auth))  # connect to rabbit
    channel = connection.channel()  # create channel
    
    queueName = 'WORD-TEMPLATE'
    # 申明队列
    channel.queue_declare(queue=queueName, durable=True, passive=True)
    
    template_list = ['company.docx', 'people.docx']
    index_map = {}
    tz = pytz.timezone('Asia/Shanghai')
    
    
    # 发送短信
    class SendMail(object):
        def __init__(self, username, passwd, recv, title, content,
                     file=None, ssl=False,
                     email_host='smtp.qq.com', port=25, ssl_port=465):
            '''
            :param username: 用户名
            :param passwd: 密码
            :param recv: 收件人，多个要传list ['a@qq.com','b@qq.com]
            :param title: 邮件标题
            :param content: 邮件正文
            :param file: 附件路径，如果不在当前目录下，要写绝对路径，默认没有附件
            :param ssl: 是否安全链接，默认为普通
            :param email_host: smtp服务器地址，默认为163服务器
            :param port: 非安全链接端口，默认为25
            :param ssl_port: 安全链接端口，默认为465
            '''
            self.username = username  # 用户名
            self.passwd = passwd  # 密码
            self.recv = recv  # 收件人，多个要传list ['a@qq.com','b@qq.com]
            self.title = title  # 邮件标题
            self.content = content  # 邮件正文
            self.file = file  # 附件路径，如果不在当前目录下，要写绝对路径
            self.email_host = email_host  # smtp服务器地址
            self.port = port  # 普通端口
            self.ssl = ssl  # 是否安全链接
            self.ssl_port = ssl_port  # 安全链接端口
    
        def send_mail(self):
            msg = MIMEMultipart()
            # 发送内容的对象
            if self.file:  # 处理附件的
                file_name = os.path.split(self.file)[-1]  # 只取文件名，不取路径
                try:
                    f = open(self.file, 'rb').read()
                except Exception as e:
                    raise Exception('附件打不开！！！！')
                else:
                    att = MIMEText(f, "base64", "utf-8")
                    att["Content-Type"] = 'application/octet-stream'
                    # base64.b64encode(file_name.encode()).decode()
                    new_file_name = '=?utf-8?b?' + base64.b64encode(file_name.encode()).decode() + '?='
                    # 这里是处理文件名为中文名的，必须这么写
                    att["Content-Disposition"] = 'attachment; filename="%s"' % (new_file_name)
                    msg.attach(att)
            msg.attach(MIMEText(self.content))  # 邮件正文的内容
            msg['Subject'] = self.title  # 邮件主题
            msg['From'] = self.username  # 发送者账号
            msg['To'] = ','.join(self.recv)  # 接收者账号列表
            if self.ssl:
                self.smtp = smtplib.SMTP_SSL(self.email_host, port=self.ssl_port)
            else:
                self.smtp = smtplib.SMTP(self.email_host, port=self.port)
            # 发送邮件服务器的对象
            self.smtp.login(self.username, self.passwd)
            try:
                self.smtp.sendmail(self.username, self.recv, msg.as_string())
                pass
            except Exception as e:
                print('发送失败', e)
            else:
                print('发送成功！')
            self.smtp.quit()
    
    
    def uploadWord(id, new_name, key):
        """
        上传文件
        :param param:
        :param temps:
        :param param1:
        :return:
        """
        # 重试三次
        for i in range(3):
            try:
                files = {"file": open(new_name, "rb")}
                # 超时时间为3分钟
                url = "http://{}/api/v2/unsecure/upload/word?key={}&id={}".format(
                    address, key, id)
                r = requests.post(url, headers="", files=files)
                if r.status_code == 200:
                    if r.json()['data']:
                        print("upload video:" + r.text)
                        return True
                    else:
                        print("失败")
                else:
                    print("upload video yes:" + r.text)
                    continue
            except Exception as e:
                print(e)
                pass
        return False
    
    
    def people_create(temp, name, jsons):
        """
        创建人员报表
        :param temp:
        :param name:
        :param jsons:
        :return:
        """
        doc = docx.Document(temp)
        # 每一段的编号、内容
        for i in range(len(doc.paragraphs)):
            print(str(i), doc.paragraphs[i].text)
    
            # # 合同编号
            # if i == 1:
            #     style = doc.paragraphs[i].style
            #     font = style.font
            #     print(style.font)
            #     text = "健康体检（单位）{}-{}号".format(str(time.strftime('%Y', time.localtime(time.time()))), jsons["number"])
            #     p = doc.paragraphs[i]
            #     p.clear()
            #     p.style = style
            #     run1 = p.add_run(text).font
            #     run1 = font
    
        tbs = doc.tables
        for tb in tbs:
            for row in range(len(tb.rows)):
                for cell in range(len(tb.rows[row].cells)):
                    print(row, cell, tb.cell(row, cell).text)
                    style = tb.cell(row, 0).paragraphs[0].style
                    p = tb.cell(row, cell).paragraphs[0]
                    p.style = style
                    if row == 0 and cell == 1:
                        p.add_run('{}'.format(jsons["xm"]))
                    if row == 0 and cell == 4:
                        p.add_run('{}'.format(jsons["xb"]))
                    if row == 0 and cell == 6:
                        p.add_run('{}'.format(jsons["nl"]))
    
                    if row == 1 and cell == 1:
                        p.add_run('{}'.format(jsons["sfz"]))
                    if row == 1 and cell == 4:
                        p.add_run('{}'.format(jsons["mob"]))
                    if row == 1 and cell == 6:
                        p.add_run('{}'.format(jsons["gzmc"]))
    
                    if row == 2 and cell == 1:
                        p.add_run('{}'.format(jsons["companyName"]))
                    if row == 3 and cell == 1:
                        p.add_run('{}'.format(jsons["address"]))
                    if row == 4 and cell == 1:
                        p.add_run('{}'.format(jsons["registration"]))
                    if row == 5 and cell == 1:
                        p.add_run('{}'.format(jsons["hosmc"]))
                    if row == 6 and cell == 1:
                        p.add_run('{}'.format(jsons["jsrq"]))
                    if row == 6 and cell == 6:
                        p.add_run('{}'.format(jsons["dqrq"]))
        doc.save(name)
    
    
    def company_create(temp, name, jsons):
        """
        创建公司报表
        :param temp: 模板
        :param name: 文件名
        :param jsons: json
        :return:
        """
        doc = docx.Document(temp)
    
        # 每一段的编号、内容
        for i in range(len(doc.paragraphs)):
            print(str(i), doc.paragraphs[i].text)
    
            # 合同编号
            if i == 1:
                style = doc.paragraphs[i].style
                font = style.font
                print(style.font)
                text = "健康体检（单位）{}-{}号".format(str(time.strftime('%Y', time.localtime(time.time()))), jsons["number"])
                p = doc.paragraphs[i]
                p.clear()
                p.style = style
                run1 = p.add_run(text).font
                run1 = font
    
        tbs = doc.tables
        for tb in tbs:
            for row in range(len(tb.rows)):
                for cell in range(len(tb.rows[row].cells)):
                    print(row, cell, tb.cell(row, cell).text)
                    style = tb.cell(row, 0).paragraphs[0].style
                    p = tb.cell(row, cell).paragraphs[0]
                    p.style = style
                    if row == 0 and cell == 1:
                        p.add_run('\n{}'.format(jsons["companyName"]))
                    if row == 1 and cell == 1:
                        p.add_run('\n{}'.format(jsons["address"]))
                    if row == 2 and cell == 1:
                        p = p.clear()
                        p.add_run('\n{}'.format(jsons["registration"]))
        doc.save(name)
    
    
    # 回调
    def callback(ch, method, properties, body):
        # 解析json
        jsons = json.loads(str(body.decode()))
        try:
            print(jsons)
            if jsons["type"] == '1':
                temp = ['people.docx']
            elif jsons["type"] == '2':
                temp = ['company.docx']
            creat_temp = []
            for itm in temp:
                docx_name = str(jsons["number"] + "_" + itm.split('.')[0]) + ".docx"
                # 生成
                if jsons["type"] == '1':
                    people_create(itm, docx_name, jsons)
                elif jsons["type"] == '2':
                    company_create(itm, docx_name, jsons)
                creat_temp.append(docx_name)
    
            for temps in creat_temp:
                data = uploadWord(jsons["id"], temps, jsons["key"])
                if data:
                    creat_temp.remove(temps)
                    os.remove(temps)
                else:
                    m = SendMail(
                        username='xxxxxxxxxx@QQ.com',
                        passwd='xxxxxxxxxx',
                        recv=['xxxxxxxxxx@QQ.com'],
                        title='Word文件生成失败',
                        content='文件信息:' + str(jsons),
                        file=r"{" + 'contract.zip' + "}",
                        ssl=True,
                    )
                    m.send_mail()
            # 删除
            for temps in creat_temp:
                os.remove(temps)
    
        except:
            m = SendMail(
                username='xxxxxxxxxx@QQ.com',
                passwd='xxxxxxxxxx',
                recv=['xxxxxxxxxx@QQ.com'],
                title='Word文件生成失败',
                content='文件信息:' + str(jsons) + "\n请手动上传",
                ssl=True,
            )
            m.send_mail()
    
    
    # no_ack 设置成 False，在调用callback函数时，未收到确认标识，消息会重回队列
    # True，无论调用callback成功与否，消息都被消费掉
    channel.basic_consume(queueName, callback,
                          auto_ack=True)
    channel.start_consuming()

@GetMapping("/health/certificate/template")
        public RestObject getHealthCertificateTemplate(JwtAuthenticationToken token,
                                                       @RequestParam(value = "type") Integer type,
                                                       @RequestParam(value = "id") Integer id
        ) {
            String url = null;
            HandleRecord handleRecord = null;
            Map<String, String> map = new HashMap<>();
            //1走人员2走公司
            if (type == 1) {
                SimpleDateFormat f = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                TairuoJkz1 tairuoJkz1 = healthCertificateService.getHealthCertificateDetail(id);
                map.put("xm", tairuoJkz1.getXm());
                map.put("xb", tairuoJkz1.getXb());
                map.put("nl", tairuoJkz1.getNl().toString());
    
                map.put("sfz", tairuoJkz1.getSfz());
                map.put("mob", tairuoJkz1.getMob());
                map.put("gzmc", tairuoJkz1.getGzmc());
    
                map.put("companyName", tairuoJkz1.getQymc());
                map.put("address", tairuoJkz1.getDz());
                map.put("registration", tairuoJkz1.getQyzzh());
    
                map.put("hosmc", tairuoJkz1.getHosmc());
                map.put("jsrq", f.format(tairuoJkz1.getJsrq()));
                map.put("dqrq", f.format(tairuoJkz1.getDqrq()));
                handleRecord = healthCertificateService.getHandleRecord(1, tairuoJkz1.getBh(), null);
            } else {
                HealthCertificateCompany healthCertificateCompany = healthCertificateService.getHealthCertificateCompanyDetail(id);
                map.put("companyName", healthCertificateCompany.getCompanyName());
                map.put("address", healthCertificateCompany.getAddress());
                map.put("registration", healthCertificateCompany.getRegistration());
                map.put("companyId", healthCertificateCompany.getCompanyId().toString());
                handleRecord = healthCertificateService.getHandleRecord(2, null, healthCertificateCompany.getCompanyId());
            }
            //如果存在健康证记录下载记录则返回文件地址
            //否则生成HandleRecord
            if (handleRecord != null && handleRecord.getFileUrl() != null && !handleRecord.getFileUrl().equals("")) {
                return new RestObject(handleRecord.getFileUrl());
            } else {
                handleRecord = healthCertificateService.createHandleRecord(type, id);
            }
            map.put("number", handleRecord.getNumber().toString());
            map.put("type", type.toString());
            map.put("id", handleRecord.getId().toString());
            //创建随机数生成唯一key
            String key = "";
            while (true) {
                key = randomCode();
                Object value = redisTemplate.opsForValue().get(key);
                if (value == null) {
                    break;
                }
            }
            map.put("key", key);
            //设置过期时间1分钟
            redisTemplate.opsForValue().set(key, key);
            redisTemplate.expire(key, 1, TimeUnit.MINUTES);
            amqpTemplate.convertAndSend(queue.getName(), JSON.toJSONString(map));
            //如果key的value未修改或过期则休眠1s
            while (redisTemplate.opsForValue().get(key) != null) {
                Object value = redisTemplate.opsForValue().get(key);
                if (value != null && String.valueOf(value).equals(key)) {
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                } else {
                    if (value != null) {
                        return new RestObject(String.valueOf(value));
                    }
                }
            }
            return new RestObject("");
        }
    
    
        /**
         * 随机数生成
         *
         * @return
         */
        private String randomCode() {
            String codeStr = Long.toString(System.currentTimeMillis());
            return codeStr.substring(codeStr.length() - 6);
        }

/**
         * word文件上传地址
         *
         * @param file
         * @return
         */
        @PostMapping("/upload/word")
        public RestObject uploadWord(@RequestParam(value = "file", required = false) MultipartFile file,
                                     @RequestParam(value = "key", required = false) String key,
                                     @RequestParam(value = "id", required = false) Integer id) {
    
            try {
                String fileName = file.getOriginalFilename();
                log.info("上传模板生成jpg文件:{}", fileName);
                //1.获取摄像头序列号
                String fileUrl = fileUploadService.uploadFile(file, "companyfile1",
                        "department/" + id.toString() + "_dept_" + UUID.randomUUID().toString() + ProcessingFileUtils.getSuffixName(file), null);
                healthCertificateService.updateHandleRecordUrl(id, fileUrl);
                redisTemplate.opsForValue().set(key, fileUrl);
                return new RestObject(true);
            } catch (Exception e) {
                log.error("上传模板生成jpg文件失败" + e.toString());
                redisTemplate.delete(key);
                return new RestObject(false);
            }
        }

**大致代码如上！**

**首先获取基础信息，然后封装成JSON发送至MQ，然后通过生成的key存入Redis中，根据key获取图片地址并返回前端。**

### **设置字体DEMO：** ###

import docx
    from docx.shared import Pt
    from docx.oxml.ns import qn
    
    temp = 'company.docx'
    doc = docx.Document(temp)
    
    # 每一段的编号、内容
    for i in range(len(doc.paragraphs)):
        print(str(i), doc.paragraphs[i].text)
        if i == 0:
            text = "测试" + doc.paragraphs[i].text
            p = doc.paragraphs[i]
            p.clear()
            run = p.add_run(text)
            # 设置字体
            run.font.size = Pt(14)
            run.font.bold = True
            run.font.name = u'楷体'
            r = run._element
            r.rPr.rFonts.set(qn('w:eastAsia'), u'楷体')
    
    tbs = doc.tables
    for tb in tbs:
        for row in range(len(tb.rows)):
            for cell in range(len(tb.rows[row].cells)):
                print(row, cell, tb.cell(row, cell).text)
                if row < 5 and cell == 1:
                    style = tb.cell(row, 0).paragraphs[0].style
                    p = tb.cell(row, cell).paragraphs[0]
                    p.style = style
                    p.add_run('\n我我我我')
    
    doc.save('demo.docx')