晨曦's Blog

import datetime

'''
UTC转北京时间
'''

def utc_format(utc_time, utc_time_format='%Y-%m-%dT%H:%M:%S.000Z'):
    utc_datetime = datetime.datetime.strptime(utc_time, utc_time_format)
    local_datetime = utc_datetime + datetime.timedelta(hours=8)
    local_time_format = "%Y-%m-%d %H:%M:%S"
    local_time = local_datetime.strftime(local_time_format)
    return local_time

print(utc_format(utc_time="2021-10-26T09:34:31.000Z"))

async def aioredlock():
    # retry_count 重试次数，这里配置为1，杜绝进程重复去获取
    lock_manager = Aioredlock([{'host': 'localhost', 'port': 6379, 'db': 1}],retry_count=1)

    while True:

        # 每秒钟去检测锁的状态，如果锁被其他进程占用就跳过
        # lock_timeout 锁的时间为 65S，相当于进程拿到锁后最大65秒后释放
        # 业务逻辑执行 小于 65秒，业务逻辑执行完后当即释放锁
        # 每60秒执行一次 sleep 59s即可，因为检测已经sleep了1s

        await asyncio.sleep(1)
        if await lock_manager.is_locked("_ztsg_auto_generate_code"):
            #logger.info('The resource is already acquired')
            continue

        try:
            async with await lock_manager.lock("_ztsg_auto_generate_code", lock_timeout=65) as lock:

                assert lock.valid is True
                assert await lock_manager.is_locked("_ztsg_auto_generate_code") is True

                # 每60s钟运行一次任务
                await asyncio.sleep(59)

                logger.info('Start Tick! The time is '+str(os.getpid())+': %s' % datetime.now())
                # 业务代码
                logger.info('Stop Tick! The time is '+str(os.getpid())+': %s' % datetime.now())
        except LockAcquiringError:
            print('Something happened during normal operation. We just log it.')
        except LockError:
            print('Something is really wrong and we prefer to raise the exception')
        except Exception as e:
            print(e)
            print('Something else went wrong')

register_tortoise(
    app, db_url=db_url, modules={"models": ["app.models"]}, generate_schemas=False
)

register_tortoise(
    app,
    config={
        'connections': {
            'default': db_url
        },
        'apps': {
            'models': {
                "models": ["app.models"],
                'default_connection': 'default',
            }
        },
        "use_tz": False,
        "timezone": "Asia/Shanghai",
    },
    generate_schemas=False
)

import json
from base64 import b64encode, b64decode
from Crypto.Cipher import AES
from Crypto.Util.Padding import pad, unpad
from Crypto.Random import get_random_bytes

class AesCbc:
    def __init__(self, key):
        self.bs = AES.block_size
        self.key = key
        self.mode = AES.MODE_CBC

    def encrypt(self, data):
        cipher = AES.new(self.key, self.mode)
        ct_bytes = cipher.encrypt(pad(data, self.bs))
        iv = b64encode(cipher.iv).decode('utf-8')
        ct = b64encode(ct_bytes).decode('utf-8')
        return json.dumps({'iv': iv, 'ciphertext': ct})

    def decrypt(self, json_input):
        try:
            b64 = json.loads(json_input)
            iv = b64decode(b64['iv'])
            ct = b64decode(b64['ciphertext'])
            cipher = AES.new(self.key, self.mode, iv)
            plaintext = unpad(cipher.decrypt(ct), self.bs)
            return plaintext.decode('utf-8')
        except (ValueError, KeyError) as err:
            print("Incorrect decryption ", err)
            return None

if __name__ == "__main__":
    data = "需要加密的数据".encode('utf-8')
    key = get_random_bytes(16)  # Bytes

    aes_cipher = AesCbc(key)

    encrypt_reuslt = aes_cipher.encrypt(data)
    print("密文: ", encrypt_reuslt)

    plaintext = aes_cipher.decrypt(encrypt_reuslt)
    print("原文: ", plaintext)

import json
from base64 import b64encode, b64decode
from Crypto.Cipher import AES
from Crypto.Random import get_random_bytes

class AesOcb:
    def __init__(self, key):
        self.bs = AES.block_size
        self.key = key
        self.mode = AES.MODE_OCB
        self.json_k = ['nonce', 'header', 'ciphertext', 'tag']

    def encrypt(self, header, data):
        header = header
        cipher = AES.new(self.key, self.mode)
        cipher.update(header)
        ciphertext, tag = cipher.encrypt_and_digest(data)
        json_v = [b64encode(x).decode('utf-8') for x in [cipher.nonce, header, ciphertext, tag]]
        return json.dumps(dict(zip(self.json_k, json_v)))

    def decrypt(self, json_input):
        try:
            b64 = json.loads(json_input)
            jv = {k: b64decode(b64[k]) for k in self.json_k}
            cipher = AES.new(self.key, self.mode, nonce=jv['nonce'])
            cipher.update(jv['header'])
            plaintext = cipher.decrypt_and_verify(jv['ciphertext'], jv['tag'])
            return plaintext.decode('utf-8')
        except (ValueError, KeyError) as err:
            # 解密错误
            print(err)
            return None

if __name__ == "__main__":
    data = "需要加密的数据".encode('utf-8')
    key = get_random_bytes(16)  # Bytes
    header = b'header'

    aes_cipher = AesOcb(key)

    encrypt_reuslt = aes_cipher.encrypt(header, data)
    print("密文: ", encrypt_reuslt)

    plaintext = aes_cipher.decrypt(encrypt_reuslt)
    print("原文: ", plaintext)

import requests
import hashlib
import time

key = "xxx" # 密钥，由服务端颁发

client = "local" # 客户端
server = "kafka" # 服务端
env = "test" # 环境
timestamp = str(time.time())[:10] # 时间
sign_str = client+server+env+timestamp+key
sign = hashlib.md5(sign_str.encode(encoding='utf-8')).hexdigest() # md5

headers = dict(client=client, server=server, env=env, timestamp=timestamp, sign=sign)

r1 = requests.get("https://www.xxx.com/hello", headers=headers)

from sanic import Sanic
from sanic.response import text, json
import hashlib

app = Sanic("My Hello, world app")

@app.get("/hello")
async def hello_world(request):

      key = "xxx"  # 密钥

      client = request.headers.get("client")
      server = request.headers.get("server")
      env = request.headers.get("env")
      timestamp = request.headers.get("timestamp")
      sign = request.headers.get("sign")

      _sign_str = client+server+env+timestamp+key
      _sign = hashlib.md5(_sign_str.encode(encoding='utf-8')).hexdigest()

      if sign==_sign:
            return json(list(request.headers.items()))
      else:
            return text("验证失败")
app.run(host='0.0.0.0',debug=True,auto_reload=True)

./server.py

from sanic import Sanic
from databases import Database

app = Sanic('jakehu')

# 数据库
database = Database('mysql://localhost/example')
# 服务开启之前
@app.listener("before_server_start")
async def startup_db(app, loop):
    await database.connect()
# 服务结束之后
@app.listener("after_server_stop")
async def shutdown_db(app, loop):
    await database.disconnect()
app.ctx.db = database

@app.get("/")
async def foo_handler(request):
    return text("Hello World!")

app.run(host='0.0.0.0', debug=True, auto_reload=True)

@app.get("/")
async def foo_handler(request):
    database= request.app.ctx.db
    query = "SELECT * FROM table"
    rows = await database.fetch_all(query=query)
    result = [dict(row._mapping) for row in rows]
    return json(result)