转载

Java程序员怎么快速掌握python？

对于Java程序员来说，已经有了一门编程语言的经验，相信入门python不是个问题，那如何能以更快的速度熟悉并掌握python呢？当然是更多的练习和使用。

工具

开始之前，当然得选择好用称手的工具。

pycharm

jetbrains系的IDE体验都不错，功能强大好用，界面美观，个人一直在用pycharm。

vscode

vscode小巧，插件丰富，界面简洁美观，很多前端开发使用，装上python环境，也能用，个人偶尔使用。

virtualenv

virtualenv用来给项目创建一个全新的虚拟python运行环境，安装的新包只用于当前项目，不会影响别的项目和全局python环境，有点像npm install是在当前项目下安装依赖，不影响全局的nodejs环境。

练习

如何更多的练习？Java程序员工作中项目用到的肯定都是Java，在工作中能用python的机会不多，但工作中总是会遇到各种重复性的事，这时就可以用python来自动化这些事。下面列举几个常见场景，辅以代码片断简要说明。

日志分析

知识点： 文件读写 、 字符串操作 、 列表操作 、 字典操作

#本例实现读取nginx的access日志文件，根据IP统计访问频次，取访问频次前10的IP打印并保存到文件。日志示例：
#127.0.0.1 - - [07/Jul/2019:22:14:04 +0800] "POST /api/test HTTP/1.1" client_body_timeout10

import collections

# 初始化一个带默认值的字典
dic = collections.defaultdict(int)
# 以UTF8编码打开日志文件，遍历日志文件的每一行
for line in open("access.log", encoding="UTF8"):
    # 每行文本使用空格分隔，生成列表
    parts = line.strip().split(" ")
    # IP是列表第一个元素
    ip = parts[0]
    # 计数加1
    dic[ip] += 1

# 字典按频次倒序排序，lambda是个匿名函数，指明排序时根据字典的值排序
sorted_list = sorted(dic.items(), key=lambda t: t[1], reverse=True)

# 打开文件用于写入
out = open("out.txt", mode="w", encoding="UTF8")

# 取前10
top10 = sorted_list[:10]

# 前10拼接成字符串
top10_str = "/n".join([str(item) for item in top10])

# 打印结果
print(top10_str)

# 结果写入到文件中
out.write(top10_str)

报表导出

知识点： 数据库操作 、 excel读写

# 本例实现从数据库查询数据，并导出到excel中
import os

import pymysql
from openpyxl import Workbook

# 创建数据库连接
# 默认如果cursorclass未指定，返回的查询结果数据类型是元组，设成DictCursor结果的数据类型会是字典{"field":"value"}，更便于使用
connection = pymysql.connect(host='localhost', user='root', passwd='admin', db='test', port=3306,
                             charset='utf8', cursorclass=pymysql.cursors.DictCursor)
# 可以理解成java里的statement，用于执行sql
cursor = connection.cursor()

# 执行sql查询
cursor.execute("select id,username from user")
# 获取返回的查询结果
users = cursor.fetchall()

out_file_name = "out.xlsx"

# 如果文件存在则删除
if os.path.exists(out_file_name):
    os.remove(out_file_name)

# 新建workbook
workbook = Workbook()
# 新建sheet
sheet = workbook.create_sheet("用户列表", index=0)
# 往sheet里添加一行标题
sheet.append(["用户ID", "用户名"])

# 往sheet里添加查询到的用户数据
for user in users:
    sheet.append([user["id"], user["username"]])

# 保存workbook并关闭
workbook.save(out_file_name)
workbook.close()

定时检测服务状态

知识点： 定时任务 、 http请求 、 钉钉通知

# 本例实现每5分钟请求一个http接口，然后把结果发到钉钉
import time

import schedule
from dingtalkchatbot.chatbot import DingtalkChatbot
import requests
from requests_toolbelt import SSLAdapter

# 钉钉的机器人webhook地址
webhook = 'https://oapi.dingtalk.com/robot/send?access_token=xxx'
# 创建钉钉机器人
bot = DingtalkChatbot(webhook)


def test():
    # https时需要这样执行，如果是http可以直接requests.get()
    adapter = SSLAdapter('TLSv1')
    session = requests.Session()
    session.mount('https://', adapter)
    # get请求，返回json数据，用字典表示
    result = session.get("https://api.bootcdn.cn/libraries/vue.min.json").json()
    # 钉钉发消息
    bot.send_text("test ok:" + result["name"])


# 每5分钟测试一下
schedule.every(5).minutes.do(test)

# schedule固定写法
while True:
    schedule.run_pending()
    time.sleep(1)

结束进程

知识点：执行命令或程序

# 本例实现检测未响应进程并重启
import os
import time

import schedule


# 解析命令输出
def parse_output(output):
    print(output)
    pid_list = []
    lines = output.strip().split("/n")
    if len(lines) > 2:
        for line in lines[2:]:
            pid_list.append(line.split()[1])
    return pid_list


def list_not_response(process_name):
    return list_process(process_name, True)


def list_process(process_name, not_respond=False):
    cmd = 'tasklist /FI "IMAGENAME eq %s"'
    if not_respond:
        cmd = cmd + ' /FI "STATUS eq Not Responding"'
    # 执行命令并获取命令的输出
    output = os.popen(cmd % process_name)
    return parse_output(output.read())


def start_program(program):
    os.popen(program)


def check_job():
    target_exe = "test.exe"
    not_respond_list = list_not_response(target_exe)
    if len(not_respond_list) <= 0:
        return
    pid_params = " ".join(["/PID " + pid for pid in not_respond_list])
    os.popen("taskkill /F " + pid_params)
    if len(list_process(target_exe)) <= 0:
        start_program(r'E:/bin/test.exe')


if __name__ == '__main__':
    schedule.every(5).seconds.do(check_job)
    while True:
        schedule.run_pending()
        time.sleep(1)