使用Flask封装集成深度学习算法

# -*- coding: utf-8 -*-

from flask import Flask, jsonify
from flask_cors import CORS
from pre_request import pre, Rule

from log import logger
from code import ResponseCode, ResponseMessage

# 创建一个服务
app = Flask(__name__)
CORS(app, supports_credentials=True)

"""
# 方法功能说明
"""
@app.route(rule='/api/moduleName/methodName', methods=['POST'])
def methodName():

    # 参数校验
    rule = {
        "text": Rule(type=str, required=True, gte=3, lte=255),
        "type": Rule(type=int, required=True, gte=1, lte=1)
    }
    try:
        params = pre.parse(rule=rule)
    except Exception as e:
        logger.error(e)
        fail_response = dict(code=ResponseCode.RARAM_FAIL, msg=ResponseMessage.RARAM_FAIL, data=None)
        logger.error(fail_response)
        return jsonify(fail_response)

    # 获取参数
    text = params.get("text")

    # 业务处理模块
    result = text + ",hello world!"
    logger.info("测试日志记录")

    # 成功的结果返回
    success_response = dict(code=ResponseCode.SUCCESS, msg=ResponseMessage.SUCCESS, data=result)
    logger.info(success_response)
    return jsonify(success_response)


if __name__ == '__main__':
    # 解决中文乱码问题
    app.config['JSON_AS_ASCII'] = False
    # 启动服务，指定主机和端口
    app.run(host='0.0.0.0', port=5000, debug=False, threaded=True)

# -*- coding: utf-8 -*-

import logging

logger = logging.getLogger(__name__)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')

# 输出到控制台
console = logging.StreamHandler()
console.setLevel(logging.INFO)
console.setFormatter(formatter)
logger.addHandler(console)

# 输出到文件
logger.setLevel(level=logging.INFO)
handler = logging.FileHandler("./server.log")
handler.setLevel(logging.INFO)
handler.setFormatter(formatter)
logger.addHandler(handler)

# -*- coding: utf-8 -*-


class ResponseCode(object):
    SUCCESS = 200
    RARAM_FAIL = 400
    BUSINESS_FAIL = 500


class ResponseMessage(object):
    SUCCESS = "请求成功"
    RARAM_FAIL = "参数校验失败"
    BUSINESS_FAIL = "业务处理失败"

# -*- coding: utf-8 -*-

from code import ResponseMessage, ResponseCode


class ResMsg(object):
    """
    封装响应文本
    """
    def __init__(self, data=None, code=ResponseCode.SUCCESS, msg=ResponseMessage.SUCCESS):
        self._data = data
        self._msg = msg
        self._code = code

    def update(self, code=None, data=None, msg=None):
        """
        更新默认响应文本
        :param code:响应状态码
        :param data: 响应数据
        :param msg: 响应消息
        :return:
        """
        if code is not None:
            self._code = code
        if data is not None:
            self._data = data
        if msg is not None:
            self._msg = msg

    def add_field(self, name=None, value=None):
        """
        在响应文本中加入新的字段，方便使用
        :param name: 变量名
        :param value: 变量值
        :return:
        """
        if name is not None and value is not None:
            self.__dict__[name] = value

    @property
    def data(self):
        """
        输出响应文本内容
        :return:
        """
        body = self.__dict__
        body["data"] = body.pop("_data")
        body["msg"] = body.pop("_msg")
        body["code"] = body.pop("_code")
        return body

# 声明处
@app.route(rule='/moduleName/methodName', methods=['GET'])
# 接参处
id = request.args.get("id")

# -*- coding: utf-8 -*-

import base64
import requests
import json

if __name__ == '__main__':
    # 测试请求
    url = 'http://127.0.0.1:5000/moduleName/methodName'
    f = open('./data/test.jpg', 'rb')
    # base64编码
    base64_data = base64.b64encode(f.read())
    f.close()
    base64_data = base64_data.decode()
    # 传输的数据格式
    data = {'img': base64_data}
    # post传递数据
    r = requests.post(url, data=json.dumps(data))
    print(r.text.encode().decode('unicode_escape')) 

# -*- coding: utf-8 -*-

import os
from uuid import uuid1
from flask import Flask, jsonify
from flask_cors import CORS
from pre_request import pre, Rule

from code import ResponseCode, ResponseMessage
from log import logger
from utils import base64_to_img

# 创建一个服务
app = Flask(__name__)
CORS(app, supports_credentials=True)

"""
# 方法功能说明
"""
@app.route(rule='/api/moduleName/methodName', methods=['POST'])
def methodName():

    # 参数校验
    rule = {
        "img": Rule(type=str, required=True),
        "ext": Rule(type=str, required=False)
    }
    try:
        params = pre.parse(rule=rule)
    except Exception as e:
        logger.error(e)
        fail_response = dict(code=ResponseCode.RARAM_FAIL, msg=ResponseMessage.RARAM_FAIL, data=None)
        logger.error(fail_response)
        return jsonify(fail_response)

    # 获取参数
    image_b64 = params.get("img")
    ext = params.get("ext")

    # 将base64字符串解析成图片保存
    if not os.path.exists('./img'):
        os.makedirs('./img')
    uuid = uuid1()
    if ext is not None:
        img_path = './img/{}.{}'.format(uuid, ext)
    else:
        img_path = './img/{}.jpg'.format(uuid)
    try:
        base64_to_img(image_b64, img_path)
    except Exception as e:
        logger.error(e)
        fail_response = dict(code=ResponseCode.BUSINESS_FAIL, msg=ResponseMessage.BUSINESS_FAIL, data=None)
        logger.error(fail_response)
        return jsonify(fail_response)

    # 下面对保存的图片进行若干处理
    result = image_b64
    logger.info("测试日志记录")

    # 处理完成后删除生成的图片文件
    os.remove(img_path)

    # 成功的结果返回
    success_response = dict(code=ResponseCode.SUCCESS, msg=ResponseMessage.SUCCESS, data=result)
    logger.info(success_response)
    return jsonify(success_response)


if __name__ == '__main__':
    # 解决中文乱码问题
    app.config['JSON_AS_ASCII'] = False
    # 启动服务 指定主机和端口
    app.run(host='0.0.0.0', port=5000, debug=False, threaded=True)

# -*- coding: utf-8 -*-

import time
import requests

if __name__ == '__main__':
    url = 'http://127.0.0.1:5000/moduleName/methodName'
    img_path = './data/test.jpg'
    files = {'image': open(img_path, "rb")}
    r = requests.post(url, files=files)
    end_time = time.time()
    print(r.text.encode().decode('unicode_escape')) 

from flask import Flask, request
from flask_cors import CORS

app = Flask(__name__)
CORS(app, supports_credentials=True)

from flask import Flask, request
from flask_cors import cross_origin

app = Flask(__name__)

@app.route('/')
@cross_origin(supports_credentials=True)
def hello():
    name = request.args.get("name", "World")
    return f'Hello, {name}!'

if __name__ == "__main__":
    app.config['JSON_AS_ASCII'] = False
    app.run(host='0.0.0.0', port='5000')

request_body = request.get_json()

request_data = request.get_data(as_text=True)
request_data = ''.join(request_data.split())  
request_body = json.loads(request_data)

# 1.threaded : 多线程支持，默认为False，即不开启多线程;
app.run(threaded=True)
# 2.processes：进程数量，默认为1.
app.run(processes=True)

# -*- coding: utf-8 -*-

import time
import grequests

start = time.time()
req_list = [grequests.post('http://httpbin.org/post', data={'a':1, 'b':2}) for i in range(10)]
res_list = grequests.map(req_list)
result_list = []
for res in res_list:
    result_list.append(res.text)
print(result_list)
print(len(result_list))
print(time.time()-start)

# -*- coding: utf-8 -*-

import base64

# 生成测试数据
before_base64 = 'abc'.encode()
after_base64 = base64.b64encode(before_base64)
print(after_base64)

# 方式一：decode为utf-8编码
method_one_base64 = after_base64.decode('utf-8')
print(method_one_base64)

# 方式二：str转化为utf-8编码
method_two_base64 = str(after_base64, 'utf-8')
print(method_two_base64)

>>> b'YWJj'
>>> YWJj
>>> YWJj

# -*- coding: utf-8 -*-

import json

d = {'s': '测试', 'd': u'\u4ea4\u6362\u673a'}
d1 = json.dumps(d1)
print(d1)
d2 = json.loads(j)
print(d2)

>>> {"s": "\u6d4b\u8bd5", "d": "\u4ea4\u6362\u673a"}
>>> {'s': '测试', 'd': '交换机'}

import logging

# 日志记录
logging.basicConfig(filename='server.log', level=logging.INFO,
                    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# -*- coding: utf-8 -*-

import logging

logger = logging.getLogger(__name__)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')

# 输出到控制台
console = logging.StreamHandler()
console.setLevel(logging.INFO)
console.setFormatter(formatter)
logger.addHandler(console)

# 输出到文件
logger.setLevel(level=logging.INFO)
handler = logging.FileHandler("./server.log")
handler.setLevel(logging.INFO)
handler.setFormatter(formatter)
logger.addHandler(handler)

# -*- coding: utf-8 -*-

class ResponseCode(object):
    SUCCESS = 200
    FAIL = 500

class ResponseMessage(object):
    SUCCESS = "请求成功"
    FAIL = "请求失败"

# -*- coding: utf-8 -*-

from code import ResponseMessage, ResponseCode


class ResMsg(object):
    """
    封装响应文本
    """
    def __init__(self, data=None, code=ResponseCode.SUCCESS, msg=ResponseMessage.SUCCESS):
        self._data = data
        self._msg = msg
        self._code = code

    def update(self, code=None, data=None, msg=None):
        """
        更新默认响应文本
        :param code:响应状态码
        :param data: 响应数据
        :param msg: 响应消息
        :return:
        """
        if code is not None:
            self._code = code
        if data is not None:
            self._data = data
        if msg is not None:
            self._msg = msg

    def add_field(self, name=None, value=None):
        """
        在响应文本中加入新的字段，方便使用
        :param name: 变量名
        :param value: 变量值
        :return:
        """
        if name is not None and value is not None:
            self.__dict__[name] = value

    @property
    def data(self):
        """
        输出响应文本内容
        :return:
        """
        body = self.__dict__
        body["data"] = body.pop("_data")
        body["msg"] = body.pop("_msg")
        body["code"] = body.pop("_code")
        return body

# -*- coding: utf-8 -*-

from flask import Flask, jsonify
from flask_cors import CORS

from code import ResponseCode, ResponseMessage

# 创建一个服务
app = Flask(__name__)
CORS(app, supports_credentials=True)


@app.route("/test",methods=["GET"])
def test():
	test_dict = dict(name="zhang",age=18)
	data = dict(code = ResponseCode.SUCCESS, msg = ResponseMessage.SUCCESS, data = test_dict)
	return jsonify(data)


if __name__ == '__main__':
    # 解决中文乱码问题
    app.config['JSON_AS_ASCII'] = False
    # 启动服务，指定主机和端口
    app.run(host='0.0.0.0', port=5000, debug=False, threaded=True)

# -*- coding: utf-8 -*-

from flask import Flask
from flask_cors import CORS
from pre_request import pre, Rule

# 创建一个服务
app = Flask(__name__)
CORS(app, supports_credentials=True)

rule = {
    "userName": Rule(type=str, required=True, gte=3, lte=20, dest="user_name"),
    "gender": Rule(type=int, required=True, enum=[1, 2]),
    "age": Rule(type=int, required=True, gte=18, lte=60),
    "country": Rule(type=str, required=False, gte=2, default="中国")
}

@app.route("/user/info", methods=["POST"])
def user_info_handler():
    params = pre.parse(rule=rule)
    userName = params.get("userName")
    gender = params.get("gender")
    age = params.get("age")
    country = params.get("country")
    return "success"


if __name__ == '__main__':
    # 解决中文乱码问题
    app.config['JSON_AS_ASCII'] = False
    # 启动服务 指定主机和端口
    app.run(host='0.0.0.0', port=5000, debug=False, threaded=True)

# 使用 CDN
# app.config["API_DOC_CDN"] = True

# 禁用文档页面
# app.config["API_DOC_ENABLE"] = False

# SHA256 加密的授权密码，例如这里是 admin
# echo -n admin | shasum -a 256
# app.config["API_DOC_PASSWORD_SHA2"] = "8c6976e5b5410415bde908bd4dee15dfb167a9c873fc4bb8a81f6f2ab448a918"

# 允许显示的方法
# app.config["API_DOC_METHODS_LIST"] = ["GET", "POST", "PUT", "DELETE", "PATCH"]

# 自定义 url_prefix
# app.config["API_DOC_URL_PREFIX"] = "/docs/api"

# 需要排除的 RESTful Api 类名
# app.config["API_DOC_RESTFUL_EXCLUDE"] = ["Todo"]

# 需要显示的 Api 蓝图名称
# app.config["API_DOC_MEMBER"] = ["api", "platform"]

# 需要排除的子成员 Api 函数名称
# app.config["API_DOC_MEMBER_SUB_EXCLUDE"] = ["delete_data"]

# 自动生成请求参数 markdown
# app.config["API_DOC_AUTO_GENERATING_ARGS_MD"] = True

# 禁止以 markdown 处理所有文档
# app.config["API_DOC_ALL_MD"] = False

# -*- coding: utf-8 -*-

def _init():  
    global _global_dict
    _global_dict = {}

def set_value(key, value):
    """ 定义一个全局变量 """
    _global_dict[key] = value

def get_value(key, defValue=None):
    """ 获得一个全局变量,不存在则返回默认值 """
    try:
        return _global_dict[key]
    except KeyError:
        return defValue

import gol

gol._init() # 先必须在主模块初始化（只需要一次即可）
gol.set_value('name', 'zhangsan')
gol.set_value('age', 23)

import gol

name = gol.get_value('name')
age = gol.get_value('age')

from paddleocr import PaddleOCR

# Paddleocr目前支持的多语言语种可以通过修改lang参数进行切换
# 例如`ch`, `en`, `fr`, `german`, `korean`, `japan`
ocr = PaddleOCR(use_angle_cls=True, lang="ch")  # need to run only once to download and load model into memory
img_path = './imgs/test.jpg'
result = ocr.ocr(img_path, cls=True)
for line in result:
    print(line)

# -*- coding: utf-8 -*-

from paddlenlp import Taskflow

# 中文分词
paddle_nlp = Taskflow("word_segmentation")
result = paddle_nlp("第十四届全运会在西安举办")
print(result)
# >>> ['第十四届', '全运会', '在', '西安', '举办']

# 词性标注
paddle_nlp = Taskflow("pos_tagging")
result = paddle_nlp("第十四届全运会在西安举办")
print(result)
# >>> [('第十四届', 'm'), ('全运会', 'nz'), ('在', 'p'), ('西安', 'LOC'), ('举办', 'v')]

# 名词短语标注
paddle_nlp = Taskflow("knowledge_mining", model="nptag")
result = paddle_nlp("红曲霉菌")
print(result)
# >>> [{'text': '红曲霉菌', 'label': '微生物'}]

# 情感分析
paddle_nlp = Taskflow("sentiment_analysis")
result = paddle_nlp("这个产品用起来真的很流畅，我非常喜欢")
print(result)
# >>> [{'text': '这个产品用起来真的很流畅，我非常喜欢', 'label': 'positive', 'score': 0.9938690066337585}]

# 文本相似度
paddle_nlp = Taskflow("text_similarity")
result = paddle_nlp([["世界上什么东西最小", "世界上什么东西最小？"]])
print(result)
# >>> [{'text1': '世界上什么东西最小', 'text2': '世界上什么东西最小？', 'similarity': 0.992725}]

# -*- coding: utf-8 -*-

import os
from gtts import gTTS

os.environ["https_proxy"] = "http://127.0.0.1:1080"

# 谷歌文字转语音API测试
text = "测试gtts文本转语音"
audio = gTTS(text=text, lang="zh-cn")
audio.save("demo.mp3")

from playsound import playsound
playsound('demo.mp3')

def winCommand(*command):
        bufLen = 600
        buf = c_buffer(bufLen)
        #command = ' '.join(command).encode('utf-16') # 1.修改前
        command = ' '.join(command) # 1.修改后
        errorCode = int(windll.winmm.mciSendStringW(command, buf, bufLen - 1, 0))  # use widestring version of the function
        if errorCode:
            errorBuffer = c_buffer(bufLen)
            windll.winmm.mciGetErrorStringW(errorCode, errorBuffer, bufLen - 1)  # use widestring version of the function
            exceptionMessage = ('\n    Error ' + str(errorCode) + ' for command:'
                                #'\n        ' + command.decode('utf-16') + # 2.修改前
                                '\n        ' + command + # 2.修改后
                                '\n    ' + errorBuffer.raw.decode('utf-16').rstrip('\0'))
            logger.error(exceptionMessage)
            raise PlaysoundException(exceptionMessage)
        return buf.value

#-*- coding:utf-8 -*-

from googletrans import Translator
import os

os.environ["https_proxy"] = "http://127.0.0.1:1080"

translator = Translator()
result = translator.translate('hello world', dest='zh-cn').text
print(result)

# -*- coding: utf-8 -*-

import cv2
import operator
import numpy as np
import os
from scipy.signal import argrelextrema

def smooth(x, window_len=13, window='hanning'):
    """使用具有所需大小的窗口使数据平滑。
    This method is based on the convolution of a scaled window with the signal.
    The signal is prepared by introducing reflected copies of the signal
    (with the window size) in both ends so that transient parts are minimized
    in the begining and end part of the output signal.
    该方法是基于一个标度窗口与信号的卷积。
    通过在两端引入信号的反射副本(具有窗口大小)来准备信号，
    使得在输出信号的开始和结束部分中将瞬态部分最小化。
    input:
        x: the input signal输入信号
        window_len: the dimension of the smoothing window平滑窗口的尺寸
        window: the type of window from 'flat', 'hanning', 'hamming', 'bartlett', 'blackman'
            flat window will produce a moving average smoothing.
            平坦的窗口将产生移动平均平滑
    output:
        the smoothed signal平滑信号

    example:
    import numpy as np
    t = np.linspace(-2,2,0.1)
    x = np.sin(t)+np.random.randn(len(t))*0.1
    y = smooth(x)

    see also:
    numpy.hanning, numpy.hamming, numpy.bartlett, numpy.blackman, numpy.convolve
    scipy.signal.lfilter

    TODO: 如果使用数组而不是字符串，则window参数可能是窗口本身
    """
    print(len(x), window_len)
    s = np.r_[2 * x[0] - x[window_len:1:-1],
              x, 2 * x[-1] - x[-1:-window_len:-1]]

    if window == 'flat':  # moving average平移
        w = np.ones(window_len, 'd')
    else:
        w = getattr(np, window)(window_len)
    y = np.convolve(w / w.sum(), s, mode='same')
    return y[window_len - 1:-window_len + 1]


class Frame:
    """class to hold information about each frame
    用于保存有关每个帧的信息
    """

    def __init__(self, id, diff):
        self.id = id
        self.diff = diff

    def __lt__(self, other):
        if self.id == other.id:
            return self.id < other.id
        return self.id < other.id

    def __gt__(self, other):
        return other.__lt__(self)

    def __eq__(self, other):
        return self.id == other.id and self.id == other.id

    def __ne__(self, other):
        return not self.__eq__(other)


def rel_change(a, b):
    x = (b - a) / max(a, b)
    print(x)
    return x


def getEffectiveFrame(videopath, dir):
    # 如果文件目录不存在则创建目录
    if not os.path.exists(dir):
        os.makedirs(dir)
    (filepath, tempfilename) = os.path.split(videopath)  # 分离路径和文件名
    (filename, extension) = os.path.splitext(tempfilename)  # 区分文件的名字和后缀
    # Setting fixed threshold criteria设置固定阈值标准
    USE_THRESH = False
    # fixed threshold value固定阈值
    THRESH = 0.8
    # Setting fixed threshold criteria设置固定阈值标准
    USE_TOP_ORDER = False
    # Setting local maxima criteria设置局部最大值标准
    USE_LOCAL_MAXIMA = True
    # Number of top sorted frames排名最高的帧数
    NUM_TOP_FRAMES = 50
    # smoothing window size平滑窗口大小
    len_window = int(50)
    print("target video :" + videopath)
    print("frame save directory: " + dir)
    # load video and compute diff between frames加载视频并计算帧之间的差异
    cap = cv2.VideoCapture(str(videopath))
    prev_frame = None
    frame_diffs = []
    frames = []
    success, frame = cap.read()
    i = 0
    while (success):
        luv = cv2.cvtColor(frame, cv2.COLOR_BGR2LUV)
        curr_frame = luv
        if curr_frame is not None and prev_frame is not None:
            # logic here
            diff = cv2.absdiff(curr_frame, prev_frame)  # 获取差分图
            diff_sum = np.sum(diff)
            diff_sum_mean = diff_sum / (diff.shape[0] * diff.shape[1])  # 平均帧
            frame_diffs.append(diff_sum_mean)
            frame = Frame(i, diff_sum_mean)
            frames.append(frame)
        prev_frame = curr_frame
        i = i + 1
        success, frame = cap.read()
    cap.release()

    # compute keyframe
    keyframe_id_set = set()
    if USE_TOP_ORDER:
        # sort the list in descending order以降序对列表进行排序
        frames.sort(key=operator.attrgetter("diff"), reverse=True)  # 排序operator.attrgetter
        for keyframe in frames[:NUM_TOP_FRAMES]:
            keyframe_id_set.add(keyframe.id)
    if USE_THRESH:
        print("Using Threshold")  # 使用阈值
        for i in range(1, len(frames)):
            if (rel_change(np.float(frames[i - 1].diff), np.float(frames[i].diff)) >= THRESH):
                keyframe_id_set.add(frames[i].id)
    if USE_LOCAL_MAXIMA:
        print("Using Local Maxima")  # 使用局部极大值
        diff_array = np.array(frame_diffs)
        sm_diff_array = smooth(diff_array, len_window)  # 平滑
        frame_indexes = np.asarray(argrelextrema(sm_diff_array, np.greater))[0]  # 找极值
        for i in frame_indexes:
            keyframe_id_set.add(frames[i - 1].id)  # 记录极值帧数

    # save all keyframes as image将所有关键帧另存为图像
    cap = cv2.VideoCapture(str(videopath))
    success, frame = cap.read()
    idx = 0
    num = 0
    while (success):
        if idx in keyframe_id_set:
            num = num + 1
            name = filename + '_' + str(num) + ".jpg"
            cv2.imwrite(dir + name, frame)
            keyframe_id_set.remove(idx)
        idx = idx + 1
        success, frame = cap.read()
    cap.release()

    
if __name__ == "__main__":
    videopath = './data/demo.mp4'    # Video path of the source file源文件的视频路径
    dir = './data/keyframe/'  # Directory to store the processed frames存储已处理帧的目录
    getEffectiveFrame(videopath, dir)

# -*- coding: utf-8 -*-

import os
import cv2

"""
# Features：使用 opencv 实现图片压缩，compress_config为图片压缩配置，说明如下：
# [cv2.IMWRITE_PNG_COMPRESSION, 9]  无损压缩（取值范围：0~9，数值越小，压缩比越低）
# [cv2.IMWRITE_JPEG_QUALITY, 80]  有损压缩（取值范围：0~100，数值越小，压缩比越高，图片质量损失越严重）
"""
class Compress_img:

    def __init__(self, img_path, compress_config):
        self.img_path = img_path
        self.img_name = img_path.split('/')[-1]
        self.compress_config = compress_config

    def compress_img_CV(self, show=False):
        old_fsize = os.path.getsize(self.img_path)
        # 读取并压缩图片
        img_resize = cv2.imread(self.img_path)
        cv2.imwrite(self.img_path, img_resize, self.compress_config)
        new_fsize = os.path.getsize(self.img_path)
        # 计算压缩率
        compress_rate = str(round(new_fsize / old_fsize * 100, 2)) + "%"
        print("%s 图片已压缩，" % (self.img_path), "压缩率为：", compress_rate)
        # 查看压缩后的图片
        if show:
            cv2.imshow(self.img_name, img_resize)
            cv2.waitKey(0)

if __name__ == '__main__':

    img_path = './test.jpg'
    compress_para = [cv2.IMWRITE_PNG_COMPRESSION, 9]
    compress = Compress_img(img_path, compress_para)
    compress.compress_img_CV()

from blind_watermark import WaterMark

# 设置密码，默认是 1
bwm1 = WaterMark(password_img=1, password_wm=1)
# 读取原始图片
bwm1.read_img('input/001.jpg')
# 定义水印文本
wm = '@eula.club'
# 合并文本并输出新的图片
bwm1.read_wm(wm, mode='str')
bwm1.embed('output/001.jpg')
# 输出结果
len_wm = len(bwm1.wm_bit)
print('Put down the length of wm_bit {len_wm}'.format(len_wm=len_wm))

bwm1 = WaterMark(password_img=1, password_wm=1)
wm_extract = bwm1.extract('output/001.jpg', wm_shape=len_wm, mode='str')
print(wm_extract)

# -*- coding: utf-8 -*-

from PIL import Image


def get_thumbnail_pic(input_img_path, output_img_path):
    im = Image.open(input_img_path)
    im.thumbnail((80, 80))
    print(im.format, im.size, im.mode)
    im.save(output_img_path, 'JPEG')


if __name__=='__main__':
    input_img_path = './input/001.jpg'
    output_img_path = './output/001.jpeg'
    get_thumbnail_pic(input_img_path, output_img_path)

# -*- coding: utf-8 -*-

import json
import time
from uuid import uuid1
from flask import Flask, jsonify, request
from flask_cors import CORS
import os
import asyncio

from log import logger
from responseCode import ResponseCode, ResponseMessage
from utils import url_save_pdf, download_file

# 创建一个服务
app = Flask(__name__)
CORS(app, supports_credentials=True)

"""
# 将任意公开访问的url转化为pdf提供下载
"""
@app.route(rule='/api/pyppeteer/urlSavePdf', methods=['POST'])
def urlToPdf():
    # 获取JSON格式的请求体，并解析
    request_data = request.get_data(as_text=True)
    request_body = json.loads(request_data)

    # 参数校验模块
    url = request_body.get("url")
    if not url:
        fail_response = dict(code=ResponseCode.RARAM_FAIL, msg=ResponseMessage.RARAM_FAIL, data=None)
        logger.error(fail_response)
        return jsonify(fail_response)
    pdf_name = request_body.get("pdf_name")
    if not pdf_name:
        pdf_name = '{}.pdf'.format(uuid1())
    '''
     resolution: 设置网页显示尺寸
         width: 网页显示宽度
         height: 网页显示高度
     '''
    resolution = request_body.get("resolution")
    if not resolution:
        resolution = {"width": 1920, "height": 1680}
    '''
     clip: 位置与图片尺寸信息
         x: 网页截图的起始x坐标
         y: 网页截图的起始y坐标
         width: 图片宽度
         height: 图片高度
     '''
    clip = request_body.get("clip")
    if not clip:
        clip = {"width": 1920, "height": 1680}

    # 创建pdf的存储目录
    now_str = time.strftime("%Y%m%d", time.localtime())
    pdf_root_path = './tmp/'
    pdf_base_path = pdf_root_path + now_str
    if not os.path.exists(pdf_base_path):
        os.makedirs(pdf_base_path)
    pdf_path = pdf_base_path + '/' + pdf_name

    # 将url保存成pdf文件
    try:
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        loop.run_until_complete(url_save_pdf(url, pdf_path, resolution, clip))
        logger.info("成功将【{}】网址保存成pdf文件【{}】！".format(url, pdf_path))
    except Exception as e:
        logger.error(e)
        fail_response = dict(code=ResponseCode.BUSINESS_FAIL, msg=ResponseMessage.BUSINESS_FAIL, data=None)
        logger.error(fail_response)
        return jsonify(fail_response)

    # 将pdf文件转成文件流提供下载
    return download_file(pdf_path)


if __name__ == '__main__':
    # 解决中文乱码问题
    app.config['JSON_AS_ASCII'] = False
    # 启动服务，指定主机和端口
    app.run(host='0.0.0.0', port=5006, debug=False, threaded=True)

# -*- coding: utf-8 -*-

import os
import urllib.parse

from PIL import Image
from pyppeteer import launch
from reportlab.pdfgen.canvas import Canvas
from reportlab.lib.utils import ImageReader
import imageio.v2 as imageio
from flask import Response


# 指定url区域截屏保存成pdf
async def url_save_pdf(url, pdf_path, resolution, clip):
    start_parm = {
        # 下列三个参数用于解决 Flask 运行 Pyppeteer 报错 "signal only works in main thread"
        "handleSIGINT": False,
        "handleSIGTERM": False,
        "handleSIGHUP": False,
        "headless": True,    # 关闭无头浏览器
        "args": [
            '--no-sandbox',  # 关闭沙盒模式
        ],
    }
    browser = await launch(**start_parm)
    page = await browser.newPage()
    # 加载指定的网页url
    await page.goto(url)
    # 设置网页显示尺寸
    await page.setViewport(resolution)
    # 设置截屏区域
    if 'x' not in clip or 'y' not in clip:
        await page.pdf({'path': pdf_path, 'width': clip['width'], 'height': clip['height']})
        await browser.close()
    else:
        img_data = await page.screenshot({'clip': clip})
        img_data_array = imageio.imread(img_data, format="png")
        im = Image.fromarray(img_data_array)
        page_width, page_height = im.size
        c = Canvas(pdf_path, pagesize=(page_width, page_height))
        c.drawImage(ImageReader(im), 0, 0)
        c.save()


# 检验是否含有中文字符
def is_contains_chinese(strs):
    for _char in strs:
        if '\u4e00' <= _char <= '\u9fa5':
            return True
    return False


# 将文件转成文件流提供下载
def download_file(file_path):

    # 文件路径、文件名、后缀分割
    file_dir, file_full_name = os.path.split(file_path)
    file_name, file_ext = os.path.splitext(file_full_name)

    # 文件名如果包含中文则进行编码
    if is_contains_chinese(file_name):
        file_name = urllib.parse.quote(file_name)
    new_file_name = file_name + file_ext

    # 流式读取下载
    def send_file():
        with open(file_path, 'rb') as targetfile:
            while 1:
                data = targetfile.read(20 * 1024 * 1024)   # 每次读取20M
                if not data:
                    break
                yield data
    response = Response(send_file(), content_type='application/octet-stream')
    response.headers["Content-disposition"] = 'attachment; filename=%s' % new_file_name
    return response

# -*- coding: utf-8 -*-

from responseCode import ResponseMessage, ResponseCode


class ResMsg(object):
    """
    封装响应文本
    """
    def __init__(self, data=None, code=ResponseCode.SUCCESS, msg=ResponseMessage.SUCCESS):
        self._data = data
        self._msg = msg
        self._code = code

    def update(self, code=None, data=None, msg=None):
        """
        更新默认响应文本
        :param code:响应状态码
        :param data: 响应数据
        :param msg: 响应消息
        :return:
        """
        if code is not None:
            self._code = code
        if data is not None:
            self._data = data
        if msg is not None:
            self._msg = msg

    def add_field(self, name=None, value=None):
        """
        在响应文本中加入新的字段，方便使用
        :param name: 变量名
        :param value: 变量值
        :return:
        """
        if name is not None and value is not None:
            self.__dict__[name] = value

    @property
    def data(self):
        """
        输出响应文本内容
        :return:
        """
        body = self.__dict__
        body["data"] = body.pop("_data")
        body["msg"] = body.pop("_msg")
        body["code"] = body.pop("_code")
        return body

# -*- coding: utf-8 -*-


class ResponseCode(object):
    SUCCESS = 200
    RARAM_FAIL = 400
    BUSINESS_FAIL = 500


class ResponseMessage(object):
    SUCCESS = "请求成功"
    RARAM_FAIL = "参数校验失败"
    BUSINESS_FAIL = "业务处理失败"

# -*- coding: utf-8 -*-

import logging

logger = logging.getLogger(__name__)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')

# 输出到控制台
console = logging.StreamHandler()
console.setLevel(logging.INFO)
console.setFormatter(formatter)
logger.addHandler(console)

# 输出到文件
logger.setLevel(level=logging.INFO)
handler = logging.FileHandler("./server.log")
handler.setLevel(logging.INFO)
handler.setFormatter(formatter)
logger.addHandler(handler)

start_parm = {
    # 下列三个参数用于解决 Flask 运行 Pyppeteer 报错 "signal only works in main thread"
    "handleSIGINT": False,
    "handleSIGTERM": False,
    "handleSIGHUP": False,
    "headless": True,    # 关闭无头浏览器
    "args": [
        '--no-sandbox',  # 关闭沙盒模式
    ],
}

# Excel转字典列表
def excel_to_dict(path):

    # 判断是否为文件路径
    if os.path.exists(path):
        workbook = xlrd.open_workbook(path)
    else:
        workbook = xlrd.open_workbook(filename=path.name, file_contents=path.read())

    # 根据sheet索引或者名称获取sheet内容
    data_sheet = workbook.sheets()[0]
    # 获取sheet名称，行数，列数据
    sheet_nrows = data_sheet.nrows
    sheet_ncols = data_sheet.ncols

    # excel转dict
    get_data = []
    for i in range(1, sheet_nrows):
        # 定义一个空字典
        sheet_data = {}
        for j in range(sheet_ncols):
            # 获取单元格数据
            c_cell = data_sheet.cell_value(i, j)
            # 循环每一个有效的单元格，将字段与值对应存储到字典中
            sheet_data[data_sheet.row_values(0)[j]] = c_cell
        # 再将字典追加到列表中
        get_data.append(sheet_data)
    # 返回从excel中获取到的数据：以列表存字典的形式返回
    return get_data


# CSV转字典列表
def csv_to_dict(path):
    get_data = []
    with open(path, 'r',encoding="GBK") as f:
        reader = csv.reader(f)
        fieldnames = next(reader)
        csv_reader = csv.DictReader(f, fieldnames=fieldnames)
        for row in csv_reader:
            d = {}
            for k, v in row.items():
                d[k] = v
            get_data.append(d)
    return get_data

# 将字典列表导出xls
def export_xls(path, dic_data):
    data_list = []
    # 循环得到每一个data
    for data in dic_data:
        # 循环得到data字典里的所有键值对的值
        for value in data.values():
            # 将得到的值放入空列表中
            data_list.append(value)
    # 创建一个新的列表生成式并赋给一个变量new_list.
    # 这个列表生成式主要是将数据每3个为一个新的元素存入新的列表中，即列表套列表
    new_list = [data_list[i:i + 3] for i in range(0, len(data_list), 3)]
    # 生成一个xlwt.Workbook对象
    xls = xlwt.Workbook()
    # 调用对象的add_sheet方法
    sheet = xls.add_sheet('Sheet1', cell_overwrite_ok=True)
    # 创建我们需要的第一行的标头数据
    heads = ['id', 'message', 'result']
    ls = 0
    # 将标头循环写入表中
    for head in heads:
        sheet.write(0, ls, head)
        ls += 1
    i = 1
    # 将数据分两次循环写入表中 外围循环行
    for list in new_list:
        j = 0
        # 内围循环列
        for data in list:
            sheet.write(i, j, data)
            j += 1
        i += 1
    # 最后将文件save保存
    xls.save(path)


# 将字典列表导出xlsx
def export_xlsx(path, dic_data):
    # 将字典列表转换为DataFrame
    pf = pd.DataFrame(list(dic_data))
    # 指定字段顺序
    order = ['id', 'message', 'result']
    pf = pf[order]
    # 指定生成的Excel表格名称
    file_path=pd.ExcelWriter(path)
    # 替换空单元格
    pf.fillna(' ', inplace=True)
    # 输出
    pf.to_excel(file_path, encoding='utf-8', index=False)
    # 保存表格
    file_path.save()


# 将字典列表导出csv
def export_csv(path, dic_data):
    with open(path, 'w', newline='') as f:
        fieldnames = ['id', 'message', 'result']
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        for item in dic_data:
            writer.writerow(item)