使用Sentry监控展示Django全局错误日志，并通过飞书实时报警
最近工作上的Django项目，使用了Sentry来做错误日志的监控报警，因为错误日志太多，排查不容易，所以还不如来个主动报警，这样定位排错也很方便，而实际使用情况来看，确实是超级方便的！Sentry在错误监控领域已经是一个很成熟的应用了，支持各种语言多种服务程序的日志监控，如下图：关于更详细的介绍，可以访问官网查看：https://sentry.io/Sentry安装Sentry的安装部分，我这里采用的额时Docker容器部署，关于Docker的安装可以看我的另一篇文章[post cid="65" /]这里具体介绍怎么部署容器版Sentry。部署步骤docker拉取redis postsql 和sentrydocker pull redis docker pull postgres docker pull sentry启动redis和sentrydocker run -d --name sentry-redis redis docker run -d --name sentry-postgres -e POSTGRES_PASSWORD=secret -e POSTGRES_USER=sentry postgres启动Sentry相关docker run --rm sentry config generate-secret-key # 得到secret-key，然后把key复制到下面四行的单引号中 #下面这步会提示输入邮箱和密码 docker run -it --rm -e SENTRY_SECRET_KEY='' --link sentry-postgres:postgres --link sentry-redis:redis sentry upgrade docker run -d -p 9000:9000 --name my-sentry -e SENTRY_SECRET_KEY='' --link sentry-redis:redis --link sentry-postgres:postgres sentry docker run -d --name sentry-cron -e SENTRY_SECRET_KEY='' --link sentry-postgres:postgres --link sentry-redis:redis sentry run cron docker run -d --name sentry-worker-1 -e SENTRY_SECRET_KEY='' --link sentry-postgres:postgres --link sentry-redis:redis sentry run worker访问测试经过上面的步骤，Sentry已经算是部署完毕了，然后打开浏览器，对应的ip地址加9000端口即可访问。打开显示如下图：登录之前设置好的邮箱和密码即可登入系统。新建项目以及基本设置修改系统语言在左侧的user_setting里面可以修改语言为中文简体新建项目在首页点击项目，会出现很多项目模板，选择Django即可Django接入SentryDjango接入Sentry相对来说比较简单，我们这里采用安装Sentry客户端的方式接入。# 首先安装raven pip install raven然后在settings文件中的INSTALLED_APPS里注册Sentry模型INSTALLED_APPS = [ ... # Sentry错误日志监控系统 'raven.contrib.django.raven_compat', ]随后我们返回Sentry网页，点击创建的项目，然后点击齿轮设置图标，点击“客户端密钥（DSN）”，复制该项目的DSN然后再将复制的Sentry DSN粘贴在Settings配置文件中# Sentry错误日志监控系统 RAVEN_CONFIG = { 'dsn': 'http://1283djasssdv2312d12bi21dib1341@192.168.1.124:9000/2', }就这样Django便接入了Sentry，我们在项目里新建一个test_sentry.py文件测试下错误是否能被捕获，内容如下：dsn = "http://1283djasssdv2312d12bi21dib1341@192.168.1.124:9000/2" from raven import Client Client = Client(dsn) try: 1 / 0 except ZeroDivisionError: Client.captureException()执行后去Sentry网页查看，如果出现下图，就说明Sentry起作用了！（因为我已经跑了一段时间了，所以页面有很多别的错误，请忽略~~~）我们点击这条记录进去，一起看看Sentry有多强大方便，访问信息、错误栈全都一览无余！太方便了！在接下来，我们来接入飞书进行实时告警接入飞书告警Sentry有一点好处就是内置了webhook方法，只要将外部申请好的webhook链接填入Sentry对应的设置，便可以实现错误日志实时告警！飞书捷径机器人设置告警捷径飞书时字节跳动旗下的办公协同软件，面世以来饱受赞誉，功能非常强大，公司里也在用，办公协同很牛，yyds~，推荐。我们要用到的时飞书的“飞书捷径”机器人，具体操作可以看官方文档，相信绝对比我讲的清楚，点击如何在飞书捷径中使用 Sentry配置Sentry webhook首先我们要设置一下Sentry的报警规则，也就是定义项目错误在什么情况下触发报警，以及报警方式，。点击项目的设置栏，点击“警报”，进入规则标签页，点击新建规则：我们需要定义的点有3个：“An event is seen” 条件，任何选项报警措施，Send a notification via webhookPerform these actions at most once every 5分钟 for an issue （修改报警间隔）这里新建的时Sentry报警的规则，然后点击保存即可，返回即可看到我们刚才新建的规则然后我们获取一下上面设置的飞书捷径的webhook，就在下图示例位置：最后，我们只需要将飞书捷径的webhook链接填入到Sentry网页设置中，具体位置在项目的设置选项最底部，点击webhook：填入链接后，点击save Changes测试告警我们现在来测试一下告警是否成功，点击上图的webhook设置页面，右上角有一个“Test Plugin”，Sentry就会往飞书捷径机器人发送一条告警，如下图所示：后记到这一步，整个配置就完成了，其实Sentry能做的还有很多，感兴趣的话，你可以自己深入了解。
django使用django-otp实现后台admin管理系统令牌二次登录，让你的系统更加安全
很多人吐槽django的后台管理系统登录模块不够安全，没有验证码，没有短信。就是裸账号密码登录。确实，django在系统定制这上面做到了极致，用户能改的地方不多，admin管理系统也是一样的被人吐槽简陋，当然，admin系统也不是不能美化，关于美化的相关介绍可以看我的另一篇介绍“simpleui美化”。这里给大家带来的是admin登录的otp二次令牌验证。具体效果如下图，otp token就是令牌验证码：安装安装的话也很简单，首先是先安装两个包，在项目环境里执行：pip install django-otp pip install qrcode然后将 django-otp 添加到 setting.py 和中间件中：INSTALLED_APPS = [ ''' # 安全认证opt 'django_otp', 'django_otp.plugins.otp_totp' ''' ] MIDDLEWARE = [ ''' 'django.contrib.auth.middleware.AuthenticationMiddleware', # 用户认证，注意放在用户认证中间件下面 'django_otp.middleware.OTPMiddleware', ''' ] # opt加密配置，自定义字符配置就行 OTP_TOTP_ISSUER = "otp解码"最后在项目的urls里面加入otp路由配置：from django_otp.admin import OTPAdminSite from django.contrib import admin admin.site.__class__ = OTPAdminSite这样就完成了基础的配置，这时候当你再次打开项目admin后台，就会出现上图出现过的登陆页面了配置登录那现在问题来了，没有otp token我该怎么登录呢，这不是把自己写死了么？别急，等我慢慢说，首先先把刚才在urls里面配置的路由注释掉：# admin.site.__class__ = OTPAdminSite然后打开admin，恢复了老版本，可以正常登录，浏览器链接部分打开“/admin/#otp_totp/totpdevice/”，就会出现如下页面（样式可能不太一样，那是因为我配置了simpleui，无关大雅）这时候点击新增，然后在user部分点击放大镜添加勾选管理员，然后name处写上备注，右下角点击保存，返回列表页面这时候可以看到列表出现了刚才添加的用户，点击右边的qrcode，就会弹出一个二维码。这时候用手机app扫码，就会把动态验证码保存到手机app上，手机app会在文末给出，安卓的是离线的app，提供安装包下载，苹果则是商店下载。这时候基础配置就都完成啦，再把刚才注释掉的urls代码放开就好了，以后每次登录都需要用户名密码和otp的密码才能登陆。关于otp验证技术关于otp验证技术相关介绍，我找了一篇网上的评论介绍，大家可以参考下离线验证码生成器的原理是什么？app下载这里分安卓和ios安卓：https://pan.baidu.com/s/1GgP1iVK7ZKHOZVI_UokiQg ios: 好了，有什么问题都可以联系我。
代码的执行性能是程序员需要特别关注的一个点，市面上有很多检测监控python代码执行性能的工具，但是都过于笨重，这里推荐一个更加轻量，粒度更细的性能测试工具包——line_profiler。原理与安装line_profiler作为一个性能测试工具来说是很轻便的，安装也很简单，正常pip安装即可pip install line_profiler安装完成后不需要做额外的过多操作，只需要在要测试的方法之上添加一个@profile装饰器。在程序执行时就会自动将测试代码带入测试程序中进行性能测试。功能测试这里简单的用一段测试代码测试下，新建test_code文件：@profile def test_code(): a = [] a.append('1') a.append('2') a.append('3') a.append('4') a.extend(['4', '5', '6']) c = '' for i in a: c += i print(c) if __name__ == '__main__': test_code()然后运行测试命令：kernprof -l -v test_code.py然后控制台就能看到输出：1234456 Wrote profile results to test_code.py.lprof Timer unit: 1e-06 s Total time: 0.0002404 s File: test_code.py Function: test_code at line 3 Line # Hits Time Per Hit % Time Line Contents ============================================================== 3 @profile 4 def test_code(): 5 1 2.8 2.8 1.2 a = [] 6 7 1 2.5 2.5 1.0 a.append('1') 8 1 1.2 1.2 0.5 a.append('2') 9 1 1.3 1.3 0.5 a.append('3') 10 1 1.2 1.2 0.5 a.append('4') 11 12 1 2.0 2.0 0.8 a.extend(['4', '5', '6']) 13 14 1 0.9 0.9 0.4 c = '' 15 8 7.0 0.9 2.9 for i in a: 16 7 8.0 1.1 3.3 c += i 17 1 213.5 213.5 88.8 print(c)从输出中可以看到每行代码的耗时以及耗时百分比，可以看到for循环和字符串拼接是比较耗时的，这就是line_profiler的整个使用流程，很简单吧。
在Web应用中，应用性能是一个值得关注的点，我们不应该仅仅局限满足于应用程序“能跑”和“能用”上，还需要考虑到应用程序的执行性能，比如sql语句执行时间，耗时部分在哪里等。简介刚好前不久Django项目里用到过一个测试模块Django-Debug-Toolbar，它是一个内嵌模组，拥有调试ui界面，在Django项目里简单配置后就能在web前端展示的性能调试工具，具体效果如图：从右侧可以看到有很多指标小模块，执行耗时，sql执行耗时，缓存，信号等等，是一个性能调试利器。接下来我们就看看怎么在系统中对此进行配置：配置在终端中执行：python -m pip install django-debug-toolbar然后在Django项目里的settings中的INSTALLED_APPS中添加：INSTALLED_APPS = [ 'debug_toolbar', ]以及添加中间件Middleware，尽量将添加的这句排前一点：MIDDLEWARE = [ # ... 'debug_toolbar.middleware.DebugToolbarMiddleware', # ... ]以及在settings文件中添加白名单配置：INTERNAL_IPS = [ # ... '127.0.0.1', # ... ]然后在settings增加DEBUG_TOOLBAR_​​PANELS基础配置：此设置允许您： #添加默认情况下未启用的内置面板，添加第三方面板，移除内置面板，更改面板的顺序。 DEBUG_TOOLBAR_PANELS = [ 'debug_toolbar.panels.history.HistoryPanel', 'debug_toolbar.panels.versions.VersionsPanel', 'debug_toolbar.panels.timer.TimerPanel', 'debug_toolbar.panels.settings.SettingsPanel', 'debug_toolbar.panels.headers.HeadersPanel', 'debug_toolbar.panels.request.RequestPanel', 'debug_toolbar.panels.sql.SQLPanel', 'debug_toolbar.panels.staticfiles.StaticFilesPanel', 'debug_toolbar.panels.templates.TemplatesPanel', 'debug_toolbar.panels.cache.CachePanel', 'debug_toolbar.panels.signals.SignalsPanel', 'debug_toolbar.panels.logging.LoggingPanel', 'debug_toolbar.panels.redirects.RedirectsPanel', 'debug_toolbar.panels.profiling.ProfilingPanel', ]记得还要引入jquery的jsDEBUG_TOOLBAR_CONFIG = { "JQUERY_URL": '//cdn.bootcss.com/jquery/2.2.4/jquery.min.js', }然后在路由urls里面添加路由：import debug_toolbar from django.conf import settings from django.urls import include, path urlpatterns = [ ... path('__debug__/', include(debug_toolbar.urls)), ]后记就这样配置就完成了。其实Django-Debug-Toolbar有很多第三方的小模组插件的，大家可以去官方文档处查看，点击链接跳转
使用python生成阿里云直播流推流拉流地址
遇到直播相关的业务时，比如手机带货直播等，我们想到的就是寻找大厂提供的相应视频直播流服务。大厂的技术是值得信赖的，以阿里云直播为例，流程是用户方发起直播任务，相应的调用自己服务接口逻辑生成直播推流和拉流地址，继而拿到推流地址进行推流。推流的视频流就经过阿里云服务处理，然后分发到阿里云自己的cdn，用户则在用户端使用拉流地址进行拉流观看。整个流程就是这样的。这里我们以python语言为例，主要实现产品服务端接收响应生成推流拉流地址的逻辑。话不多说，直接上代码：import time import datetime import hashlib import config def md5_encode(hash_str: str): """""" m = hashlib.md5() m.update(hash_str.encode("utf8")) return m.hexdigest() def create_live_stream_url(app_name: str, stream_name: str): time_ex = datetime.datetime.now() + datetime.timedelta(hours=config.hours) time_sta = time.mktime(time_ex.timetuple()) time_fin = str(int(time_sta)) # 加密key，直播鉴权 push_key = config.push_key pull_key = config.pull_key str_push = "/"+app_name+"/"+stream_name+"-"+time_fin+"-0-0-"+push_key push_url = "rtmp://"+config.push_host+"/"+app_name+"/"+stream_name+"?auth_key="+time_fin+"-0-0-"+md5_encode(str_push) str_view_rtmp = "/"+app_name+"/"+stream_name+"-"+time_fin+"-0-0-"+pull_key str_view_flv = "/"+app_name+"/"+stream_name+".flv"+"-"+time_fin+"-0-0-"+pull_key str_view_m3u8 = "/"+app_name+"/"+stream_name+".m3u8"+"-"+time_fin+"-0-0-"+pull_key rtmp_url = "rtmp://"+config.pull_host+"/"+app_name+"/"+stream_name+"?auth_key="+time_fin+"-0-0-"+md5_encode(str_view_rtmp) flv_url = "http://"+config.pull_host+"/"+app_name+"/"+stream_name+".flv"+"?auth_key="+time_fin+"-0-0-"+md5_encode(str_view_flv) m3u8_url = "http://"+config.pull_host+"/"+app_name+"/"+stream_name+".m3u8"+"?auth_key="+time_fin+"-0-0-"+md5_encode(str_view_m3u8) # print("推流地址: " + push_url) # print("rtmp播流地址: " + rtmp_url) # print("flv播流地址: " + flv_url) # print("m3u8播流地址: " + m3u8_url) return {'push_url': push_url, 'rtmp_url': rtmp_url, 'flv_url': flv_url, 'm3u8_url': m3u8_url} 传入阿里云视频服务中配置的app_name和生成的stream_name，处理生成的4个流地址，注意，地址格式必须按照上述规则生成，否则阿里云在验证地址时候会无法通过，也就是生成的流地址是没用的。
记得之前工作业务上有实现生成小程序分享海报的需求，但是相比小程序端canvas生成海报还是复杂的多，这里使用的是PIL来画图拼接，最后生成图片，最后保存在本地或者上传到云存储以便后续使用，具体看业务需求。话不多说，这里使用的是python语言，以下为代码import uuid import json import os import requests import config import base64 from PIL import Image, ImageDraw, ImageFont class PosterImage: """海报""" qr_code_base_url = 'https://api.weixin.qq.com/wxa/getwxacodeunlimit?access_token=' def __init__(self, avatar, backdrop_path): self.avatar = avatar self.backdrop_path = backdrop_path self.qr_code_url = self.qr_code_base_url + （你的access_token） self.backdrop = Image.open(self.backdrop_path) self.generate_copy_backdrop() def get_qr_code(self, content: str): """生成二维码 pil对象""" def get_stream(self, url, user_id, role, kwargs=None): """获取二维码字节码""" json_data = {'scene': '', 'page': '', 'width': 280, 'auto_color': False, 'is_hyaline': False} # 提交到微信地址 result = requests.post(url, json=json_data, timeout=3, verify=False) self.stream = result.content def get_avatar(self): """获取微信头像""" self.stream = requests.get(url=self.avatar).content def save_stream(self): """将流数据保存至本地文件内,返回文件名""" prefix = uuid.uuid1().hex file_name = f'{prefix}.png' file_path = 'qr_code/' + file_name import os path = os.getcwd()+'/plugins/public/' file_path = os.path.join(path, file_path) with open(file_path, 'wb') as f: f.write(self.stream) return file_path def open_material(self, path): """打开素材 qr_code:背景 qr_code:小程序页面二维码 """ try: self.material = Image.open(path) return self.material except(OSError, NameError): print('OSError, Path:', path) def generate_copy_backdrop(self): """生成背景图拷贝对象,对背景图的操作不会影响原对象""" self.copy_backdrop = self.backdrop.copy().convert('RGB') self.copy_backdrop_draw = ImageDraw.Draw(self.copy_backdrop) def set_font(self, css_path: str = None, fill: tuple = (0, 0, 0), size: int = 36): """设定字体""" import os path = os.getcwd() path = os.path.join(path, "static/Alibaba-PuHuiTi-Bold.ttf") self.font = ImageFont.truetype(path, size=size) self.font_fill = fill def synthesis(self, img, x: float, y: float, size=None): """合成图片 :param img: pil对象 :param size: 图形尺寸 :param x: 横向坐标 :param y: 纵向坐标 """ if size: img = img.resize((size, size), Image.ANTIALIAS) self.copy_backdrop.paste(img, (x, y)) def write(self, content, x: float, y: float): """写入内容,写入前可以通过self.set_font 设定字体 :param content: 文字内容 :param x: 横向坐标 :param y: 纵向坐标 """ self.copy_backdrop_draw.text((x, y), text=content, fill=self.font_fill, font=self.font) def create_post(): """生成分享海报""" import os path = os.getcwd() path = os.path.join(path, "static/backdrop.png") poster_image = PosterImage(avatar=avatar, backdrop_path=path) # 获取二维码/合成二维码 poster_image.get_stream(url=poster_image.qr_code_url, user_id=user_id, role=role) poster_image.open_material(poster_image.save_stream()) poster_image.synthesis(poster_image.material, x=505, y=855, size=195) # 获取微信头像/合成头像 if user.avatar: poster_image.get_avatar() poster_image.open_material(poster_image.save_stream()) poster_image.synthesis(poster_image.material, x=65, y=895, size=120) # 获取昵称 if user.nickname: poster_image.set_font() poster_image.write(nickname, x=205, y=900) filename = f'{uuid.uuid1().hex}.jpg' path = os.getcwd()+'/plugins/public/' poster_image_path = os.path.join(path, 'post_image/' + filename) poster_image.copy_backdrop.save(poster_image_path) # todo 将海报直传oss 或者保存本地 其实这段代码没有很复杂，只是将构建好的图像不断的合成，难点的话就是相对应的方法调用，还有代码里出现的很多x和y坐标，这个是你的图片元素的位置，你可以放置任意的图像进行合成，相当于不断的在一个画布上堆东西，api还是很丰富的，可以旋转，遮罩，放大缩小等。具体使用还需要你自己摸索。
之前在做后台管理系统时候，有导出和导入excel的需求，其实不难，很快就用xlrd和xlwt实现了。[scode type="blue"]首先先安装xlrd和xlwt[/scode]import xlrd import xlwt[scode type="blue"]导入excel业务实现[/scode]def xls_import(): """学生导入""" form = validators.XlsImportStudent().validate_() file = request.files.get('file') filename = f'{uuid.uuid1().hex}.{form.file_extension.data}' # 保存文件 file_path = 'inxlsx/' + filename path = os.getcwd() + '/plugins/public/' file_path = os.path.join(path, file_path) file.save(file_path + filename) data = open_excel(file=file_path + filename).sheets()[0] nrows = data.nrows pass_num = 0 success_num = 0 for i in range(1, nrows): # 判断学生是否存在 student = Student.query.filter_by(name=data.row_values(i)[0], phone=data.row_values(i)[1]).first() if student: pass_num += 1 pass else: # 数据处理 # 性别 gender = data.row_values(i)[5] if gender == "男": gender_int = 1 else: gender_int = 2 # 政治面貌 political = data.row_values(i)[10] if political == "群众": political_int = 1 elif political == "中共共青团员": political_int = 2 elif political == "中共预备党员": political_int = 3 elif political == "中共党员": political_int = 4 else: political_int = 5 # 最高学历 education = data.row_values(i)[15] if education == "小学": education_int = 1 elif education == "初中": education_int = 2 elif education == "高中": education_int = 3 elif education == "大专": education_int = 4 elif education == "本科": education_int = 5 elif education == "硕士研究生": education_int = 6 elif education == "博士研究生": education_int = 7 else: education_int = 8 # 日期处理 birthday = data.row_values(i)[4] # print(birthday) birthday = datetime.date.fromordinal(int(birthday) + 693594) # dt = datetime.date(1993, 4, 27) # print(dt.toordinal()) graduated_time = data.row_values(i)[17] graduated_time = datetime.date.fromordinal(int(graduated_time) + 693594) # 学校和专业 profession_code = data.row_values(i)[3] profession = Profession.query.filter_by(profession_code=profession_code).first() if profession: Student( bh=uuid.uuid4().hex, name=data.row_values(i)[0], phone=data.row_values(i)[1], id_card=data.row_values(i)[2], school=profession.school_id, profession=profession.id, birthday=birthday, gender=gender_int, mail=data.row_values(i)[6], clan=data.row_values(i)[7], hometown=data.row_values(i)[8], address=data.row_values(i)[9], political=political_int, health=data.row_values(i)[11], birth_place=data.row_values(i)[12], home_address=data.row_values(i)[13], work_unit=data.row_values(i)[14], education=education_int, graduated_school=data.row_values(i)[16], graduated_time=graduated_time, avatar=data.row_values(i)[18], source=2 ).direct_add_() success_num += 1 else: return result_format(data={'fail': '第' + str(i+1) + '条数据的专业编码不存在，请检查！'}) Student.static_commit_() return ''[scode type="blue"]导出excel业务实现[/scode]def xls_out_port(): """学生导出""" form = validators.XlsOutportStudent(request.args).validate_() query = Student.query.filter_by(status=1).order_by(Student.id.desc()) if form.profession.data: query = query.filter_by(profession=form.profession.data) paginate = query.paginate(page=form.page.data, per_page=form.limit.data, error_out=True) funcs = [orm_func('change_school_info')] data = paginate_info(paginate=paginate, items=[item.serialization(funcs=funcs) for item in paginate.items]) # 生成excel对象 filename = xlwt.Workbook() sheet = filename.add_sheet("Sheet1") # 写入第一行数据 row0 = ['姓名', '手机号', '身份证号', '报名学校', '报名专业', '出生日期', '性别', '邮箱', '民族', '籍贯', '通信地址', '政治面貌', '身体状况', '出生地', '家庭地址', '工作单位', '最高学历', '毕业学校', '毕业日期', '用户头像'] for i in range(0, len(row0)): sheet.write(0, i, row0[i], set_style('sky_blue')) for index, item in enumerate(data['items']): row_start = 1 + index sheet.write(row_start, 0, item['name'], set_style()) sheet.write(row_start, 1, item['phone'], set_style()) sheet.write(row_start, 2, item['id_card'], set_style()) sheet.write(row_start, 3, item['school'], set_style()) sheet.write(row_start, 4, item['profession'], set_style()) sheet.write(row_start, 5, item['birthday'].split(' ')[0], set_style()) # 性别 if item['gender'] == 1: gender = "男" else: gender = "女" sheet.write(row_start, 6, gender, set_style()) sheet.write(row_start, 7, item['mail'], set_style()) sheet.write(row_start, 8, item['clan'], set_style()) sheet.write(row_start, 9, item['hometown'], set_style()) sheet.write(row_start, 10, item['address'], set_style()) # 政治面貌 if item['political'] == 1: political = "群众" elif item['political'] == 2: political = "中共共青团员" elif item['political'] == 3: political = "中共预备党员" elif item['political'] == 4: political = "中共党员" else: political = "其他" sheet.write(row_start, 11, political, set_style()) sheet.write(row_start, 12, item['health'], set_style()) sheet.write(row_start, 13, item['birth_place'], set_style()) sheet.write(row_start, 14, item['home_address'], set_style()) sheet.write(row_start, 15, item['work_unit'], set_style()) # 学历 if item['education'] == 1: education = "小学" elif item['education'] == 2: education = "初中" elif item['education'] == 3: education = "高中" elif item['education'] == 4: education = "大专" elif item['education'] == 5: education = "本科" elif item['education'] == 6: education = "硕士研究生" elif item['education'] == 7: education = "博士研究生" else: education = "其他" sheet.write(row_start, 16, education, set_style()) sheet.write(row_start, 17, item['graduated_school'], set_style()) sheet.write(row_start, 18, item['graduated_time'].split(' ')[0], set_style()) sheet.write(row_start, 19, item['avatar'], set_style()) # 设置单元格宽度 sheet.col(1).width = 4500 sheet.col(2).width = 6500 sheet.col(3).width = 3000 sheet.col(4).width = 3000 sheet.col(5).width = 3700 sheet.col(7).width = 5500 sheet.col(10).width = 5000 sheet.col(11).width = 3000 sheet.col(12).width = 3000 sheet.col(13).width = 3000 sheet.col(14).width = 5000 sheet.col(15).width = 5000 sheet.col(16).width = 3000 sheet.col(17).width = 5000 sheet.col(18).width = 3700 sheet.col(19).width = 15000 file_name = '奥翔教育学生信息-' + datetime.datetime.now().strftime("%Y%m%d%H%M%S%f") + '.xlsx' # 保存文件 file_path = 'outxlsx/' + file_name path = os.getcwd() + '/plugins/public/' file_path = os.path.join(path, file_path) filename.save(file_path) return ''
后端配合微信小程序实现消息订阅
消息通知是微信小程序里面必不可少的功能，在功能开发中是有很多此类的需求，比如下单成功发送付款通知，付款成功发送订单付款成功通知等。微信小程序的消息通知其实也经历过接口变更以及调用方式变更，以前是直接调用，现在变成了订阅式调用，一次订阅对应一次推送。具体可以查看微信接口文档：微信消息订阅文档具体流程是小程序进行消息订阅：后端使用对应的模板id实现相应的消息调用：[scode type="blue"]后端实现[/scode]后端使用python的flask进行逻辑实现。def send_template_msg(openid: str, template_id: str, value): """ 发送模板消息 :param openid:接收者（用户）的 openid :param template_id:微信消息提醒模板编号 :param value:模板内容，不填则下发空模板 :return: """ # 发送模板消息 url = f'https://api.weixin.qq.com/cgi-bin/message/subscribe/send?access_token=xxxxxx' data = {"touser": openid, "template_id": template_id, "lang": "zh_CN", "data": value} r = requests.post(url, json=data).content return r def send()： value = { "thing2": { "value": name # 申请人 }, "phone_number3": { "value": phone # 手机号 }, "time4": { "value": create_time.strftime('%Y-%m-%d') # 申请时间 } } result = send_template_msg(user.user_openid, config.TEMPLATE_ID_1, value) result = json.loads(result.decode()) if result.get('errcode') != 0: raise ViewException(error_code=result.get('errcode'), message=result.get('errmsg'))
在微信开发中，会有保存手机号的需求，流程是小程序端调用api获取手机号，再发送到后端保存，注意这个手机号是加密的，这里需要后端在获取后解密。这里用python的flask作为后端实现。[scode type="blue"]表单验证：[/scode]这里需要传入验证参数encryptedData和vinumclass DecodeUserPhoneForm(BaseForm): """解密用户手机号""" encryptedData = wtforms.StringField( validators=[ DataRequired()]) vinum = wtforms.StringField( validators=[ DataRequired()])[scode type="blue"]然后是逻辑处理：[/scode]class WXBizDataCrypt: def __init__(self, appId, sessionKey): self.appId = appId self.sessionKey = sessionKey def decrypt(self, encryptedData, iv): # base64 decode sessionKey = base64.b64decode(self.sessionKey) encryptedData = base64.b64decode(encryptedData) iv = base64.b64decode(iv) cipher = AES.new(sessionKey, AES.MODE_CBC, iv) decrypted = json.loads(self._unpad(cipher.decrypt(encryptedData)).decode()) if decrypted['watermark']['appid'] != self.appId: raise Exception('Invalid Buffer') return decrypted def _unpad(self, s): return s[:-ord(s[len(s)-1:])] def decode_user_phone(): """解密用户手机号""" form = validators.DecodeUserPhoneForm(request.args).validate_() user = get_user_model() encryptedData = form.encryptedData.data + '==' iv = form.vinum.data + '==' pc = WXBizDataCrypt(config.MINI_PROGRAM_APPID, user.user_session_key) mobile_obj = pc.decrypt(encryptedData, iv) mobile = mobile_obj['phoneNumber'] print(mobile) data = { 'mobile': mobile } # 存储号码 todo return ''随即就能获取到解密后的手机号，再进行逻辑处理。
面试是一门学问，考验的是你的知识储备，临场发挥。对自己心仪的公司面试成功与否很大程度上靠的是对面试准备与否，是否熟悉相应的回答套路等。面试python岗位也是一样。一 面试流程介绍1.1 python后端职位分析职位分析：招聘信息看什么？1.岗位职责（业务是否感兴趣）2.职位要求（自己是否掌握，查漏补缺）3.公司技术栈（公司使用到哪些技术）看招聘描述，挖掘信息从招聘信息中我们能挖掘到什么？1.你对公司做的业务是否感兴趣2.职位要求中的知识技能是否掌握，面试有多大概率成功3.自己还有哪些知识技能需要查漏补缺有的放矢，针对性准备提升面试成功率1.针对公司的技术栈和要求编写不同的简历2.表现出对职位和公司业务的兴趣3.突出自己的技能优势，提升匹配度（技能与公司要求比较符合）1.2 面试流程与环节后端面试流程学生重基础，社招重项目1.一面问基础2.二面问项目3.三面问设计（设计后端）学生重基础项目经验少，基础很重要1.学历和成绩2.大学所学的计算机课程3.在校项目。实习经验社招重项目社招重视项目和设计1.参与过哪些项目？有没有知名项目2.在项目中承担的职责3.有没有系统设计经验？行为面试非技术性问题1.自我介绍2.口头表达能力3.沟通交流能力HR面试到这里拿到offer可能就很大啦1.薪资待遇（锚定效应，可以提出比期望薪资稍高的待遇）2.职业规划3.自我介绍，沟通交流等1.3 python后端技术栈python语言基础1.语言特点2.语法基础3.高级特性算法与数据结构1.常用算法和数据结构2.分析时间，空间复杂度3.实现常见数据结构和算法编程范式1.面向对象编程2.常用设计模式3.函数式编程操作系统1.常用的linux命令2.进程，线程3.内存管理网络编程1.常用TCP/IP/HTTP协议2.socket编程基础3.python并发库数据库1.mysql常考，索引优化2.关系型和Nosql的使用场景3.redis缓存Python web框架1.常用框架对比，restful2.wsgi原理3.web安全问题系统设计1.设计原则，如何分析2.后端系统常用组件（缓存、数据库、消息队列等）3.技术选型和实现（短网址、feed流系统）技术之外，软实力1.学习能力2.业务理解能力，沟通交流能力3.心态1.4 python初中级工程师技能要求和面试标准初级工程师1.扎实的计算机理论基础2.代码规范，风格良好3.能在指导下靠谱的完成业务需求中级工程师1.扎实的计算机基础和丰富的项目经验2.能独立设计和完成项目需求3.熟悉常用web组件（缓存、消息队列等），具有一定的系统设计能力软实力1.具有产品意识，技术引导产品2.沟通交流能力，团队协作能力3.技术领导能力和影响力面试造核弹，工作拧螺丝1.工作内容和业务紧密相关，需要先了解公司是做什么的2.平台决定成长（业务体量），还是要看公司的规模成都3.准备面试需要有的放矢，跟职位相匹配才行1.5简历书写与自我介绍表现出个人优势，突出关键信息1.基本信息（姓名，学校，学历，联系方式等）2.职业技能（编程语言，框架，数据库，开发工具等）3.关键项目经验（承担职责，用到了哪些技术）简历自我评价1.简历自我评价可有可无2.最终还是面试官评价3.如果要写保证尽量内容简介、态度真诚简历加分项1.知名项目经验2.技术栈比较匹配3.开源项目（github、技术博客、linux等）简历需要注意的1.内容精简，突出重点，一页就好2.注意格式，推荐用pdf，避免hr打开格式乱掉3.信息真实，不弄虚作假。技能要与岗位相符自我介绍说什么1.个人信息2.掌握的技术，参与过的项目3.应聘的岗位，表达对该岗位的看法和兴趣举例：个人信息：你好我是~~工作项目经历：之前就职于~，担任什么，参与过~，对~~比较熟悉求职意向：我希望应聘到这个岗位，~~自我介绍的准备1.早准备2.准备开场白讲稿，面试前多练习3.找一个同伴好友模拟面试，消除紧张心理1.6行为面试常见问题与回答技巧面试官会根据候选人过去的行为评测其胜任能力1.理论依据：行为的连贯性，在旧公司是这样，很可能在新公司也会这样2.人在相似的场景时会倾向于重复以往的行为模式3.评判人的业务能力，沟通交流能力，语言表达能力，抗压能力等行为面试的套路（需要重点注意，一般就是问简历上的项目）1.提问方式：说说你曾经~~~2.说说你做过的这个项目3.说说你碰到过的技术难题？你是如何解决的？有何收获？**star模型（需要按照这个来准备）1.情景（situation）：什么情景下发生的2.任务（task）：你是如何明确你的任务的3.行动（action）：采取了什么样的行动4.结果（result）：结果怎么样？学到了什么？**常见问题：面试官一般会问，你还有什么要问我的吗？1.你可千万别说没了，直接说没了说明你对这个岗位缺乏了解和兴趣2.表现出兴趣：问问工作内容（业务），技术栈（用到了什么技术），团队（团队有几个人），项目（这是什么项目，项目在公司的份量等）3.问自己的感兴趣的一些技术问题，架构问题聊天是个重要的软技能1.态度真诚，力求真实，不要弄虚作假2.言简意赅，突出重点，省略细枝末节。适当模拟练习3.采用STAR模型让回答更有条理二、python语言基础考察点 2.1 python语言基础常考题python是静态还是动态类型？是强类型还是弱类型？1.动态强类型语言（不少人误认为是弱类型）2.动态还是静态指的是编译器还是运行期确定类型3.强类型指的是不会发生隐式类型转换（python属于动态强类型语言）python作为后端语言的优缺点1.胶水语言，轮子多，应用广泛2.语言灵活，生产力高3.（缺点）性能不是很高，代码不是很好维护，python2和3的兼容性问题什么是鸭子类型？1.关注点在对象的行为，而不是类型（duck typeing）2.比如file,StringIO,socket都支持read/write方法，都能操作文件3.实现了相同的魔法方法，一定程度上来说，就是一个类型的，如定义了__iter__魔法方法，就可以用for来迭代什么是monkey patch（猴子补丁）？ 哪些地方用到了？自己如何实现？1.所谓的monkey patch就是运行时替换2.比如gevent库需要修改内置的socket3.form gevent import monkey; maonkey.patch_socket()(也可以自己实现，也就是运行过程中替换原来的部分功能，实现别的功能)什么是python的自省（Introspection）？1.运行时判断一个对象的类型的能力2.python一切皆对象，用type；id；isinstance获取对象类型信息3.Inspect模块提供了更多获取对象信息的函数列表或者字典生成器1.比如[i for i in range(10) if i % 2 == 0] 生成0到10 里能被2整除的数2.一种快速生成list/dict/set的方式，用来替代map/filter3.（i for i in range(10) if i % 2 == 0）生成的就是生成器2.2 python2和3差异Python3的改进1.print变成了函数而不是关键字2.编码问题。Python3不再有unicode对象，默认str就是unicode3.除法变化。Python3除号返回浮点数。而不像python2的地板除法4.优化的super（）直接调用父类的方法5.高级解包操作 如：a,b,*rest = range(10)6.限定关键字参数，在方法参数特别多的情况下，可以指定参数，一面搞混def test(a, b, *, c){} test(1,2,c=3)7.python3重新抛出异常不会丢失原来栈的信息8.range，zip，map，dict.values都是返回迭代器Python3新增的1.yield from链接子生成器2.asyncio内置库，async/await原生协程支持异步编程3.新的内置库enum,mock,asyncio,ipaddress,concurrent.futures等Python3改进的1.生成的pyc文件统一放到__pycache__里2.一些内置库的修改。urllib，selector等3.性能优化等熟悉一些兼容2/3的工具1.six模块2to3等工具转换代码__future__模块2.3 python函数常考题python如何传参？1.是传递值还是传递引用呢？都不是。唯一支持的参数传递是共享传参2.call by object（共享对象），就是形参实参都指向同个对象3.call by sharing（共享传参），函数形参获得函数实参上各个引用的副本。4.python里面传递参数都是通过对象引用来传递的，也就是通过传递对象引用，如果操作的是可变对象，就可以直接在原来对象的基础上修改；如果操作的是不可变对象，那就相当于对对象进行一次复制拷贝，每次都不回修改到原来的对象数据。函数传递中args，kwargs含义是什么*1.用来处理可变参数2.*args被打包成tuple3.**kwargs被打包成dict（关键字参数要放在最后） 2.4 python异常机制常考题什么时候需要捕获异常？1.网络请求（超时，连接错误）2.资源访问（权限问题，资源不存在）3.代码逻辑（越界访问，KeyError）如何自定义自己的异常？为什么需要钉子自己的异常？1.因为有些业务异常找不到合适的异常来抛。2.继承Exception实现自定义的异常，不要继承BaseException3.给异常加上一些附加信息4.处理一些业务相关的特定异常（raise MyException）#Eg： Class MyException(Exception): Pass Try: Raise MyException(‘MyException’) Except MyException as e: Print(e)2.5 python性能剖析与优化，GIL常考题什么是GIL？1.Cpython解释器的内存管理并不是线程安全的2.保护多线程情况下对python对象的访问3.Cpython使用简单的锁机制避免多个线程同时执行字节码GIL的影响（限制了程序的多喝执行）1.同一个时间只能有一个线程执行字节码2.CPU密集程序难以利用多核优势，因为多核为了争抢执行，会互相拥挤争抢，导致白白浪费资源3.IO期间会释放GIL，所以对IO密集程序影响不大如何规避GIL的影响1.CPU密集型可以使用多进程+进程池2.IO密集可以使用多线程/协程（多进程其实系统开销更大，不一定比多线程效率高）3.使用cpython扩展（转化为c语言来执行）即便有了GIL还要关注线程安全，非原子操作都要注意（什么操作才是原子的？）1.一个操作如果是一个字节指令可以完成的就是原子的2.原子的是可以保证线程安全的，因为不是一步到位的话，中间数据还是可能会被其他线程给污染3.可以使用dis操作来分析字节码使用各种profile工具（内置或第三方）剖析程序性能（了解）1.二八定律，大部分时间耗时在少量代码上2.内置的profile/cprofile等工具3.使用pyflame（uber开源）的火焰图工具服务端性能优化措施，也就是着陆点注意：Web应用语言不会成为瓶颈1.数据结构与算法优化2.数据库层：索引优化，慢查询消除，批量操作减少IO，NoSql3.网络IO：批量操作，pipeline操作减少IO4.缓存：使用内存数据库redis/memcached5.异步：asyncio，celery6.并发：gevent/多线程2.6 python生成器与协程什么是生成器（Generator）？1.生成器就是可以生成值得函数2.当一个函数里有了yield关键字就成了生成器3.生成器可以挂起执行并且保持当前执行的状态，也就是说区别于普通函数的return，生成器返回值后，记录了函数执行的状态Python2没有原生的协程，只有基于生成器的协程1.生成器可以通过yield暂停执行和产出数据2.同时支持send()向生成器发送数据throw()向生成器抛出异常，如果生成器函数中yield是在一个等式中，这代表着可以send发送数据过去协程的注意点1.协程需要使用send(None)或者next(coroutine)来预激(prime)才能启动2.在yield处协程会暂停执行3.单独的yield会产出值给调用方4.可以通过coroutine.send(value)来给协程发送值，发送的值会赋值给yield表达式左边的变量value=yield5.协程执行完成后(再继续调用，没有遇到下一个yield语句)会抛出StopIteration异常Python3原生协程Python3.5引入async/await支持原生协程，结合asyncio3.7 python单元测试什么是单元测试？1.针对程序模块进行正确性检验2.一个函数，一个类进行验证3.自底向上保证程序正确性，从一个函数一个类最细粒度来写单元测试为什么要写单元测试？注意：三无代码不可取（无文档，五注释，无单测）1.保证代码逻辑的正确性2.单测影响设计，一般来说容易测试的到吗往往是高内聚低耦合的，容易测试的代码质量更好3.回归测试，防止改一处整个服务整个服务不可用单元测试相关的库1.nose/pytest比较常用2.mock模块用来模拟替换网络请求3.coverage用来统计测试覆盖率如下测试案例：三。python算法与数据结构考察点 3.1 python常用内置算法与数据结构常考题仔细回想下你用过的哪些内置的算法和数据结构1.sorted2.dict，list，set，tuple具体分类如下图：collections模块提供了一些内置数据结构的扩展（重要）1.namedtuple：元组元素命名方式来定义Import collections Point = collections.namedtuple(‘Point’, ‘x,y’) p = Point(1,2) print(p.x)2.deque：实现队列de collections.deque() de.append(1) de.appendleft(0) de.pop() de.popleft()3.Counter：Counter类的目的是用来跟踪值出现的次数。它是一个无序的容器类型，以字典的键值对形式存储，其中元素作为key，其计数作为value。计数值可以是任意的Interger（包括0和负数）。Counter类和其他语言的bags或multisets很相似。c = collections.Counter() c = collections.Counter(‘abcad’) print(c) print(c.most_common())4.OrderedDict：记住字典添加顺序od = collections.OrderedDict() od[‘c’] = ‘c’ od[‘a’] = ‘a’ od[‘b’] = ‘b’ print(list(od.keys())) #[‘c’, ‘a’, ‘b’]5.defaultdict：字典具有默认值dd = collections.defaultdict(int) dd[‘a’] #0 dd[‘b’] += 1 print(dd) #defaultdict(int, {‘a’:0, ‘b’:1})python中dict的底层结构使用的是哈希表1.为了支持快速查找使用了哈希表作为底层结构2.哈希表平均查找时间复杂度为O(1)3.Cpython解释器使用二次探查解决哈希冲突问题**注意后续延展：如何解决哈希冲突问题？探查法，二次计算查找可用的内存块；链接法，同一个内存块做链表链接哈希表如何扩容的？**python中的list和tuple的异同1.都是线性结构，都支持下表访问2.都是序列类型，都能进行遍历3.list是可变对象，tuple是不可变对象，其保存到饿引用不可变4.list没法作为dict的键，tuple可以（可变对象不可hash）LRUCache，替换掉最近最少使用的对象（redis应用）1.缓存剔除策略，当缓存空间不够用的时候需要一种方式剔除key2.常见的有LRU,LFU等3.LRU通过使用一个循环双端队列不断把最新访问的key放到表头，这样剔除的时候只需要剔除最远处的key就好了实现如下图：4.利用python内置的dict+collections.OrderedDict实现5.dict用来做k/v键值对的缓存6.OrderedDict用来实现更新最新最近访问的key下面是具体代码实现：from collections import OrderedDict class LRUCache: def __init__(self, capacity=128): self.od = OrderedDict() self.capacity = capacity def get(self, key): # 每次访问更新最新使用的 key if key in self.od: val = self.od[key] self.od.move_to_end(key) return val else: return -1 def put(self, key, value): # 更新 k/v if key in self.od: del self.od[key] self.od[key] = value # 更新 key 到表头 else: # insert self.od[key] = value # 判断当前容量是否已经满了 if len(self.od) > self.capacity: self.od.popitem(last=False) 3.2 python面试常考算法（排序+查找，重中之重）1.常考排序算法：冒泡排序，快速排序，归并排序，堆排序2.线性查找，二分查找等3.能独立实现代码（手写），能够分析时间空间复杂度排序算法中的稳定性，什么是排序算法的稳定性？1.相同大小的元素在排序之后依然保持相对位置不变，就是稳定的2.r[i]=r[j]且r[i]在r[j]之前，排序之后r[i]依然在r[j]之前3.稳定性对于排序一个复杂结构，并且需要保持原有排序才有意义写出快速排序（步骤，示例）1.partiton：选择基准分割数组为两个子数组，小于基准和大于基准的2.对两个子数组分别快排3.递归合并结果def quicksort(array): # 递归出口 if len(array) < 2: return array else: pivot_index = 0 # 第一个元素作为pivot pivot = array[pivot_index] less_part = [ i for i in array[pivot_index+1:] if i <= pivot ] great_part = [ i for i in array[pivot_index+1:] if i > pivot ] return quicksort(less_part) + [pivot] + quicksort(great_part) def test_quicksort(): import random ll = list(range(10)) random.shuffle(ll) print(ll) assert quicksort(ll) == sorted(ll) test_quicksort() 写出归并排序，归并两个有序数组def merge_sorted_list(sorted_a, sorted_b): length_a, length_b = len(sorted_a), len(sorted_b) a = b = 0 new_sorted_seq = [] while a < length_a and b < length_b: if sorted_a[a] < sorted_b[b]: new_sorted_seq.append(sorted_a[a]) a += 1 else: new_sorted_seq.append(sorted_b[b]) b += 1 if a < length_a: new_sorted_seq.extend(sorted_a[a:]) else: new_sorted_seq.extend(sorted_b[b:]) return new_sorted_seq def test_merge_sorted_list(): a = [1, 2, 5] b = [0, 3, 4, 8] print(merge_sorted_list(a, b)) # 分治法三步走。注意递归出口 def mergesort(array): # 递归出口 if len(array) <= 1: return array else: mid = int(len(array)/2) left_half = mergesort(array[:mid]) right_half = mergesort(array[mid:]) return merge_sorted_list(left_half, right_half) def test_mergesort(): import random ll =list(range(10)) random.shuffle(ll) print(ll) assert mergesort(ll) == sorted(ll) test_mergesort() 实现堆排序请写出二分查找递归方式实现二分，注意递归出口def binary_search(array, target): # 二分查找 if not array: return -1 beg, end = 0, len(array) while beg < end: mid = beg + (end - beg) // 2 # py3 if array[mid] == target: return mid elif array[mid] > target: end = mid else: beg = mid + 1 return -1 def test(): """ 如何设计测试用例：(等价类划分) - 正常值功能测试 - 边界值（比如最大最小，最左最右值） - 异常值（比如 None，空值，非法值） """ # 正常值，包含有和无两种结果 assert binary_search([0, 1, 2, 3, 4, 5], 1) == 1 assert binary_search([0, 1, 2, 3, 4, 5], 6) == -1 assert binary_search([0, 1, 2, 3, 4, 5], -1) == -1 # 边界值 assert binary_search([0, 1, 2, 3, 4, 5], 0) == 0 assert binary_search([0, 1, 2, 3, 4, 5], 5) == 5 assert binary_search([0], 0) == 0 # 异常值 assert binary_search([], 1) == -1 3.3 python数据结构常考题常考题型1.常见的数据结构链表，队列，栈，二叉树，堆2.使用内置结构实现高级数据结构，比如内置的list/deque实现栈3.Leetcode或者《剑指offer》上的常见题常考数据结构之链表（链表有单链表，双链表，循环双端链表）1.如何使用python来表示链表结构2.实现链表常见操作，比如插入节点，反转链表，合并多个链表等3.Leetcode联系常见链表题目如何实现反转链表（单链表倒置）# Definition for singly-linked list. # class ListNode: # def __init__(self, x): # self.val = x # self.next = None class Solution: def reverseList(self, head): """ :type head: ListNode :rtype: ListNode """ pre = None cur = head while cur: nextnode = cur.next cur.next = pre pre = cur cur = nextnode return pre 常见数据结构之队列（队列queue是先进先出的结构）1.如何使用python实现队列？2.实现队列的apend和pop操作，如何做到先进先出3.使用python的list或者collections.deque实现队列实现队列（利用collections.deque）# 实现队列。使用 deque from collections import deque class Queue: def __init__(self): self.items = deque() def append(self, val): return self.items.append(val) def pop(self): return self.items.popleft() def empty(self): return len(self.items) == 0 def test_queue(): q = Queue() q.append(0) q.append(1) q.append(2) print(q.pop()) print(q.pop()) print(q.pop()) test_queue() 常考数据结构之栈（栈stack是后进先出结构）1.如何使用python实现栈？2.实现栈的push和pop操作，如何做到后进先出的3.同样也可以用python list 或者collections.deque实现栈实现一个栈思考练习题：如何用两个栈实现队列？实现获取最小值的栈MIniStack常考数据结构之字典与集合（python dict/set底层都是哈希表）1.哈希表的实现原理，底层其实就是一个数组2.根据哈希函数快速定位一个元素，平均查找O(1)3.不断加入元素会引起哈希表重新开辟空间，拷贝之前元素到新数组哈希表如何解决冲突（链接法和开放寻址法）1.元素key冲突之后使用一个链表填充相同key的元素2.开放寻址法是冲突之后根据一种方式（二次探查）寻找下一个可用的槽3.cpython使用的是二次探查常考数据结构之二叉树（先序，中序，后序遍历）1.先(根)序：先处理根，之后是左子树，然后是右子树2.中(根)序：先处理左子树，然后是根，然后是右子树3.后(根)序：先处理左子树，然后是右子树，最后是根树的遍历方式先(根)序遍历：中序遍历：后序遍历：最后遍历根，上图挪下位置常考数据结构之堆（堆其实是完全二叉树，有最大堆和最小堆）1.最大堆：对于每个非叶子节点V，V的值都比它的两个孩子大2.最大堆支持每次pop操作获取最大的元素，最小堆获取最小元素3.常见问题：用堆来完成topk问题，从海量数字中寻找最大的k个使用堆解决TOPK问题import heapq class TopK: """获取大量元素 topk 大个元素，固定内存 思路： 1. 先放入元素前 k 个建立一个最小堆 2. 迭代剩余元素： 如果当前元素小于堆顶元素，跳过该元素（肯定不是前 k 大） 否则替换堆顶元素为当前元素，并重新调整堆 """ def __init__(self, iterable, k): self.minheap = [] self.capacity = k self.iterable = iterable def push(self, val): if len(self.minheap) >= self.capacity: min_val = self.minheap[0] if val < min_val: # 当然你可以直接 if val > min_val操作，这里我只是显示指出跳过这个元素 pass else: heapq.heapreplace(self.minheap, val) # 返回并且pop堆顶最小值，推入新的 val 值并调整堆 else: heapq.heappush(self.minheap, val) # 前面 k 个元素直接放入minheap def get_topk(self): for val in self.iterable: self.push(val) return self.minheap def test(): import random i = list(range(1000)) # 这里可以是一个可迭代元素，节省内存 random.shuffle(i) print(i) _ = TopK(i, 10) print(_.get_topk()) # [990, 991, 992, 996, 994, 993, 997, 998, 999, 995] test() 3.4 python数据结构常考题之链表链表涉及到指针操作较为复杂，容易出错，经常用作考题1.熟悉链表的定义和常见操作2.常考题：删除一个链表节点3.常考题：合并两个有序链表删除一个链表节点# Definition for singly-linked list. # class ListNode: # def __init__(self, x): # self.val = x # self.next = None class Solution: def deleteNode(self, node): """ :type node: ListNode :rtype: void Do not return anything, modify node in-place instead. """ nextnode = node.next after_next_node = node.next.next node.val = nextnode.val node.next = after_next_node 合并两个有序链表# Definition for singly-linked list. class ListNode: def __init__(self, x): self.val = x self.next = None class Solution: def mergeTwoLists(self, l1, l2): """ :type l1: ListNode :type l2: ListNode :rtype: ListNode """ root = ListNode(None) cur = root while l1 and l2: if l1.val < l2.val: node = ListNode(l1.val) l1 = l1.next else: node = ListNode(l2.val) l2 = l2.next cur.next = node cur = node # l1 或者 l2 可能还有剩余元素 cur.next = l1 or l2 return root.next 3.5 python数据结构常考题之二叉树二叉树涉及到递归和指针操作，常结合递归考察1.二叉树的操作很多可以用递归的方式解决，不了解递归会比较吃力2.常考题：二叉树的镜像（反转二叉树）3.常考题：如何层序遍历二叉树（广度优先）实现反转二叉树# Definition for a binary tree node. # class TreeNode: # def __init__(self, x): # self.val = x # self.left = None # self.right = None class Solution: def invertTree(self, root): """ :type root: TreeNode :rtype: TreeNode """ if root: root.left, root.right = root.right, root.left self.invertTree(root.left) self.invertTree(root.right) return root 实现层序遍历二叉树# Definition for a binary tree node. # class TreeNode: # def __init__(self, x): # self.val = x # self.left = None # self.right = None class Solution: def levelOrder(self, root): """ :type root: TreeNode :rtype: List[List[int]] """ if not root: # NOTE: 注意 root 可能为空 return [] res = [] cur_nodes = [root] next_nodes = [] res.append([i.val for i in cur_nodes]) # [3] while cur_nodes or next_nodes: for node in cur_nodes: if node.left: next_nodes.append(node.left) if node.right: next_nodes.append(node.right) if next_nodes: res.append( [i.val for i in next_nodes] ) cur_nodes = next_nodes next_nodes = [] return res 3.6 python数据结构常考题之栈与队列1.熟练掌握用python的list或者collections.deque实现栈和队列2.常考题：用栈实现队列使用栈实现队列from collections import deque class Stack: def __init__(self): self.items = deque() def push(self, val): return self.items.append(val) def pop(self): return self.items.pop() def top(self): # 返回栈顶值 return self.items[-1] def empty(self): return len(self.items) == 0 class MyQueue: def __init__(self): """ Initialize your data structure here. """ self.s1 = Stack() self.s2 = Stack() def push(self, x): """ Push element x to the back of queue. :type x: int :rtype: void """ self.s1.push(x) def pop(self): """ Removes the element from in front of queue and returns that element. :rtype: int """ if not self.s2.empty(): return self.s2.pop() while not self.s1.empty(): val = self.s1.pop() self.s2.push(val) return self.s2.pop() def peek(self): """ Get the front element. :rtype: int """ if not self.s2.empty(): return self.s2.top() while not self.s1.empty(): val = self.s1.pop() self.s2.push(val) return self.s2.top() def empty(self): """ Returns whether the queue is empty. :rtype: bool """ return self.s1.empty() and self.s2.empty() def test(): q = MyQueue() q.push(1) q.push(2) q.push(3) print(q.pop()) print(q.pop()) print(q.pop()) test() 3.7 python数据结构常考题之堆堆的常考题基本围绕在合并多个有序（数组/链表）topk问题1.理解堆的概念，堆是完全二叉树，有最大堆和最小堆2.会使用python内置的heapq模块实现堆的操作3.常考题：合并k个有序链表合并k个有序链表# Definition for singly-linked list. # class ListNode: # def __init__(self, x): # self.val = x # self.next = None from heapq import heapify, heappop class Solution: def mergeKLists(self, lists): """ :type lists: List[ListNode] :rtype: ListNode """ # 读取所有节点值 h = [] for node in lists: while node: h.append(node.val) node = node.next # 构造一个最小堆 if not h: return None heapify(h) # 转换成最小堆 # 构造链表 root = ListNode(heappop(h)) curnode = root while h: nextnode = ListNode(heappop(h)) curnode.next = nextnode curnode = nextnode return root 3.8 python字符串常考算法题了解常用的字符串操作1.python内置了很多字符串操作，比如split，upper，replace等2.常考题：翻转一个字符串3.常考题：判断一个数字是否是回文数翻转一个字符串（使用的是双端遍历法）class Solution: def reverseString(self, s): """ :type s: List[str] :rtype: void Do not return anything, modify s in-place instead. """ beg = 0 end = len(s)-1 while beg < end: s[beg], s[end] = s[end], s[beg] beg += 1 end -= 1 判断一个数字是否是回文数（也是双端遍历）class Solution: def isPalindrome(self, x): """ :type x: int :rtype: bool """ if x < 0: return False s = str(x) beg, end = 0, len(s)-1 while beg < end: if s[beg] == s[end]: beg += 1 end -= 1 else: return False return True def test(): s = Solution() assert s.isPalindrome(121) is True assert s.isPalindrome(-1) is False assert s.isPalindrome(1) is True test() 四.编程范式考察点 4.1 面向对象基础及python类常考问题什么是面向对象（oop）1.把对象作为基本单元，把对象抽象成类（Class），包含成员（属性）和方法2.数据封装（包含属性和方法），继承（继承父类方法），多态（重写父类方法）3.python中使用类来实现面向对象python中如何创建类组合和继承，优先使用组合而非继承1.组合是使用其他的类实例作为自己的一个属性（Has-a的关系）2.子类继承父类的属性和方法（Is-a 的关系）3.优先使用组合保持代码简单使用组合的例子：课程源码：inherit_combination.py类变量和实例变量的区别1.类变量由所有实例共享2.实例变量由实例单独享有，不同实例之间不影响3.当我们需要在一个类的不同实例之间共享变量的时候使用类变量classmethod/staticmethod的异同1.都可以通过Class.method()方式使用2.classmethod第一个参数是cls，可以引用类变量3.staticmethod使用起来和普通函数一样，只不过放在类里面组织什么是元类，使用场景是什么？（元类-Meta Class是创建类的类）1.元类允许我们控制类的生成，比如修改类的属性等2.使用type来定义元类3.元类最常见的一个使用场景就是ORM框架class Base: pass class Child(Base): pass # 等价定义 注意Base后要加上逗号否则就不是tuple了 SameChild = type('Child', (Base,), {}) # 加上方法 class ChildWithMethod(Base): bar = True def hello(self): print('hello') def hello(self): print('hello') # 等价定义 ChildWithMethod = type( 'ChildWithMethod', (Base,), {'bar': True, 'hello': hello} ) # 元类继承自 type class LowercaseMeta(type): """ 修改类的属性名称为小写的元类 """ def __new__(mcs, name, bases, attrs): lower_attrs = {} for k, v in attrs.items(): if not k.startswith('__'): # 排除magic method lower_attrs[k.lower()] = v else: lower_attrs[k] = v return type.__new__(mcs, name, bases, lower_attrs) class LowercaseClass(metaclass=LowercaseMeta): # py3 BAR = True def HELLO(self): print('hello') print(dir(LowercaseClass)) # 你会发现"BAR"和"HELLO"都变成了小写 # 用一个类的实例调用hello方法，我们修改了类定义时候的属性名!!! LowercaseClass().hello() 4.2 装饰器面试常考问题什么是装饰器（Decorator）1.python中一切皆对象，函数也可以当做参数传递2.在不修改原函数的情况下，使用装饰器给该函数添加其他功能，有利于代码重用3.装饰器是接受函数作为参数，添加功能后返回一个新函数的函数（类）4.python中通过@使用装饰器编写一个记录函数耗时的装饰器import time def log_time(func): # 接受一个函数作为参数 def _log(*args, **kwargs): beg = time.time() res = func(*args, **kwargs) print('ues time:{}'.format(time.time()-beg)) return res return _log # @log_time # @装饰器语法糖 def mysleep(): time.sleep(1) newsleep = log_time(mysleep) newsleep() 如何使用类编写装饰器，以及给装饰器添加参数？（使用__call__魔法函数）import time class LogTime: def __init__(self, use_int=False): self.use_int = use_int def __call__(self, func): def _log(*args, **kwargs): beg = time.time() res = func(*args, **kwargs) if self.use_int: print('use time: {}'.format( int(time.time()-beg)) ) else: print('use time: {}'.format( time.time()-beg) ) return res return _log @LogTime(True) def mysleep(): time.sleep(1) if __name__ == '__main__': mysleep() 4.3 设计模式：创建型模式python应用面试题常见的创建型设计模式1.工厂模式(Factory)：解决对象创建问题（重点）2.构造模式(Builder)：控制复杂对象的创建（重点）3.原型模式(Prototype):通过原型的克隆创建新的实例4.单例模式(Borg/Singleton)：一个类只能有一个实例，只能创建同一个对象（重点）5.对象池模式(Pool)：预先分配同一类型的一组实例6.惰性计算模式(Lazy Evaluation)：延迟计算(python 的property)工厂模式什么是工厂模式（Factory）1.解决对象创建问题2.解耦对象的创建和使用3。包括工厂方法和抽象工厂构造模式什么是构造模式（Builder）1.用来控制复杂对象的构造2.创建和表示分离，比如你要买电脑。工厂模式直接给你电脑，但构造模式允许你自定义电脑的配置，组装完成后给你原型模式什么是原型模式（Prototype）1.通过克隆原型来创建新的实例2.可以使用相同的原型，通过修改部分属性来创建新的实例3.用途：对于一些创建实例开销比较大的地方可以使用原型模式单例模式单例模式是一种常用的软件设计模式。在它的核心结构中只包含一个被称为单例类的特殊类。通过单例模式可以保证系统中一个类只有一个实例而且该实例易于外界访问，从而方便对实例个数的控制并节约系统资源。如果希望在系统中某个类的对象只能存在一个，单例模式是最好的解决方案。单例模式的实现有很多方式：__new__()在__init__()之前被调用，用于生成实例对象。利用这个方法和类的属性的特点可以实现设计模式的单例模式。单例模式是指创建唯一对象，单例模式设计的类只能实例 （这个绝对常考啊.绝对要记住1~2个方法,当时面试官是让手写的）1.使用__new__方法class Singleton(object): def __new__(cls, *args, **kw): if not hasattr(cls, '_instance'): orig = super(Singleton, cls) cls._instance = orig.__new__(cls, *args, **kw) return cls._instance class MyClass(Singleton): a = 12.共享属性创建实例时把所有实例的__dict__指向同一个字典,这样它们具有相同的属性和方法.class Borg(object): _state = {} def __new__(cls, *args, **kw): ob = super(Borg, cls).__new__(cls, *args, **kw) ob.__dict__ = cls._state return ob class MyClass2(Borg): a = 13.装饰器版本def singleton(cls): instances = {} def getinstance(*args, **kw): if cls not in instances: instances[cls] = cls(*args, **kw) return instances[cls] return getinstance @singletonclass MyClass: ...4.import方法作为python的模块是天然的单例模式# mysingleton.pyclass My_Singleton(object): def foo(self): pass my_singleton = My_Singleton() # to usefrom mysingleton import my_singleton my_singleton.foo()5.共享同一个实例的方法源码：# 一个工厂方法的例子 class DogToy: def speak(self): print("wang wang") class CatToy: def speak(self): print("miao miao") def toy_factory(toy_type): if toy_type == 'dog': return DogToy() elif toy_type == 'cat': return CatToy() # 一个构造模式的例子 class Computer: def __init__(self, serial_number): self.serial = serial_number self.memory = None # in gigabytes self.hdd = None # in gigabytes self.gpu = None def __str__(self): info = ('Memory: {}GB'.format(self.memory), 'Hard Disk: {}GB'.format(self.hdd), 'Graphics Card: {}'.format(self.gpu)) return '\n'.join(info) class ComputerBuilder: def __init__(self): self.computer = Computer('AG23385193') def configure_memory(self, amount): self.computer.memory = amount def configure_hdd(self, amount): self.computer.hdd = amount def configure_gpu(self, gpu_model): self.computer.gpu = gpu_model class HardwareEngineer: def __init__(self): self.builder = None def construct_computer(self, memory, hdd, gpu): self.builder = ComputerBuilder() [step for step in (self.builder.configure_memory(memory), self.builder.configure_hdd(hdd), self.builder.configure_gpu(gpu))] @property def computer(self): return self.builder.computer # 使用buidler，可以创建多个builder类实现不同的组装方式 engineer = HardwareEngineer() engineer.construct_computer(hdd=500, memory=8, gpu='GeForce GTX 650 Ti') computer = engineer.computer print(computer) # 单例模式 class Singleton: def __new__(cls, *args, **kwargs): if not hasattr(cls, '_instance'): _instance = super().__new__(cls, *args, **kwargs) cls._instance = _instance return cls._instance class MyClass(Singleton): pass c1 = MyClass() c2 = MyClass() assert c1 is c2 # 单例的，c1 c2 同一个实例 4.4 设计模式：结构型模式python应用面试题常见结构型设计模式1.装饰器模式(Decorator)：无需子类化扩展对象功能 （重要）2.代理模式(Proxy)：把一个对象的操作代理到另一个对象（重要）3.适配器模式(Adapter)：通过一个间接层适配统一接口（重要）4.外观模式(Facade)：简化复杂对象的访问问题5.享元模式(Flyweight)：通过对象复用(池)改善资源利用，比如连接池6.MVC模式(Model-View-Controller)：解耦展示逻辑和业务逻辑 （重要）代理模式什么是代理模式（Proxy）1.把一个对象的操作代理到另一个对象2.这里又要提到之前实现的Stack/Queue，把操作代理到deque3.通常使用has-a组合关系4.可用作校验用途from collections import deque class Stack(object): # 使用组合的例子 def __init__(self): self._deque = deque() # has a deque() def push(self, value): return self._deque.append(value) def pop(self): return self._deque.pop() def empty(self): return len(self._deque) == 0 def __iter__(self): res = [] for i in self._deque: res.append(i) for i in reversed(res): yield i s = Stack() s.push(1) s.push(2) for i in s: print(i) 适配器模式什么是适配器模式（Adapter）1.把不同对象的接口适配到同一接口2.想象一个多功能充电头，可以给不同的电器充电，充当了适配器3.当我们需要给不同的对象统一接口# 适配器模式的例子 class Dog(object): def __init__(self): self.name = "Dog" def bark(self): return "woof!" class Cat(object): def __init__(self): self.name = "Cat" def meow(self): return "meow!" class Adapter: def __init__(self, obj, **adapted_methods): """We set the adapted methods in the object's dict""" self.obj = obj self.__dict__.update(adapted_methods) def __getattr__(self, attr): """All non-adapted calls are passed to the object""" return getattr(self.obj, attr) objects = [] dog = Dog() objects.append(Adapter(dog, make_noise=dog.bark)) cat = Cat() objects.append(Adapter(cat, make_noise=cat.meow)) for obj in objects: print("A {0} goes {1}".format(obj.name, obj.make_noise())) 4.5 设计模式：行为型模式python应用面试题常见行为型设计模式1.迭代器模式(Iterator)：通过统一的接口迭代对象2.观察者模式(Observer)：对象发生改变的时候，观察者执行相应动作（常见订阅模式，事先订阅监听）3.策略模式(Strategy)：针对不同规模输入使用不同的策略迭代器模式迭代器模式(Iterator)1.python内置对迭代器模式的支持2.比如我们可以用for遍历各种Iterable的数据类型3.python里可以实现__next__和__iter__实现迭代器观察者模式1.发布订阅是一种最常用的实现方式2.发布订阅用于解耦逻辑3.可以通过回调等方式实现，当发生事件时，调用相应的回调函数策略模式策略模式（Strategy）1.根据不同的输入采用不同的策略2.比如买东西超过10个打八折，超过20个打七折3.对外暴露统一的接口，内部采用不同的策略计算# 发布订阅模式 class Publisher: # 发布者 def __init__(self): self.observers = [] # 观察者 def add(self, observer): # 加入观察者 if observer not in self.observers: self.observers.append(observer) else: print('Failed to add : {}').format(observer) def remove(self, observer): # 移除观察者 try: self.observers.remove(observer) except ValueError: print('Failed to remove : {}').format(observer) def notify(self): # 调用观察者的回调 [o.notify_by(self) for o in self.observers] class Formatter(Publisher): # 继承自发布者 def __init__(self, name): super().__init__() self.name = name self._data = 0 @property def data(self): return self._data @data.setter def data(self, new_value): self._data = int(new_value) self.notify() # data 在被合法赋值以后会执行notify class BinaryFormatter: """ 订阅者 """ def notify_by(self, publisher): print("{}: '{}' has now bin data = {}".format( type(self).__name__, publisher.name, bin(publisher.data)) ) def test(): df = Formatter('formatter') # 发布者 bf = BinaryFormatter() # 订阅者 df.add(bf) # 添加订阅者 df.data = 3 # 设置的时候调用订阅者的notify_by # 策略模式 class Order: def __init__(self, price, discount_strategy=None): self.price = price self.discount_strategy = discount_strategy def price_after_discount(self): if self.discount_strategy: discount = self.discount_strategy(self) else: discount = 0 return self.price - discount def __repr__(self): fmt = "<Price: {}, price after discount: {}>" return fmt.format( self.price, self.price_after_discount() ) def ten_percent_discount(order): return order.price * 0.10 def on_sale_discount(order): return order.price * 0.25 + 20 def main(): order0 = Order(100) order1 = Order(100, discount_strategy=ten_percent_discount) order2 = Order(1000, discount_strategy=on_sale_discount) print(order0) print(order1) print(order2) 4.6 python函数式编程常考题python支持部分函数式编程特性1.把电脑的运算视作数学上的函数计算（lambda演算）2.高阶函数：map/reduce/filter3.无副作用，相同的参数调用始终产生同样的结果什么是闭包(Closure)？1.绑定了外部作用域的变量的函数2.即使程序离开外部作用域，如果闭包仍然可见，绑定变量不会被销毁3.每次运行外部函数都会重新创建闭包五 操作系统考察点 5.1面试常考linux命令为什么要学linux？（大部分企业应用跑在linux server上）1.熟练在linux服务器上操作2.了解linux工作原理和常用工具3.需要了解查看文件，进程，内存相关的一些命令，用来调试和排查如何查询linux命令1.使用man命令查询用法。但是man手册比较晦涩2.使用工具自带的help，比如pip --help3.这里介绍一个man的替代工具tldr。pip install tldr文件/目录操作命令（掌握常见的文件操作工具）1.chown/chmod/chgrp2.ls/rm/cd/cp/mv/touch/rename/ln(软链接和硬链接等)3.locate/find/grep 定位查找和搜索文件查看（文件或者日志查看工具）1.编辑器 vi/nano2.cat/head/tail 查看文件3.more/less 交互式查看文件进程操作命令1.ps 查看进程（ps -ef|grep tomcat）2.kill -9 杀死进程3.top/htop 监控进程内存操作命令1.free 查看可用内存2.了解每一列的具体含义3.排查内存泄露问题掌握常见的网络工具1.ifconfig 查看网卡信息2.lsof/netstat 查看端口信息3.ssh/scp远程登录/复制。tcpdump抓包掌握常见用户和组的操作1.useradd/usermod2.groupadd/groupmod5.2 操作系统线程和进程常考面试题进程和线程对比1.进程是对运行时程序的封装，是系统资源调度和分配的基本单位2.线程是进程的子任务，cpu调度和分配的基本单位，实现进程内并发（python同一时间内只能执行一个线程，无法利用多核的优势，所以还是cpu不断切换线程并发执行）3.一个进程可以包含多个线程，线程依赖进程存在，并共享进程内存python哪些操作是线程安全的？1.一个操作可以在多线程环境中安全使用，获取正确的结果2.线程安全的操作好比线程是顺序执行的而不是并发执行的，达到线程隔离3.非原子性操作的线程容易使得数据变得不安全4.一般如果涉及到写操作需要考虑如何让多个线程安全的访问数据线程同步的方式1.互斥量（锁）：通过互斥机制防止多个线程同时访问公共资源2.信号量（Semphare）：控制同一时刻多个线程访问同一个资源的线程数3.事件（信号）：通过通知的方式保持线程同步（使用较少）进程间通信的方式1.管道/匿名管道/有名管道(pipe)2.信号(Signal)：比如用户使用Ctrl+c产生SIGINT程序终止信号3.消息队列(Message)4.共享内存(share memory)5.信号量(Semaphore)6.套接字(Socket)：最常用的方式python中如何使用多线程（threading模块）1.threading.Thread类用来创建线程2.start()方法启动线程3.可以用join()等待线程结束python中如何使用多进程（python有GIL，可以使用多进程实现cpu密集程序）1.multiprocessing多进程模块2.multiprocessing.Process类实现多进程3.一般用在cpu密集程序里，避免GIL的影响课程源码-多进程：multiprocess_test.py