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

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

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

基礎整理

數據是運營的基礎，人員數據是基礎數據，首先需要搞定人員信息。

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

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

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

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

這里重點說一下數據庫。

使用數據庫的好處是，方便數據整合、統計和更新。但是數據庫一般比較重，維護部署都是問題，于是選用了文本數據庫 SQLite作為數據庫。

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

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

pip install sqlite3

創建數據庫鏈接：

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

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

有了鏈接，就可以執行數據庫操作了，比如創建一個庫表,插入數據：

# 創建一個游標 cur = con.cursor() # 執行SQl 語句創建庫表 cur.execute('''CREATE TABLE stocks
               (date text, trans text, symbol text, qty real, price real)''') # 向庫表中插入數據 cur.execute("INSERT INTO stocks VALUES ('2006-01-05','BUY','RHAT',100,35.14)") # 提交更改結果 con.commit() # 關閉鏈接 con.close()

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

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

處理好基礎數據，也有了數據庫工具，就可以開始構建系統了。

結構設計

系統再小，也需要先從結構設計入手。

這個系統只是單機版（實際上可以做 Web 擴展，下篇展開），所以省去了網絡和 API 設計，直接從庫表設計開始。

首先分析業務。

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

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

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

人員表

• mixin_id 是用戶注冊App的id
• std_id 為打卡系統的id
• team 為小組名
• title 為職務

然后設置一個活動類型表，并指定活動與職務的關系：

活動表

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

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

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

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

除了基本的數據結構，還有積分統計明細和積分合計表，這里不再贅述，會在核算部分提及。

數據收集

現在數據框架有了，數據從何而來呢？

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

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

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

好問題：為什么不直接處理 Excel 呢？

1. 因為Excel 處理需要安裝額外庫，也沒有文本文件處理方便。
2. 另外未來考慮做成 Web 系統，所以沒有做 Excel 的進一步擴展。

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

所以就研究了下打開管理后臺的請求，分析了一下，請求中有個 cookie 值是關鍵，于是，復制請求，轉化為 Python 代碼

收集到的數據是 JSON 格式的，將其轉化為 List，插入數據庫：

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 方法是用來記錄打開記錄的，參數 rows 是從打開后臺抓取的數據
• get_user 是可以根據打卡用戶的 id，從用戶表中找到用戶記錄，然后結合打卡記錄，補全打卡記錄
• db 是 上面提到的 DBSqlite 的一個實例，調用其 insert 方法將數據插入數據庫

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

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

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

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

容易想到的方法是，為數據設置聯合主鍵，然后對數據做增量式更新。

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

從業務上分析可知：活動數據并不多，學員個數不過一百。

那么不妨每次重算??？

即每次執行時，先庫表數據刪除，然后重新插入一遍。

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

自動核算

數據統計收集完畢，就需要根據活動積分，計算每個人的積分明細合計。

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

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

統計普通成員積分明細的語句如下：

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) 為一個成員的當日積分合計，日期 體現在 group by 的條件中了

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

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

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

是不優雅多了？

打卡率是通過統計活動記錄計算的：

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

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

全部清除重算

其他數據處理也類似。

報表導出

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

本著一切從簡的原則（主要是需要盡快提供結果），選擇也 Excel 呈現統計結果。

要輸出哪些內容呢？

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

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

寫個 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 語句中，使用了行轉列的技巧，使得結果為 第一列為日期，后面列為各個組，這樣是為了繪制成圖表方便

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

1. 操作 Excel 比較費勁，調試工作量大
2. 我有更大的打算，即最終實現為在線版的，所以花費大量時間不值得

因此我直接將數據輸出到文本文件里了。

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

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

方法執行的結果，寫入文本文件：

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 方法將返回的行寫入文件中
• 這里還可以調用其他產生輸出方法，將結果寫入文件

最后，文件中數據如下：

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%

復制到 Excel 的圖表數據中就會形成打卡率圖表：

打卡率圖表

日常維護

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

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

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

于是先將獲取項動態獲取到，然后合成為行轉列的語句，再和查詢語句合并為完整的 Sql 語句，這樣活動再有調整時，只管添加數據項就好了，代碼如下：

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 函數中，每天執行一次，將輸出的文本文件中的數據復制到 Excel 中，就完成當日報表了，整個操作耗時不到十分鐘，還算滿意。

總結

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

雖然讓一切程序化是一個理想，在實現的道路上會有很多阻礙，所以還需要找到落地的平衡點，需要接受不完美，需要已實用為導向 —— 先實現，再完美。想要了解更多Python教程知識與技能歡迎持續關注編程學習網