YYDS! Python 幫我扛起運(yùn)營(yíng)大旗！

文 | 李曉飛

來(lái)源：Python 技術(shù)「ID: pythonall」

最近參加了一個(gè)訓(xùn)練營(yíng)，作為副教練，承擔(dān)起訓(xùn)練營(yíng)的運(yùn)營(yíng)工作。事不大，活不少，打卡記錄、活動(dòng)積分、獎(jiǎng)勵(lì)制度、評(píng)優(yōu)方案、趨勢(shì)對(duì)比，應(yīng)有盡有……

開始認(rèn)為 Excel 就足夠應(yīng)付，沒(méi)想到第一項(xiàng)工作 —— 人員匯總，就把我難倒了，于是果斷拎起 Python 這把大刀，披荊斬棘，利用業(yè)余時(shí)間，不到一周竟然打造出了一套運(yùn)營(yíng)管理系統(tǒng)，到底是如何做的呢？一起來(lái)看。

基礎(chǔ)整理

數(shù)據(jù)是運(yùn)營(yíng)的基礎(chǔ)，人員數(shù)據(jù)是基礎(chǔ)數(shù)據(jù)，首先需要搞定人員信息。

訓(xùn)練營(yíng)里的人員信息來(lái)自多個(gè)渠道，有通過(guò) APP 報(bào)名的，有調(diào)查問(wèn)卷收集的，還有人工錄取的。

加上同一個(gè)可能在不太的地方用不一樣的名字，以及不同渠道收集的數(shù)據(jù)完整性不同，所以整理基礎(chǔ)數(shù)據(jù)工作耗費(fèi)了將近兩天時(shí)間。

最初用 Excel 的 VLookup 做數(shù)據(jù)合并，但靈活度小，限制大，放棄了。

最后使用 Python 處理各個(gè)渠道的數(shù)據(jù)，再錄入了數(shù)據(jù)庫(kù)，完成了數(shù)據(jù)整理工作。

這里重點(diǎn)說(shuō)一下數(shù)據(jù)庫(kù)。

使用數(shù)據(jù)庫(kù)的好處是，方便數(shù)據(jù)整合、統(tǒng)計(jì)和更新。但是數(shù)據(jù)庫(kù)一般比較重，維護(hù)部署都是問(wèn)題，于是選用了文本數(shù)據(jù)庫(kù) SQLite[1] 作為數(shù)據(jù)庫(kù)。

SQLite 很輕，不需要服務(wù)器，但功能與 MySQL[2] 類似。

使用起來(lái)安裝 Python 的 SQLite 模塊就可以了：

pip install sqlite3

創(chuàng)建數(shù)據(jù)庫(kù)鏈接：

import sqlite3
conn = sqlite3.connect('database.db')

其中 database.db 就是一個(gè)普通文件，如果沒(méi)有，會(huì)自動(dòng)創(chuàng)建一個(gè)。

有了鏈接，就可以執(zhí)行數(shù)據(jù)庫(kù)操作了，比如創(chuàng)建一個(gè)庫(kù)表,插入數(shù)據(jù)：

# 創(chuàng)建一個(gè)游標(biāo)
cur = con.cursor()

# 執(zhí)行SQl 語(yǔ)句創(chuàng)建庫(kù)表
cur.execute('''CREATE TABLE stocks
               (date text, trans text, symbol text, qty real, price real)''')

# 向庫(kù)表中插入數(shù)據(jù)
cur.execute("INSERT INTO stocks VALUES ('2006-01-05','BUY','RHAT',100,35.14)")

# 提交更改結(jié)果
con.commit()

# 關(guān)閉鏈接
con.close()

因?yàn)樾枰l繁地操作數(shù)據(jù)庫(kù)，所以將這些操作寫成一個(gè)類：

class DBSqlite:
    def __init__(self, db):
        super().__init__()
        self.db = db
        self._conn = None

    def __del__(self):
        if self._conn:
            self._conn.close()
        
    def _get_conn(self):
        if not self._conn:
            self._conn = sqlite3.connect(self.db)
            self._conn.row_factory = sqlite3.Row
        return self._conn

    def _query(self, sql):
        conn = self._get_conn()
        cur = conn.cursor()
        rows = []
        for row in cur.execute(sql):
            rows.append(row)
        cur.close()
        return rows

    def de(self, sql):
        conn = self._get_conn()
        cur = conn.cursor()
        for s in sql.split(";"):
            cur.execute(s)
        conn.commit()
        cur.close()
        return True

    def insert(self, table, rows):
        conn = self._get_conn()
        cur = conn.cursor()
        cur.executemany("insert into %s values (%s)" % (table, ("?,"*len(rows[0]))[:-1]), rows)
        conn.commit()
        cur.close()
        return True

    def query(self, sql):
        conn = self._get_conn()
        cur = conn.cursor()
        cur.execute(sql)
        rows = cur.fetchall()
        cur.close()
        return rows

• 封裝了基本操作，de 為執(zhí)行一個(gè)數(shù)據(jù)庫(kù)操作，insert 為插入數(shù)據(jù)，query 執(zhí)行一個(gè)查詢
• 需要注意的是 _get_conn 中的 self._conn.row_factory = sqlite3.Row 語(yǔ)句，作用時(shí)執(zhí)行查詢后，返回的結(jié)果會(huì)被轉(zhuǎn)化為 sqlite.Row 對(duì)象，以便通過(guò)字段名來(lái)讀取值，詳細(xì)參加 row_factory[3]。

處理好基礎(chǔ)數(shù)據(jù)，也有了數(shù)據(jù)庫(kù)工具，就可以開始構(gòu)建系統(tǒng)了。

結(jié)構(gòu)設(shè)計(jì)

系統(tǒng)再小，也需要先從結(jié)構(gòu)設(shè)計(jì)入手。

這個(gè)系統(tǒng)只是單機(jī)版（實(shí)際上可以做 Web 擴(kuò)展，下篇展開），所以省去了網(wǎng)絡(luò)和 API 設(shè)計(jì)，直接從庫(kù)表設(shè)計(jì)開始。

首先分析業(yè)務(wù)。

訓(xùn)練營(yíng)的運(yùn)營(yíng)數(shù)據(jù)包括，打卡數(shù)據(jù)，開單數(shù)據(jù)，組長(zhǎng)日常工作，以及成員積分（通過(guò)積分規(guī)則，再自動(dòng)核算部分展開）。

另外，成員有職務(wù)之分：普通成員 和 組長(zhǎng)。規(guī)則是：組長(zhǎng)可以作為普通成員，普通成員不能作為組長(zhǎng)。

那么人員庫(kù)表中，加入了職務(wù)，和組別，以便區(qū)分人員角色：

• mixin_id 是用戶注冊(cè)App的id
• std_id 為打卡系統(tǒng)的id
• team 為小組名
• title 為職務(wù)

然后設(shè)置一個(gè)活動(dòng)類型表，并指定活動(dòng)與職務(wù)的關(guān)系：

• type 為活動(dòng)類型
• value 為活動(dòng)積分
• tilte 為該活動(dòng)對(duì)于的職務(wù)

接下來(lái)就是活動(dòng)記錄表了，由于已經(jīng)定義了活動(dòng)與職務(wù)的對(duì)于關(guān)系，所以，活動(dòng)記錄表中，只需記錄活動(dòng)類型即可：

• mixin_id 為用戶id，std_id 其實(shí)是沒(méi)必要的，不過(guò)錄入打卡記錄時(shí)順帶記錄了
• date 為活動(dòng)發(fā)生的日期
• type 為活動(dòng)內(nèi)容

如果同一個(gè)人同一天同一個(gè)活動(dòng)出現(xiàn)多次，就會(huì)有重復(fù)記錄，那么如何區(qū)分是否真的重復(fù)呢？在 數(shù)據(jù)收集 中展開。

除了基本的數(shù)據(jù)結(jié)構(gòu)，還有積分統(tǒng)計(jì)明細(xì)和積分合計(jì)表，這里不再贅述，會(huì)在核算部分提及。

數(shù)據(jù)收集

現(xiàn)在數(shù)據(jù)框架有了，數(shù)據(jù)從何而來(lái)呢？

這個(gè)訓(xùn)練營(yíng)的數(shù)據(jù)主要來(lái)自兩個(gè)地方，第一是打卡數(shù)據(jù)，第二是日常記錄數(shù)據(jù)。

打卡數(shù)據(jù)由鯨打卡提供，可以在瀏覽器中查看，并且提供了導(dǎo)出打卡 Excel 的功能。

不過(guò)操作比較麻煩：首先登錄后臺(tái)（用微信掃碼登錄），再先選擇導(dǎo)出條件（一般為時(shí)間區(qū)間），下載Excel，然后打開 Excel，才能復(fù)制其中的打卡信息，存入文本文件，最后才能執(zhí)行腳本處理。

好問(wèn)題：
為什么不直接處理 Excel 呢？

1. 因?yàn)镋xcel 處理需要安裝額外庫(kù)，也沒(méi)有文本文件處理方便。
2. 另外未來(lái)考慮做成 Web 系統(tǒng)，所以沒(méi)有做 Excel 的進(jìn)一步擴(kuò)展。

不選擇導(dǎo)出，就得用程序請(qǐng)鯨魚打卡上抓取了。

所以就研究了下打開管理后臺(tái)的請(qǐng)求，分析了一下，請(qǐng)求中有個(gè) cookie 值是關(guān)鍵，于是，復(fù)制請(qǐng)求，轉(zhuǎn)化為 Python 代碼，詳細(xì)描述見 自動(dòng)預(yù)約程序

收集到的數(shù)據(jù)是 JSON 格式的，將其轉(zhuǎn)化為 List，插入數(shù)據(jù)庫(kù)：

def record_check(rows):
    dbrows = []
    for row in rows:
        u = get_user(std_id=int(row[0]))
        if u:
            if row[2] != "×":
                dbrows.append((u['mixin_id'], u['std_id'], row[1], "打卡", 1, row[2], None))
        else:
            print("沒(méi)有找到用戶：", row)
    if len(dbrows) > 0:
        db.insert("tprj_activity", dbrows)
    return dbrows

• record_check 方法是用來(lái)記錄打開記錄的，參數(shù) rows 是從打開后臺(tái)抓取的數(shù)據(jù)
• get_user 是可以根據(jù)打卡用戶的 id，從用戶表中找到用戶記錄，然后結(jié)合打卡記錄，補(bǔ)全打卡記錄
• db 是 上面提到的 DBSqlite 的一個(gè)實(shí)例，調(diào)用其 insert 方法將數(shù)據(jù)插入數(shù)據(jù)庫(kù)

日常記錄，需要根據(jù)訓(xùn)練營(yíng)中的實(shí)際情況做記錄，比如成員開單，組長(zhǎng)輪值等，記錄在 Excel 中比較方便。每日統(tǒng)計(jì)一次，所以我直接將數(shù)據(jù)復(fù)制處理，也存放到文本文件中，用程序解析成記錄行，插入庫(kù)表，展示一下解析方法：

def merge_activity(datafilename):
    rows = []
    with open(datafilename, 'r', encoding='utf-8') as check_f:
        data = {}
        for line in check_f:
            linedata = line[:-1].split('\t')
            date = linedata[0].replace("/","-")
            userinfo = linedata[1].split("/")
            team = userinfo[0]
            name, mixin_id, std_id = userinfo[1].split('-')
            atype = linedata[2]
            rows.append((mixin_id, date, atype))
    ...

可以看到，通過(guò)讀入文本行，再拆分成對(duì)于字段，合成活動(dòng)記錄。

這樣兩個(gè)數(shù)據(jù)收集工作就做好了，這里還需要解決一個(gè)問(wèn)題 —— 避免數(shù)據(jù)重復(fù)。

容易想到的方法是，為數(shù)據(jù)設(shè)置聯(lián)合主鍵，然后對(duì)數(shù)據(jù)做增量式更新。

但是這樣做需要做更多的工作，而且還要很好的測(cè)試。

從業(yè)務(wù)上分析可知：活動(dòng)數(shù)據(jù)并不多，學(xué)員個(gè)數(shù)不過(guò)一百。

那么不妨每次重算?。?/p>

即每次執(zhí)行時(shí)，先庫(kù)表數(shù)據(jù)刪除，然后重新插入一遍。

雖然效率了不高，也算是用框架換時(shí)間吧，換的不出機(jī)器時(shí)間，而是我的工作時(shí)間哈哈。

自動(dòng)核算

數(shù)據(jù)統(tǒng)計(jì)收集完畢，就需要根據(jù)活動(dòng)積分，計(jì)算每個(gè)人的積分明細(xì)合計(jì)。

既然我們選用了數(shù)據(jù)庫(kù)，就直接用 Sql 語(yǔ)句搞定吧。

相對(duì)程序處理來(lái)說(shuō)，Sql 更容易做統(tǒng)計(jì)類的事情。

統(tǒng)計(jì)普通成員積分明細(xì)的語(yǔ)句如下：

INSERT INTO tprj_user_score_detail 
SELECT a.mixin_id, sum(s.value), u.team, '成員', a.date 
FROM
    tprj_activity a
    LEFT JOIN tprj_user u ON a.mixin_id = u.mixin_id
    LEFT JOIN tbas_score s ON a.type = s.type 
WHERE s.title = '成員' 
GROUP BY
    a.mixin_id,
    u.team,
    u.title,
    a.date 

• 查詢所有職務(wù)屬于 成員 的活動(dòng)積分，插入成員積分明細(xì)表
• tprj_activity 為活動(dòng)記錄表，與 tprj_user 用戶表鏈接，然后再鏈接上活動(dòng)表 tbas_score，作用是對(duì)活動(dòng)類做約束
• where 條件中，限制活動(dòng)類型必須為 成員 活動(dòng)
• sum(s.value) 為一個(gè)成員的當(dāng)日積分合計(jì)，日期 體現(xiàn)在 group by 的條件中了

類似的需要寫很多統(tǒng)計(jì)語(yǔ)句，比如組長(zhǎng)的，小組的，以及各自的積分合計(jì)，不再逐個(gè)展示了。

由于 sql 語(yǔ)句較多，為了便于管理，將 sql 語(yǔ)句整理到 sql.py 文件中，在導(dǎo)入主程序代碼，最后調(diào)用 DBSqlite 工具方法執(zhí)行，例如：

import sql
...
db.de(sql.user_score_detail)
...

是不優(yōu)雅多了？

打卡率是通過(guò)統(tǒng)計(jì)活動(dòng)記錄計(jì)算的：

def cal_check_rate():
    ## 計(jì)算打卡率
    team_member = {}
    for r in db.query(sql.team_member_count):
        team_member[r['team']] = r['mcount']
    dbrows = []
    for r in db.query(sql.team_check_count):
        dbrows.append((r['team'], r['date'], round((r['checkcount']/team_member[r['team']])*100)))
    if len(dbrows) > 0:
        db.insert("tprj_team_check_rate", dbrows)
    return dbrows

• team_member_count 語(yǔ)句語(yǔ)句獲取各組的人數(shù)，因?yàn)榭赡苡腥藳](méi)有注冊(cè)打卡。只通過(guò)打卡記錄獲取組內(nèi)人數(shù)，不嚴(yán)謹(jǐn)。
• team_check_count 語(yǔ)句是按組和日期分類核算出的組打卡數(shù)
• 打卡率公式為：（打卡個(gè)數(shù)/組內(nèi)人數(shù)） * 100%
• 將計(jì)算好的打卡率，按日期存入 dbrows，最后插入數(shù)據(jù)庫(kù)

這里還需要注意的是重復(fù)數(shù)據(jù)問(wèn)題，處理方法簡(jiǎn)單粗暴：

全部清除重算

其他數(shù)據(jù)處理也類似。

報(bào)表導(dǎo)出

數(shù)據(jù)處理做好了，要讓發(fā)揮數(shù)據(jù)的作用，就需要制作成報(bào)表，才能讓其他人利用。

本著一切從簡(jiǎn)的原則（主要是需要盡快提供結(jié)果），選擇也 Excel 呈現(xiàn)統(tǒng)計(jì)結(jié)果。

要輸出哪些內(nèi)容呢？

打卡率、成員積分、組排名等，是需要的。

對(duì)于打卡率，需要按組分類，這樣就有讀出小組成員的作用，如何抽取數(shù)據(jù)呢？

寫個(gè) Sql 就好了, 獲取打卡率的語(yǔ)句 check_rate_show 如下：

SELECT 
    date,
    max(case when team ='1組' then rate else 0 end) as '1組',
    max(case when team ='2組' then rate else 0 end) as '2組',
    max(case when team ='3組' then rate else 0 end) as '3組',
    max(case when team ='4組' then rate else 0 end) as '4組',
    max(case when team ='5組' then rate else 0 end) as '5組'
FROM tprj_team_check_rate
GROUP BY date

• tprj_team_check_rate 是用于按組和日期存放打卡率
• select 語(yǔ)句中，使用了行轉(zhuǎn)列的技巧，使得結(jié)果為 第一列為日期，后面列為各個(gè)組，這樣是為了繪制成圖表方便

其實(shí)結(jié)果可以導(dǎo)入 Excel ，生成報(bào)表，更方便一些，但是我沒(méi)這樣做，因?yàn)椋?/p>

1. 操作 Excel 比較費(fèi)勁，調(diào)試工作量大
2. 我有更大的打算，即最終實(shí)現(xiàn)為在線版的，所以花費(fèi)大量時(shí)間不值得

因此我直接將數(shù)據(jù)輸出到文本文件里了。

例如對(duì)打卡率的輸出是這樣的：

def show_check_rate():
    data = db.qj(sql.check_rate_show)
    result = []
    # 處理表頭
    line = '\t'.join(data[0].keys()) + "\n"
    result.append(line)

    # 生成表頭
    for d in data:
        row = []
        for k in d.keys():
            if k != 'date':
                row.append(str(d[k]) + "%")
            else:
                row.append(d[k])
        line = '\t'.join(row) + "\n"
        result.append(line)
    result.append('\n')
    return result

• check_rate_show 執(zhí)行 Sql 獲得數(shù)據(jù)
• 從數(shù)據(jù)中獲取表頭信息，做成一行記錄，請(qǐng)注意字段的分隔為 tab 符，這樣是為了方便直接粘貼到 Excel 中
• 取出數(shù)據(jù)中的每一行，做成表體數(shù)據(jù)行
• 最后再加入一個(gè)回車，這是為了和其他的輸出分隔開

方法執(zhí)行的結(jié)果，寫入文本文件：

filename = "result_%s.txt" % today.strftime("%Y-%m-%d %H_%M_%S")
with open(filename, 'w', encoding='utf-8') as r:
    r.writelines(show_check_rate()) # 打卡率
    r.writelines(show_member_score()) # 成員積分
    ...

• filename 為要寫入的文本文件，這里利用當(dāng)前時(shí)間作為文件名，是為了不重復(fù)
• 打開文件，用 writelines 方法將返回的行寫入文件中
• 這里還可以調(diào)用其他產(chǎn)生輸出方法，將結(jié)果寫入文件

最后，文件中數(shù)據(jù)如下：

date	1組	2組	3組	4組	5組
2021-08-01	65%	90%	79%	85%	72%
2021-08-02	75%	90%	79%	85%	67%
2021-08-03	55%	90%	84%	75%	67%
2021-08-04	60%	95%	74%	75%	61%

復(fù)制到 Excel 的圖表數(shù)據(jù)中就會(huì)形成打卡率圖表：

日常維護(hù)

運(yùn)營(yíng)工作不是一成不變的，比如為了激勵(lì)成員對(duì)提出的問(wèn)題進(jìn)行整理，新增了一個(gè)積分點(diǎn)叫 解答整理。

就得調(diào)整積分項(xiàng)，因?yàn)橹耙呀?jīng)將積分項(xiàng)用庫(kù)表存儲(chǔ)了，現(xiàn)在只需要增加一條記錄，并指明該積分適用于成員角色就可以了。

另外，在 活動(dòng)詳情 報(bào)表中，需要按活動(dòng)名稱記錄每個(gè)人的數(shù)據(jù)，也是個(gè)行轉(zhuǎn)列的操作，但麻煩的是活動(dòng)項(xiàng)是會(huì)變的。

于是先將獲取項(xiàng)動(dòng)態(tài)獲取到，然后合成為行轉(zhuǎn)列的語(yǔ)句，再和查詢語(yǔ)句合并為完整的 Sql 語(yǔ)句，這樣活動(dòng)再有調(diào)整時(shí)，只管添加數(shù)據(jù)項(xiàng)就好了，代碼如下：

score_type_temp = "max(case when type ='{atype}' then num else 0 end) as '{atype}'"

types = db.query("select type, value from tbas_score where title='%s'" % title)
temps = []
for t in types:
    temps.append(sql.score_type_temp.format(atype=t['type']))

allsql = sql.member_score.format(",\n".join(temps))

最后，將各部分的代碼集成起來(lái)，放在一個(gè) main 函數(shù)中，每天執(zhí)行一次，將輸出的文本文件中的數(shù)據(jù)復(fù)制到 Excel 中，就完成當(dāng)日?qǐng)?bào)表了，整個(gè)操作耗時(shí)不到十分鐘，還算滿意。

總結(jié)

促使我這么做的是，不想在機(jī)械的事情上耗費(fèi)時(shí)間，所以會(huì)盡可能的將能自動(dòng)處理的，讓程序處理。

雖然讓一切程序化是一個(gè)理想，在實(shí)現(xiàn)的道路上會(huì)有很多阻礙，所以還需要找到落地的平衡點(diǎn)，需要接受不完美，需要已實(shí)用為導(dǎo)向 —— 先實(shí)現(xiàn)，再完美。

下期，在實(shí)現(xiàn)基本功能的基礎(chǔ)上，我們聊聊如何將這個(gè)平臺(tái) Web 化。

比心！

參考

PS：公號(hào)內(nèi)回復(fù)「Python」即可進(jìn)入Python 新手學(xué)習(xí)交流群，一起 100 天計(jì)劃！

老規(guī)矩，兄弟們還記得么，右下角的 “在看” 點(diǎn)一下，如果感覺文章內(nèi)容不錯(cuò)的話，記得分享朋友圈讓更多的人知道！

【代碼獲取方式】