VB联合Python开发

用到 Python，首先你肯定得装一个Python吧，用3.x的

创建一个.py文件

说明：需要引用pythoncom  (这个需要pip install pywin32 )

import pythoncom
class PythonUtilities:

    _public_methods_=['SplitString']
    _public_methods_ = ['httpBom']
    _reg_progid_='PythonDemos.Utilities'
    _reg_clsid_=pythoncom.CreateGuid()

    def SplitString(self, val, item=None):
        import string
        if item !=None:
            item=str(item)
        val=str(val)
        return val.split(item)

    def httpBom(self):
        return  "hhh"

if __name__=='__main__':
    print ('Registering COM server...')
    import win32com.server.register
    win32com.server.register.UseCommandLine(PythonUtilities)
然后在VB中就可以使用Python中的方法了

Private Sub Command1_Click()
    Set PythonUtils = CreateObject("PythonDemos.Utilities")
'    response = PythonUtils.SplitString("Hello from VB My name is Leo Liu")
'    For Each Item In response
'        List1.AddItem Item
'    Next
    MsgBox PythonUtils.httpBom
End Sub
最终的使用方法

cmd 中 输入 ： Python PythonDemos.py

然后VB就可以使用Python这个类中发方法了

如果想要注销这个COM

python PythonDemos.py --unregister 

然后将这个文件打包成 exe:

安装 pin install pyinstaller

Pyinstaller -F WFJieKouPY.py    #打包成多文件
Pyinstaller -D WFJieKouPY.py    #打包成单文件

注意：需要引用 import win32timezone  否则报错

import pypyodbc
import os
import requests
import pythoncom
import json
import win32com.server.register
import win32timezone
class PythonUtilities:
    _public_methods_ = ['FindBoms', 'SendBoms', 'test', 'GetData']
    _reg_progid_ = 'WFJieKouPY.Utilities'
    _reg_clsid_ = pythoncom.CreateGuid()

    accountId = ""
    tenantId = "sxqcdev"
    appId = "DEVTEST"
    appSecuret = "!QAZ2wsx#EDC4rfv"
    language = "zh_CN"

    user = "SQESB"
    usertype = "UserName"

    appTokenUrl = r'http://172.16.127.102:8081/ierp/api/getAppToken.do'
    accessTokenUrl = r'http://172.16.127.102:8081/ierp/api/login.do'
    pushBomUrl = r'http://172.16.127.102:8081/ierp/kapi/app/iscb/sf_bom_save'
    findBomUrl = r'http://172.16.127.102:8081/ierp/kapi/app/iscb/sf_bom_query'
    # 获取密码
    def GetKey(self):
        #####获取app_token
        data = {
            "accountId": self.accountId,
            "tenantId": self.tenantId,
            "appId": self.appId,
            "appSecuret": self.appSecuret,
            "language": self.language
        }
        headers = {
            "Content-Type": "application/json",
        }
        res = requests.post(url=self.appTokenUrl, json=data, headers=headers)
        res = res.json()
        if (res['state'] == "success"):
            app_token = res['data']['app_token']
            # print("app_token:" + app_token)
        else:
            return ""

        #########获取access_token
        data = {
            "user": self.user,
            "apptoken": app_token,
            "tenantId": self.tenantId,
            "accountId": self.accountId,
            "usertype": self.usertype
        }
        headers = {
            "Content-Type": "application/json",
        }
        res = requests.post(url=self.accessTokenUrl, json=data, headers=headers)
        res = res.json()
        if (res['state'] == "success"):
            access_token = res['data']['access_token']
            # print("access_token:" + access_token)
        else:
            return ""

        return access_token  # 返回密钥

    # # 取数据整合出成json格式
    # def GetData(self, path=os.getcwd() + r'\data\data.mdb'):
    #
    #     # 连接数据库
    #     # path = os.getcwd() + r'\data\data.mdb'  # 数据库文件
    #     # path = 'E:\万方接口程序\data\data.mdb'
    #
    #     conn = pypyodbc.connect(r"Driver={Microsoft Access Driver (*.mdb, *.accdb)};DBQ=" + path + ";Uid=;Pwd=;")
    #     cursor = conn.cursor()
    #     # ===========================获取某个表数据内容======================================================
    #     SQL_value = "SELECT * FROM 万方BOM表  ORDER BY material"  # 查表
    #     cursor.execute(SQL_value)
    #     result = cursor.fetchall()  # 生成表格
    #     data = {
    #         "boms": []
    #     }
    #     lastBom = ""
    #     if (len(result) > 0):
    #         for row in result:  # 按行输出
    #
    #             if (len(data['boms']) == 0 or lastBom != row[3]):  # 空的直接添加)
    #
    #                 data['boms'].append({
    #                     "createorg_number": row[1],
    #                     "version_number": row[2],
    #                     "material": row[3],
    #                     "wzta_sccj": row[4],
    #                     "wzta_versionremark": row[5],
    #                     "wzta_luxian": row[6],
    #                     "wzta_chexing": row[7],
    #                     "entry": []
    #                 })
    #                 data['boms'][len(data['boms']) - 1]['entry'].append({
    #                     "entrymaterial_number": row[8],
    #                     "wzta_tzyl": float(row[9]),
    #                     "wzta_gongweihao": row[10],
    #                     "wzta_zijiaoxianhao": row[11],
    #                     "wzta_weizhihao": row[12],
    #                     "wzta_tuzhidaihao": row[13],
    #                     "wzta_tuzhichicun": row[14],
    #                     "wzta_gongyiremark": row[15],
    #                     "entryqtynumerator": float(row[16]),
    #                     "entryoperationnumber": row[17],
    #                 })
    #             else:
    #                 data['boms'][len(data['boms']) - 1]['entry'].append({
    #                     "entrymaterial_number": row[8],
    #                     "wzta_tzyl": float(row[9]),
    #                     "wzta_gongweihao": row[10],
    #                     "wzta_zijiaoxianhao": row[11],
    #                     "wzta_weizhihao": row[12],
    #                     "wzta_tuzhidaihao": row[13],
    #                     "wzta_tuzhichicun": row[14],
    #                     "wzta_gongyiremark": row[15],
    #                     "entryqtynumerator": float(row[16]),
    #                     "entryoperationnumber": row[17],
    #
    #                 })
    #             lastBom = row[3]
    #
    #         # print(json.dumps(data,sort_keys=True,indent=4,ensure_ascii=False))
    #
    #     cursor.close()
    #     conn.close()
    #     return data
    #     # return json.dumps(data,ensure_ascii=False)

    # 查询物料是否存在
    def FindBoms(self, material, createorg_number='351'):
        # arr = []
        # arr.append(material)
        if material == "" :
            return
        data = {
            "number": material.split(","),
            "createorg_number": createorg_number
        }
        accessToken = self.GetKey()
        # print(accessToken)
        headers = {
            "Content-Type": "application/json",
            "accessToken": accessToken
        }
        res = requests.post(url=self.findBomUrl, json=data,
                            headers=headers)
        return json.dumps(res.json(), ensure_ascii=False)

    # # 推送BOM
    # def SendBoms(self, path=os.getcwd() + r'\data\data.mdb'):
    #     data = self.GetData(path)
    #     if (len(data["boms"]) <= 0):
    #         return {
    #             "status": False,
    #             "message": "数据表中没有待发送的物料。"
    #         }
    #
    #     accessToken = self.GetKey()
    #     # print(accessToken)
    #     headers = {
    #         "Content-Type": "application/json",
    #         "accessToken": accessToken
    #     }
    #     res = requests.post(url=self.pushBomUrl, json=data,
    #                         headers=headers)
    #     return json.dumps(res.json(), ensure_ascii=False)


    # # 推送BOM
    def SendBoms(self, bomStr):

        if bomStr == "":
            return
        boms = [] #存放推送数组
    #     先分组
        materialFZArr = bomStr.split("‖")
        for fzStr in materialFZArr:
            # 头部分组
            singleArr = fzStr.split("※")
            headArr = singleArr[0].split("→")
            boms.append({
                "createorg_number": headArr[0],
                "version_number":  headArr[1],
                "material":  headArr[2],
                "wzta_sccj":  headArr[3],
                "wzta_versionremark":  headArr[4],
                "wzta_luxian":  headArr[5],
                "wzta_chexing":  headArr[6],
                "entry": []
            })
            for zjStr in singleArr[1].split("▷"):
                zjStrArr = zjStr.split("→")
                boms[len(boms)-1]["entry"].append({
                    "entryoperationnumber": zjStrArr[0],
                    "entrymaterial_number": zjStrArr[1],
                    "entryqtynumerator": float(zjStrArr[2]),
                    "wzta_tzyl": float(zjStrArr[3]),
                    "wzta_gongweihao": zjStrArr[4],
                    "wzta_zijiaoxianhao": zjStrArr[5],
                    "wzta_weizhihao": zjStrArr[6],
                    "wzta_tuzhidaihao": zjStrArr[7],
                    "wzta_tuzhichicun": zjStrArr[8],
                    "wzta_gongyiremark":zjStrArr[9]
                })

        #print(boms)
        # return boms

        accessToken = self.GetKey()
        # print(accessToken)
        headers = {
            "Content-Type": "application/json",
            "accessToken": accessToken
        }
        res = requests.post(url=self.pushBomUrl, json=boms,
                            headers=headers)
        return json.dumps(res.json(), ensure_ascii=False)

    def test(self, str='zhw'):
        return str


print('Registering COM server...')


win32com.server.register.UseCommandLine(PythonUtilities)

input("运行成功！Press <Enter>")

# if __name__ == '__main__':
#     print('Registering COM server...')
#     import win32com.server.register
#
#     win32com.server.register.UseCommandLine(PythonUtilities)
#
#     input("运行成功！")