增加飞书写入

main.py

@@ -1,113 +0,0 @@
-import os
-import json
-import asyncio
-import datetime
-import pandas as pd
-from playwright.async_api import async_playwright
-
-from applications import check_account, get_info, Mysql
-
-platform_name = None
-data_frame = []
-lock = asyncio.Lock()
-
-
-async def handle_download(download):
-    global platform_name
-    global data_frame
-
-    async with lock:
-        today = datetime.datetime.today().__str__().split(" ")[0].replace("-", "")
-        temp = [platform_name]
-
-        download_path = await download.path()
-        # 指定新文件名
-        new_file_path = r"temp_file\temp.xlsx"
-        # 重命名文件
-        if os.path.exists(new_file_path):
-            os.remove(new_file_path)
-        os.rename(download_path, new_file_path)
-        if os.path.exists(download_path):
-            os.remove(download_path)
-        df = pd.read_excel(new_file_path)
-        df = df.values.tolist()
-        # 实验名称
-        task_name = df[0][1]
-        temp.append(task_name)
-        # 数据指标
-        task_index = df[5][1]
-        temp.append(task_index)
-        title_row = df[8]
-        data_line = df[-1]
-        # if data_line[0] != today:
-        #     print("Not today")
-        # else:
-        temp += data_line
-        w = {temp: data_line[index] for index, temp in enumerate(title_row)}
-        print(platform_name, "\t", task_name, "\t", task_index)
-        print(json.dumps(w, ensure_ascii=False, indent=4))
-        while len(temp) < 29:
-            temp.append("NULL")
-        Mysql().insert_line(data=temp)
-        data_frame.append(temp)
-        os.remove(new_file_path)
-
-
-async def run(playwright):
-    """
-    小程序：
-        -1 票圈内容精选   ok
-        -2 票圈 l 3亿人喜欢的视频平台  ok
-        -3 票圈 l 视频精选
-        -4 票圈 l 祝福
-        -5 票圈 l 福年
-        -6 票圈 l 信仰之路    ok
-        -7 票圈视频  ok
-        -8 票圈短视频
-        -9 老好看视频    ok
-        -10 票圈最惊奇   ok
-        -11 票圈视频+
-    """
-    global platform_name
-    global data_frame
-
-    # 初始化浏览器
-    browser = await playwright.chromium.launch(
-        headless=False,
-        downloads_path="temp_file",
-    )
-    context = await browser.new_context(accept_downloads=True)
-    page = await context.new_page()
-    await page.set_viewport_size({"width": 1680, "height": 1080})
-
-    page.on("download", handle_download)
-
-    # 登陆，需要扫码
-    await page.goto("https://wedata.weixin.qq.com/mp2/login")
-    await page.goto("https://wedata.weixin.qq.com/mp2/?source=0")
-    await page.goto("https://wedata.weixin.qq.com/mp2/basic-data/core-data?source=0")
-
-    # # 默认先进入票圈内容精选
-    platform_name = '票圈视频+'
-    await get_info(page, platform_name)
-
-    # columns = ["小程序", "实验", "指标", "日期", "对照组"] + [
-    #     "实验组{}".format(i) for i in range(1, 25)
-    # ]
-    # print(len(columns))
-    # out_df = DataFrame(data_frame, columns=columns)
-    # out_df.to_excel("{}.xlsx".format("2024-01-02-result"), index=False)
-
-    # await asyncio.sleep(100)
-
-    # 关闭浏览器上下文和浏览器
-    await context.close()
-    await browser.close()
-
-
-async def main():
-    async with async_playwright() as playwright:
-        await run(playwright)
-
-
-asyncio.run(main())

main_single_server.py

@@ -8,7 +8,6 @@ from playwright.async_api import async_playwright
 from applications import get_info, Mysql
 
 platform_name = None
-data_frame = []
 lock = asyncio.Lock()
 
 
@@ -102,15 +101,6 @@ async def run(playwright):
     platform_name = platform[Id]
     await get_info(page, platform_name)
 
-    # columns = ["小程序", "实验", "指标", "日期", "对照组"] + [
-    #     "实验组{}".format(i) for i in range(1, 25)
-    # ]
-    # print(len(columns))
-    # out_df = DataFrame(data_frame, columns=columns)
-    # out_df.to_excel("{}.xlsx".format("2024-01-02-result"), index=False)
-
-    # await asyncio.sleep(100)
-
     # 关闭浏览器上下文和浏览器
     await context.close()
     await browser.close()

manager.py

@@ -0,0 +1,76 @@
+"""
+往飞书内写数据
+"""
+import numpy as np
+
+from feishu.feishu import Feishu
+from applications.mysql import Mysql
+
+
+class FeishuManager(object):
+    """
+    写入飞书管理器
+    """
+    def __init__(self):
+        self.F = Feishu()
+        self.M = Mysql()
+        self.map = {
+            "产品功能实验": "ca8499",
+            "产品实验": "gwzBOM",
+            "广告实验": "Pt6JUj",
+            "产品/广告实验": "avHzKO",
+            "算法实验": "MFwgBr",
+            "广告类型策略实验": "XCCAOP"
+        }
+
+    def select(self, minigram, date, task_name):
+        """
+        从mysql中把数据读出来
+        :param task_name: 实验名称
+        :param minigram: 小程序名称
+        :param date: 日期
+        :return: data_list
+        """
+        select_sql = f"""SELECT * FROM we_analysis_results where mini_program = '{minigram}' and date = '{date}' and task_name = '{task_name}';"""
+        we_tuple = self.M.select(select_sql)
+        init_list = []
+        for line in we_tuple:
+            result = [i for i in ([line[3]] + list(line[5:]))]
+            init_list.append(result)
+        # We分析矩阵
+        we_array = np.transpose(init_list)
+        we_array_v2 = we_array[1:, :]
+        # 去掉全部是0的行
+        non_zero_rows = np.any(we_array_v2 != 0, axis=1)
+        we_array_v3 = we_array_v2[non_zero_rows]
+        we_list = we_array_v3.tolist()
+        return we_list
+
+    def insert(self, we_list, date, mini_program, task_name):
+        """
+        处理矩阵，并且插入飞书表
+        :param task_name: 实验名称
+        :param date: 日期
+        :param mini_program: 小程序
+        :param we_list: we分析矩阵
+        :return: None
+        """
+        rows = len(we_list)
+        datas = [
+            [date, 0, mini_program, "ab{}".format(i), "00:00", "", "实验名称"] + we_list[i]
+            for i in range(rows)
+        ]
+        self.F.insert_value(
+            sheet_id=self.map[task_name],
+            values=datas,
+            ranges="A4:AW24"
+        )
+
+
+if __name__ == '__main__':
+    Fm = FeishuManager()
+    account_name = "票圈内容精选"
+    date_str = "2024-01-17"
+    temp = Fm.select(account_name, date_str)
+    Fm.insert(temp, date_str, account_name)
+

run_all.py

@@ -1,3 +0,0 @@
-import os
-
-os.system("python main.py")

run_single.py

@@ -1,4 +0,0 @@
-import os
-
-
-os.system("python main_single_server.py")

test.py

@@ -1,47 +0,0 @@
-"""
-测试代码
-"""
-
-import numpy as np
-from pandas import DataFrame
-
-from feishu.feishu import Feishu
-from applications.mysql import Mysql
-
-
-F = Feishu()
-M = Mysql()
-
-sql = "SELECT * FROM we_analysis_results where mini_program = '票圈视频+' and date = \'2024-01-15\';"
-
-if __name__ == '__main__':
-    data_lines = M.select(sql)
-    L = []
-    title_list = []
-    for line in data_lines:
-        result = [i for i in ([line[3]] + list(line[5:]))]
-        L.append(result)
-    w = transposed_matrix = np.transpose(L)
-    first_row = w[0, :]
-    w = w[1:, :]
-    non_zero_rows = np.any(w != 0, axis=1)
-    w_filtered = w[non_zero_rows]
-    my_list = w_filtered.tolist()
-    rows = len(my_list)
-    datas = [["2024-01-15", "票圈视频+", "", "ab{}".format(i)] + my_list[i] for i in range(rows)]
-    F.insert_value(
-        sheet_id="gwzBOM",
-        values=datas,
-        ranges="A2:AH22"
-    )
-    # # 插入日期
-    # F.insert_value(
-    #     sheet_id="gwzBOM",
-    #     values=[["2024-01-15", "票圈视频+"] for i in range(rows)],
-    #     ranges="A2:B22"
-    # )
-    # titles = [0] + list(first_row)
-    # print(titles)
-    # DF = DataFrame(w_with_labels, columns=titles)
-    # DF.to_excel("test.xlsx", index=False)
-