使用python作為TCP服務(wù)端 物聯(lián)網(wǎng)應(yīng)用

說明

• 因公司自主研發(fā)的電噠噠智慧能量柜需要通過TCP與服務(wù)器進(jìn)行通訊，機(jī)柜定時(shí)傳輸硬件設(shè)備信息，用戶掃碼后通過服務(wù)端下發(fā)“租/還”命令，控制機(jī)柜開關(guān)鎖等。
• 服務(wù)端通過pymysql直連數(shù)據(jù)庫(kù)操作
• 僅做為項(xiàng)目記錄，供參考勿噴

附上完整代碼

            
              import socketserver
import _thread
import time
import logging
from threading import Thread
from base import base
from helper import helper
from logger import Logger
from mysql import DB

class Myserver(socketserver.BaseRequestHandler):
    tcp_data = ""
    #主程序
    def handle(self):
        conn = self.request
        conn.sendall(bytes("connection success",encoding="ANSI"))
        # Logger('runtime.log', level='info').logger.info('設(shè)備已連接>>>' + str(conn))
        while True:
            try:
                ret_bytes = conn.recv(1024 * 2)
                ret_str = str(ret_bytes, encoding="ANSI")
                tcp_data = ret_str
                # 斷開連接命令
                if ret_str == "q":
                    # Logger('runtime.log', level='info').logger.info('設(shè)備已斷開>>>' + str(conn))
                    conn.sendall(bytes("connection dropped", encoding="utf-8"))
                    break
                    //通訊建立后創(chuàng)建線程
                _thread.start_new_thread(Myserver.task, (self,conn,ret_str))
            except OSError as e:
                # Logger('error.log', level='info').logger.info(str(e))
                break
            time.sleep(0.1)

    #線程
    def task(self,conn,data):
  		if data:
                try:
                    while True:
						//這里可以寫自己的業(yè)務(wù)邏輯
                        time.sleep(1000)
                except IndexError as e:
                    Logger('error.log', level='info').logger.info(e)

if __name__ == "__main__":
    print('*' * 50)
    print('溫州引力聚合信息科技有限公司')
    print('DIDADA Socket Service v1.0')
    print('2019年1月16日')
    print('*' * 50)
    logging.basicConfig(format='【%(asctime)s - %(pathname)s[line:%(lineno)d] - %(levelname)s:%(message)s】',level=logging.INFO)
    ip = str(base.GetConfig('service','ip'))
    port = int(base.GetConfig('service','port'))
    base.KillPort(port)
    logging.info('服務(wù)開啟>>>地址>>>'+ ip +'>>>端口號(hào)>>>'+ str(port))
    server = socketserver.ThreadingTCPServer((ip,port),Myserver)
    server.serve_forever()


            
          

Logger類

            
              import logging
from logging import handlers
import os
import ctypes

FOREGROUND_WHITE = 0x0007
FOREGROUND_BLUE = 0x01  # text color contains blue.
FOREGROUND_GREEN = 0x02  # text color contains green.
FOREGROUND_RED = 0x04  # text color contains red.
FOREGROUND_YELLOW = FOREGROUND_RED | FOREGROUND_GREEN

STD_OUTPUT_HANDLE= -11
std_out_handle = ctypes.windll.kernel32.GetStdHandle(STD_OUTPUT_HANDLE)
def set_color(color, handle=std_out_handle):
    bool = ctypes.windll.kernel32.SetConsoleTextAttribute(handle, color)
    return bool

class Logger(object):
    level_relations = {
        'debug':logging.DEBUG,
        'info':logging.INFO,
        'warning':logging.WARNING,
        'error':logging.ERROR,
        'crit':logging.CRITICAL
    }#日志級(jí)別關(guān)系映射

    def __init__(self,filename,level='info',when='D',backCount=3,fmt='%(asctime)s - %(pathname)s[line:%(lineno)d] - %(levelname)s: %(message)s'):
        set_color(FOREGROUND_GREEN)
        self.logger = logging.getLogger(filename)
        format_str = logging.Formatter(fmt)#設(shè)置日志格式
        self.logger.setLevel(self.level_relations.get(level))#設(shè)置日志級(jí)別
        sh = logging.StreamHandler()#往屏幕上輸出
        sh.setFormatter(format_str) #設(shè)置屏幕上顯示的格式
        th = handlers.TimedRotatingFileHandler(filename=filename,when=when,backupCount=backCount,encoding='utf-8')#往文件里寫入#指定間隔時(shí)間自動(dòng)生成文件的處理器
        #實(shí)例化TimedRotatingFileHandler
        #interval是時(shí)間間隔，backupCount是備份文件的個(gè)數(shù)，如果超過這個(gè)個(gè)數(shù)，就會(huì)自動(dòng)刪除，when是間隔的時(shí)間單位，單位有以下幾種：
        # S 秒
        # M 分
        # H 小時(shí)、
        # D 天、
        # W 每星期（interval==0時(shí)代表星期一）
        # midnight 每天凌晨
        th.setFormatter(format_str)#設(shè)置文件里寫入的格式
        self.logger.addHandler(sh) #把對(duì)象加到logger里
        self.logger.addHandler(th)


            
          

DB類

            
              import pymysql.cursors


class DB(object):
    # 連接數(shù)據(jù)庫(kù)
    def __init__(self, host='localhost', port=3306, user='root', passwd='', db='mysql', charset='utf8'):
        self.host = host
        self.port = port
        self.user = user
        self.passwd = passwd
        self.db = db
        self.charset = charset
        self.connect = pymysql.Connect(
            host=self.host,
            port=self.port,
            user=self.user,
            passwd=self.passwd,
            db=self.db,
            charset=self.charset
        )
        # 使用cursor()方法獲取操作游標(biāo)
        self.cursor = self.connect.cursor()

    def select(self, sql):
        # SQL 查詢語(yǔ)句
        try:
            # 執(zhí)行SQL語(yǔ)句
            self.cursor.execute(sql)
            # 獲取所有記錄列表
            results = self.cursor.fetchall()
            return results
        except Exception as e:
            return e
        # 關(guān)閉數(shù)據(jù)庫(kù)連接
        self.connect.close()


            
          

Base類

            
              
import logging
import configparser
import os
import re
import requests
import json
class base:
    def __init__(self):
        pass

    # 端口占用清理
    def KillPort(port):
        # # 查找端口的pid
        find_port= 'netstat -aon | findstr %s' % port
        result = os.popen(find_port)
        text = result.read()
        logging.info('清理占用端口>>>' + find_port)
        logging.info('清理占用端口>>>' + text)
        # logging.info(text [9:23])
        if text[9:23] =='127.0.0.1:%s' % port:
            pid = text[70:76]
            logging.info('占用端口的pid:' + pid)
            find_kill= 'taskkill -f -pid %s' %pid
            result = os.popen(find_kill)
            logging.info(result.read())
            pass

    #獲取配置文件信息
    def GetConfig(section,key,path='config.ini'):
        config = configparser.ConfigParser()
        config.read(path)  # 配置文件的路徑
        v = config.get(section,key)
        return v

    #Post請(qǐng)求
    def http_post(url,parameter={}):
        return requests.post(url, params=parameter)

    #判斷字符串是否包含另外一個(gè)字符串
    def find_string(s, t):
        return t in s


            
          

附上效果圖

項(xiàng)目結(jié)構(gòu)