易小灯塔发布的文章 - 易小灯塔的博客

2022-02-04

FastApi快速入门 03: 路由管理 4.路由管理4.1 FastApi 路由管理fastapi路由管理，和GIN的框架思想一致。类似于flask的蓝图入口函数----主路由---控制器---服务4.1.1 main入口函数需要include api_routerfrom router.api import api_router app = FastAPI() app.include_router(api_router, prefix="/api") if __name__ == "__main__": uvicorn.run(app, host="127.0.0.1", port=8081) 4.1.2 路由文件必须导入实际的控制文件，每个控制文件设置一个routerfrom fastapi import APIRouter from control import user api_router = APIRouter() api_router.include_router(user.router, prefix="/user") 4.1.3 实际的控制文件，设置好子路由from typing import List from fastapi import Depends, FastAPI, HTTPException, APIRouter from pydantic import BaseModel from services import user_services from models import user_models router = APIRouter() class UserSchemas(BaseModel): username: str fullname: str password: str @router.get("/{username}") def get_user(username: str): print(username) user = user_services.get_user_by_username(username) return user 4.2 跨域设置from fastapi.middleware.cors import CORSMiddleware app = FastAPI() origins = [ "http://localhost", "http://localhost:8090", ] app.add_middleware( CORSMiddleware, allow_origins=origins, allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) 5.依赖注入FastAPI 提供了简单易用，但功能强大的依赖注入系统，可以让开发人员轻松地把组件集成至FastAPI。什么是「依赖注入」？依赖注入是一种消除类之间依赖关系的设计模式。把有依赖关系的类放到容器中，解析出这些类的实例，就是依赖注入。目的是实现类的解耦。示例：from typing import Optional from fastapi import Depends, FastAPI app = FastAPI() async def common_parameters(q: Optional[str] = None, skip: int = 0, limit: int = 100): return {"q": q, "skip": skip, "limit": limit} @app.get("/items/") async def read_items(commons: dict = Depends(common_parameters)): return commons @app.get("/users/") async def read_users(commons: dict = Depends(common_parameters)): return commons 本例中的依赖项预期接收如下参数：类型为 str 的可选查询参数 q 类型为 int 的可选查询参数 skip，默认值是 0 类型为 int 的可选查询参数 limit，默认值是 100 然后，依赖项函数返回包含这些值的 dict。使用Class作为依赖：from typing import Optional from fastapi import Depends, FastAPI app = FastAPI() fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}] class CommonQueryParams: def __init__(self, q: Optional[str] = None, skip: int = 0, limit: int = 100): self.q = q self.skip = skip self.limit = limit @app.get("/items/") async def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)): response = {} if commons.q: response.update({"q": commons.q}) items = fake_items_db[commons.skip : commons.skip + commons.limit] response.update({"items": items}) return response 使用嵌套子依赖：from typing import Optional from fastapi import Cookie, Depends, FastAPI app = FastAPI() def query_extractor(q: Optional[str] = None): return q def query_or_cookie_extractor( q: str = Depends(query_extractor), last_query: Optional[str] = Cookie(None) ): if not q: return last_query return q @app.get("/items/") async def read_query(query_or_default: str = Depends(query_or_cookie_extractor)): return {"q_or_cookie": query_or_default} 在路径中使用依赖：from fastapi import Depends, FastAPI, Header, HTTPException app = FastAPI() async def verify_token(x_token: str = Header(...)): if x_token != "fake-super-secret-token": raise HTTPException(status_code=400, detail="X-Token header invalid") async def verify_key(x_key: str = Header(...)): if x_key != "fake-super-secret-key": raise HTTPException(status_code=400, detail="X-Key header invalid") return x_key @app.get("/items/", dependencies=[Depends(verify_token), Depends(verify_key)]) async def read_items(): return [{"item": "Foo"}, {"item": "Bar"}] 全局依赖项，可以为所有路径操作应用该依赖项：from fastapi import Depends, FastAPI, Header, HTTPException async def verify_token(x_token: str = Header(...)): if x_token != "fake-super-secret-token": raise HTTPException(status_code=400, detail="X-Token header invalid") async def verify_key(x_key: str = Header(...)): if x_key != "fake-super-secret-key": raise HTTPException(status_code=400, detail="X-Key header invalid") return x_key app = FastAPI(dependencies=[Depends(verify_token), Depends(verify_key)]) @app.get("/items/") async def read_items(): return [{"item": "Portal Gun"}, {"item": "Plumbus"}] @app.get("/users/") async def read_users(): return [{"username": "Rick"}, {"username": "Morty"}]

2022年02月04日
3,063 阅读
0 评论
0 点赞

2022-02-03

FastApi快速入门 02 : 使用SQLAlchemy orm框架 3.使用数据库示例3.1 使用SQLAlchemy orm框架官方代码里面使用的是sqlalchemy，异步也是使用的这个3.1.1 准备工作安装 SQLAlchemypip install sqlalchemy 创建数据库 fastapi_dbcreate database fastapi_db; 3.1.2 配置数据库连接创建文件database.pyfrom sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from sqlalchemy.ext.declarative import declarative_base # 数据库连接配置 SQLALCHEMY_DATABASE_URI = ( "mysql+pymysql://root:123456@localhost/fastapi_db?charset=utf8mb4" # 用户:密码@服务器/数据库?参数 ) # 创建数据库引擎 engine = create_engine(SQLALCHEMY_DATABASE_URI) # 创建数据库会话 SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) # 声明基类 Base = declarative_base() 3.1.3 创建数据模型创建文件models.pyfrom sqlalchemy import Boolean, Column, ForeignKey, Integer, String from sqlalchemy.orm import relationship from .database import Base # 定义 User 类 class User(Base): __tablename__ = 'users' # 定义表名 id = Column(Integer, primary_key=True, index=True) email = Column(String(255), unique=True, index=True) hashed_password = Column(String(255)) is_active = Column(Boolean, default=True) items = relationship("Item", back_populates="owner") # 关联 Item 表 # 定义 Item 类 class Item(Base): __tablename__ = "items" id = Column(Integer, primary_key=True, index=True) title = Column(String(255), index=True) description = Column(String(255), index=True) owner_id = Column(Integer, ForeignKey('users.id')) owner = relationship("User", back_populates="items") # 关联 User 表 3.1.4 创建 Pydantic 模型创建文件schemas.pyfrom typing import List from pydantic import BaseModel class ItemBase(BaseModel): title: str description: str = None class ItemCreate(ItemBase): pass class Item(ItemBase): id: int owner_id: int class Config: orm_mode = True class UserBase(BaseModel): email: str class UserCreate(UserBase): password: str class User(UserBase): id: int is_active: bool items: List[Item] = [] class Config: orm_mode = True 复制3.1.5 crud 工具创建文件crud.pyfrom sqlalchemy.orm import Session from . import models, schemas def get_user(db: Session, user_id: int): """ 根据id获取用户信息 :param db: 数据库会话 :param user_id: 用户id :return: 用户信息 """ return db.query(models.User).filter(models.User.id == user_id).first() def get_user_by_email(db: Session, email: str): """ 根据email获取用户信息 :param db: 数据库会话 :param email: 用户email :return: 用户信息 """ return db.query(models.User).filter(models.User.email == email).first() def get_users(db: Session, skip: int = 0, limit: int = 100): """ 获取特定数量的用户 :param db: 数据库会话 :param skip: 开始位置 :param limit: 限制数量 :return: 用户信息列表 """ return db.query(models.User).offset(skip).limit(limit).all() def create_user(db: Session, user: schemas.UserCreate): """ 创建用户 :param db: 数据库会话 :param user: 用户模型 :return: 根据email和password登录的用户信息 """ fake_hashed_password = user.password + "notreallyhashed" db_user = models.User(email=user.email, hashed_password=fake_hashed_password) db.add(db_user) # 添加到会话 db.commit() # 提交到数据库 db.refresh(db_user) # 刷新数据库 return db_user def get_items(db: Session, skip: int = 0, limit: int = 100): """ 获取指定数量的item :param db: 数据库会话 :param skip: 开始位置 :param limit: 限制数量 :return: item列表 """ return db.query(models.Item).offset(skip).limit(limit).all() def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int): """ 创建用户item :param db: 数据库会话 :param item: Item对象 :param user_id: 用户id :return: Item模型对象 """ db_item = models.Item(**item.dict(), owner_id=user_id) db.add(db_item) db.commit() db.refresh(db_item) return db_item 3.1.6 入口main函数修改from typing import List from fastapi import Depends, FastAPI, HTTPException from sqlalchemy.orm import Session from . import crud, models, schemas from .database import SessionLocal, engine models.Base.metadata.create_all(bind=engine) app = FastAPI() # 依赖 def get_db(): try: db = SessionLocal() yield db finally: db.close() @app.post("/users/", response_model=schemas.User) def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)): # 根据email查找用户 db_user = crud.get_user_by_email(db, email=user.email) # 如果用户存在，提示该邮箱已经被注册 if db_user: raise HTTPException(status_code=400, detail="Email already registered") # 返回创建的user对象 return crud.create_user(db=db, user=user) @app.get("/users/", response_model=List[schemas.User]) def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)): # 读取指定数量用户 users = crud.get_users(db, skip=skip, limit=limit) return users @app.get("/users/{user_id}", response_model=schemas.User) def read_user(user_id: int, db: Session = Depends(get_db)): # 获取当前id的用户信息 db_user = crud.get_user(db, user_id=user_id) # 如果没有信息，提示用户不存在 if db_user is None: raise HTTPException(status_code=404, detail="User not found") return db_user @app.post("/users/{user_id}/items/", response_model=schemas.Item) def create_item_for_user( user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db) ): # 创建该用户的items return crud.create_user_item(db=db, item=item, user_id=user_id) @app.get("/items/", response_model=List[schemas.Item]) def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)): # 获取所有items items = crud.get_items(db, skip=skip, limit=limit) return items (pt19) D:\web_python_dev>uvicorn fastapi_mysql.main:app --reload INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit) INFO: Started reloader process [6988] using watchgod INFO: Started server process [2112] INFO: Waiting for application startup. INFO: Application startup complete. mysql> use fastapi_db Database changed mysql> show tables; +----------------------+ | Tables_in_fastapi_db | +----------------------+ | items | | users | +----------------------+ 2 rows in set (0.00 sec) 3.2 其他orm框架除了sqlalchemy, 还有fastapi-async-sqlalchemy, tortoise-orm等其他框架, 网上可以搜搜

2022年02月03日
4,858 阅读
0 评论
3 点赞

2022-01-18

FastApi快速入门 01: 基础使用 FastApi Web后端开发框架快速入门01 简介FastAPI 是一个用于构建 API 的现代、快速（高性能）的 web 框架，使用基于类型提示的 Python 3.6 及更高版本。关键特性: 快速：可与 NodeJS 和 Go 比肩的极高性能（归功于 Starlette 和 Pydantic）。最快的 Python web 框架之一。高效编码：提高功能开发速度约 200％至 300％。更少bug：减少约 40％的人为（开发者）导致错误。智能：极佳的编辑器支持。处处皆可自动补全，减少调试时间。简单：设计的易于使用和学习，减少阅读文档时间。简短：减少代码重复。通过不同的参数声明实现丰富功能。bug 更少。健壮：生产可用级别的代码。以及自动生成的交互式文档。标准化：基于 API 的相关开放标准并完全兼容：OpenAPI (以前被称为 Swagger) 和 JSON Schema。官方文档：https://fastapi.tiangolo.com 源码地址：https://github.com/tiangolo/fastapi 安装直接使用pip命令进行安装：pip install fastapi 需要一个ASGI 服务器，生产环境可以使用 Uvicorn 或者 Hypercornpip install uvicorn 可选的其他组件# 从表单获取参数 pip install python-multipart # 资源文件管理 pip install aiofiles # jinja2模板引擎 pip install jinja2 # orm操作数据库 pip install tortoise-orm #异步orm框架 pip install sqlalchemy # 有名的ORM框架 pip install pymysql # MySQL驱动 pip install fastapi-async-sqlalchemy # 异步sqlalchemy 1.hello world 1.1 创建文件main.pyfrom fastapi import FastAPI import uvicorn app = FastAPI() @app.get("/") async def index(): return {"msg":"hello world"} if __name__ == '__main__': uvicorn.run(app,host="127.0.0.1",port=8000) 1.2 启动项目如果是用虚拟环境先进入虚拟环境activate # 进入虚拟环境 deactivate # 退出虚拟环境运行脚本uvicorn main:app # 普通启动 uvicorn main:app --reload # 自动重载方式启动 uvicorn main:app --port 8000 --reload # 指定端口启动 1.3 接口文档fastapi会自动生成接口文档, 启动后可以查看接口文档http://127.0.0.1:8000/docs 交互式 API 文档http://127.0.0.1:8000/redoc 2.常规请求示例2.1 单个POST请求@app.post("/login") def login(): return {"msg":"login success"} 2.2 同时支持支持post, get和put请求@app.api_route("/list", methods=["GET","POST","PUT"]) def list(): return {"list":"data"} 2.3 从URL获取参数@app.get("/page/{n}") def page(n): """从URL获取参数""" return n 2.4 从请求头获取参数from fastapi import FastAPI,Header @app.get("/user") def user(id,token=Header(None)): '''从请求头获取参数''' return {"id":id,"token":token} 2.5 从请求体获取参数from fastapi import FastAPI,Header,Body @app.get("/getdata") def getdata(data=Body(None)): """从请求体获取参数""" return {"data":data} 2.6 从请求表单获取参数安装处理包pip install python-multipart from fastapi import FastAPI,Header,Body,Form @app.post("/login2") def login2(username=Form(None),passwd=Form(None)): """从请求表单获取参数""" return {"data":{"username":username,"passwd":passwd}} 2.7 定制响应头from fastapi.responses import JSONResponse @app.get("/user") def user2(): '''定制响应头''' return JSONResponse(content={"msg":"get user"}, status_code=202, headers={"a":"b"}) 2.8 返回HTML页面from fastapi.responses import JSONResponse,HTMLResponse @app.get("/html") def html(): '''返回HTML页面''' html_content = ''' <html> <body> <p> html page </p> </body> </html> ''' return HTMLResponse(content=html_content) 2.9 返回资源文件安装附加包pip install aiofiles from fastapi.responses import JSONResponse,HTMLResponse,FileResponse @app.get("/files") def files(): '''返回资源文件''' files = 'static/mage.jpg' return FileResponse(files,filename="mage.jpg") 2.10 使用模板使用jinja2模板引擎pip install jinja2 from fastapi import FastAPI,Request from fastapi.templating import Jinja2Templates template = Jinja2Templates("pages") @app.get("/app1") def app1(req:Request): """返回html页面""" return template.TemplateResponse("index.html",context={"request":req}) 模板语法略

2022年01月18日
1,362 阅读
0 评论
0 点赞

2021-12-05

Python 多线程之 Redis 分布式锁前言在很多互联网产品应用中，有些场景需要加锁处理，例如：双11秒杀，全局递增ID，楼层生成等等。大部分的解决方案是基于 DB 实现的，Redis 为单进程单线程模式，采用队列模式将并发访问变成串行访问，且多客户端对 Redis 的连接并不存在竞争关系。其次 Redis 提供一些命令SETNX，GETSET，可以方便实现分布式锁机制。Python代码实现import time import redis import threading #使用连接池方式连接redis redis_pool=redis.ConnectionPool(host="127.0.0.1",port=6379) redis_conn=redis.Redis(connection_pool=redis_pool) #定义redis类 class RedisLock(): def __init__(self): self.redis_conn = redis_conn print("init the redis connection") #获取锁 def get_lock(self,name,value): while True: # set(name, value, ex=None, px=None, nx=False, xx=False) # nx - 如果设置为True，则只有name不存在时，当前set操作才执行 # ex - 过期时间（秒） result=self.redis_conn.set(name,value,nx=True,ex=3) # print(result) if(result): # 获取到result后就终止while循环 break # time.sleep(0.5) #释放锁 def release_lock(self,name,value): #获取原name key对应的value old_value = redis_conn.get(name) print("--------------------------------the key =%s ;the name=%s"%(name,old_value)) #判断原value 与要释放的值是否相同 if(old_value == value): #相同就从redis里面释放 self.redis_conn.delete(name) print("release the lock is success") def redis_lock_test(lock,name,value): try: print("% --start to work"%name) print("% --ready get the lock and execute lock operation"%name) lock.get_lock(name,value)#这里是获取锁操作 print("% --get the lock and continue to operation"%name) except Exception as e: print("the exception is：%s"%str(e)) finally: print("% --ready release the lock"%name) lock.release_lock(name,value)#最终必须释放锁操作 print("% --release the lock is over"%name) if __name__ == '__main__': start_time=time.time() rs=RedisLock() tasks=[] for i in range(1,3): # 创建线程 t = threading.Thread(target=redis_lock_test(rs,"task-name%"%i,"lock%d"%i)) # 将创建的线程放入列表 tasks.append(t) # 启动线程 t.start() #这里没有设置守护线程且没有设置join函数的timeout参数时，主线程将会一直等待，直到子线程全部结束，主线程才结束，程序退出 [t.join() for t in tasks] print("total waster time is:",time.time()-start_time)注意：thread.setDaemon(True) 当设置守护线程 join 函数的参数 timeout=2 时，主线程将会等待多个子线程 timeout 的累加和这样的一段时间，时间一到，主线程结束，杀死未执行完的子线程，程序退出。当没有设置守护线程且 join 函数的参数 timeout=2 时，主线程将会等待多个子线程 timeout 的累加和这样的一段时间，时间一到主线程结束，但是并没有杀死子线程，子线程依然可以继续执行，直到子线程全部结束，程序退出。参考 https://blog.csdn.net/weixin_41754309/article/details/121419465

2021年12月05日
751 阅读
0 评论
0 点赞

2021-06-21

Django项目使用Elasticsearch全文搜索引擎实现搜索优化 Django项目使用Elasticsearch全文搜索引擎实现搜索优化简介由于项目需要用到模糊搜索, Mysql模糊查询正常情况下在数据量小的时候，速度还是可以的，数据表数据量大的时候就会查询时效率低下, 对此, 使用全文搜索引擎就是一个很好的解决方法.Elasticsearch 是一个分布式、高扩展、高实时的搜索与数据分析引擎。它能很方便的使大量数据具有搜索、分析和探索的能力。在 Django 项目中使用Elasticsearch有很多现成的库, 比如elasticsearch,django-elasticsearch-dsl, django-haystack , drf_haystack 等。Haystack 是在 Django 中对接搜索引擎的框架，搭建了用户和搜索引擎之间的沟通桥梁。我们在 Django 中可以通过使用 Haystack 来调用 Elasticsearch 搜索引擎。 Haystack 可以在不修改代码的情况下使用不同的搜索后端（比如 Elasticsearch、Whoosh、Solr等等）。需要注意的是Haystack只支持es6以下的版本准备下载elasticsearch 2.4.6版本, 建议使用docker方式安装, 注意版本号下载kibana 4.6.4版本，用于管理下载对应版本的中文分词插件elasticsearch-analysis-ik-1.10.6.zip, 在es安装好下载的插件下载地址https://www.elastic.co/cn/elasticsearch/https://www.elastic.co/cn/downloads/kibanahttps://github.com/medcl/elasticsearch-analysis-ik/releases?page=13 docker镜像市场: http://hub.daocloud.io/Docker安装在服务器上新建配置文件docker-compose.ymlversion: "2" services: elasticsearch: image: daocloud.io/library/elasticsearch:2.6.4 restart: always container_name: elasticserach ports: - "9200:9200" kibana: image: daocloud.io/library/kibana:4.6.4 restart: always container_name: kibana ports: - "5601:5601" environment: - elasticsearch_url=http://127.0.0.1:9200 depends_on: - elasticsearch 输入命令安装docker-compose up -d 安装分词插件docker ps # 查看当前运行进程 docker exec -it es镜像id bash cd bin ./elasticsearch-plugin install https://github.com/medcl/elasticsearch-analysis-ik/releases/download/v2.6.4/elasticsearch-analysis-ik-1.10.6.zip 重启esdocker restart es镜像id 安装好Python需要的库pip install django-haystack pip install elasticsearch==2.4.1 # 注意版本2.4.1 pip install drf_haystack drf_haystack 文档https://drf-haystack.readthedocs.io/en/latest/index.html 示例1.配置再设置里配置添加appINSTALLED_APPS = [ 'haystack', ... ] 配置链接HAYSTACK_CONNECTIONS = { 'default': { 'ENGINE': 'haystack.backends.elasticsearch_backend.ElasticsearchSearchEngine', 'URL': 'http://127.0.0.1:9200/', # 此处为elasticsearch运行的服务器ip地址和端口 'INDEX_NAME': 'gameprop', # 指定elasticserach建立的索引库名称 }, } # 搜索结果每页显示数量 HAYSTACK_SEARCH_RESULTS_PER_PAGE = 5 # 实时更新index HAYSTACK_SIGNAL_PROCESSOR = 'haystack.signals.RealtimeSignalProcessor' 2.创建haystack数据模型在app名/模块名目录下创建search_indexes.py文件，注意文件名必须使用search_indexes.py，代码如下：from haystack import indexes from Headlines.models import Article class ArticleIndex(indexes.SearchIndex, indexes.Indexable): """文章索引模型类""" text = indexes.CharField(document=True, use_template=True) id = indexes.IntegerField(model_attr='id') createtime = indexes.DateTimeField(model_attr='createtime') content = indexes.CharField(model_attr='content') title = indexes.CharField(model_attr='title') def get_model(self): """返回建立索引的模型类""" return Article def index_queryset(self, using=None): """返回要建立索引的数据查询集""" return self.get_model().objects.all() 3.创建 text 字段索引值模板文件创建 text 字段索引值模板文件templates/search/indexes/APP名/模块名_text.txt{{ object.id }} {{ object.title }} {{ object.channel }} {{ object.labels }} {{ object.content }} 4.手动生成初始索引在命令行中添加如下命令, 来手动生成索引表:python manage.py rebuild_index 5.编辑视图文件views.pyfrom drf_haystack.viewsets import HaystackViewSet # 搜索文章 class SearchArticleViewSet(HaystackViewSet): # GET /Headlines/search/?text=<搜索关键字> # 这里可以写多个模型，相应的：serializer里也可以写多个index_classes index_models = [Article] serializer_class = SearchArticleIndexSerializer 6.编辑序列化器serializers.pyfrom drf_haystack import serializers as HSER # 搜索文章 class SearchArticleIndexSerializer(HSER.HaystackSerializer): class Meta: index_classes = [ArticleIndex] # 索引类的名称,可以有多个 fields = ('text', 'content', 'id', 'title', 'createtime') 7.编辑路由urls.py# 搜索文章路由 from rest_framework.routers import DefaultRouter from Headlines.views import SearchArticleViewSet router = DefaultRouter() router.register(r'Headlines/search', SearchArticleViewSet, base_name='search') urlpatterns += router.urls

2021年06月21日
2,178 阅读
2 评论
5 点赞

2021-03-08

自动化运维 Python Netmiko库的使用 Netmiko用于简化paramiko与网络设备之间的ssh连接,可在windows与Unix平台使用 , Netmiko 脚本比 Paramiko 脚本短很多安装pip install netmiko 实例执行单条命令from netmiko import ConnectHandler cisco = { 'device_type':'cisco_ios', 'host':'ip地址', 'username':'用户名', 'password':'密码' } net_connect = ConnectHandler(**cisco) ##或者 # net_connect = ConnectHandler(device_type='cisco_ios',host='IP地址',username='用户名',password='密码') #找到目前所在视图 current_view = net_connect.find_prompt() print(current_view) #执行命令，返回结果为字符串，赋值给output output = net_connect.send_command('show ip int brief') print(output) # #此为在windows里如果\n不能显示回车，则进行如下语句格式化 # o_list = output.split("\n") # for line in o_list: # print(line) 执行配置命令：手动关闭接口G1/0/29from netmiko import ConnectHandler cisco = { 'device_type':'cisco_ios', 'host':'ip地址', 'username':'用户名', 'password':'密码' } net_connect = ConnectHandler(**cisco) ##或者 # net_connect = ConnectHandler(device_type='cisco_ios',host='IP地址',username='用户名',password='密码') #要配置的命令 config_commands = ['interface GigabitEthernet1/0/29','shutdown'] #提交要配置的命令，input为提交的真实内容 input = net_connect.send_config_set(config_commands) #验证shutdown是否执行成功 output = net_connect.send_command('show run inter gi1/0/29') print(output) # #此为在windows里如果\n不能显示回车，则进行如下语句格式化 # o_list = output.split("\n") # for line in o_list: # print(line) 执行多条命令from netmiko import ConnectHandler as ch # 通过字典方式定义设备登录信息 host = { 'device_type': 'hp_comware', 'host': '192.168.56.20', 'username': 'netdevops', 'password': 'netdevops', 'port': 22, 'secret': '', # enable密码，没有可以不写这行 } # 连接设备 conn = ch(**host) # 定义一个命令列表，比如为G0/1配置一个IP地址 commands = ['int g0/1', 'ip add 1.1.1.1 30', 'desc netmiko_config'] # 这个时候可以使用 send_config_set 方法执行多条命令 output = conn.send_config_set(commands) print(output) 常用方法net_connect.send_command() # 向下发送命令，返回输出（基于模式）net_connect.send_command_timing() # 沿通道发送命令，返回输出（基于时序）net_connect.send_config_set() # 将配置命令发送到远程设备net_connect.send_config_from_file() # 发送从文件加载的配置命令net_connect.save_config() # 将running#config保存到startup#confignet_connect.enable() # 输入启用模式net_connect.find_prompt() # 返回当前路由器提示符net_connect.commit() # 在Juniper和IOS#XR上执行提交操作net_connect.disconnect() # 关闭连接net_connect.write_channel() # 通道的低级写入net_connect.read_channel() # 通道的低级写入

2021年03月08日
2,136 阅读
0 评论
2 点赞

2021-03-08

自动化运维 Python Paramiko库的使用 Python自动化运维 Paramiko库的使用简介Paramiko是用Python语言写的一个库，遵循SSH2协议，支持以加密和认证的方式，进行远程服务器的连接。paramiko支持Linux, Solaris, BSD, MacOS X, Windows等平台通过SSH从一个平台连接到另外一个平台。利用该模块，可以方便的进行ssh连接和sftp协议进行sftp文件传输paramiko包含两个核心组件：SSHClient和SFTPClient。 SSHClient的作用类似于Linux的ssh命令，是对SSH会话的封装，该类封装了传输(Transport)，通道(Channel)及SFTPClient建立的方法(open_sftp)，通常用于执行远程命令。 SFTPClient的作用类似与Linux的sftp命令，是对SFTP客户端的封装，用以实现远程文件操作，如文件上传、下载、修改文件权限等操作。这两个组件实现了常见ssh客户端的一些功能, 通过Paramiko编写脚本任务, 就可以实现自动化运维使用在命令行下输入下面命令即可安装pip install paramiko 密码远程连接单个连接import paramiko ##1.创建一个ssh对象 client = paramiko.SSHClient() #2.解决问题:如果之前没有，连接过的ip，会出现选择yes或者no的操作， ##自动选择yes client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) #3.连接服务器 client.connect(hostname='172.25.254.31', port=22, username='root', password='westos') #4.执行操作 stdin,stdout, stderr = client.exec_command('hostname') #5.获取命令执行的结果 result=stdout.read().decode('utf-8') print(result) #6.关闭连接 client.close() 批量远程密码连接from paramiko.ssh_exception import NoValidConnectionsError from paramiko.ssh_exception import AuthenticationException def connect(cmd,hostname,port=22,username='root',passwd='westos'): import paramiko ##1.创建一个ssh对象 client = paramiko.SSHClient() #2.解决问题:如果之前没有，连接过的ip，会出现选择yes或者no的操作， ##自动选择yes client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) #3.连接服务器 try: client.connect(hostnamehostname=hostname, portport=port, usernameusername=username, password=passwd) print('正在连接主机%s......'%(hostname)) except NoValidConnectionsError as e: ###用户不存在时的报错 print("连接失败") except AuthenticationException as t: ##密码错误的报错 print('密码错误') else: #4.执行操作 stdin,stdout, stderr = client.exec_command(cmd) #5.获取命令执行的结果 result=stdout.read().decode('utf-8') print(result) #6.关闭连接 finally: client.close() with open('ip.txt') as f: #ip.txt为本地局域网内的一些用户信息 for line in f: lineline = line.strip() ##去掉换行符 hostname,port,username,passwd= line.split(':') print(hostname.center(50,'*')) connect('uname', hostname, port,username,passwd) 基于公钥密钥连接import paramiko from paramiko.ssh_exception import NoValidConnectionsError, AuthenticationException def connect(cmd, hostname, port=22, user='root'): client = paramiko.SSHClient() private_key = paramiko.RSAKey.from_private_key_file('id_rsa') ###id_rsa为本地局域网密钥文件 client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) try: client.connect(hostnamehostname=hostname, portport=port, userusername=user, pkey=private_key ) stdin, stdout, stderr = client.exec_command(cmd) except NoValidConnectionsError as e: print("连接失败") except AuthenticationException as e: print("密码错误") else: result = stdout.read().decode('utf-8') print(result) finally: client.close() for count in range(254): host = '172.25.254.%s' %(count+1) print(host.center(50, '*')) connect('uname', host) sftp文件操作上传文件import paramiko #获取Transport实例 tran = paramiko.Transport("172.25.254.31",22) #连接SSH服务端 tran.connect(username = "root", password = "westos") #获取SFTP实例 sftp = paramiko.SFTPClient.from_transport(tran) #设置上传的本地/远程文件路径 localpath="passwd.html" ##本地文件路径 remotepath="/home/kiosk/Desktop/fish" ##上传对象保存的文件路径 #执行上传动作 sftp.put(localpath,remotepath) tran.close() 下载文件import paramiko #获取SSHClient实例 client = paramiko.SSHClient() client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) #连接SSH服务端 client.connect("172.25.254.31",username="root",password="westos") #获取Transport实例 tran = client.get_transport() #获取SFTP实例 sftp = paramiko.SFTPClient.from_transport(tran) remotepath='/home/kiosk/Desktop/fish' localpath='/home/kiosk/Desktop/fish' sftp.get(remotepath, localpath) client.close() 密钥的上传和下载import paramiko private_key = paramiko.RSAKey.from_private_key_file('id_rsa') tran = paramiko.Transport('172.25.254.31',22) tran.connect(username='root',password='westos') #获取SFTP实例 sftp = paramiko.SFTPClient.from_transport(tran) remotepath='/home/kiosk/Desktop/fish8' localpath='/home/kiosk/Desktop/fish1' sftp.put(localpath,remotepath) sftp.get(remotepath, localpath) paramiko的再封装import os import paramiko from paramiko.ssh_exception import NoValidConnectionsError, AuthenticationException, SSHException class SshRemoteHost(object): def __init__(self, hostname, port, user, passwd, cmd): self.hostname = hostname self.port = port self.user = user self.passwd = passwd self.cmd = cmd def run(self): """默认调用的内容""" # cmd hostname # put # get cmd_str = self.cmd.split()[0] # cmd # 类的反射，判断类里面是否可以支持该操作 if hasattr(self, 'do_'+ cmd_str): # do_cmd getattr(self, 'do_'+cmd_str)() else: print("目前不支持该功能") def do_cmd(self): client = paramiko.SSHClient() client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) try: client.connect(hostname=self.hostname, port=self.port, username=self.user, password=self.passwd) print("正在连接%s......." % (self.hostname)) except NoValidConnectionsError as e: print("连接失败") except AuthenticationException as e: print("密码错误") else: # 4. 执行操作 cmd = ''.join(self.cmd.split()[1:]) ##将输入的后面的取出，作为 stdin, stdout, stderr = client.exec_command(cmd) # 5.获取命令执行的结果 result = stdout.read().decode('utf-8') print(result) finally: # 6.关闭连接 client.close() def do_put(self): ###put /tmp/passwd ###将本地的/tmp/passwd上传到远端/tmp/passwd print('正在上传...') try: #获取Transport实例 tran = paramiko.Transport(self.hostname,int(self.port)) ##由于端口为整形，而我们用split方法得到的是str #连接SSH服务端 tran.connect(username = self.user, password = self.passwd) except SSHException as e: print('连接失败') else: #获取SFTP实例 sftp = paramiko.SFTPClient.from_transport(tran) newCmd = self.cmd.split()[1:] if len(newCmd) == 2: #设置上传的本地/远程文件路径 localpath=newCmd[0] remotepath=newCmd[1] #执行上传动作 sftp.put(localpath,remotepath) print('%s文件上传到%s主机的%s文件成功' %(localpath,self.hostname,remotepath)) else: print('上传文件信息错误') tran.close() def do_get(self): print('正在下载...') try: # 获取Transport实例 tran = paramiko.Transport(self.hostname, int(self.port)) ##由于端口为整形，而我们用split方法得到的是str # 连接SSH服务端 tran.connect(username=self.user, password=self.passwd) except SSHException as e: print('连接失败') else: # 获取SFTP实例 sftp = paramiko.SFTPClient.from_transport(tran) newCmd = self.cmd.split()[1:] if len(newCmd) == 2: # 设置下载的本地/远程文件路径 localpath = newCmd[1] remotepath = newCmd[0] # 执行上传动作 sftp.get( remotepath,localpath) print('%s主机的%s文件下载到%s文件成功' % (self.hostname,remotepath,localpath)) else: print('上传文件信息错误') tran.close() import paramiko import os # 1.选择操作的主机组:eg:mysql,web,ftp groups=[file.rstrip('.conf') for file in os.listdir('conf')] print("主机组显示：".center(50,'*')) for group in groups: print('\t',group) choiceGroup = input("选择批量操作的主机组(eg:mysql):") ##2.根据选择的主机组，显示包含的主机IP/主机名 # 1).打开文件conf/choiceGroup.conf # 2).依次读取文件每一行 # 3).只拿出 print("主机组包含的主机：".center(50,'*')) with open('conf/%s.conf' %(choiceGroup)) as f: for line in f: print(line.split(':')[0]) f.seek(0,0) ##把指针移动到文件最开始 hostinfos = [line.strip() for line in f.readlines()] ###3.让用户确认信息，选择需要批量执行的命令； ## -cmd shell 命令 ## -put 本地文件远程文件 ## -get 远程文件本地文件 print("批量执行脚本".center(50,"*")) while True: cmd = input('>>:').strip() if cmd : if cmd == 'exit' or cmd == "quit": print("执行完毕，正在退出") break for info in hostinfos: host,port,user,passwd = info.split(':') clientObj = SshRemoteHost(host,port,user,passwd,cmd) clientObj.run()

2021年03月08日
3,421 阅读
0 评论
0 点赞

2021-02-05

为什么MySQL单表不不建议超过2000W? 经常听说单表不能存太多数据, 如果单表数据太多，性能就会下降得比较厉害本就探索精神, 去搜索了下由来.对于MySQL数据库单表最大行数的说法，常见的限制是约为2千万行（或2000万行），但这个数字并不是MySQL官方的硬性限制。实际上，MySQL的表的最大行数和大小受到多个因素的影响，包括硬件资源、操作系统限制、配置设置以及数据类型和索引的使用等。这个2kw是个建议值，我们要来看下这个2kw是怎么来的单表行数限制单表行数理论最大值是多少。我们设计表的时候, 往往需要设置一个主键, 常见的id就是主键。如果主键声明为int大小，也就是32位，那么能支持2^32-1，也就是21个亿左右。如果是bigint，那就是2^64-1，但这个数字太大，一般还没到这个限制之前，磁盘先受不了。如果我把主键声明为 tinyint，一个字节，8位，最大2^8-1，也就是255。如果我想插入一个id=256的数据，那就会报错。也就是说，tinyint主键限制表内最多255条数据。结论: 主键本身唯一，也就是说主键的大小可以限制表的上限。表的索引索引内部是用的B+树，最大行数2kw的由来是基于B+树索引结构和页的大小进行估算的。下面是重新整理的说明：假设使用B+树作为索引结构，并且每个页的大小为15KB。在B+树中，叶子节点存储实际的行数据，每条行数据大小假设为1KB。一个页可以容纳的行数为y=15。我们可以使用公式 (x ^ (z-1)) * y 来计算行总数，其中x为每个非叶子节点的分支因子（即每个非叶子节点的子节点数），z为B+树的层级数。假设B+树是两层，即z=2，我们需要确定分支因子x的值。假设每个页的大小是15KB，由于叶子节点和非叶子节点的数据结构相同，我们假设剩下的15KB可以用于非叶子节点。因此，x可以计算为15KB除以每个非叶子节点大小的估计值。假设每个非叶子节点大小与叶子节点大小相同，即1KB。则x=15KB/1KB=15。已知x=15，y=15，z=2，代入公式 (x ^ (z-1)) y，可以计算出行总数为 (15 ^ (2-1)) 15 = 2万行。因此，这个2万行是指在B+树索引结构下，每个表的建议最大行数。在两层B+树的结构中，通过合理的分支因子和页大小设置，建议的最大行数为2万行。超过这个行数，可能会影响查询性能和磁盘IO次数。参考资料https://juejin.cn/post/7116381265300815903

2021年02月05日
583 阅读
0 评论
0 点赞

2021-01-03

分布式唯一ID生成规则SnowFlake雪花算法Python实现分布式唯一ID生成规则SnowFlake雪花算法Python实现简介在复杂分布式系统中，往往需要对大量的数据和消息进行唯一标识。在好多系统随着数据的日渐增长，对数据分库分表后需要有一个唯一ID来标识一条数据或消息，数据库的自增ID显然不能满足需求。此时一个能够生成全局唯一ID的系统是非常必要的。雪花算法和美团Leaf都是用于分布式系统中生成唯一ID的工具雪花算法: SnowFlake是Twitter公司采用的一种算法，目的是在分布式系统中产生全局唯一且趋势递增的ID。美团Leaf: Leaf是美团基础研发平台推出的一个分布式ID生成服务当单表数据量急剧上升后，表的查询性能会逐渐下降，会涉及到分库分表操作，如何确保数据均匀分布，可以通过全局唯一的ID。全局唯一的ID生成规则要求：全局唯一性有序递增高可用时间上的特性（例如互联网订单，从订单号可以看出具体时间的信息）雪花算法Python实现 1.第一位占用1bit，其值始终是0，没有实际作用。2.时间戳占用41bit，精确到毫秒，总共可以容纳约69年的时间。3.工作机器id 占用10bit，其中高位5bit是数据中心ID，低位5bit是工作节点ID，做多可以容纳1024个节点。4.序列号占用12bit，每个节点每毫秒0开始不断累加，最多可以累加到4095，一共可以产生4096个ID。SnowFlake算法在同一毫秒内最多可以生成多少个全局唯一ID呢：：同一毫秒的ID数量 = 1024 X 4096 = 4194304 import time class InvalidSystemClock(Exception): """ 时钟回拨异常 """ pass # 64位ID的划分 WORKER_ID_BITS = 5 DATACENTER_ID_BITS = 5 SEQUENCE_BITS = 12 # 最大取值计算 MAX_WORKER_ID = -1 ^ (-1 << WORKER_ID_BITS) # 2**5-1 0b11111 MAX_DATACENTER_ID = -1 ^ (-1 << DATACENTER_ID_BITS) # 移位偏移计算 WOKER_ID_SHIFT = SEQUENCE_BITS DATACENTER_ID_SHIFT = SEQUENCE_BITS + WORKER_ID_BITS TIMESTAMP_LEFT_SHIFT = SEQUENCE_BITS + WORKER_ID_BITS + DATACENTER_ID_BITS # 序号循环掩码 SEQUENCE_MASK = -1 ^ (-1 << SEQUENCE_BITS) # 开始时间截 (2015-01-01) TWEPOCH = 1420041600000 class IdWorker(object): """ 用于生成IDs """ def __init__(self, datacenter_id, worker_id, sequence=0): """ 初始化 :param datacenter_id: 数据中心（机器区域）ID :param worker_id: 机器ID :param sequence: 其实序号 """ # sanity check if worker_id > MAX_WORKER_ID or worker_id < 0: raise ValueError('worker_id值越界') if datacenter_id > MAX_DATACENTER_ID or datacenter_id < 0: raise ValueError('datacenter_id值越界') self.worker_id = worker_id self.datacenter_id = datacenter_id self.sequence = sequence self.last_timestamp = -1 # 上次计算的时间戳 def _gen_timestamp(self): """ 生成整数时间戳 :return:int timestamp """ return int(time.time() * 1000) def get_id(self): """ 获取新ID :return: """ timestamp = self._gen_timestamp() # 时钟回拨 if timestamp < self.last_timestamp: raise InvalidSystemClock if timestamp == self.last_timestamp: self.sequence = (self.sequence + 1) & SEQUENCE_MASK if self.sequence == 0: timestamp = self._til_next_millis(self.last_timestamp) else: self.sequence = 0 self.last_timestamp = timestamp new_id = ((timestamp - TWEPOCH) << TIMESTAMP_LEFT_SHIFT) | (self.datacenter_id << DATACENTER_ID_SHIFT) | \ (self.worker_id << WOKER_ID_SHIFT) | self.sequence return new_id def _til_next_millis(self, last_timestamp): """ 等到下一毫秒 """ timestamp = self._gen_timestamp() while timestamp <= last_timestamp: timestamp = self._gen_timestamp() return timestamp if __name__ == '__main__': worker = IdWorker(0, 0) print(worker.get_id())

2021年01月03日
1,458 阅读
0 评论
0 点赞

2020-12-28

Ant Design Pro 环境搭建简介ant design蚂蚁金服基于react打造的一个服务于企业级产品的UI框架。ant design pro 就是基于Ant Design这个框架搭建的中后台管理控制台的脚手架官网 https://pro.ant.design/文档 https://pro.ant.design/docs/getting-started-cnhttps://gitee.com/ant-design/ant-design-pro环境安装Nodejshttp://nodejs.cn/npm修改配置$ npm config set prefix"D:\Devs\nodejs\node_modules\npm\node_global_modules" $ npm config set cache"D:\Devs\nodejs\node_modules\npm\node_cache"修改npm源为淘宝源$ npm config set registry https://registry.npm.taobao.org切换回官方源$ npm config set registry http://www.npmjs.orgcnpm$ npm install -g cnpm --registry=https://registry.npm.taobao.orgcreate-react-app 项目构建工具$ cnpm install -g create-react-appUmi 乌米可插拔的企业级 react 应用框架$ npm install -g umicreate-umi 项目构建工具$ npm install -g create-umiyarn 比npm更好用的包管理$ npm install -g yarntyarn 国内源版本的tyarn$ npm install yarn tyarn -g开发工具React Developer Tools谷歌浏览器插件使用代理进入chrome插件商店安装Umi UI 可视化项目的本地研发工作台项目管理。集中式管理本地项目。配置管理。umi / bigfish 常用项目配置。任务管理。集成启动、构建、测试、代码规范检查、重新安装依赖等常用操作。资产、区块管理在项目中执行$ yarn add @umijs/preset-ui -D $ umi dev项目创建react项目使用create-react-app创建react项目$ create-react-app my-app启动项目$ cd my-app $ npm start编译生产版本$ npm run buildAnt Design 组件库$ yarn create react-app 项目 # 进入到项目目录, 引入 antd $ yarn add antd # 初始化环境 $ yanr # 运行项目 $ yarn startAnd Design Pro V4创建项目$ yarn create umi antd_pro $ # 或者 $ npm create umi antd_pro安装Umi UI 可视化项目的本地研发工作台$ yarn add @umijs/preset-ui -D # 或者 $ npm install --save @umijs/preset-ui进入到项目目录antd_pro , 初始化环境$ npm install $ # 或者 $ yarn运行项目$ npm start $ # 或者 $ yarn start运行Umi UI 管理工作台$ umi dev

2020年12月28日
1,085 阅读
0 评论
0 点赞

2020-12-22

MySQL超大分页优化 MySQL超大分页优化在数据库开发过程中我们经常会使用分页，核心技术是使用用limit start, count分页语句进行数据的读取MySQL分页起点越大查询速度越慢直接用limit start, count分页语句，表示从第start条记录开始选择count条记录：select * from product limit start, count 当起始页较小时，查询没有性能问题，我们分别看下从10， 1000， 10000， 100000开始分页的执行时间（每页取20条）。select * from product limit 10, 20 0.002秒 select * from product limit 1000, 20 0.011秒 select * from product limit 10000, 20 0.027秒 select * from product limit 100000, 20 0.057秒我们已经看出随着起始记录的增加，时间也随着增大，这说明分页语句limit跟起始页码是有很大关系的，那么我们把起始记录改为100w看下:select * from product limit 1000000, 20 0.682秒我们惊讶的发现MySQL在数据量大的情况下分页起点越大查询速度越慢，300万条起的查询速度已经需要1.368秒钟。这是为什么呢?因为limit 3000000,10的语法实际上是mysql扫描到前3000020条数据,之后丢弃前面的3000000行，这个步骤其实是浪费掉的。select * from product limit 3000000, 20 1.368秒从中我们也能总结出两件事情： limit语句的查询时间与起始记录的位置成正比 mysql的limit语句是很方便，但是对记录很多的表并不适合直接使用。二、 limit大分页问题的性能优化方法（1）利用表的覆盖索引来加速分页查询MySQL的查询完全命中索引的时候，称为覆盖索引，是非常快的。因为查询只需要在索引上进行查找,之后可以直接返回,而不用再回表拿数据。在我们的例子中，我们知道id字段是主键，自然就包含了默认的主键索引。现在让我们看看利用覆盖索引的查询效果如何。select id from product limit 1000000, 20 0.2秒那么如果我们也要查询所有列，如何优化？优化的关键是要做到让MySQL每次只扫描20条记录，我们可以使用limit n，这样性能就没有问题，因为MySQL只扫描n行。我们可以先通过子查询先获取起始记录的id,然后根据Id拿数据：select * from vote_record where id>=(select id from vote_record limit 1000000,1) limit 20; （2）用上次分页的最大id优化先找到上次分页的最大ID，然后利用id上的索引来查询，类似于：select * from user where id>1000000 limit 100

2020年12月22日
663 阅读
0 评论
0 点赞

2020-11-20

使用AI实现高精度钢琴曲转谱Piano Transcription简明使用教程简介此项目可将钢琴录音(mp3, wav等格式)转录成MIDI文件, 识别的精度非常之高, 扒谱党的福利。相对于人工扒谱效率高了不知道多少倍, AI扒谱的时代已经到来。前段时间字节跳动发布了全球最大的古典钢琴MIDI数据集 GiantMIDI-Piano, GiantMIDI-Piano 的特点是使用钢琴转谱技术，通过计算机将音频文件自动转为 MIDI 文件，并通过该技术转谱了大规模的 MIDI 数据集。研究者万余首钢琴作品+一千多小时训练，开发并开源了一套高精度钢琴转谱系统piano_transcription，将所有音频转谱成 MIDI 文件，进而构建了 GiantMIDI-Piano 数据库。该转谱系统的特点包括：能够将任意声部数目、任意复杂度，甚至双钢琴、多钢琴的钢琴音频转谱为 MIDI 文件。实现了任意时间精度的音符检测，突破了之前算法 32 毫秒识别精度的限制。对每个音符实现了 128 个粒度的力度识别。同时包含了钢琴音符和钢琴踏板的识别。在 MAESTRO 评测数据集上取得 96.72% 的 F1 值，超越了 Google 系统的 94.80%。预训练模型的代码以 Apache 2.0 协议开源。详细介绍见https://zhuanlan.zhihu.com/p/269218623GiantMIDI-Pianohttps://github.com/bytedance/GiantMIDI-Pianopiano_transcriptionhttps://github.com/bytedance/piano_transcription一网友在此基础制作的修改版https://zhuanlan.zhihu.com/p/270999354在此基础上, 我再稍微加工了下, 可批量转置, 使得更易使用下面是一些转置后制作的一些视频2011年Zetacola《赤色要塞》的钢琴演奏音质修复+特效钢琴版使用方法只需三步即可下载资源安装环境使用下载下面的资源包, 解压链接: https://pan.baidu.com/s/1aqnlgfFCjB0KIlPEB8RcRg 提取码: xubj资源包内包括piano_transcription项目，项目所使用的已训练完好模型，python安装包和ffmpeg安装环境以64位的windows版本为例, Mac和linux的步骤也是一样的, 只需安装对应系统的软件版本安装下列环境01 Python 3.7使用到了f-string, 需要Python3.6以上版本才支持, 这里选择安装Python3.7.9版本02 项目所依赖的库03 ffmpeg用于读取媒体文件, 使用资源包里的版本即可04 cuda限NVIDIA显卡, 使用cuda转置速度更快, 没有可以不装, 使用CPU转置05 pytorch需要看Python版本和cuda版本, 选择相应的版本下载01 安装 Python 3.7双击附件里边的Python 3.7.9安装即可在cmd里输入python出现下面提示及安装成功02 项目所依赖的库在项目目录根下按住shift在空白处点击powershell打开输入下面命令, 升级pip版本(我装有多个python版本, 所以图片显示命令不一样, 新手直接输入下面命令即可)python -m pip install --upgrade pip 安装依赖环境pip install -r .\requirements.txt 03 安装ffmpeg需要把ffmpeg的bin目录添加到环境变量右键点击此电脑--属性 , 点击高级系统设置, 高级里边的环境变量, 双击Path, 选择新建,将ffmpeg下的bin目录路径复制的里边,如 F:\钢琴转谱资源包\ffmpeg\bin(将路径改为你所解压的ffmpeg的bin目录路径)然后点击确定在cmd里输入ffmpeg 有显示ffmpeg version 4.3.1字样的即添加好了环境变量04 安装cuda (使用CPU跳过此步骤)CUDA 是 NVIDIA 的一种并行计算平台和编程模型。使用显卡转置比用CPU快很多cuda版本和驱动支持有关,显卡驱动会影响支持cuda的版本 , 为避免不必要的错误, 先去官网更新驱动到最新版本,这里以win10 1909版本的系统为例win7等其他系统去NVIDIA控制面板查看支持cuda的最高版本, 选择相应的cuda版本驱动下载地址https://www.nvidia.cn/geforce/drivers/安装显卡最新驱动, 再去下载cuda工具包11.1安装(有3G大)下载地址https://developer.nvidia.com/zh-cn/cuda-downloads在安装界面选择自定义, 如图只勾选cuda当中的那几项, 其他不必勾选, 选了有可能安装失败安装完成后显示05 安装pytorch到官网下载自己系统合适的版本, 如果使用CPU选择cuda版本选择nonehttps://pytorch.org/这里选择windows系统python3.7 使用pip安装cuda11为例, 按下图选择好版本后下面会有对应的安装命令在cmd命令行里输入安装pytorch, 文件挺大有差不多2G, 如果网络不好安装失败, 请看下边的离线包安装pip install torch===1.7.0+cu110 torchvision===0.8.1+cu110 torchaudio===0.7.0 -f https://download.pytorch.org/whl/torch_stable.html 文件比较大差不多2G, 如果网络不好可以下载离线安装包https://download.pytorch.org/whl/cu110/torch-1.7.0%2Bcu110-cp37-cp37m-win_amd64.whl下载好后再目录下按住shift在空白处右键点击, 选择powershell打开命令行, 输入下面安装命令 pip install .\torch-1.7.0+cu110-cp37-cp37m-win_amd64.whl 使用把MP3,或wav文件放入input文件夹, 可以放多个音频文件在项目piano_transcription按住shift在空白处右键点击, 选择powershell打开命令行, 输入以下命令, 等待程序跑完即可 python .\start.py 使用GPU进行转置还是挺快的, 大概10秒~1分多钟一首跑完后可在output文件夹得到转置好的mid文件, 使用播放器播放即可, 可使用Pianoteq的音源生成高音质的mp3, 如果效果不是很好, 可以使用midi编辑软件进行进一步的修改。midi文件还可以通过一些软件转成琴谱如果使用CPU转置可以修改start.py文件把第19行里边的cuda改为cpu即可

2020年11月20日
37,794 阅读
6 评论
114 点赞

2020-10-12

2020年10月12日
6,623 阅读
0 评论
4 点赞

2020-10-05

unravel 让图片唱歌详细教程 Real Time Image Animation 项目 01 下载资源项目资源原始项目Real Time Image Animation项目地址：https://github.com/anandpawara/打包修改好的代码、权重文件、视频图片素材, 下载下面这个即可下载链接（密码:amz5）：https://pan.baidu.com/s/1TEd7SOaO5mzPaxpOh2pALQ资源来源于Jack-Cui 知乎文章https://zhuanlan.zhihu.com/p/19311921602 环境安装1. 安装Python为避免报错, 选择和项目一样的Python版本Python3.7https://www.python.org/ftp/python/3.7.9/python-3.7.9-amd64.exe安装过程: 略2.安装依赖包安装项目依赖包, 进入到进入到项目目录Real_Time_Image_Animation-master, 按住shift右键点击目录空白处在此处打开powershell窗口, 输入下面命令安装python -m pip install -r requirements.txt 3.安装pytorch进入到pytorch官网https://pytorch.org/get-started/locally/ 安装合适的版本我选择了最新的版本在命令行输入pip install torch===1.6.0 torchvision===0.7.0 -f https://download.pytorch.org/whl/torch_stable.html 这个包比较大, 大约有1个G, 如果网络不好可以下载包再安装4.安装CUDA安装CUDA必须是nvidia显卡建议先升级显卡驱动, 驱动版本越新能安装的cuda版本越高, 也能安装旧版本的cuda我安装的是CUDA Toolkit 11.1 实测也能兼容pytorch 1.6.0https://developer.nvidia.com/cuda-toolkit-archive安装时选择自定义安装, 为避免不必要的错误, 只勾选如图所CUDA下那四项就可以了5.安装ffmpeg下载https://www.gyan.dev/ffmpeg/builds/packages/ffmpeg-4.3.1-2020-10-01-full_build.7z或蓝奏云盘https://wws.lanzous.com/iRzjbhg3xod下载好后解压把bin目录添加到环境变量cmd输入ffmpeg -version 显示版本号等信息即添加成功使用在项目Real_Time_Image_Animation-master文件夹下打开命令行运行python `项目主程序` -i `需要制作的源图路径` -c `权重文件路径` -v `输入的视频文件(unravel源视频, 也可以自己制作)` 例如python image_animation.py -i Inputs/trump2.png -c checkpoints/vox-cpk.pth.tar -v 1.mp4 注意输入的源图最好是长宽比1:1的正方形分辨率256x256像素以上, 并且五官清晰自己做的输入源视频分辨率是256x256像素, 时长2分钟内, 不然会报内存不足

2020年10月05日
27,386 阅读
2 评论
74 点赞

2020-09-21

2020年09月21日
1,410 阅读
0 评论
0 点赞