一次使用Python连接数据库生成二维码并安装为windows服务的工作任务

最近有一个需求，在现有生产系统上的人员库中增加一个此人员关键信息的二维码，支持文字版和跳转版两种方式，与报表工具关联，可打印。以windows服务方式，定时检查，只要发现某人员没有此二维码信息，就生成并写入人员库。 

　　决定使用Python。

　　在此之前，没有用Python实现过其中的任何一个工作。对于习惯于微软开发环境下的程序员，使用开源系统，几乎一步一个坎，不过确实挺简单的。

　　整体设想，主要工作有以下几个步骤：

 1.生成二维码

 2.建立连接，读取数据

 3.根据读取的数据生成二维码

 4.二维码写入数据库

 5.与报表工具关联，显示在报表上

 6.写日志

7.生成windows服务

下面分步叙述过程。

1.   生成二维码

使用QRCode

安装命令pip install QRCode，会自动下载并安装与当前版本匹配插件。

 

代码如下（来自网络，稍加改动，保留原作者声明，下同）：

# coding: utf-8
"""
filename: qrcode.py
Created by Tacey Wong at 16-9-22 下午10:34 
"""

#import zbar
import qrcode
from PIL import Image
import os, sys


def gen_qrcode(string, path, logo=""):
    """
    生成中间带logo的二维码
    需要安装qrcode, PIL库
    @参数 string: 二维码字符串
    @参数 path: 生成的二维码保存路径
    @参数 logo: logo文件路径
    @return: None
    """
    
    qr = qrcode.QRCode(
        version=2,
        error_correction=qrcode.constants.ERROR_CORRECT_H,
        box_size=8,
        border=1
    )
    qr.add_data(string)
    qr.make(fit=True)
    img = qr.make_image()
    #img = img.convert("RGBA")
    img = img.convert("RGB")
    if logo and os.path.exists(logo):
        try:
            icon = Image.open(logo)
            img_w, img_h = img.size
        except(Exception) as e:
            print(e)
            sys.exit(1)
        factor = 4
        size_w = int(img_w / factor)
        size_h = int(img_h / factor)

        icon_w, icon_h = icon.size
        if icon_w > size_w:
            icon_w = size_w
        if icon_h > size_h:
            icon_h = size_h
        icon = icon.resize((icon_w, icon_h), Image.ANTIALIAS)

        w = int((img_w - icon_w) / 2)
        h = int((img_h - icon_h) / 2)
        #icon = icon.convert("RGBA") #png图像使用
        icon = icon.convert("RGB")
        img.paste(icon, (w, h), icon)
    img.save(path)
    # 调用系统命令打开图片
    # xdg - open(opens a file or URL in the user's preferred application)
    #os.system('xdg-open %s' % path)


##def decode_qrcode(path):
##    """
##    解析二维码信息
##    @参数 path: 二维码图片路径
##    @return: 二维码信息
##    """
##    # 创建图片扫描对象
##    scanner = zbar.ImageScanner()
##    # 设置对象属性
##    scanner.parse_config('enable')
##    # 打开含有二维码的图片
##    img = Image.open(path).convert('L')
##    # 获取图片的尺寸
##    width, height = img.size
##    # 建立zbar图片对象并扫描转换为字节信息
##    qrCode = zbar.Image(width, height, 'Y800', img.tobytes())
##    scanner.scan(qrCode)
##    # 组装解码信息
##    data = ''
##    for s in qrCode:
##        data += s.data
##    # 删除图片对象
##    del img
##    # 输出解码结果
##    return data

def SetQRCode(info,imgFileName=""):
    #if __name__ == "__main__":
    #info = """"""
    #pic_path = "QR" + imgFileName + ".jpg"
    pic_path = "QR.jpg"
    icon_path = "logo.png"
    logo_path = ""
    gen_qrcode(info, pic_path,logo_path )
    #print(decode_qrcode(pic_path))

2.   建立连接，读取数据

本步骤使用cx_Oracle，安装方法同上

>>>pip install cx_Oracle

 需要oracle客户端，下载地址：https://oracle.github.io/odpi/doc/installation.html#windows，找到适合你的版本，下载zip版本以后，解压，把oracle客户端路径加到系统路径中。

 

代码如下：

import cx_Oracle
import logging
import inspect
import os



#建立和数据库系统的连接

conn = cx_Oracle.connect('用户名', '密码', '数据库服务器IP/oracle服务名') #建立连接，3 个参数分开写

cursor = conn.cursor()

def updateData(fid,value):
    param={'fieldname1': fieldname1, 'fieldname2': fieldname2}
    
    cursor.execute("""UPDATE tablename
                        SET FQRCODE=:fieldname1
                        WHERE DICT_OID=:fieldname2 """,param)
    #cursor.close();
    conn.commit();
    #conn.close();
 
def openImage():

    file=open("qr.jpg",'rb');
    img=file.read();
    file.close()
    
    return img
    

def searchAll():
    
    import gqrcode
    
    cursor.execute('''
                        SELECT *
                            
                      FROM tablename n
                ''')
    #personInfoTuple=cursor.fetchone() #获取一条

    #personInfoTuple=cursor.fetchmany(2) #获取2条

    personInfoTuple = cursor.fetchall() #获取全部
    
    import datetime;
    import time;

    listIds=[]
    for x in personInfoTuple:
        
        authcode = x[5]
        strInfo = "网址”

        now = datetime.datetime.now();
        now = now.strftime('%Y%m%d%H%M%S%MS')
        clock = time.clock();
        filename=x[4]+now
       
        gqrcode.SetQRCode(strInfo) #生成二维码

3.   根据读取的数据生成二维码

此处需要安装PIL，即pillow

>>>pip install pillow

 imgStream=openImage(); #以二进制流打开图片

 listIds.append(x[0])

4.   二维码写入数据库

        updateData(x[0], imgStream);#入库

5.   与报表工具关联，显示在报表上

以二进制流的方式入库以后，在报表工具中，以图片控件读取字段即可，大多数报表工具都支持。我用的是如意报表。

6.   写日志

Python里面另有loggning模块，因为任务的时间紧迫，我没有仔细研究，直接写文本文件。
//日志
def logger(msg):
    try:    
        logFileName="logging_test.log"
        logging.basicConfig(filename=logFileName,filemode='a',level=logging.DEBUG)
        logging.debug(msg)
    except Exception as ex:
        logging.debug(ex);

7.   生成windows服务

这一步我浪费了很多时间。启动服务时，在有些机器上报错，有些不报错，没有找到规律，如果有人在事件查看器中发现“等待服务的连接超时（30000）毫秒”，我没有解决。

这一步需要安装pywin32，理论上用pip install pywin32即可，但是，我用的是p3.6.3，PyInstaller的官方网站似乎没有此版本，我没找到。我在CSDN上找到的此版本，说是官方的。链接：http://download.csdn.net/download/battlewolf_hua/9827297

 

我为了检测此服务需要多少内存，需要下载安装psutil

 

>>>pip install psutil

 代码如下：

#encoding=utf-8  
import win32serviceutil   
import win32service   
import win32event  
import os   
import logging  
import inspect
import servicemanager
import sys
import datetime
import oracleTest
import psutil
  
class ServiceTest (win32serviceutil.ServiceFramework):   

    info = psutil.virtual_memory()
    _totalMemory = round(info.total /1024 /1024/1024,2) #总内存GB
    
    _thisMemory = psutil.Process(os.getpid()).memory_info().rss #本程序使用内存KB
    
    _now = datetime.datetime.now();
    _prefix = _now.strftime('%Y%m%d')
    _svc_name_ = "ConstructorNoBuildingService"  #服务名
    #_svc_name_ = "PythonService"  #服务名
    
    _svc_display_name_ = "ServiceTest"  #服务在windows系统中显示的名称
    _svc_description_ = "二维码服务"  #服务的描述
  
    def __init__(self, args):   
        win32serviceutil.ServiceFramework.__init__(self, args)   
        self.hWaitStop = win32event.CreateEvent(None, 0, 0, None)  
        self.logger = self._getLogger()
        
        self.run = True  
          
    def _getLogger(self):  
         
        logger = logging.getLogger('[ServiceTest]')  
        
        this_file = inspect.getfile(inspect.currentframe())  
        dirpath = os.path.abspath(os.path.dirname(this_file))
        handler = logging.FileHandler(os.path.join(dirpath, "service" + self._prefix + ".log"))  
          
        formatter = logging.Formatter('%(asctime)s %(name)-12s %(levelname)-8s %(message)s')  
        handler.setFormatter(formatter)  
          
        logger.addHandler(handler)  
        logger.setLevel(logging.DEBUG)
 
        
        return logger

    _hour = _now.strftime('%H')
    def SvcDoRun(self):  
        import time
 
##            self.logger.info("service is run....")
        while self.run :  
    ##                self.logger.info("I am runing....")
            
            listUpdatedIds = oracleTest.searchAll()
            if(len(listUpdatedIds)==0):
                self.logger.info("无数据，服务空载")
            else:
                self.logger.info(listUpdatedIds)

            self.logger.info("使用内存"+str(self._thisMemory/1024)+"KB")
            time.sleep(30) #执行任务间隔，相当于timer控件中的interval
               
    def SvcStop(self):   
        self.logger.info("service was stopped.")  
        self.ReportServiceStatus(win32service.SERVICE_STOP_PENDING)   
        win32event.SetEvent(self.hWaitStop)   
        self.run = False

  
if __name__=='__main__':
    print("主程序启动")
##    print(len(sys.argv));
    logger(len(sys.argv));
    if len(sys.argv) == 1:
        try:
            evtsrc_dll = os.path.abspath(servicemanager.__file__)
            servicemanager.PrepareToHostSingle(ServiceTest)
            servicemanager.Initialize(ServiceTest, evtsrc_dll)
            servicemanager.StartServiceCtrlDispatcher()
        except win32service.error as details:
            import winerror
            if details == winerror.ERROR_FAILED_SERVICE_CONTROLLER_CONNECT:
                print("错误发生")
                win32serviceutil.usage()
    else:
        print("否则，执行win32serviceutil.HandleCommandLine(ServiceTest)")
        try:
            win32serviceutil.HandleCommandLine(ServiceTest)
        except Exception as e:
            print(e)
            win32serviceutil.usage()

8.   安装服务

命令行，源文件所在目录

Python serviceTest.py install

Good luck

 

 

 

【推广】 免费学中医，健康全家人
原文地址：https://www.cnblogs.com/Sabre/p/8076733.html