Python 分析在德的中国程序员，告别 996 ？

出品 | 程序人生（ID:coder_life）

今年5月27日， 一位据说在德国的中国程序员@将记忆深埋在微博公布：

“半年时间，100多TB数据， 利用1024、91、sex8、PornHub、xvideos 等 站采集的数据对比Facebook、instagram、TikTok 、抖音、微博等 交媒体。我们在全球范围内成功识别了10多万从事不可描述行业的小姐姐。”

热炒之下，这套Deep Learning系统瞬间炸了锅，顺便炸翻了在德国处于懵逼状态的一众平时安安静静老老实实的程序员：我们身边竟然藏着这样一个人？！

这引起了我对在德中国籍程序员的行业分类以及专业方向等相关数据的好奇。长期以来德国一直面临着劳动力短缺，特别是工程技术方向，尤其是IT专业人才的极度缺乏，以至于德国政府将这些专业的人才获得欧盟蓝卡的最低年薪标准降到了税前41808欧元（2019）。换句话说，软件信息专业的同学毕业后在德国很容易找到工作，并且获得蓝卡工作居留许可。近年来身边来自印度，俄罗斯，中国的程序员也在逐年增加。那么中国程序员在德国到底从事那些行业呢？

蓝卡和德国程序员数据

先在 上找了一圈，没有找到特别针对中国籍程序员的数据分析，只找到关于蓝卡和在德国工作的程序员的数据分析。

蓝卡数据

2013-2018年，超过76000外籍人员持蓝卡在德国工作。2017年德国共有21727外国人申请蓝卡工作签证，其中中国国籍申请者占了近10%。这说明仅2017年，就有二千多中国籍雇员申请了蓝卡，这其中IT从业者占比未知。假设IT软件信息领域的中国雇员只占比其中10%，那么过去五年中就有约800名中国籍程序员拿到蓝卡。实际上根据生活和工作的接触，我保守估计在德中国籍程序员数量超过1500人。

据2016年数据，欧盟蓝卡签证的所在申请国，84%位于德国，可以说几乎整个欧盟的外国工程师都来德国找工作了。

在德国工作的程序员数据

据来自Stack Overflow的德国IT数据分析，2016年全德国有超过120000软件开发人员，2017年暴增超过820000。不过82万这个数字不可信，毕竟德国总人口才八千多万，如果是将近1%的占比，德国不至于一直闹码农荒。可信的十几万程序员中，软件开发方向数据如下：Web开发占比65.51%，系统管理员位居第二，数据库管理员第三。仅仅这三个方向就吃掉了75%的占比，为什么德国程序员看起来很偏科，爆火的机器学习和数据分析才各占4%左右。

这是因为德国IT行业大多为德国的支柱产业服务，如汽车、制药、机械、电子等，这些公司所需的企业内部管理软件如今多为SaaS构架，同时因为传统行业对云服务的怀疑和不信任态度，亦或安全原因，他们又维护着大量的企业私有服务器，和企业级数据库。所以不难理解前三甲总合占比之大。

虽说国内的移动开发趋势这两年有点弱，但德国的iOS和Android移动开发就从来没有强过，因为缺乏B2C土壤，传统企业一般也不重视移动开发（未必需要），相关产业很多都外包于东欧或者印度，中国的团队。

在德中国程序员数据分析

络上暂时没有发现任何关于这些可能存在的1500名中国程序员的数据，这就尴尬了，没数据怎么分析？

等等，平时管理的几个德国的IT行业微信群不就是最好的数据源？群友加起来也有500多人了，样本虽不大，但毕竟还是遵循正态分布的。不过必须用Python 3开发一套脚本来收集和处理相关数据。

在德中国程序员做什么

专业方向&工作领域&开发语言和框架的数据采集

如果使用匿名调查 告方式，扰民且又费时费力，此类信息只能从群昵称上打主意了，首先是发群公告规范群友昵称标准：

昵称|行业或专业领域|擅长开发框架或语言

举例：

小呆|学生|想找数据分析工作

中二|前端|nodejs, react

大傻|机器学习|nlp

老痴|自动驾驶|c++

大部分群友按标准改了昵称，但是还有一部分死硬派坚决不改，又不能经常发群消息提醒，只能开发机器人自动提醒了。微信机器人Wxpy是一个包装得非常简洁的微信个人 API， 在 itchat 的基础上，通过大量接口优化提升了模块的易用性，并进行丰富的功能扩展，一些常见的场景：

运行脚本时自动把日志发送到你的微信

群成员信息获取，邀请或者踢人

跨 或跨群转发消息

通过脚本和第三方API交互，比如图灵聊天机器人、智能办公、智能家居

开发需求

获取群聊群成员信息

找出所有群昵称不符合标准的群友

随机抽取5人，在群里发布改昵称提醒消息

同时将这次提醒的5人，存储进数据库

每天早八点晚八点两次定时启动昵称检查脚本

某人在将来被提醒次数超过10次，还不予配合不改昵称时，将自动踢出群

新群友被邀请进入群时，立刻发送群规提示改昵称

该任务所需第三方库如下：

pip3 installwxpy

pip3 installapscheduler

pip3 installpymysql

pip3 installDBUtils

1. 建库建表

因为需要存储微信表情字符集，所以表的默认编码采用utf8mb4_unicode_ci。

DROPTABLEIFEXISTS`wx_chat_group`;

CREATETABLE`wx_chat_group`(

`id`int(11) NOTNULLAUTO_INCREMENT,

`name`VARCHAR(64) COLLATEutf8mb4_unicode_ci NOTNULLDEFAULT”,

PRIMARY KEY`id`(`id`)

)

ENGINE= InnoDB

DEFAULTCHARSET= utf8mb4 COLLATEutf8mb4_unicode_ci;

INSERTINTO`wx_chat_group`(`id`, `name`) VALUES(1, ‘德国IT职业信息分享群’);

— 每次抽取的不合规格的昵称将存储如表以供计数

DROPTABLEIFEXISTS`wx_chat_nickname_check`;

CREATETABLE`wx_chat_nickname_check`(

`id`BIGINT(20) NOTNULLAUTO_INCREMENT,

`group_id`int(9) UNSIGNEDNOTNULL,

`wx_puid`VARCHAR(16) COLLATEutf8_unicode_ci NOTNULLDEFAULT”,

`nickname`VARCHAR(64) CHARACTERSETutf8mb4 COLLATEutf8mb4_unicode_ci NOTNULLDEFAULT”,

`create_time`timestampNOTNULLDEFAULTCURRENT_TIMESTAMPCOMMENT‘Create time’,

PRIMARY KEY`id`(`id`),

INDEX`idx_group_id`(`group_id`),

INDEX`idx_create_time`(`create_time`)

)

ENGINE= InnoDB

DEFAULTCHARSET= utf8mb4 COLLATEutf8mb4_unicode_ci;

2. 用户设置

所有用户自定义变量存入conf文件里，如群名、临时存储路径、数据库接入信息，踢人阈值：

[wechat]

group_name_1=德国IT职业信息分享群

group_id_1=1

path_tmp=/opt/tmp/

notice_random=5

kick_max=10

tuling_api_key=xxxxx

[mysql]

mysql_host=localhost

mysql_port=3306

mysql_user=root

mysql_pwd=xxxx

mysql_database=wechat_group_ibot

3. 监听群消息

初始化群聊对象，并且监听群消息

# 查找群聊，并且设置附加属性，以备后用

def init_group(group_name, group_id):

group = ensure_one(bot.groups.search(group_name))

group.ext_attr = lambda: None

setattr(group.ext_attr, ‘group_id’, group_id)

setattr(group.ext_attr, ‘group_name’, group_name)

returngroup

# 初始化微信机器人bot

bot = Bot(cache_path=True, console_qr=True)

# unique chat person’s id

bot.enable_puid

# 读取自定义参数

cf = configparser.ConfigParser

cf.read(‘wechat.conf’)

group_name_1 = cf.get(‘wechat’, ‘group_name_1’)

group_id_1 = cf.get(‘wechat’, ‘group_id_1’)

# 初始化群聊对象

group_1 = init_group(group_name_1, group_id_1)

# 监听类型为NOTE的群消息，如：”aa”邀请”bbb”加入了群聊

@bot.register(group_1, NOTE)

def welcome_for_group(msg):

try:

new_member_name = re.search(r’邀请”(.+?)”|”(.+?)”通过’, msg.text).group(1)

exceptAttributeError:

return

group_1.send(welcome_text.format(new_member_name, space_after_chat_at))

# 保持bot持续运行

bot.join

4. 昵称检查

检查群友昵称，存入数据库并且发送提醒, 具体逻辑代码这里不予累述。

def check_nickname(nickname):

# 正则检验群昵称是否标准

ifre.match(r'([一-龥]|[ -~]|[sS])+|([一-龥]|[ -~])+|([一-龥]|[ -~])+’, nickname):

returnTrue

else:

returnFalse

……

# 检查群友昵称

def process_group_members(group):

# 每次检查前先刷新群成员信息，避免用户改了昵称后再次被提醒

# 但刷新会改变成员临时的内部puid，所以检查昵称必须同时结合puid和nickname

group.update_group(members_details=False)

……

formember ingroup:

nickname = member.name

wx_puid = member.puid

ifnotcheck_nickname(nickname):

invalid_member = GroupMember(nickname, wx_puid, 0)

invalid_members.append(invalid_member)

…..

# 随机抽取不合格的5人

random_members = random.sample(invalid_members, k=5)

……

# 将本次提醒群友存入数据库，供下次计数

def insert_invalid_name(group_id, wx_puid, nickname):

bot_db.execute(“INSERT INTO wx_chat_nickname_check (`group_id`, `wx_puid`, `nickname`)”

” VALUES (%s, %s, %s)”,

(group_id, wx_puid, nickname))

# 获取昵称不合规群友被提醒计数

def get_invalid_name_count(group_id, wx_puid, nickname):

result = bot_db.get_count(“SELECT id FROM wx_chat_nickname_check “

“WHERE group_id = %s and (wx_puid = %s or nickname = %s)”, (group_id, wx_puid, nickname))

returnresult

5. 数据库连接池

这里的数据库连接使用了数据库连接池：DBUtils.PersistentDB

DBUtils.PooledDB: 适用于多线程频繁开启关闭数据库连接

DBUtils.PersistentDB：适用于单线程多次频繁连接数据库

如果不采用线程池而是采取直连，那么运行一段时间后，脚本将出现该错误

pymysql.err.OperationalError: 2006

这里将DBUtils再次封装了一下，写了一个单例模式BotDatabase， 提供了query(select), execute(update, delete) 以及批处理execute等常用接口。

6. 启动定时器

# 早八点晚八点各执行检查一次

def start_schedule_for_checking_member(group):

scheduler = BlockingScheduler

scheduler.add_job(lambda: process_group_members(group), ‘cron’, hour=8, minute=1, timezone=“Europe/Paris”)

scheduler.add_job(lambda: process_group_members(group), ‘cron’, hour=20, minute=1, timezone=“Europe/Paris”)

最终成果

已知问题

在消息中输入 @群员昵称 并不能真正让该群友收到@提示（显示推送提示），微信App里是在@群员昵称后自动加上了一个特殊的显示空白的字符u’?′。但是经测试，加上这个符 也不行，推测是微信Web API基于防范垃圾推送，屏蔽了群提示接口。

wxpy的bot在运行一段时间后会停止工作，出现连接服务器错误，必须重新登录，推测是微信Web API的Session安全机制导致的问题。

数据清洗

一段时间后大部分群友修改了昵称，于是有了在德中国程序员职业和专业方向的数据，经清洗后，导出CSV规格如下。

数据分析

该任务所需第三方库如下：

pip3 installpandas

pip3 installmatplotlib

pip3 installjieba

pip3 installwordcloud

pip3 installseaborn

pip3 installpalettable

开发需求

在德程序员男女比例

在德IT软件专业在职和学生比例

分析在德程序员所处行业和专业方向

程序猿和程序媛所处行业和专业方向对比

分析在德程序员常用开发语言和框架

程序猿和程序媛常用开发语言和框架对比

在职和学生常用开发语言和框架对比

1. 在德程序员男女比例，输出Pie Chart

defgen_pie_member_gender(self, csv_file):

df = pd.read_csv(csv_file, delimiter=‘ ‘, encoding=‘utf-8’)

genders = df[‘gender’]

col = [0, 0, 0]

forg ingenders:

ifg == 1:

col[0] = col[0] + 1

elifg == 2:

col[1] = col[1] + 1

else:

col[2] = col[2] + 1

perccent_male = ‘{0:.2f}%’.format((col[0]/len(genders) * 100))

perccent_female = ‘{0:.2f}%’.format((col[1]/len(genders) * 100))

perccent_unknown = ‘{0:.2f}%’.format((col[2]/len(genders) * 100))

labels = [r’Male %s’% perccent_male,

r’Female %s’% perccent_female,

r’Unknown %s’% perccent_unknown]

colors = [‘lightskyblue’, ‘pink’, ‘gold’]

plt.figure(figsize=(8, 6))

patches, texts = plt.pie(col, colors=colors, startangle=90)

plt.legend(patches, labels, loc=“best”)

plt.title(‘Gender of Member’)

# Set aspect ratio to be equal so that pie is drawn as a circle.

plt.axis(‘equal’)

plt.tight_layout

path_image = os.path.join(self.path_analyse,

‘%s_member_gender_pie.png’% self.group_id)

plt.savefig(path_image, format=‘png’, dpi=100)

plt.close

returnpath_image

分析：

在德中国程序猿和程序媛比率约为2：1，这个比例基本和中国籍蓝卡申请人男女比率持平。但是根据2018年中国程序员数据调查表，中国程序员群体中男女比例接近12:1。德国的各位猿，你们就偷乐吧。

2. 在德IT软件专业在职人员和学生比例，输出Pie Chart

代码和上面雷同。

分析：

IT信息行业在职工作人员和在读学生比率为9比1，绝大部分人是在职工作的。

3. 在德程序员所处行业和专业方向，输出词云

# 这里采用一个汉字停词库，近两千词

@staticmethod

defload_stopwords:

filepath = os.path.join(‘./assets’, r’stopwords_cn.txt’)

stopwords = [line.strip forline inopen(filepath, encoding=‘utf-8’).readlines]

returnstopwords

defgen_wordcloud_info_nicknames(self, csv_file, column=‘branch’, gender=‘all’):

df = pd.read_csv(csv_file, delimiter=‘ ‘, encoding=‘utf-8’)

stopwords = set(STOPWORDS)

stopwords.update(self.load_stopwords)