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

Python技術(shù) 2021-08-13

展開(kāi)全文

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

金屬質(zhì)感分割線

最近參加了一個(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)有盡有……

開(kāi)始認(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ù),，所以將這些操作寫(xiě)成一個(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ù)工具,，就可以開(kāi)始構(gòu)建系統(tǒng)了。

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

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

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

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

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

另外,，成員有職務(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ù)收集中展開(kāi)。

除了基本的數(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，然后打開(kāi) Excel,，才能復(fù)制其中的打卡信息,，存入文本文件，最后才能執(zhí)行腳本處理,。

好問(wèn)題：
為什么不直接處理 Excel 呢,？
因?yàn)镋xcel 處理需要安裝額外庫(kù),，也沒(méi)有文本文件處理方便。
另外未來(lái)考慮做成 Web 系統(tǒng),，所以沒(méi)有做 Excel 的進(jìn)一步擴(kuò)展,。

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

所以就研究了下打開(kāi)管理后臺(tái)的請(qǐng)求,，分析了一下，請(qǐng)求中有個(gè) cookie 值是關(guān)鍵,，于是,，復(fù)制請(qǐng)求，轉(zhuǎn)化為 Python 代碼,，詳細(xì)描述見(jiàn) 自動(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)記錄打開(kāi)記錄的,，參數(shù) rows 是從打開(kāi)后臺(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í)際情況做記錄,，比如成員開(kāi)單,，組長(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 的條件中了

類似的需要寫(xiě)很多統(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ù)呢,？

寫(xiě)個(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>

操作 Excel 比較費(fèi)勁,，調(diào)試工作量大
我有更大的打算，即最終實(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è)回車(chē),，這是為了和其他的輸出分隔開(kāi)

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


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

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

date1組2組3組4組5組
2021-08-0165%90%79%85%72%
2021-08-0275%90%79%85%67%
2021-08-0355%90%84%75%67%
2021-08-0460%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í)不到十分鐘,，還算滿意,。