Loguru是一个功能强大且易于使用的开源Python日志记录库。它建立在Python标准库中的logging模块之上，并提供了更加简洁直观、功能丰富的接口。Logging模块的使用见： Python日志记录库logging总结 。Loguru官方仓库见：loguru，loguru官方文档见： loguru-doc。Loguru的主要特点包括：简单易用：无需复杂的配置和定制即可实现基本的日志记录和输出。灵活的日志格式：支持自定义日志格式，并提供丰富的格式化选项。丰富的日志级别：支持多种日志级别，例如DEBUG、INFO、WARNING、ERROR和CRITICAL。多种日志目标：可以将日志输出到终端、文件、电子邮件、网络服务器等目标。强大的日志处理功能：支持日志过滤、格式化、压缩、旋转等功能。支持异步日志记录：能够极大地提升日志记录的性能。支持跨进程、跨线程的日志记录：可以安全地记录多进程、多线程应用程序的日志。Loguru与logging是Python中常用的两个日志记录库，但两者在功能和易用性方面存在一些差异，如下所示：特性Logurulogging易用性更简单易用相对复杂日志格式更灵活较简单日志级别更丰富较少日志目标更多种类较少日志处理功能更强大较弱异步日志记录支持不支持跨进程、跨线程支持支持支持总的来说，loguru在易用性、功能性和性能方面都优于logging。如果要一个简单、强大且易于使用的日志系统，loguru是一个很好的选择。而如果只是需要快速输出一些调试信息，print可能就足够了。不过，对于生产环境，使用loguru或其他日志系统通常会更加合适。Loguru安装命令如下：pip install loguru# 查看loguru版本 import loguru print(loguru.__version__) # 输出：0.7.21 使用说明1.1 基础用法简单使用 Loguru的核心概念是只有一个全局的日志记录器，也就是logger。这个设计使得日志记录变得非常简洁和一致。使用Loguru时，你不需要创建多个日志实例，而是直接使用这个全局的logger来记录信息。这不仅减少了配置的复杂性，也使得日志管理更加集中和高效。from loguru import logger logger.debug("这是一个调试信息")输出：2024-06-29 19:57:44.506 | DEBUG | __main__::3 - 这是一个调试信息Loguru日志输出默认格式如下：时间戳：表示日志记录的具体时间，格式通常为年-月-日 时:分:秒.毫秒。日志级别：表示这条日志的严重性级别。进程或线程标识：表示日志来自哪个模块或脚本。 main 表示日志来自主模块。如果是其他文件会显示文件名。文件名和行号：记录日志消息的函数名和行号。日志消息：实际的日志内容，此外loguru支持使用颜色来区分不同的日志级别，使得日志输出更加直观.日志等级Loguru可以通过简单的函数调用来记录不同级别的日志，并自动处理日志的格式化和输出。这一特点可以让使用者专注于记录重要的信息，而不必关心日志的具体实现细节。Loguru支持的日志级别，按照从最低到最高严重性排序：TRACE: 最详细的日志信息，用于追踪代码执行过程。Loguru默认情况下使用DEBUG级别作为最低日志记录级别，而不是TRACE级别。这是因为TRACE级别会产生大量的日志信息。DEBUG: 用于记录详细的调试信息，通常只在开发过程中使用，以帮助诊断问题。INFO: 用于记录常规信息，比如程序的正常运行状态或一些关键的操作。SUCCESS: 通常用于记录操作成功的消息，比如任务完成或数据成功保存。WARNING: 用于记录可能不是错误，但需要注意或可能在未来导致问题的事件。ERROR: 用于记录错误，这些错误可能会影响程序的某些功能，但通常不会导致程序完全停止。CRITICAL: 用于记录非常严重的错误，这些错误可能会导致程序完全停止或数据丢失。from loguru import logger logger.debug("这是一条跟踪消息") logger.debug("这是一条调试信息") logger.info("这是一条普通信息") logger.success("操作成功完成") logger.warning("这是一条警告信息") logger.error("这是一条错误信息") logger.critical("这是一条严重错误信息")输出：2024-06-29 19:58:11.535 | DEBUG | __main__:<module>:3 - 这是一条跟踪消息 2024-06-29 19:58:11.536 | DEBUG | __main__:<module>:4 - 这是一条调试信息 2024-06-29 19:58:11.536 | INFO | __main__:<module>:5 - 这是一条普通信息 2024-06-29 19:58:11.537 | SUCCESS | __main__:<module>:6 - 操作成功完成 2024-06-29 19:58:11.537 | WARNING | __main__:<module>:7 - 这是一条警告信息 2024-06-29 19:58:11.538 | ERROR | __main__:<module>:8 - 这是一条错误信息 2024-06-29 19:58:11.538 | CRITICAL | __main__:<module>:9 - 这是一条严重错误信息1.2 日志配置在loguru中，add函数用于添加日志处理器。这个函数用于指定日志消息应该被发送到何处，例如控制台、文件或其他自定义的目的地。add函数主要参数介绍如下：sink: 定义日志消息的输出位置，可以是文件路径、标准输出（stdout）、标准错误（stderr，默认）或其他自定义的输出位置。format: 指定日志消息的格式，可以是简单的字符串，也可以是格式化字符串，支持各种字段插值。level: 设置处理程序处理的日志消息的最低级别。比如设置为DEBUG，则处理程序将处理所有级别的日志消息。filter: 可选参数，用于添加过滤器，根据特定的条件过滤掉不需要的日志消息。colorize: 布尔值，指定是否对日志消息进行着色处理，使日志在控制台中更易于区分。serialize: 布尔值，指定是否对日志消息进行序列化处理，通常与enqueue=True一起使用，以确保多线程安全。enqueue: 布尔值，指定是否将日志消息放入队列中处理，用于多线程应用中避免阻塞。backtrace: 布尔值或字符串，指定是否记录回溯信息，默认为False。diagnose: 布尔值，启用后，会在处理程序内部出现错误时记录诊断信息。rotation: 日志文件轮换的配置，支持按大小或时间进行日志文件的轮换。retention: 用于设置日志文件的保留时间。compression: 布尔值，指定是否对轮换后的日志文件进行压缩处理。from loguru import logger import sys # 终端显示不受该段代码设置 # 添加一个日志处理器，输出到文件 # 设置日志最低显示级别为INFO，format将设置sink中的内容 # sink链接的本地文件，如不存在则新建。如果存在则追写 logger.add(sink="myapp.log", level="INFO", format="{time:HH:mm:ss} | {message}| {level}") # debug结果不被显示到本地文件 logger.debug("这是一条调试信息") logger.info("这是一条普通信息")输出：2024-06-29 19:58:56.159 | DEBUG | __main__:<module>:11 - 这是一条调试信息 2024-06-29 19:58:56.159 | INFO | __main__:<module>:12 - 这是一条普通信息当连续两次调用 add 函数时，loguru 会将新的日志处理器添加到处理器列表中，而不是覆盖之前的处理器。这意味着所有添加的处理器都会接收到日志消息，并且按照它们被添加的顺序来处理这些消息。from loguru import logger logger.add(sink="myapp1.log", level="INFO") logger.add(sink="myapp2.log", level="INFO") # 会同时存入所有add添加日志处理器 logger.info("这是一条普通信息，存入myapp2")如果想删除所有已添加的日志处理器，loguru运行使用 logger.remove()方法不带任何参数来移除所有日志处理器。from loguru import logger import sys # 移除所有日志处理器（包括终端输出） logger.remove() logger.add(sink="myapp3.log", level="INFO", format="{time:HH:mm:ss} | {message}| {level}") logger.debug("这是一条调试信息存入myapp3") logger.info("这是一条普通信息存入myapp3")注意调用logger.remove()之后的所有日志将不会被记录，因为没有处理器了。from loguru import logger # 移除所有日志处理器（包括终端输出） logger.remove() # 没有输出 logger.info("这是一条普通信息存入myapp3")如果希望移除某些日志处理器，而不是从所有日志器中移除，代码如下：from loguru import logger # 移除默认终端logger，如果终端存在。 # logger.remove(0) # 添加多个文件处理器，enqueu设置异步日志记录 handler1 = logger.add("myapp1.log", enqueue=True) print(handler1) # handler_id是移除的处理器的唯一标识符 handler2 = logger.add("myapp2.log") # 记录一些日志 logger.info("这些信息会被记录到两个文件中") # 移除特定的文件处理器 logger.remove(handler1) # 现在只有myapp2.log 会记录日志 logger.info("这条信息只会记录在myapp2.log 中")如果想将日志输出到日志台，代码如下：from loguru import logger import sys logger.remove() # 移除默认输出 # 添加一个日志处理器，输出到控制台，使用自定义格式 logger.add( sink=sys.stdout, level="DEBUG", # green表示颜色 format="<green>{time:HH:mm}</green> <level>{message}</level>" ) # 注意终端显示会同步显示 logger.debug("这是一条调试信息") logger.info("这是一条普通信息")时间自定义 可以使用datatime库来自定义日志时间格式。from datetime import datetime from loguru import logger # 自定义时间格式 # time_format = "%Y-%m-%d %H:%M:%S,%f" # 包括微秒 time_format = "%H:%M:%S,%f" # 包括微秒但不含年月日 # 定义日志格式，使用 datetime.now().strftime() 来格式化时间 log_format = "{time:" + time_format + "} - {level} - {message}" logger.add("myapp.log", format=log_format, level="DEBUG") # 记录一条日志 logger.debug("这是一个带有微秒的测试日志")日志轮换from loguru import logger # 当文件大小达到100MB时创建新的日志文件，旧文件保留并重命名，用于防止单个日志文件变得过大。 logger.add("file_1.log", rotation="100 MB") # 每天中午12时创建新的日志文件，旧文件保留并重命名 logger.add("file_2.log", rotation="12:00") # 当日志文件存在超过一周时创建新的日志文件，旧文件保留并重命名 logger.add("file_3.log", rotation="1 week") # 设置日志文件保留10天 logger.add("file_4.log", retention="10 days") # 当文件大小达到100MB时创建新的日志文件，旧文件保留压缩为zip文件 logger.add('file_{time}.log', rotation="100 MB", compression='zip')1.3 进阶使用异常捕获 @logger.catch装饰器可以用来装饰my_function函数，并将这些异常信息记录到日志中。from loguru import logger logger.add(sink='myapp.log') @logger.catch def my_function(x, y): return x / y res = my_function(0,0)过滤 使用loguru库进行Python日志记录时，可以通过自定义的filter函数来筛选并记录特定的日志信息。此函数接收一个记录对象作为参数，根据日志消息内容(message)、级别(level)或其他日志属性，返回布尔值以决定是否记录该条日志。如果函数返回True，则日志被记录；若返回False，则忽略该日志。from loguru import logger # 定义一个过滤器函数 def my_filter(record): # 只记录包含 "第一" 的日志 return "第一" in record["message"] # 使用过滤器 logger.add("myapp.log", filter=my_filter) # 记录一些日志 logger.info("第一个记录") logger.info("第二个记录")此外可以结合bind方法进行过滤，bind方法用于向日志记录器添加额外的上下文信息。这些信息将被包含在每条日志消息中，但不会改变日志消息本身。如下所示：from loguru import logger def filter_user(record): return record["extra"].get("user") =="A" logger.add("myapp.log", filter=filter_user) # 绑定user logger.bind(user="A").info("来自A") logger.bind(user="B").info("来自B")原文出处：https://www.cnblogs.com/luohenyueji/p/18276299
1.创建单表class Test(Base): __tablename__ = 'user' nid = Colume(Integer,primary_key=True,autoincrement=True) name = Colume(String(32))2.创建一对多class Team(Base): __tablename__ = 'team' tid = Colume(Integer,primary_key=True,autoincrement=True) caption = Colume(String(32)) class user(Base): __tablename__ = 'user' nid = Colume(Integer,primary_key=True,autoincrement=True) name = Colume(String(32)) team_id = Colume(Integer,ForeignKey('team.gid'))写完类，接下来就是把类转化为数据库表了。3.生成表、删除表def init_db(): #根据Base去找它的子类，把所有的子类生成表。 Base.metadata.create_all(engine) def drop_db(): #把Base所有的子类对应表删除。 Base.metadata.drop_all(engine)from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index from sqlalchemy.orm import sessionmaker, relationship from sqlalchemy import create_engine engine = create_engine("mysql+pymysql://python:python@192.168.0.57:3306/python_mysql", max_overflow=5) Base = declarative_base() # 单表 class Test(Base): __tablename__ = 'test' nid = Column(Integer, primary_key=True, autoincrement=True) name = Column(String(32)) # 一对多 class Team(Base): __tablename__ = 'team' tid = Column(Integer, primary_key=True, autoincrement=True) caption = Column(String(32)) class user(Base): __tablename__ = 'user' nid = Column(Integer, primary_key=True, autoincrement=True) name = Column(String(32)) team_id = Column(Integer, ForeignKey('team.tid')) def init_db(): Base.metadata.create_all(engine) def drop_db(): Base.metadata.drop_all(engine) init_db()执行完上面代码后，就会在对应库生成test、user、group三张表，user表的group_id以group表的gid为外键。4.生成表后开始操作表，添加team表数据from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index from sqlalchemy.orm import sessionmaker, relationship from sqlalchemy import create_engine engine = create_engine("mysql+pymysql://python:python@192.168.0.57:3306/python_mysql", max_overflow=5) Base = declarative_base() # 单表 class Test(Base): __tablename__ = 'test' nid = Column(Integer, primary_key=True, autoincrement=True) name = Column(String(32)) # 一对多 class Team(Base): __tablename__ = 'team' tid = Column(Integer, primary_key=True, autoincrement=True) caption = Column(String(32)) class user(Base): __tablename__ = 'user' nid = Column(Integer, primary_key=True, autoincrement=True) name = Column(String(32)) team_id = Column(Integer, ForeignKey('team.tid')) def init_db(): Base.metadata.create_all(engine) def drop_db(): Base.metadata.drop_all(engine) # init_db() # drop_db() Session = sessionmaker(bind=engine) session = Session() #往team表里插入两条数据 session.add(Team(caption='dba')) session.add(Team(caption='ddd')) session.commit()5. 添加user表数据Session = sessionmaker(bind=engine) session = Session() #批量添加数据；user表的team_id与team表的tid是有外键的，按理来说要插入的team_id的值必须在team表里有对应的tid值，比如这里插入的tead_id是1、2、3，则team表里的tid至少要有1、2、3，不然会插入失败。 #但是，我发现插入没有对应键值的team_id也不会报错。 session.add_all([ User(name='zzz',team_id=1), User(name='sss',team_id=2), User(name='ccc',team_id=3), ]) session.commit()6.查询单表如果仅仅是查询user表的name值，那不需要联合别的表，直接查询单表即可ret = session.query(User).filter(User.name=='zzz').all() obj = ret[0] print(obj.name) #上面的代码等价于这个： ret = session.query(User.name).filter(User.name=='zzz').all() print(ret)7.通过__repr__()方法改变返回值from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index from sqlalchemy.orm import sessionmaker, relationship from sqlalchemy import create_engine engine = create_engine("mysql+pymysql://python:python@192.168.0.57:3306/python_mysql", max_overflow=5) Base = declarative_base() # 创建单表 class Users(Base): __tablename__ = 'users' id = Column(Integer, primary_key=True) name = Column(String(32)) extra = Column(String(16)) __table_args__ = ( UniqueConstraint('id', 'name', name='uix_id_name'), Index('ix_id_name', 'name', 'extra'), ) #__repr__方法是注释的，看print(ret)的输出 #def __repr__(self): # return "%s-%s" %(self.id, self.name) def init_db(): Base.metadata.create_all(engine) def drop_db(): Base.metadata.drop_all(engine) init_db() Session = sessionmaker(bind=engine) session = Session() session.add(Users(id=1,name='zsc')) session.commit() ret = session.query(Users).all() print(ret) #结果： [<__main__.Users object at 0x7f1836e80630>] 没有User类里没有__repr__方法时，session.query(Users).all()返回的是类的对象。from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index from sqlalchemy.orm import sessionmaker, relationship from sqlalchemy import create_engine engine = create_engine("mysql+pymysql://python:python@192.168.0.57:3306/python_mysql", max_overflow=5) Base = declarative_base() # 创建单表 class Users(Base): __tablename__ = 'users' id = Column(Integer, primary_key=True) name = Column(String(32)) extra = Column(String(16)) __table_args__ = ( UniqueConstraint('id', 'name', name='uix_id_name'), Index('ix_id_name', 'name', 'extra'), ) #__repr__方法取消注释 def __repr__(self): return "%s-%s" %(self.id, self.name) def init_db(): Base.metadata.create_all(engine) def drop_db(): Base.metadata.drop_all(engine) init_db() Session = sessionmaker(bind=engine) session = Session() session.add(Users(id=1,name='zsc')) session.commit() ret = session.query(Users).all() print(ret) #结果： [1-zsc] User类里定义了__repr__方法时，session.query(Users).all()返回的是定义的返回结果。8.联合查询#创建表时指定了外键 ret = session.query(User.name).join(Team).all()等价于SELECT user.name AS FROM user INNER JOIN team ON team.tid = user.team_id #用select的话需要用on指定约束条件，用SQLAlchemy就不用指定了。 #用“isouter=True”指定left join ret = session.query(User.name).join(Team,isouter=True).all()上面的查询，随便是依赖到了别的表，但是结果只是显示了user表的数据，如果想同时显示user和team表的数据，就得用下面的方法了，ret = session.query(User.name,Team.caption).join(Team).all() print(ret) #结果： [('zzz', 'dba'), ('sss', 'ddd')]虽然上面的联合查询已经比直接用select简单了，但是还是很麻烦，所以就有了下面的方法。9.relationship9.1 利用ralationship正向查询正向查询即是使用做外链的表来查询被外链里的数据from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index from sqlalchemy.orm import sessionmaker, relationship from sqlalchemy import create_engine engine = create_engine("mysql+pymysql://python:python@192.168.0.57:3306/python_mysql", max_overflow=5) Base = declarative_base() # 一对多 class Team(Base): __tablename__ = 'team' tid = Column(Integer, primary_key=True, autoincrement=True) caption = Column(String(32)) class User(Base): __tablename__ = 'user' nid = Column(Integer, primary_key=True, autoincrement=True) name = Column(String(32)) team_id = Column(Integer, ForeignKey('team.tid')) #加上底下这行后，不用使用.join()也可实现联表查询 #哪个表做外链，就把relationship加到哪个表 favor = relationship("Team", backref='uuu') def init_db(): Base.metadata.create_all(engine) def drop_db(): Base.metadata.drop_all(engine) # init_db() # drop_db() Session = sessionmaker(bind=engine) session = Session() ret = session.query(User).all() for obj in ret: print(obj.nid,obj.name,obj.favor,obj.favor.tid,obj.favor.caption) #结果： 1 zzz <__main__.Team object at 0x7f5c10d02a20> 1 dba 2 sss <__main__.Team object at 0x7f5c10d026a0> 2 ddd #可见，ret仅仅是User的query结果，而使用obj.favor就相当于是使用Team表，即可直接操作team表。9.2利用ralationship实现反向查询反向查询即是使用被外链的表查询到做外链的数据class Test(Base): __tablename__ = 'test' nid = Column(Integer, primary_key=True, autoincrement=True) name = Column(String(32)) # 一对多 class Team(Base): __tablename__ = 'team' tid = Column(Integer, primary_key=True, autoincrement=True) caption = Column(String(32)) class User(Base): __tablename__ = 'user' nid = Column(Integer, primary_key=True, autoincrement=True) name = Column(String(32)) team_id = Column(Integer, ForeignKey('team.tid')) favor = relationship("Team", backref='uuu') def init_db(): Base.metadata.create_all(engine) def drop_db(): Base.metadata.drop_all(engine) # init_db() # drop_db() Session = sessionmaker(bind=engine) session = Session() ret = session.query(Team).filter(Team.caption == 'dba').all() print(ret[0].tid) print(ret[0].caption) print(ret[0].uuu) #结果： 1 dba [<__main__.User object at 0x7f7d3fa5ba20>] #favor = relationship("Team", backref='uuu')里的uuu的作用就是存储着对应的做外链里的数据；比如user里有7个人是dba组的，这时候print(ret[0].uuu)就会返回7个用户的信息；user里有3个dbb组的，这时候print(ret[0].uuu)就会返回3个相关用户的信息。除了一对多还是多对多关系，多对多是专门建一个中间表来存储两张表的关联关系。SQLAlchemy看着麻烦，其实就是记语法而已，多用即可；先建表，再操作单表，再用连表，在整关系，一对多，多对多。原文转载：https://www.cnblogs.com/fuckily/p/6042743.html
{callout color="#999999"}为了运维方便，选择直接用宝塔来管理python fastapi的项目，虽然直接部署可能性能更好更灵活，但是我选择了低层本，每个人的选择可能是不一样的，各有 考虑吧。本文的大逻辑是先写一个helloworld的程序，然后再部署到服务器上{/callout}步骤一：先本地运行一个基于fastapi的helloWorld例子，方便后面在服务器上验证编写基于FastApi的Hello World文件main.pyimport uvicorn from fastapi import FastAPI app = FastAPI() @app.get("/") def sayHi(): return {"message":"Hello world!"} # 启动uvicorn服务，默认端口8000 uvicorn myapi:api --reload if __name__ == '__main__': uvicorn.run('main:app')显示本地运行跑通，本地可以使用vscode编译器，并在运行dos命令pip install fastapi[all] uvicorn main:app --reloadreload参数是为了修改代码后的热部署，运行没有报错后可以浏览器访问： http://127.0.0.1:8000如果看到打印信息则说明OK步骤二：在宝塔上部署python的环境商店安装插件python进程管理插件Python项目管理器：管理应用实例进程守护管理器：实例进程的守护进入Python项目管理器，首先安装python版本，尽量与本地的一致，避免出现本地好的，服务器上出现问题，特别怕依赖包不一致的问题。本地查看版本的命令是python --version在本地生成requirements.txt，否则宝塔创建项目会报错。创建命令如下pip freeze >requirements.txt pip install -r requirements.txt将代码上传到宝塔的/www/wwwpython/helloworld在python进程管理插件创建项目，具体参数如下，记得选择gunicon在配置修改参数，重启。默认为worker_class = 'geventwebsocket.gunicorn.workers.GeventWebSocketWorker' 修改为worker_class = 'uvicorn.workers.UvicornWorker'这里有一个坑，启动后系统会自动暂停，日志报错如下的话，是2个问题，一是启动文件名为main.py，二是启动文件是app，即在main.py中启动命令是：uvicorn.run('main:app')具体报错信息如下：Worker failed to boot.Traceback (most recent call last): File "/www/wwwpython/helloworld/helloworld_venv/lib/python2.7/site-packages/gunicorn/arbiter.py", line 586, in spawn_worker worker.init_process() File "/www/wwwpython/helloworld/helloworld_venv/lib/python2.7/site-packages/gunicorn/workers/ggevent.py", line 203, in init_process super(GeventWorker, self).init_process() File "/www/wwwpython/helloworld/helloworld_venv/lib/python2.7/site-packages/gunicorn/workers/base.py", line 135, in init_process self.load_wsgi() File "/www/wwwpython/helloworld/helloworld_venv/lib/python2.7/site-packages/gunicorn/workers/base.py", line 144, in load_wsgi self.wsgi = self.app.wsgi() File "/www/wwwpython/helloworld/helloworld_venv/lib/python2.7/site-packages/gunicorn/app/base.py", line 67, in wsgi self.callable = self.load() File "/www/wwwpython/helloworld/helloworld_venv/lib/python2.7/site-packages/gunicorn/app/wsgiapp.py", line 52, in load return self.load_wsgiapp() File "/www/wwwpython/helloworld/helloworld_venv/lib/python2.7/site-packages/gunicorn/app/wsgiapp.py", line 41, in load_wsgiapp return util.import_app(self.app_uri) File "/www/wwwpython/helloworld/helloworld_venv/lib/python2.7/site-packages/gunicorn/util.py", line 350, in import_app __import__(module) File "/www/wwwpython/helloworld/helloworld_venv/lib/python2.7/site-packages/gevent/builtins.py", line 96, in __import__ result = _import(*args, **kwargs) File "/www/wwwpython/helloworld/myapi.py", line 9 SyntaxError: Non-ASCII character '\xe5' in file /www/wwwpython/helloworld/myapi.py on line 9, but no encoding declared; see http://www.python.org/peps/pep-0263.html for details [2023-03-25 09:26:44 +0000] [7373] [INFO] Worker exiting (pid: 7373) [2023-03-25 09:26:44 +0000] [7368] [INFO] Shutting down: Master [2023-03-25 09:26:44 +0000] [7368] [INFO] Reason: Worker failed to boot. [2023-03-25 09:26:45 +0000] [7385] [INFO] Starting gunicorn 19.10.0 [2023-03-25 09:26:45 +0000] [7385] [INFO] Listening at: http://0.0.0.0:12345 (7385) [2023-03-25 09:26:45 +0000] [7385] [INFO] Using worker: geventwebsocket.gunicorn.workers.GeventWebSocketWorker [2023-03-25 09:26:45 +0000] [7394] [INFO] Booting worker with pid: 7394 [2023-03-25 09:26:45 +0000] [7394] [ERROR] Exception in worker process Traceback (most recent call last): File "/www/wwwpython/helloworld/helloworld_venv/lib/python2.7/site-packages/gunicorn/arbiter.py", line 586, in spawn_worker worker.init_process() File "/www/wwwpython/helloworld/helloworld_venv/lib/python2.7/site-packages/gunicorn/workers/ggevent.py", line 203, in init_process super(GeventWorker, self).init_process() File "/www/wwwpython/helloworld/helloworld_venv/lib/python2.7/site-packages/gunicorn/workers/base.py", line 135, in init_process self.load_wsgi() File "/www/wwwpython/helloworld/helloworld_venv/lib/python2.7/site-packages/gunicorn/workers/base.py", line 144, in load_wsgi self.wsgi = self.app.wsgi() File "/www/wwwpython/helloworld/helloworld_venv/lib/python2.7/site-packages/gunicorn/app/base.py", line 67, in wsgi self.callable = self.load() File "/www/wwwpython/helloworld/helloworld_venv/lib/python2.7/site-packages/gunicorn/app/wsgiapp.py", line 52, in load return self.load_wsgiapp() File "/www/wwwpython/helloworld/helloworld_venv/lib/python2.7/site-packages/gunicorn/app/wsgiapp.py", line 41, in load_wsgiapp return util.import_app(self.app_uri) File "/www/wwwpython/helloworld/helloworld_venv/lib/python2.7/site-packages/gunicorn/util.py", line 350, in import_app __import__(module) File "/www/wwwpython/helloworld/helloworld_venv/lib/python2.7/site-packages/gevent/builtins.py", line 96, in __import__ result = _import(*args, **kwargs) File "/www/wwwpython/helloworld/myapi.py", line 9 SyntaxError: Non-ASCII character '\xe5' in file /www/wwwpython/helloworld/myapi.py on line 9, but no encoding declared; see http://www.python.org/peps/pep-0263.html for details [2023-03-25 09:26:45 +0000] [7394] [INFO] Worker exiting (pid: 7394) [2023-03-25 09:26:45 +0000] [7385] [INFO] Shutting down: Master [2023-03-25 09:26:45 +0000] [7385] [INFO] Reason: Worker failed to boot.设置映射域名，比如设置了 demo.xxx.com，即系统会自动创建这个域名的网站，并设置了反向代理通过 http://demo.xxx.com 返回正常即配置完成。原文出处：https://www.cnblogs.com/jpeanut/p/17254756.html
逻辑结构特点：只是描述数据结构中数据元素之间的联系规律是从具体问题中抽象出来的数学模型，是独立于计算机存储器的(与机器无关)逻辑结构分类线性结构集合中必存在唯一的一个 "第一个元素"集合中必存在唯一的一个 "最后的元素"除最后元素之外，其他数据元素均有唯一的 "后继"除第一元素之外，其他数据元素均有唯一的 "前驱"树形结构(层次结构)树形结构指的是数据元素之间存在着"一对多"的树形关系的数据结构，是一类重要的非线性数据结构。在树形结构中，树根结点没有前驱结点，其余每个结点有且只有一个前驱结点。叶子结点没有后续结点，其余每个结点的后续节点数可以是一个也可以是多个。图状结构（网状结构)图是一种比较复杂的数据结构。在图结构中任意两个元素之间都可能有关系，也就是说这是一种多对多的关系。其他结构除了以上几种常见的逻辑结构外，数据结构中还包含其他的结构，比如集合等。有时根据实际情况抽象的模型不止是简单的某一种，也可能拥有更多的特征。存储结构特点:是数据的逻辑结构在计算机存储器中的映象（或表示)存储结构是通过计算机程序来实现的，因而是依赖于具体的计算机语言的。存储结构分类顺序存储顺序存储( Sequential Storage ):将数据结构中各元素按照其逻辑顺序存放于存储器一片连续的存储空间中。链式存储链式存储(Linked Storage )∶将数据结构中各元素分布到存储器的不同点，用记录下一个结点位置的方式建立它们之间的联系，由此得到的存储结构为链式存储结构。索引存储索引存储（ lndexed Storage ):在存储数据的同时，建立一个附加的索引表，即索引存储结构=数据文件+索引表。线性表的链式存储只能操作头部节点指向下一个定义将线性表L=(a0,a1.......an-1)中各元素分布在存储器的不同存储块，称为结点，每个结点（尾节点除外）中都持有一个指向下一个节 点的引用，这样所得到的存储结构为链表结构。特点逻辑上相邻的元素ai, ai+1，其存储位置也不一定相邻;存储稀疏，不必开辟整块存储空间。对表的插入和删除等运算的效率较高。逻辑结构复杂，不利于遍历。栈先进后出 --> 水桶定义栈是限制在一端进行插入操作和删除操作的线性表（俗称堆栈），允许进行操作的一端称为"栈顶”，另一固定端称为"栈底”，当栈中没有元素时称为空栈"。特点栈只能在一端进行数据操作栈模型具有先进后出或者叫做后进先出的规律队列先进先出 --> 排队定义队列是限制在两端进行插入操作和删除操作的线性表，允许进行存入操作的一端称为队尾”，允许进行删除操作的一健称为“队就-b特点:队列只能在队头和队尾进行数据操作队列模型具有先进先出或者叫做后进后出的规律树形结构定义树( Tree )是n ( n≥0 )个节点的有限集合T，它满足两个条件∶有且仅有一个特定的称为根(Root )的节点;其余的节点可以分为m ( m≥0 )个互不相交的有限集合T1、T2、......、Tm，其中每一个集合又是一棵树，并称为其根的子树( Subtree ) 。基本概念一个节点的子树的个数称为该节点的度数，一棵树的度数是指该树中节点的最大度数。度数为零的节点称为树叶或终端节点，度数不为零的节点称为分支节点，除根节点外的分支节点称为内部节点。一个节点的子树之根节点称为该节点的子节点，该节点称为它们的父节点，同一节点的各个子节点之间称为兄弟节点。一棵树的根节点没有父节点，叶节点没有子节点。一个节点系列k1,k2，......,ki,ki+1,......,kj,并满足ki是ki+1的父节点，就称为一条从k1到kj的路径，路径的长度为j-1,即路径中的边数。路径中前面的节点是后面节点的祖先，后面节点是前面节点的子孙。节点的层数等于父节点的层数加一，根节点的层数定义为树中节点层数的最大值称为该树的高度或深度。m ( m≥0 ）棵互不相交的树的集合称为森林。树去掉根节点就成为森林，森林加上一个新的根节点就成为树。二叉树定义二叉树( Binary Tree )是n ( n≥0 )个节点的有限集合，它或者是空集( n=0 ），或者是由一个根节点以及两棵互不相交的、分别称为左子树和右子树的二叉树组成。二叉树与普通有序树不同，二叉树严格区分左孩子和右孩子，即使只有一个子节点也要区分左右。二叉树的特征工叉树第i ( i≥1)层上的节点最多为2i-1个。深度为k ( k≥1)的二叉树最多有2k一1个节点。在任意一棵二叉树中，树叶的数目比度数为2的节点的数目多一。满二叉树︰深度为k ( k≥1)时有2k一1个节点的二叉树。完全二叉树∶只有最下面两层有度数小于2的节点，且最下面一层的叶节点集中在最左边的若干位置上。二叉树的遍历沿某条搜索路径周游二叉树，对树中的每一个节点访问一次且仅访问一次。先序遍历︰先访问树根，再访问左子树，最后访问右子树; --> 根左右中序遍历︰先访问左子树，再访问树根，最后访问右子树; --> 左根右后序遍历︰先访问左子树，再访问右子树，最后访问树根﹔ --> 左右根 层次遍历:从根节点开始，逐层从左向右进行遍历。递归思想什么是递归?所谓递归函数是指一个函数的函数体中直接调用或间接调用了该函数自身的函数。这里的直接调用是指一个函数的函数体中含有调用自身的语句，间接调用是指一个函数在函数体里有调用了其它函数，而其它函数又反过来调用了该函数的情况。 递归函数调用的执行过程分为两个阶段递推阶段︰从原问题出发，按递归公式递推从未知到已知,最终达到递归终止条件。回归阶段:按递归终止条件求出结果，逆向逐步代入递归公式，回归到原问题求解。 优点与缺点优点︰递归可以把问题简单化，让思路更为清晰,代码更简洁缺点︰递归因系统环境影响大，当递归深度太大时，可能会得到不可预知的结果算法基础基础概念特征定义算法( Algorithm )是一个有穷规则（或语句、指令)的有序集合。它确定了解决某一问题的一个运算序列。对于问题的初始输入，通过算法有限步的运行，产生一个或多个输出。数据的逻辑结构与存储结构密切相关算法设计:取决于选定的逻辑结构算法实现:依赖于采用的存储结构算法的特性有穷性—～算法执行的步骤（或规则)是有限的;确定性——每个计算步骤无二义性;可行性——每个计算步骤能够在有限的时间内完成;输入，输出——存在数据的输入和出输出评价算法好坏的方法正确性︰运行正确是一个算法的前提。可读性︰容易理解、容易编程和调试、容易维护。健壮性:考虑情况全面，不容以出现运行错误。时间效率高︰算法消耗的时间少。储存量低︰占用较少的存储空间。排序和查找排序排序(Sort)是将无序的记录序列(或称文件）调整成有序的序列。常见排序方法∶冒泡排序冒泡排序是一种简单的排序算法。它重复地走访过要排序的数列，一次比较两个元素，如果他们的顺序错误就把他们交换过来。走访数列的工作是重复地进行直到没有再需要交换，也就是说该数列已经排序完成。选择排序工作原理为，首先在未排序序列中找到最小元素，存放到排序序列的起始位置，然后，再从剩余未排序元素中继续寻找最小元素，然后放到排序序列末尾。以此类推，直到所有元素均排序完毕
《Python利用psutil库进行监控进程和资源》psutil是Python系统和进程工具库,它提供了一种跨平台的方式来获取系统信息、管理系统进程、监控系统性能、操作系统资源等,下面就跟随小编一起来... python作为一门强大的编程语言，拥有丰富的标准库和第三方库，使得开发者可以轻松完成各种任务。在系统监控和管理方面，psutil库无疑是一个强大的工具，它可以轻松地获取系统信息、管理进程、监控系统性能等。本文将深入探讨psutil库，介绍它的基本概念、用法和示例代码，帮助大家更好地理解和利用这个强大的库。什么是psutil库psutil是Python系统和进程工具库（Python System and Process Utilities）。它提供了一种跨平台的方式来获取系统信息、管理系统进程、监控系统性能、操作系统资源等。无论是在Windows、linux还是MACOS，psutil都能提供一致的API，使得开发者可以轻松地编写跨平台的代码。安装psutil库要开始使用psutil库，首先需要安装它。可以使用pip来进行安装：pip install psutil安装完成后，就可以导入并开始使用psutil库了。import psutil获取系统信息获取CPU信息import psutil # 获取CPU核心数 cpu_count = psutil.cpu_count(logical=False) print(f"Physical CPU Count: {cpu_count}") # 获取逻辑CPU核心数（包括超线程） logical_cpu_count = psutil.cpu_count(logical=True) print(f"Logical CPU Count: {logical_cpu_count}") # 获取CPU使用率 cpu_usage = psutil.cpu_percent(interval=1, percpu=True) for i, usage in enumerate(cpu_usage, 1): print(f"CPU {i} Usage: {usage}%")获取内存信息import psutil # 获取内存总量 total_memory = psutil.virtual_memory().total print(f"Total Memory: {total_memory / (1024 ** 3):.2f} GB") # 获取可用内存 available_memory = psutil.virtual_memory().available print(f"Available Memory: {available_memory / (1024 ** 3):.2f} GB") # 获取内存使用率 memory_usage = psutil.virtual_memory().percent print(f"Memory Usage: 1.15MB%")获取磁盘信息import psutil # 获取所有磁盘分区信息 disk_partitions = psutil.disk_partitions() for partition in disk_partitions: print(f"Device: {partition.device}") print(f"Mountpoint: {partition.mountpoint}") print(f"Fstype: {partition.fstype}") print("") # 获取指定分区的使用情况 disk_usage = psutil.disk_usage("/") print(f"Total: {disk_usage.total / (1024 ** 3):.2f} GB") print(f"Used: {disk_usage.used / (1024 ** 3):.2f} GB") print(f"Free: {disk_usage.free / (1024 ** 3):.2f} GB") print(f"Usage: {disk_usage.percent}%")获取网络信息import psutil # 获取网络接口列表 network_interfaces = psutil.net_if_addrs() for interface, addresses in network_interfaces.items(): print(f"Interface: {interface}") for address in addresses: print(f" Family: {address.family}") print(f" Address: {address.address}") print(f" Netmask: {address.netmask}") print(f" Broadcast: {address.broadcast}") print("")管理进程查询进程信息import psutil # 获取所有进程列表 all_processes = psutil.process_iter(attrs=['pid', 'name']) for process in all_processes: print(f"Process ID: {process.info['pid']}, Name: {process.info['name']}") # 根据进程名称查询进程 process_name = "python3" process_list = [p.info for p in all_processes if p.info['name'] == process_name] for process in process_list: print(f"Process ID: {process['pid']}, Name: {process['name']}")创建新进程import psutil import subprocess # 创建新进程 subprocess.Popen(["notepad.exe"]) # 检查进程是否存在 process_name = "notepad.exe" is_running = any(process.info['name'] == process_name for process in psutil.process_iter(attrs=['name'])) print(f"{process_name} is running: {is_running}")终止进程import psutil # 根据进程名称终止进程 process_name = "notepad.exe" for process in psutil.process_iter(attrs=['name']): if process.info['name'] == process_name: process.terminate() print(f"Terminated {process_name}")监控系统性能监控CPU利用率import psutil import time # 监控CPU利用率 for _ in range(5): cpu_percent = psutil.cpu_percent(interval=1) print(f"CPU Usage: {cpu_percent}%") time.sleep(1)监控内存使用量import psutil import time # 监控内存使用量 for _ in range(5): memory_usage = psutil.virtual_memory().percent print(f"Memory Usage: 1.15MB%") time.sleep(1)监控磁盘IOimport psutil import time # 监控磁盘IO for _ in range(5): disk_io = psutil.disk_io_counters() print(f"Read Count: {disk_io.read_count}") print(f"Write Count: {disk_io.write_count}") time.sleep(1)监控网络流量import psutil import time # 监控网络流量 for _ in range(5): network_io = psutil.net_io_counters() print(f"Bytes Sent: {network_io.bytes_sent} bytes") print(f"Bytes Received: {network_io.bytes_recv} bytes") time.sleep(1)操作系统资源获取用户信息import psutil # 获取当前用户信息 user_info = psutil.users() for user in user_info: print(f"User: {user.name}") print(f"Terminal: {user.terminal}") print(f"Host: {user.host}") print("")获取虚拟内存信息import psutil # 获取虚拟内存信息 virtual_memory = psutil.virtual_memory() print(f"Total: {virtual_memory.total / (1024 ** 3):.2f} GB") print(f"Used: {virtual_memory.used / (1024 ** 3):.2f} GB") print(f"Free: {virtual_memory.free / (1024 ** 3):.2f} GB") print(f"Usage: {virtual_memory.percent}%")安全性和权限考虑 在使用psutil库时，需要注意安全性和权限问题。某些功能可能需要管理员权限或超级用户权限才能访问，例如终止进程、查询其他用户的进程等。确保你的程序以合适的权限运行，以避免潜在的问题。总结psutil库是Python中用于系统监控和管理的强大工具，它提供了丰富的功能来获取系统信息、管理进程、监控系统性能等。本文介绍了psutil库的基本用法和示例代码，帮助大家更好地理解和利用这个库。无论是用于系统管理、性能监控、自动化任务还是系统信息收集，psutil都是一个不可或缺的Python神器。
首先，pip 是个啥？简单来说，pip 就是 Python 的包管理器，就像手机里的应用商店一样，可以方便地帮你安装、升级或卸载 Python 的各种第三方库。命令 1：安装 pip如果你是Python新手，很可能系统已经自带pip。验证是否已安装，只需在命令行输入：python -m pip --version若未安装，可参考官方指引进行安装。命令 2：安装 Python 库想装个requests库试试水？一句命令搞定！pip install requests这就是pip最基础也是最常见的用法，它会自动处理库及其依赖关系。命令 3：升级库世界日新月异，库也要保持最新状态。要升级已安装的所有库：pip list --outdated # 查看待升级库 pip upgrade <package_name> # 升级指定库命令 4：卸载库不再需要某个库时，我们可以这样卸载：pip uninstall <package_name>命令 5：搜索库在决定安装某个库之前，你可能想先了解下有哪些可用的库能满足你的需求。这时可以使用 search 命令：pip search <keyword>比如查找与机器学习相关的库，你可以输入 pip search machine learning 。命令 6：查看已安装库详细信息想知道已安装的库具体版本和其他详情？ show 命令可以帮助你：pip show <package_name>命令 7：只下载库而不安装有时你可能需要离线环境安装包或者备份当前环境的依赖，那么可以使用 download 命令只下载不安装：pip download <package_name>命令 8：创建requirements文件在项目开发中，为了方便团队成员统一环境，我们可以创建一个包含所有依赖的 requirements 文件：pip freeze > requirements.txt这会列出当前环境中所有已安装库及其版本，并保存到 requirements.txt 文件中。而要根据这个文件安装所有依赖，只需：pip install -r requirements.txt命令 9：指定库版本安装在某些情况下，你可能需要安装特定版本的库，比如安装 requests 库的2.25.1版本：pip install requests==2.25.1命令 10：检查是否存在安全漏洞pip 配合 Safety 工具可以检查已安装库的安全性：pip install safety safety check --full-report这会扫描所有已安装的库，并报告是否有已知的安全漏洞。命令 11：创建虚拟环境并激活为了避免不同项目间依赖冲突，我们通常会在每个项目下创建独立的虚拟环境，然后使用 pip 进行管理：python -m venv my_project_env # 创建虚拟环境 source my_project_env/bin/activate # Linux/Mac激活环境 my_project_env\Scripts\activate.bat # Windows激活环境在虚拟环境中，你可以放心使用 pip 安装和管理项目的专属依赖。命令 12：清理未使用的库或缓存随着项目迭代，有些不再使用的库可能会被遗忘，这时可以使用 autoremove 命令来清除：pip-autoremove <package_name>同时，你也可以清理 pip 下载缓存以释放磁盘空间：pip cache purge命令 13：查看 pip 自身的版本信息及更新 pip要查看 pip 本身的版本，只需运行：pip --version为了确保 pip 始终保持最新，可使用如下命令升级 pip 自身：python -m pip install --upgrade pip当然，pip 的功能确实丰富多样，让我们继续探索：命令 14：指定源安装库在某些网络环境下，可能需要从国内镜像或者其他自定义源下载和安装库。例如使用阿里云的Python 镜像源：pip install -i https://mirrors.aliyun.com/pypi/simple/ <package_name>或永久更改 pip 默认源（推荐在配置文件中修改）：pip config set global.index-url https://mirrors.aliyun.com/pypi/simple/命令 15：分发本地构建的库如果你自己开发了一个 Python 包，并希望在本地测试安装，可以先打包成 whl 或 tar.gz 格式，然后通过 pip 进行安装：# 假设你已经将项目打包为my_package-0.1.0.whl pip install ./my_package-0.1.0.whl命令 16：查看库安装路径如果想知道某个库具体安装在系统哪个位置，可以使用 show --files 选项：pip show --files <package_name>命令 17：在安装时跳过测试有些库在安装过程中会执行单元测试，若想快速安装可选择跳过这些测试：pip install --no-deps --ignore-installed --no-cache-dir --disable-pip-version-check --no-compile <package_name>命令 18：列出所有全局安装的库获取全局环境下的所有已安装Python库列表：pip list命令 19：在安装时指定额外选项某些库可能在安装时需要额外参数，比如 numpy 、 scipy 等科学计算库，你可以直接在 pip 命令中传递这些选项：pip install numpy --install-option="--openblas"这里的 --openblas 是一个示例，实际选项请根据库文档进行设置。命令 20：查看依赖树要了解一个包及其所有依赖关系，可以使用 deptree 第三方工具：pip installdeptree deptree -l <package_name>这将展示指定包及其所有依赖项之间的层级关系。命令 21：修复损坏的库如果某个库在安装或升级过程中出现问题导致无法正常使用，可以尝试修复它：pip install --force-reinstall <package_name>命令 22：获取库的源码如果你对某个库的实现细节感兴趣，可以通过 pip 下载其源代码：pip download <package_name> --no-binary :all:然后在下载目录中找到对应的 tar.gz 或 whl 文件解压查看。命令 23：执行pip的自定义脚本有些开发者可能会编写自定义的 pip 脚本来自动化一些任务，你可以通过 run 命令执行：pip run my_script.py命令 24：在安装时指定 Python 版本如果你有多个Python版本并希望为特定版本安装库：python3.7 -m pip install <package_name>这里 python3.7 替换成你想要使用的Python解释器路径。命令 25：检查依赖冲突在项目中可能存在不兼容的依赖版本，可以使用 pipdeptree 第三方工具来检测：pip install pipdeptree pipdeptree --packages <package_name>这将展示指定包及其依赖项之间是否存在版本冲突。命令 26：创建独立可执行文件借助 pyinstaller 等工具，你可以通过pip将Python程序打包成一个独立可执行文件：pip install pyinstaller pyinstaller your_script.py命令 27：管理用户级别的包默认情况下，pip 会安装全局系统级别的包。若想为当前用户安装不影响系统的包，可以加上 --user 选项：pip install --user <package_name>命令 28：只升级指定的包而不升级所有包当只需要更新某个特定的库时，可以使用 --upgrade 选项：pip install --upgrade <package_name>命令 29：批量卸载多个包如果需要一次性卸载多个Python包，可以通过列表形式提供包名：pip uninstall <package1> <package2> ...命令 30：冻结当前环境下的所有包及其版本为了能够复制或记录当前环境中所有已安装包的状态，可以生成一个 requirements 文件：pip freeze > requirements.txt这将把所有已安装包及其版本号写入 requirements.txt 文件中，便于在其他环境下复现相同的软件环境。命令 31：根据requirements文件创建虚拟环境并安装包结合虚拟环境（如venv或conda），可以从requirements文件重新构建环境：python -m venv my_venv source my_venv/bin/activate # Windows: my_venv\Scripts\activate.bat pip install -r requirements.txt
1、批量文件重命名神器在工作中，我们常常需要对大量文件进行批量重命名，Python帮你轻松搞定！ import os def batch_rename(path, prefix='', suffix=''): for i, filename in enumerate(os.listdir(path)): new_name = f"{prefix}{i:03d}{suffix}{os.path.splitext(filename)[1]}" old_file = os.path.join(path, filename) new_file = os.path.join(path, new_name) os.rename(old_file, new_file) # 使用示例： batch_rename('/path/to/your/directory', 'file_', '.txt')2、自动发送邮件通知告别手动发送，用Python编写定时发送邮件的自动化脚本。import smtplib from email.mime.text import MIMEText def send_email(to_addr, subject, content): smtp_server = 'smtp.example.com' username = 'your-email@example.com' password = 'your-password' msg = MIMEText(content) msg['Subject'] = subject msg['From'] = username msg['To'] = to_addr server = smtplib.SMTP(smtp_server, 587) server.starttls() server.login(username, password) server.sendmail(username, to_addr, msg.as_string()) server.quit() # 使用示例： send_email('receiver@example.com', '每日报告提醒', '今日报告已生成，请查收。')3、定时任务自动化执行使用Python调度库，实现定时执行任务的自动化脚本。import schedule import time def job_to_schedule(): print("当前时间：", time.ctime(), "任务正在执行...") # 定义每天9点执行任务 schedule.every().day.at("09:00").do(job_to_schedule) while True: schedule.run_pending() time.sleep(1) # 使用示例： # 运行此脚本后，每天上午9点会自动打印当前时间及提示信息4、数据库操作自动化简化数据库管理，Python帮你自动化执行CRUD操作。import sqlite3 def create_connection(db_file): conn = None try: conn = sqlite3.connect(db_file) print(f"成功连接到SQLite数据库：{db_file}") except Error as e: print(e) return conn def insert_data(conn, table_name, data_dict): keys = ', '.join(data_dict.keys()) values = ', '.join(f"'{v}'" for v in data_dict.values()) sql = f"INSERT INTO {table_name} ({keys}) VALUES ({values});" try: cursor = conn.cursor() cursor.execute(sql) conn.commit() print("数据插入成功！") except sqlite3.Error as e: print(e) # 使用示例： conn = create_connection('my_database.db') data = {'name': 'John Doe', 'age': 30} insert_data(conn, 'users', data) # 在适当时候关闭数据库连接 conn.close()5、网页内容自动化抓取利用BeautifulSoup和requests库，编写Python爬虫获取所需网页信息。import requests from bs4 import BeautifulSoup def fetch_web_content(url): response = requests.get(url) if response.status_code == 200: soup = BeautifulSoup(response.text, 'html.parser') # 示例提取页面标题 title = soup.find('title').text return title else: return "无法获取网页内容" # 使用示例： url = 'https://example.com' web_title = fetch_web_content(url) print("网页标题：", web_title)6、数据清洗自动化使用Pandas库，实现复杂数据处理和清洗的自动化。import pandas as pd def clean_data(file_path): df = pd.read_csv(file_path) # 示例：处理缺失值 df.fillna('N/A', inplace=True) # 示例：去除重复行 df.drop_duplicates(inplace=True) # 示例：转换列类型 df['date_column'] = pd.to_datetime(df['date_column']) return df # 使用示例： cleaned_df = clean_data('data.csv') print("数据清洗完成，已准备就绪！")7、图片批量压缩用Python快速压缩大量图片以节省存储空间。from PIL import Image import os def compress_images(dir_path, quality=90): for filename in os.listdir(dir_path): if filename.endswith(".jpg") or filename.endswith(".png"): img = Image.open(os.path.join(dir_path, filename)) img.save(os.path.join(dir_path, f'compressed_{filename}'), optimize=True, quality=quality) # 使用示例： compress_images('/path/to/images', quality=80)8、文件内容查找替换Python脚本帮助你一键在多个文件中搜索并替换指定内容。import fileinput def search_replace_in_files(dir_path, search_text, replace_text): for line in fileinput.input([f"{dir_path}/*"], inplace=True): print(line.replace(search_text, replace_text), end='') # 使用示例： search_replace_in_files('/path/to/files', 'old_text', 'new_text')9、日志文件分析自动化通过Python解析日志文件，提取关键信息进行统计分析。def analyze_log(log_file): with open(log_file, 'r') as f: lines = f.readlines() error_count = 0 for line in lines: if "ERROR" in line: error_count += 1 print(f"日志文件中包含 {error_count} 条错误记录。") # 使用示例： analyze_log('application.log')10、数据可视化自动化利用Matplotlib库，实现数据的自动图表生成。import matplotlib.pyplot as plt import pandas as pd def visualize_data(data_file): df = pd.read_csv(data_file) # 示例：绘制柱状图 df.plot(kind='bar', x='category', y='value') plt.title('数据分布') plt.xlabel('类别') plt.ylabel('值') plt.show() # 使用示例： visualize_data('data.csv')11、邮件附件批量下载通过Python解析邮件，自动化下载所有附件。import imaplib import email from email.header import decode_header import os def download_attachments(email_addr, password, imap_server, folder='INBOX'): mail = imaplib.IMAP4_SSL(imap_server) mail.login(email_addr, password) mail.select(folder) result, data = mail.uid('search', None, "ALL") uids = data[0].split() for uid in uids: _, msg_data = mail.uid('fetch', uid, '(RFC822)') raw_email = msg_data[0][1].decode("utf-8") email_message = email.message_from_string(raw_email) for part in email_message.walk(): if part.get_content_maintype() == 'multipart': continue if part.get('Content-Disposition') is None: continue filename = part.get_filename() if bool(filename): file_data = part.get_payload(decode=True) with open(os.path.join('/path/to/download', filename), 'wb') as f: f.write(file_data) mail.close() mail.logout() # 使用示例： download_attachments('your-email@example.com', 'your-password', 'imap.example.com')12、定时发送报告自动化根据数据库或文件内容，自动生成并定时发送日报/周报。import pandas as pd import smtplib from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart def generate_report(source, to_addr, subject): # 假设这里是从数据库或文件中获取数据并生成报告内容 report_content = pd.DataFrame({"Data": [1, 2, 3], "Info": ["A", "B", "C"]}).to_html() msg = MIMEMultipart() msg['From'] = 'your-email@example.com' msg['To'] = to_addr msg['Subject'] = subject msg.attach(MIMEText(report_content, 'html')) server = smtplib.SMTP('smtp.example.com', 587) server.starttls() server.login('your-email@example.com', 'your-password') text = msg.as_string() server.sendmail('your-email@example.com', to_addr, text) server.quit() # 使用示例： generate_report('data.csv', 'receiver@example.com', '每日数据报告') # 结合前面的定时任务脚本，可实现定时发送功能13、自动化性能测试使用Python的 locust 库进行API接口的压力测试。from locust import HttpUser, task, between class WebsiteUser(HttpUser): wait_time = between(5, 15) # 定义用户操作之间的等待时间 @task def load_test_api(self): response = self.client.get("/api/data") assert response.status_code == 200 # 验证返回状态码为200 @task(3) # 指定该任务在总任务中的执行频率是其他任务的3倍 def post_data(self): data = {"key": "value"} response = self.client.post("/api/submit", json=data) assert response.status_code == 201 # 验证数据成功提交后的响应状态码 # 运行Locust命令启动性能测试： # locust -f your_test_script.py --host=http://your-api-url.com14、自动化部署与回滚脚本使用Fabric库编写SSH远程部署工具，这里以部署Django项目为例：from fabric import Connection def deploy(host_string, user, password, project_path, remote_dir): c = Connection(host=host_string, user=user, connect_kwargs={"password": password}) with c.cd(remote_dir): c.run('git pull origin master') # 更新代码 c.run('pip install -r requirements.txt') # 安装依赖 c.run('python manage.py migrate') # 执行数据库迁移 c.run('python manage.py collectstatic --noinput') # 静态文件收集 c.run('supervisorctl restart your_project_name') # 重启服务 # 使用示例： deploy( host_string='your-server-ip', user='deploy_user', password='deploy_password', project_path='/path/to/local/project', remote_dir='/path/to/remote/project' ) # 对于回滚操作，可以基于版本控制系统实现或创建备份，在出现问题时恢复上一版本的部署。
这篇文章主要介绍了Python脚本利用adb进行手机控制的方法，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧一. adb 相关命令：关闭adb服务：adb kill-server启动adb服务 adb start-server查询当前运行的所有设备 adb devices可能在adb中存在多个虚拟设备运行 可以指定虚拟设备运行 -s 虚拟设备名称重启设备 adb reboot --指定虚拟设备 adb -s 设备名称 reboot查看日志 adb logcat 清除日志 adb logcat -c进入linux shell下 adb shell 其中常用的linux命令 cd cat 等等 输入su可以获取超级管理员名称了 要确定是否有哪些命令 进入 system/bin目录 就知道了传入文件到设备中 adb push 本地文件 远程目录从设备中拷贝文件到本地 adb -s emulator-5556 pull /data/config.ini d:/显示当前运行的全部模拟器： adb devices安装应用程序： adb install -r 123.apk获取模拟器中的文件： adb pull 向模拟器中写文件： adb push 进入模拟器的shell模式： adb shell卸载apk包： adb shell cd data/app rm 123.apk exit adb uninstall 123.apk adb install -r 123.apk查看adb命令帮助信息： adb help删除系统应用： adb remount （重新挂载系统分区，使系统分区重新可写）。 adb shell cd system/app rm 123.apk获取管理员权限： adb root19、复制文件：复制一个文件或目录到设备：adb push 如：adb push update.zip /sdcard/从设备上复制一个文件或目录：adb pull 如：adb pull /sdcard/update.zip.20、取得当前运行设备的实例的列表及每个实例的状态：adb devices21：这条命令模拟android手机在屏幕坐标（X,Y）处进行了点击操作。adb shell input tap22：这条命令模拟Android手机从屏幕坐标（X1,Y1）滑动到坐标（X2,Y2）的操作。adb shell input swipe 23:Luiautomator dump dump: creates an XML dump of current UI hierarchy 这个命令是用来成成当前界面的UI层次，并用XML格式进行展示 。这样就可以获取各个组件的位置了注：如果PC要想同时控制多台Android手机，必须在adb 后面添加-s例如：adb -s 13b6e4c4 shell input tap 400 400表示对13b6e4c4这台Android手机进行在屏幕上（400,400）坐标位置进行模拟的点击事件。24.能看到设备信息就代表设备已经连接成功了，接下来的命令就是adb install 路径+包名.apk例如我的安装包放在桌面，那么命令就是adb install C:\Users\hyh\Desktop\XXX.apk* adb shell uiautomator dump /mnt/sdcard/window_dump.xml 获得手机当前界面的UI信息，生成window_dump.xml* adb shell input text “123” 输入text举例：1、打开cmd，进入到当前文件夹中，输入命令 adb devices 查看当前与电脑连接的设备（前提是，手机打开usb调试模式），可以查看已连接成功的手机。2、 若手机成功连接，输入命令 adb shell input tap 100 100 , 表示点击屏幕上坐标为（100，100）的点，如果不知道需要点击的点的具体位置的话可以在手机开发者模式中设置。二. adb 模拟按键：比如使用 adb shell input keyevent 命令，不同的 keycode 能实现不同的功能，完整的 keycode 列表详见 KeyEvent，摘引部分我觉得有意思的如下:Keycode含义3HOME 键4返回键5打开拨号应用6挂断电话24增加音量25降低音量26电源键27拍照（需要在相机应用里）64打开浏览器82菜单键85播放/暂停86停止播放87播放下一首88播放上一首122移动光标到行首或列表顶部123移动光标到行末或列表底部126恢复播放127暂停播放164静音176打开系统设置187切换应用207打开联系人208打开日历209打开音乐210打开计算器220降低屏幕亮度221提高屏幕亮度223系统休眠224点亮屏幕231打开语音助手276如果没有 wakelock 则让系统休眠input 命令的一些用法举例电源键命令：adb shell input keyevent 26执行效果相当于按电源键。菜单键命令：adb shell input keyevent 82HOME 键命令：adb shell input keyevent 3返回键命令adb shell input keyevent 4音量控制增加音量：adb shell input keyevent 24降低音量：adb shell input keyevent 25静音：adb shell input keyevent 164媒体控制播放/暂停：adb shell input keyevent 85停止播放：adb shell input keyevent 86播放下一首：adb shell input keyevent 87播放上一首：adb shell input keyevent 88恢复播放：adb shell input keyevent 126暂停播放：adb shell input keyevent 127点亮/熄灭屏幕可以通过上文讲述过的模拟电源键来切换点亮和熄灭屏幕，但如果明确地想要点亮或者熄灭屏幕，那可以使用如下方法。点亮屏幕：adb shell input keyevent 224熄灭屏幕：adb shell input keyevent 223三、使用python脚本自动运行cmd 命令1.在adb文件夹下建立一个python文件import os os.system('adb shell input tap 100 100');运行脚本，发现与在命令行输入相同语句有同样的效果。2、也可以使用subprocess.Popen，最简单使用方式如下，设置shell=True，就不会弹出cmd框process = subprocess.Popen('adb shell input tap 14 1402',shell=True)程序实例：#coding:utf-8 #程序功能：可以实现抖音个人页面视频的自动点击，从而自动增加访问量 #思路：抖音主页中两个视频切换点击，可以实现访问量的增加 #使用ADB程序，视频的屏幕坐标可以使用adb shell uiautomator dump命令，获取该页面xml源码后查得 #下述是小米MIX2抖音主页第一个视频和第二个视频的坐标位置 #缺点：运行时不能移动屏幕，后续可以采用获取模块ID号的方式去点击相应的位置 import time import subprocess i = 0 #每次操作的间隔时间取决于手机配置，配置越高时间越短 sleep_time = 0.5 while 1: #用popen设置shell=True不会弹出cmd框 process = subprocess.Popen('adb shell input tap 14 1402',shell=True) time.sleep(sleep_time) process = subprocess.Popen('adb shell input keyevent KEYCODE_BACK', shell=True) time.sleep(sleep_time) process = subprocess.Popen('adb shell input tap 375 1402', shell=True) time.sleep(sleep_time) process = subprocess.Popen('adb shell input keyevent KEYCODE_BACK', shell=True) time.sleep(sleep_time) #os.system('adb shell input tap 14 1402') #os.system('adb shell input keyevent KEYCODE_BACK') #os.system('adb shell input tap 375 1402') i+=1 print str(i) + 'clicks have been completed'实现原理Hierarchy Viewer：获得当前手机实时的UI信息，方便用于手机的自动化测试；python中的subprocess.Popen() 或 Python os模块：调用系统命令；uiautomator工具：获取界面控件信息；adb命令：对手机进行操作；原文出处： Python脚本利用adb进行手机控制的方法
这篇文章主要为大家详细介绍了python3.6根据m3u8下载mp4视频，具有一定的参考价值，感兴趣的小伙伴们可以参考一下需要下载某网站的视频，chrome浏览器按F12打开开发者模式，发现视频链接是以"blob:http"开头的链接，打开这个链接后找不到网页，网上查了下，找到了下载方法，在这里做个记录，如果有错误，欢迎指出。程序在Windows 10下运行，不过Linux应该也没问题。使用到的有re模块，requests模块和Crypto模块，其中requests模块和Crypto模块如果没安装可以使用pip命令安装。（Crypto模块安装感觉比较坑，我是从anaconda里拷贝了一份）下面开始正题：注：以下使用的m3u8文件所在的网站是自己搭建用来测试的，链接可能会失效。首先在chrome的network里找到一个m3u8文件的请求，可以通过它下载视频。通过Preview可以看到m3u8文件的内容。这里m3u8文件比较重要的内容有两个。一个是URI后面的链接，这个是加密和解密的秘钥，如果m3u8文件里有这个URI，那么这个视频就是经过了加密的，加密的方法可以看URI前面，这里是AES-128加密算法。另一个是以ts结尾链接，这个是视频片段，如果是没进行加密的，可以直接打开这个链接下载视频片段，下载下来的是后缀为ts的文件，一般可以直接播放，不过时间很短。如果是加密过的，下载后播放会提示视频文件已损坏。知道这些后，就可以进行下载了，先获取m3u8文件的内容，然后解析出秘钥（key）和ts的链接，然后下载key对每一个ts进行解密，保存到一个mp4文件里。下载用的是requests模块，解析key和ts的链接是用的re模块，解密用的是Crypto模块。使用Crypto需要注意三个地方，一个是安装和导入，第二个是Crypto里AES.new的参数，第三个是decrypt方法的参数。下面先看代码。导入模块：import re import requests from Crypto.Cipher import AES初始设置：# 保存的mp4文件名 name = "dream_it_possible.mp4" # m3u8文件的url url = "http://www.jiuyi.info/static/video/secret/dream_it_possible.m3u8" # 请求头，不一定需要，看网站更改 headers = { "Referer": "http://www.jiuyi.info/video/dream_it_possible", "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/71.0.3578.80 Safari/537.36", }获取m3u8文件内容，并解析出key和ts文件的url。# 获取m3u8文件内容 r = requests.get(url) # 通过正值表达式获取key和ts的url k = re.compile(r"http://.*?\.key") # key的正则匹配 t = re.compile(r"http://.*?\.ts") # ts的正则匹配 key_url = k.findall(r.text)[0] # key的url ts_urls = t.findall(r.text) # ts的url列表下载并解密ts文件，保存为mp4文件。# 下载key key = requests.get(key_url).content # 解密并保存ts for ts_url in ts_urls: ts_name = ts_url.split("/")[-1] # ts文件名 # 解密，new有三个参数， # 第一个是秘钥（key）的二进制数据， # 第二个使用下面这个就好 # 第三个IV在m3u8文件里URI后面会给出，如果没有，可以尝试把秘钥（key）赋值给IV sprytor = AES.new(key, AES.MODE_CBC, IV=key) # 获取ts文件二进制数据 ts = requests.get(ts_url).content # 密文长度不为16的倍数，则添加二进制"0"直到长度为16的倍数 while len(ts) % 16 != 0: ts += b"0" # 写入mp4文件 with open(name, "ab") as file: # decrypt方法的参数需要为16的倍数，如果不是，需要在后面补二进制"0" file.write(sprytor.decrypt(ts)) print(name, "下载完成")到这里就下载完成了。完整代码：import re import requests from Crypto.Cipher import AES # 保存的mp4文件名 name = "dream_it_possible.mp4" # m3u8文件的url url = "http://www.jiuyi.info/static/video/secret/dream_it_possible.m3u8" # 请求头，不一定需要，看网站更改 headers = { "Referer": "http://www.jiuyi.info/video/dream_it_possible", "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/71.0.3578.80 Safari/537.36", } print("正在解析：" + url.split("/")[-1]) # 获取m3u8文件内容 r = requests.get(url) # 通过正值表达式获取key和ts的链接 k = re.compile(r"http://.*?\.key") # key的正则匹配 t = re.compile(r"http://.*?\.ts") # ts的正则匹配 key_url = k.findall(r.text)[0] # key的url ts_urls = t.findall(r.text) # ts的url列表 # 下载key的二进制数据 print("正在下载key") key = requests.get(key_url).content # 解密并保存ts for ts_url in ts_urls: ts_name = ts_url.split("/")[-1] # ts文件名 # 解密，new有三个参数， # 第一个是秘钥（key）的二进制数据， # 第二个使用下面这个就好 # 第三个IV在m3u8文件里URI后面会给出，如果没有，可以尝试把秘钥（key）赋值给IV sprytor = AES.new(key, AES.MODE_CBC, IV=key) # 获取ts文件二进制数据 print("正在下载：" + ts_name) ts = requests.get(ts_url).content # 密文长度不为16的倍数，则添加b"0"直到长度为16的倍数 while len(ts) % 16 != 0: ts += b"0" print("正在解密：" + ts_name) # 写入mp4文件 with open(name, "ab") as file: # # decrypt方法的参数需要为16的倍数，如果不是，需要在后面补二进制"0" file.write(sprytor.decrypt(ts)) print("保存成功：" + ts_name) print(name, "下载完成")
python爬虫爬取视频时，实现实时进度条显示
一、全部代码展示from contextlib import closing from requests import get url = 'https://v26-web.douyinvod.com/57cdd29ee3a718825bf7b1b14d63955b/615d475f/video/tos/cn/tos-cn-ve-15/72c47fb481464cfda3d415b9759aade7/?a=6383&br=2192&bt=2192&cd=0%7C0%7C0&ch=26&cr=0&cs=0&cv=1&dr=0&ds=4&er=&ft=jal9wj--bz7ThWG4S1ct&l=021633499366600fdbddc0200fff0030a92169a000000490f5507&lr=all&mime_type=video_mp4&net=0&pl=0&qs=0&rc=ank7OzU6ZnRkNjMzNGkzM0ApNmY4aGU8MzwzNzo3ZjNpZWdiYXBtcjQwLXNgLS1kLTBzczYtNS0tMmE1Xi82Yy9gLTE6Yw%3D%3D&vl=&vr=' with closing(get(url, stream=True)) as response: chunk_size = 1024 # 单次请求最大值 # response.headers['content-length']得到的数据类型是str而不是int content_size = int(response.headers['content-length']) # 文件总大小 data_count = 0 # 当前已传输的大小 with open('文件名.mp4', "wb") as file: for data in response.iter_content(chunk_size=chunk_size): file.write(data) done_block = int((data_count / content_size) * 50) # 已经下载的文件大小 data_count = data_count + len(data) # 实时进度条进度 now_jd = (data_count / content_size) * 100 # %% 表示% print("\r [%s%s] %d%% " % (done_block * '█', ' ' * (50 - 1 - done_block), now_jd), end=" ")注：上面的url已过期，需要各位自己去找网页上的视频url二、解释1.with closing我们在日常读取文件资源时，经常会用到with open() as f:的句子。但是使用with语句的时候是需要条件的，任何对象，只要正确实现了上下文管理，就可以使用with语句，实现上下文管理是通过__enter__和__exit__这两个方法实现的。with用法（没有实现上下文管理）class Door(): def open(self): print('Door is opened') def close(self): print('Door is closed') with Door() as d: d.open() d.close()结果报错了：with用法（实现上下文管理）用__enter__和__exit__实现了上下文管理class Door(): def open(self): print('Door is opened') def close(self): print('Door is closed') with Door() as d: d.open() d.close() 结果没报错： closing用法（完美解决上述问题）一个对象没有实现上下文，我们就不能把它用于with语句。这个时候，可以用 contextlib 中的closing()来把该对象变为上下文对象。class Door(): def __enter__(self): print('Begin') return self def __exit__(self, exc_type, exc_value, traceback): if exc_type: print('Error') else: print('End') def open(self): print('Door is opened') def close(self): print('Door is closed') with Door() as d: d.open() d.close()例如：用with语句使用requests中的get(url) 也就是本文中的案例，使用with closing（）下载视频(在网页中)2.文件流stream想象一下，如果把文件读取比作向池子里抽水，同步会阻塞程序，异步会等待结果，如果池子非常大呢？因此有了文件流，它就好比你一边抽一边取，不用等池子满了再用，所以对于一些大型文件（几个G的视频）一般会用到这个参数。 （对小型文件也可以使用）3.response.headers['content-length']这表示获取文件的总大小，但是它得到的结果的数据类型是str而不是int，因此需要进行数据类型转换。4.response.iter_content()该方法一般用于从网上下载文件和网页（需要用到requests.get(url)）其中chunk_size表示单次请求最大值。5.\r和%\r表示回车（回到行首）%是一种占位符而对于%%，第一个%起到了转义的作用，使结果输出为百分号%