2 posts tagged with "Django"

class student():
    # 类变量，可在所有实例共享
    age = 20
    name = 'qiuwen'
    
    def __init__(self,age,name):
        # 实例变量，每个实例独有的
        self.age = age
        self.name = name

class Student:
    def __init__(self, id, name, age, sex):
        self.id = id
        self.name = name
        self.age = age
        self.sex = sex
    def test(self):
        print("name:{} age:{} sex:{}".format(self.name, self.age,self.sex))

class BoyStudent(Student):
    def test(self):
        super().test()  # super：子类重写父类方法后还需调用父类方法
        print('男学生方法')
class GirlStudent(Student):
    def __init__(self, id, name, age, sex, grade):
        super().__init__(id, name, age, sex)
        self.grade = grade
    def test(self):
        print('女学生方法'+ self.grade)
        
        
s = Student(1, 'yuqiuwen', '20', 'male')
s.test()
bs = BoyStudent(1, 'yuqiuwen', '20', 'male')
bs.test()
gs = GirlStudent(2, 'zhongxin', '18', 'female','3')
gs.test()
    

class MyClass:
    def __init__(self, filename, mode='r', encoding='utf-8'):
        self.filename = filename
        self.mode = mode
        self.encoding = encoding

    def __enter__(self):
        self.file = open(self.filename, mode=self.mode, encoding=self.encoding)
        return self.file
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.file.close()


with MyClass('test.txt', 'w', encoding='utf-8') as f:
    f.write("test msg")

__new__(cls)：构造方法，创建实例时首先调用的方法
__init__(self)：初始化方法，new方法返回对象后进行属性初始化
__del__(self)：析构方法，实例化对象被彻底销毁时调用
__len__(self)：被len()调用时的操作
__bool__(self)：被bool()调用时的行为
__str__(self)：被str()调用时的操作，将值转化为适于人阅读的形式
__repr__(self)：repr()调用时的操作，将值转化为供解释器读取的形式
__getattr__(self,name)：获取一个不存在的属性时的行为
__setattr__(self,name,value)：当一个属性被设置时的行为
__getattribute__(self,name)：当该类的属性被访问时首先调用的方法，可作为属性访问拦截器
__setitem__(self, key, value)：同setattr方法类似，区别在于如果将对象当作字典操作，设置键值对时会触发该方法
__call__(self)：允许将类实例当做函数调用

__dict__：包含一个字典，由类的数据属性组成
__doc__：类的文档字符串
__name__：类名
__module__：类定义所在的模块
__bases__：类的所有父类构成元素

import abc

class All_file(metaclass=abc.ABCMeta):
    all_type='file'
    
    @abc.abstractmethod #定义抽象方法，无需实现功能
    def read(self):
        '子类必须定义读功能'
        pass

    @abc.abstractmethod #定义抽象方法，无需实现功能
    def write(self):
        '子类必须定义写功能'
        pass
    
class Txt(All_file): #子类继承抽象类，但是必须定义read和write方法
    def read(self):
        print('文本数据的读取方法')

    def write(self):
        print('文本数据的读取方法')

class Sata(All_file): #子类继承抽象类，但是必须定义read和write方法
    def read(self):
        print('硬盘数据的读取方法')

    def write(self):
        print('硬盘数据的读取方法')

from socket import *


"""客户端"""

IP = '127.0.0.1'
SERVER_PORT = 8006
BUFLEN = 1024

dataSocket = socket(AF_INET, SOCK_STREAM)

# 连接服务端
dataSocket.connect((IP, SERVER_PORT))

while True:
    toSend = input('>> ')
    if toSend == '':
        break
    dataSocket.send(toSend.encode())

    recved = dataSocket.recv(BUFLEN)
    if not recved:
        break

    print(recved.decode())

dataSocket.close()

from socket import *


"""
服务端
"""

IP = '127.0.0.1'
PORT = 8006
BUFLEN = 1024    # 一次从socket缓冲区最多读入512个字节数据


"""
实例化一个socket对象
AF_INET表示该socket网络层使用ip协议
SOCK_STREAM 表示socket传输层使用tcp协议
"""
listenSocket = socket(AF_INET, SOCK_STREAM)

# socket绑定地址和端口
listenSocket.bind((IP, PORT))

# 使socket处于监听状态，等待客户端的连接请求
# 5表示最多接受5个等待连接的客户端
listenSocket.listen(5)
print(f"服务端启动成功，在{PORT}端口等待客户端连接...")

dataSocket, addr = listenSocket.accept()
print("接收一个客户端连接：", addr)

while True:
    # 读取对方发送的信息，最多读取BUFLEN字节
    recved = dataSocket.recv(BUFLEN)

    # 返回空字节，表示对方关闭了连接
    if not recved:
        break

    info = recved.decode()
    print(f"收到对方信息：{info}")

    dataSocket.send(f"服务端接收到了信息：{info}".encode())

dataSocket.close()
listenSocket.close()

def timeit(max_time):
    def inner(f):
        def wrapper(*args, **kw):
            start = time.perf_counter()
            res = f(*args, **kw)
            end = time.perf_counter()
            execute_time = end - start
            if execute_time > max_time:
                print(f""""func" function exceeds max_time""")
            else:
                print(f""""func" function ran for {execute_time}""")
            return res
        return wrapper
    return inner


@timeit(max_time=1)
def func(*args, **kw):
    time.sleep(2)

func(1)

"""相当于执行func = timeit(1)(func)，即先执行timeit(1)返回一个inner函数的引用，然后将func传给inner函数，返回wrapper保存到func对象"""

def timeit(f):
    # 这里使用*args和**kwargs使得函数可传递变长参数
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        f(*args, **kw)
        end = time.perf_counter()
        execute_time = end - start
        if execute_time > 2:
            print(f""""func" function exceeds max_time""")
        else:
            print(f""""func" function ran for {execute_time}""")
    return wrapper
    
@timeit
def func(*args, **kw):
    time.sleep(2)
    
"""相当于执行func = timeit(func)"""

class Timer:
    def __init__(self, func):
        self.func = func
    
    def __call__(self, *args, **kwargs):
        start = time.perf_counter()
        ret = self.func(*args, **kwargs)
        end = time.perf_counter()
        execute_time = end - start
        if execute_time > 2:
            print(f""""func" function exceeds max_time""")
        else:
            print(f""""func" function ran for {execute_time}""")
        return ret
        
@Timer
def func(*args, **kw):
    time.sleep(2)
    
"""相当于执行func=Timer(func)"""

class Timer:
    def __init__(self, max_time):
        self.max_time = max_time
        
    def __call__(self, func):
        def wrapper(*args, **kwargs):
            start = time.perf_counter()
            ret = func(*args, **kwargs)
            end = time.perf_counter()
            execute_time = end - start
            if execute_time > 2:
                print(f""""func" function exceeds max_time""")
            else:
                print(f""""func" function ran for {execute_time}""")
        	return ret
       	return wrapper
    
@Timer(max_time=2)
def func(*args, **kw):
    time.sleep(2)
    
"""相当于执行func=Timer(max_time=2)(func)"""

import time
from threading import Thread


def display(content):
   while True:
       print(content, end='', flush=True)
       time.sleep(0.1)


def main():
   Thread(target=display, args=('Ping', ), daemon=True).start()
   Thread(target=display, args=('Pong', ), daemon=True).start()
   time.sleep(5)


if __name__ == '__main__':
   main()

import time

from concurrent.futures import ThreadPoolExecutor
from threading import RLock


class Account(object):
    """银行账户"""
    def __init__(self):
        self.balance = 0.0
        self.lock = RLock()

    def deposit(self, money):
        # 通过上下文语法获得锁和释放锁
        with self.lock:
            new_balance = self.balance + money
            time.sleep(0.01)
            self.balance = new_balance
            
def main():
    """主函数"""
    account = Account()
    with ThreadPoolExecutor(max_workers=16) as pool:
        for _ in range(100):
            pool.submit(account.deposit, 1)
            print(account.balance)

if __name__ == '__main__':
    main()

class ZeroEvenOdd:
    def __init__(self, n):
        self.n = n
        self.lock_zero = threading.Lock()
        self.lock_even = threading.Lock()
        self.lock_odd = threading.Lock()
		
        # 初始锁住奇偶，保证先输出0
        self.lock_even.acquire()
        self.lock_odd.acquire()

    def zero(self) -> None:
        for i in range(1, self.n + 1):
            self.lock_zero.acquire()
            print(0, end="")
            # 上次输出偶数，则此次释放奇数锁，使下次输出奇数
            if i & 1 == 0:
                self.lock_even.release()
            else:
                self.lock_odd.release()

    # 偶数
    def even(self) -> None:
        for i in range(2, self.n+1, 2):
            self.lock_even.acquire()
            print(i, end="")
            self.lock_zero.release()
	
    # 奇数
    def odd(self) -> None:
        for i in range(1, self.n + 1, 2):
            self.lock_odd.acquire()
            print(i, end="")
            self.lock_zero.release()

"""启用两个线程实现交替打印foobar"""

class FooBar:
    def __init__(self, n):
        self.n = n
        self.queue_bar = Queue()
        self.queue_foo = Queue()
        self.queue_foo.put('')	# 初始foo队列放入一个元素保证首先输出foo

    def foo(self):


        for i in range(self.n):     
            self.queue_foo.get()	# 默认阻塞，队列为空时会一直阻塞
            print("foo", end="")
            self.queue_bar.put('')	# 默认会阻塞直至队列有空位

    def bar(self):

        for i in range(self.n):
            self.queue_bar.get()
            print("bar", end="")
            self.queue_foo.put('')


if __name__ == "__main__":
    n = int(input())
    c = FooBar(n)
    foo = threading.Thread(target=c.foo, name='thread_foo')
    bar = threading.Thread(target=c.bar, name='thread_bar')

    foo.start()
    bar.start()

from threading import RLock, Lock
from concurrent.futures import ThreadPoolExecutor, as_completed


class FooBar:
    def __init__(self, n):
        self.n = n
        self.queue_bar = Queue()
        self.queue_foo = Queue()
        self.queue_foo.put('')

    def foo(self):
        for i in range(self.n):
            # 默认阻塞，队列为空时会一直
            self.queue_foo.get()
            time.sleep(0.3)
            print("foo", end="")
            self.queue_bar.put('')

        return "print foo has completed"

    def bar(self):
        for i in range(self.n):
            # 队列为空时会阻塞
            self.queue_bar.get()
            time.sleep(0.3)
            print("bar", end="")
            self.queue_foo.put('')
        return "print bar has completed"


def print_threading():
    n = int(input())
    c = FooBar(n)
    target = [c.foo, c.bar]
    futures = []
    with ThreadPoolExecutor(max_workers=2) as pool:
        futures = [pool.submit(i) for i in target]
        
        # 获取返回结果，as_completed不用等待所有任务完成即可返回结果
        for f in as_completed(futures):
            data = f.result()
            print(data)

"""
timeit库的用法
	stmt:传入要测试的代码或函数
	setip:运行时的变量、模块等，多行语句用;隔开
	number:运行次数，默认100万次
"""
execute_time = timeit.timeit(stmt='print_threading()', setup='from __main__ import print_threading', number=1,)
print(execute_time)

import asyncio


async def task1(x):
    print("task1开始")
    # 执行到await时任务被挂起，跳到下一个任务
    await asyncio.sleep(1)
    print("task1完成")
    return x ** 2


async def task2(y):
    print("task2开始")
    await asyncio.sleep(2)
    print("task2完成")
    return y ** 2


async def main():
    # 协程对象，该对象不能直接运行
    coroutine1 = task1(2)
    coroutine2 = task2(3)
    await asyncio.gather(coroutine1, coroutine2)

    
# python3.7版本及之后实现协程
asyncio.run(main())

# python3.7之前
# # 事件循环对象
# loop = asyncio.get_event_loop()
# # 将协程对象注册到事件循环中，并执行
# ret = loop.run_until_complete(coroutine1)
# print(ret)

import os  
from zipfile import ZipFile  
  
  
def make_zip(target, output):  
    """  
    压缩文件夹或单个文件  
    :param target: 要打包的文件路径. /xx/xx/file  
    :param output: 输出路径. /xx/xx/file.zip  
    :return:  
    """    if not os.path.exists(target):  
        raise FileNotFoundError  
    with ZipFile(output, 'w') as zf:  
        if os.path.isfile(target):  # 打包单个文件  
            zf.write(target, os.path.relpath(target, start=os.path.dirname(target)))  
        else:                       # 打包目录  
            for parent, dirnames, filenames in os.walk(target):  
                for filename in filenames:  
                    filepath = os.path.join(parent, filename)  
                    relpath = os.path.relpath(filepath, start=os.path.dirname(target))  
                    zf.write(filepath, relpath)  
  
  
def unzip(target, output):  
    """  
    解压文件夹  
    :param target: 压缩文件路径  
    :param output: 输出路径  
    :return:  
    """    if not os.path.exists(output):  
        os.makedirs(output)  
    with ZipFile(target, mode='r') as zf:  
        zf.extractall(output)  
  
  
# make_zip("E:/uploads/files", "E:/uploads/files.zip")  
unzip("E:/uploads/files.zip", "E:/uploads/output")

path1 = r"E:\Projects\PythonProjects\flask\test.py"
path2 = Path(r"E:\Projects\PythonProjects\flask\test.py")

# 当前目录
print(Path.cwd())
print(os.getcwd())

# 家目录
print(Path.home())

# 绝对路径
print(os.path.abspath("目标绝对路径"))
print(os.path.realpath("返回软链接的源地址"))
print(path2.resolve())
print(os.path.isabs(path1))
print(path2.is_absolute())

# 相对路径
Path("/etc/passwd").relative_to("/etc")     # passwd
os.path.relpath("/etc/passwd", start=os.curdir)

# 文件名
print(os.path.basename(path1))
print(path2.name)

# 目录名
print(os.path.dirname(path1))
print(path2.parent)

# 判断文件是否存在
print(os.path.isfile(path1))
print(path2.is_file())

# 判断目录是否存在
print(os.path.isdir(path1))
print(path2.is_dir())

# 判断路径是否存在
print(os.path.exists(path1))
print(path2.exists())

# 路径拼接
print(os.path.join("path1", "path2"))	# 若path2也为绝对路径，则返回path2的路径
print(path2.joinpath(path3))
print(Path(path2, 'path3', 'file.py'))
print(path2/'file.py')

# 路径分割
print(os.path.split("/a/b/c.py.go"))		# ('/a/b', 'c.py')
print(os.path.splitext("/a/b/c.py.go"))		# ('/a/b/c.py', '.go')
print(Path("/a/b/c.py.go").suffix)			# ['.go']
print(Path("/a/b/c.py.go").suffixes)		# ['.py', '.go']
print(Path("/a/b/c.py.go").stem)			# c.py

# 路径比较
os.path.commonpath(['/usr/lib', '/usr/local/lib'])  # '/usr'
os.path.commonprefix(['/usr/lib', '/usr/local/lib'])  # '/usr/l'

# 替换用户家目录,~会被替换为系统当前的用户名
os.path.expanduser("~/test.py")  # /{user}/test.py
Path("~/test.py").expanduser()   # /{user}/test.py

# 替换windows反斜杠
path2.as_posix()

# 路径模式匹配，若pattern是绝对路径，path也必须为绝对路径
path2.match('/*.py')

# 创建
os.mkdir("创建目录")
os.makedirs("支持多级创建", exist_ok=True)   # exist_ok=True 存在就不创建
Path("path").mkdir(exist_ok=True, parents=True)  # parents=True 多级创建
Path("在指定路径创建文件").touch()

# 删除，删除的目录必须是空目录，否则会报错
os.remove("只能是删除文件")
os.rmdir("删除目录")
os.removedirs("可以删除多级目录")
Path("删除目录").rmdir()
Path("删除文件，若路径是目录，则删除目录").unlink()

# 重命名
os.rename("旧文件名", "新文件名")
os.renames("旧路径", "新路径")        # rename的扩展，还可以重命名路径（目录名）
Path("旧路径").rename("新路径")       # 只能重命名文件，无法重命名路径
Path("c:/Downloads/pathlib.tar.gz").with_name("setup.py")   # c:/Downloads/setup.py
Path("c:/Downloads/pathlib.tar.gz").with_suffix(".bz2")     # c:/Downloads/pathlib.tar.bz2

Path("返回文件信息").stat()
with path2.open() as f:
    ...

# 遍历目录
for root, dirs, files in os.walk("深度遍历", followlinks=False):
# root-目录路径，dirs-当前路径下的所有目录名，files当前目录下的所有文件名
	for filename in files:
		file_path = os.path.join(root, filename)
            yield file_path
            
files = [path for path in os.listdir("只遍历一级的目录和文件")]
files = [path for path in Path("返回生成器, 只遍历一级目录和文件").iterdir()]
files = [path for path in Path("返回生成器, 只遍历一级目录和文件，可正则过滤").glob('*.txt')]
files = [path for path in Path("返回生成器, 深度遍历，[!.]会排除以点开头的文件").rglob('[!.]*')]

from functools import wraps
from threading import RLock


def singleton(cls):
    """线程安全的单例装饰器"""
    instances = {}
    lock = RLock()
	
    # 使用wraps可保证不会修改被装饰函数的结构，如name等
    @wraps(cls)
    def wrapper(*args, **kwargs):
         
        """
        判断是否已有实例,如无,则新建一个实例并返回;
        先做不带锁的判断，再做带锁的检查，因为如果对象已经创建就没必要加锁可直接返回；第二个if是因为多个线程抢占锁，若第一个线程创建了实例，那么下一个线程进入就需要再判断一次
        instance = {<class '__main__.Single'>: <__main__.Single object at 0x000001A96F26FE20>}
        """
        if cls not in instances:
            with lock:
                if cls not in instances:
                    instances[cls] = cls(*args, **kwargs)
        return instances[cls]
    
   	return wrapper

class Singleton:
    instance = None
    lock = threading.RLock()
    
    def __init__(self, *args, **kwargs):
        pass
    
    def __new__(cls, *args, **kwargs):
        # 返回空对象
        if not cls.instance:
            with cls.lock:
                if not cls.instance:
                    cls.instance = object.__new__(cls)
        return cls.instance

创建项目  django-admin startproject keep_healthy
创建app	python manage.py startapp diabetes
后台启动  nohup python3 manage.py runserver 0.0.0.0:8082> /dev/null 2> /dev/null &
         nohup flask run -h 0.0.0.0 -p 5000 >/dev/null 2>&1 &

#SPIDER_MIDDLEWARES = {
#    'weibo_spider.middlewares.WeiboSpiderSpiderMiddleware': 543,
#}

s1 = 'bar ber bir bor bur'
s2 = 'abecdfghijklmnopqrstuvwxyz'
regex = ["b[ieo]r","b[^ieo]r","[e-o]"]
re.findall(regex[0],string=s1)  # ['ber', 'bir', 'bor']
re.findall(regex[1],string=s1) # ['bar', 'bur']
re.findall(regex[2],string=s2) # 匹配字符范围，数字同理

* 不出现、出现一次或多次（贪婪匹配）

s = 'br ber beer beeer'
re.findall("be*r", string=s)

+ 出现一次或多次

re.findall("be+r", string=s)

? 可选字符(懒惰匹配)

re.findall("be?r", string=s) # ['br', 'ber']

{} 出现次数

re.findall("be{1}r", string=s)  # 出现一次
re.findall("be{1,}r", string=s)  # 至少出现一次
re.findall("be{1,3}r", string=s)  # 出现1~3次

()

s="ha-ha,haa-haa"

re.findall(r"(ha)-\1,(haa)-\2",string=s)  # \1引用分组
re.findall(r"(?:ha)-ha,(haa)-\1",string=s)  # (?:)非引用分组

| 或

s = 'cat Cat rat'
re.findall(r"(C|c)at|rat", string=s)

^ $行首 行尾

s = '1. 3 eggs, beaten3'
re.findall('^[0-9]', string=s)
re.findall('[0-9]$', string=s)

s = 'Date: 4 Aug 3PM'

正向先行断言: (?=)
#只匹配后面有PM的数值
re.findall('\d+(?=PM)',string=s)


负向先行断言: (?!)
# 数字右边不是PM
re.findall('\d+(?!PM)',string=s)


正向后行断言: (?<=)
# 数字左边为$
s='Product Code: 1064 Price: $5'
re.findall("(?<=\$)\d+",string=s)


负向后行断言
# 数字左边不包含$
re.findall("(?<!\$)\d+",string=s)

s = 'cAt cat cat'

regex = re.compile(r'cat', re.IGNORECASE)
re.findall(regex, string=s)


s = "42 11.1   -2.0  rng 24"
re.findall("(?:-?\d+\.?\d+\s+)+(?=rng)", s)
?: 意思是非捕获分组，可以取消分组功能，若有'()',findall默认会捕获括号分组，search默认捕获所有分组，

import re

"""match ip"""

# 如47.98.97.198

1. 匹配0-255，包括0-199和200-255：([0-1]?\d{1,2})|(2((5[0-5])|([0-4]\d)))
2. ‘.XXX’匹配三次：(\.(([0-1]?\d{1,2})|(2((5[0-5])|([0-4]\d))))){3}

reg = re.compile("^(([0-1]?\d{1,2})|(2((5[0-5])|([0-4]\d))))(\.(([0-1]?\d{1,2})|(2((5[0-5])|([0-4]\d))))){3}$")

import hashlib

m = hashlib.md5()
m.update(str.encode("utf8"))
print(m.hexdigest())


sha1 = hashlib.sha1()
data = 'helloword'
sha1.update(data.encode('utf-8'))
sha1_data = sha1.hexdigest()
print(sha1_data)


def get_rsa(string):
    #1-1 实例化加密对象
    (pubkey,privkey)=rsa.newkeys(1024)
    #1-2 公钥加密1
    pwd1=rsa.encrypt(string,pubkey)
    print('加密后结果1为：',pwd1.hex())
    #1-3 私钥解密1
    depwd1=rsa.decrypt(pwd1,privkey)
    print('解密后的结果1为：',depwd1.decode())
get_rsa( b"15863274538")