用本地标签打印机 自动打印订单

来源:17-1 课程总结

慕妹4270370

2019-08-14

文件结构
图片描述

sql_util.py

# 导包
import pymysql


class SQLTool:
    def __init__(self, user, pwd, host, port, db):
        self.client = pymysql.connect(
            user=user,
            password=pwd,
            host=host,
            port=port,
            database=db
        )

        self.cursor = self.client.cursor()

    def query(self, sql):
        # 调用原生sql语句
        self.cursor.execute(sql)
        aaa = self.fetch_all_to_dict(self.cursor)
        return aaa

    def fetch_all_to_dict(self, cursor):

        print(cursor.description)

        desc = [i[0] for i in cursor.description]

        result = [dict(zip(desc, col)) for col in cursor.fetchall()]
        return result


import datetime
from application import app, db
from common.models.pay.PayOrder import PayOrder


class JobTask():
    """
    python manager.py runjob -m mytask/index
    """

    def __init__(self):
        pass

    def run(self, params):

        from .sql_util import SQLTool
        # 调用SQLTool这个类
        tool = SQLTool(
            "root",
            "123456",
            "127.0.0.1",
            3306,
            "food_db"
        )
        # print(tool)
        # 原生sql语句
        sql = "select * from pay_order;"
        # 调用类里的方法
        res = tool.query(sql)
        print(res)
        return "ok"

执行

python manager.py runjob -m mytask/index

成功执行 任务且 打印了订单信息
效果图
图片描述

这个是我的自动5秒 调用标签打印机的代码 且目前能5秒打印一次
print.py

from flask import Flask
from flask_apscheduler import APScheduler
import tempfile
import win32api
import win32print


class Config(object):  # 创建配置用类
    # 任务列表
    JOBS = [
        # {  # 第一个任务
        #     'id': 'job1',
        #     'func': '__main__:job_1',
        #     'args': (1, 2),
        #     'trigger': 'cron', # cron表示定时任务
        #     'hour': 19,
        #     'minute': 27
        # },
        {  # 第二个任务每隔5S执行一次
            'id': 'job2',
            'func': '__main__:method_test',  # 方法名
            'args': (1, 2),  # 入参
            'trigger': 'interval',  # interval表示循环任务
            'seconds': 5,
        }
    ]


def method_test(a, b):

    filename = tempfile.mktemp("订单信息.txt")
    open(filename, "w").write("
"
                              "订单编号2019071913272800037
"
                              "会员姓名坚持下去
"
                              "订单总价128.60
"
                              "订单状态待发货
"
                              "创建时间2019-07-19 13:27:28
"
                              "收货地址15821063464上海市徐汇区虹桥路693弄11号楼104
"
                              ""
                              "商品 	数量 	金额
"
                              "橙子 	1 	8.80
"
                              "鲜活小龙虾 1 	75.00
"
                              "基围虾 	1 	28.90
"
                              "猪肉精肉 	1 	15.90
"
                              ""
                              "")
    win32api.ShellExecute(0, "printto", filename, '"%s"' % win32print.GetDefaultPrinter(), ".", 0)


app = Flask(__name__)
app.config.from_object(Config())  # 为实例化的flask引入配置


##
@app.route("/hello", methods=["POST", "GET"])
def check():
    return "success", 200


if __name__ == '__main__':
    scheduler = APScheduler()
    scheduler.init_app(app)
    scheduler.start()
    app.run(debug=False)

我想实现的是 我本地能自动获取阿里云服务器上数据库的数据调用本地标签打印机打印订单信息

from flask import Flask
from flask_apscheduler import APScheduler
import tempfile
import win32api
import win32print


class Config(object):  # 创建配置用类
    # 任务列表
    JOBS = [
        # {  # 第一个任务
        #     'id': 'job1',
        #     'func': '__main__:job_1',
        #     'args': (1, 2),
        #     'trigger': 'cron', # cron表示定时任务
        #     'hour': 19,
        #     'minute': 27
        # },
        {  # 第二个任务每隔5S执行一次
            'id': 'job2',
            'func': '__main__:method_test',  # 方法名
            'args': (1, 2),  # 入参
            'trigger': 'interval',  # interval表示循环任务
            'seconds': 5,
        }
    ]


def method_test(a, b):

    # 我想在这里获取 阿里云服务器上数据库的新下的订单信息 ~~~~~~~
    # start ~~~~~~~
    from .sql_util import SQLTool
    # 调用SQLTool这个类
    tool = SQLTool(
        "root",
        "123456",
        "127.0.0.1",
        3306,
        "food_db"
    )
    # print(tool)
    # 原生sql语句
    sql = "select * from pay_order;"
    # 调用类里的方法
    res = tool.query(sql)
    print(res)
    return "ok"
    # end  ~~~~~~~

    filename = tempfile.mktemp("订单信息.txt")
    open(filename, "w").write("
"
                              "订单编号2019071913272800037
"
                              "会员姓名坚持下去
"
                              "订单总价128.60
"
                              "订单状态待发货
"
                              "创建时间2019-07-19 13:27:28
"
                              "收货地址15821063464上海市徐汇区虹桥路693弄11号楼104
"
                              ""
                              "商品 	数量 	金额
"
                              "橙子 	1 	8.80
"
                              "鲜活小龙虾 1 	75.00
"
                              "基围虾 	1 	28.90
"
                              "猪肉精肉 	1 	15.90
"
                              ""
                              "")
    win32api.ShellExecute(0, "printto", filename, '"%s"' % win32print.GetDefaultPrinter(), ".", 0)


app = Flask(__name__)
app.config.from_object(Config())  # 为实例化的flask引入配置


##
@app.route("/hello", methods=["POST", "GET"])
def check():
    return "success", 200


if __name__ == '__main__':
    scheduler = APScheduler()
    scheduler.init_app(app)
    scheduler.start()
    app.run(debug=False)

报错

C:UsershuangAppDataLocalProgramsPythonPython36python.exe "C:Program FilesJetBrainsPyCharm 2017.2.7helperspydevpydevd.py" --multiproc --qt-support=auto --client 127.0.0.1 --port 55343 --file C:/Users/huang/vincent/code/mooc/order/jobs/tasks/mytask/print.py
pydev debugger: process 10900 is connecting

Connected to pydev debugger (build 172.4574.37)
 * Serving Flask app "print" (lazy loading)
 * Environment: production
   WARNING: This is a development server. Do not use it in a production deployment.
   Use a production WSGI server instead.
 * Debug mode: off
 * Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)
Job "job2 (trigger: interval[0:00:05], next run at: 2019-08-14 14:56:51 CST)" raised an exception
Traceback (most recent call last):
  File "C:UsershuangAppDataLocalProgramsPythonPython36libsite-packagesapschedulerexecutorsase.py", line 125, in run_job
    retval = job.func(*job.args, **job.kwargs)
  File "C:/Users/huang/vincent/code/mooc/order/jobs/tasks/mytask/print.py", line 33, in method_test
    from .sql_util import SQLTool
ModuleNotFoundError: No module named '__main__.sql_util'; '__main__' is not a package

Process finished with exit code 1

接下来 我就不知道怎么处理了

写回答

1回答

编程浪子

2019-08-14

你好

你这代码不都报错了? sql_util 这个包在哪里?

0
1
慕妹4270370
我怎么才能实现 本地自动打印数据库的订单呢?
2019-08-14
共1条回复

Python Flask构建微信小程序订餐系统(可用于毕设)

微信小程序 + Python Flask 打造订餐系统全栈应用,可用于毕设。

1709 学习 · 1904 问题

查看课程