【原】YYDS! Python 幫我扛起運營大旗！

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

展開全文

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

金屬質(zhì)感分割線

最近參加了一個訓練營，作為副教練，承擔起訓練營的運營工作。事不大，活不少，打卡記錄、活動積分、獎勵制度、評優(yōu)方案、趨勢對比，應有盡有……

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

基礎(chǔ)整理

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

訓練營里的人員信息來自多個渠道，有通過 APP 報名的，有調(diào)查問卷收集的，還有人工錄取的。

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

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

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

這里重點說一下數(shù)據(jù)庫。

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

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

使用起來安裝 Python 的 SQLite 模塊就可以了：

pip install sqlite3

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

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

其中 database.db 就是一個普通文件，如果沒有，會自動創(chuàng)建一個。

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

# 創(chuàng)建一個游標
cur = con.cursor()

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

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

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

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

因為需要頻繁地操作數(shù)據(jù)庫，所以將這些操作寫成一個類：

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í)行一個數(shù)據(jù)庫操作，insert 為插入數(shù)據(jù)，query 執(zhí)行一個查詢
需要注意的是 _get_conn 中的 self._conn.row_factory = sqlite3.Row 語句，作用時執(zhí)行查詢后，返回的結(jié)果會被轉(zhuǎn)化為 sqlite.Row 對象，以便通過字段名來讀取值，詳細參加 row_factory[3]。

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

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

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

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

首先分析業(yè)務。

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

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

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

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

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

type 為活動類型
value 為活動積分
tilte 為該活動對于的職務

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

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

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

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

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

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

這個訓練營的數(shù)據(jù)主要來自兩個地方，第一是打卡數(shù)據(jù)，第二是日常記錄數(shù)據(jù)。

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

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

好問題：
為什么不直接處理 Excel 呢？
因為Excel 處理需要安裝額外庫，也沒有文本文件處理方便。
另外未來考慮做成 Web 系統(tǒng)，所以沒有做 Excel 的進一步擴展。

不選擇導出，就得用程序請鯨魚打卡上抓取了。

所以就研究了下打開管理后臺的請求，分析了一下，請求中有個 cookie 值是關(guān)鍵，于是，復制請求，轉(zhuǎn)化為 Python 代碼，詳細描述見自動預約程序

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

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("沒有找到用戶：", row)
    if len(dbrows) > 0:
        db.insert("tprj_activity", dbrows)
    return dbrows

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

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

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))
    ...

可以看到，通過讀入文本行，再拆分成對于字段，合成活動記錄。

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

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

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

從業(yè)務上分析可知：活動數(shù)據(jù)并不多，學員個數(shù)不過一百。

那么不妨每次重算??？

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

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

自動核算

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

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

相對程序處理來說，Sql 更容易做統(tǒng)計類的事情。

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

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

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

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

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

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

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

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

def cal_check_rate():
    ## 計算打卡率
    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 語句語句獲取各組的人數(shù)，因為可能有人沒有注冊打卡。只通過打卡記錄獲取組內(nèi)人數(shù)，不嚴謹。
team_check_count 語句是按組和日期分類核算出的組打卡數(shù)
打卡率公式為：（打卡個數(shù)/組內(nèi)人數(shù)） * 100%
將計算好的打卡率，按日期存入 dbrows，最后插入數(shù)據(jù)庫

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

全部清除重算

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

報表導出

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

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

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

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

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

寫個 Sql 就好了, 獲取打卡率的語句 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 語句中，使用了行轉(zhuǎn)列的技巧，使得結(jié)果為第一列為日期，后面列為各個組，這樣是為了繪制成圖表方便

其實結(jié)果可以導入 Excel ，生成報表，更方便一些，但是我沒這樣做，因為：

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

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

例如對打卡率的輸出是這樣的：

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ù)中獲取表頭信息，做成一行記錄，請注意字段的分隔為 tab 符，這樣是為了方便直接粘貼到 Excel 中
取出數(shù)據(jù)中的每一行，做成表體數(shù)據(jù)行
最后再加入一個回車，這是為了和其他的輸出分隔開

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

最后，文件中數(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%

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

打卡率圖表

日常維護

運營工作不是一成不變的，比如為了激勵成員對提出的問題進行整理，新增了一個積分點叫 解答整理。

就得調(diào)整積分項，因為之前已經(jīng)將積分項用庫表存儲了，現(xiàn)在只需要增加一條記錄，并指明該積分適用于成員角色就可以了。

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

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

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))

最后，將各部分的代碼集成起來，放在一個 main 函數(shù)中，每天執(zhí)行一次，將輸出的文本文件中的數(shù)據(jù)復制到 Excel 中，就完成當日報表了，整個操作耗時不到十分鐘，還算滿意。

總結(jié)

促使我這么做的是，不想在機械的事情上耗費時間，所以會盡可能的將能自動處理的，讓程序處理。

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

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

比心！

參考

[1]

SQLite: https://www./index.html

[2]

MySQL: https://www./cn/

[3]

conn.row_factory: https://docs./3/library/sqlite3.html

贊賞

共11人贊賞

【原】YYDS! Python 幫我扛起運營大旗！