初始化项目

check_account.py

@@ -0,0 +1,8 @@
+import asyncio
+
+
+async def check_account(page, account_name):
+    await page.get_by_text("账号", exact=True).click()
+    await page.get_by_text("切换账号切换账号").click()
+    await page.locator(f'text=/^{account_name}$/').click()
+    await asyncio.sleep(5)

config.py

@@ -0,0 +1,29 @@
+"""
+小程序：
+    -1 票圈内容精选
+    -2 票圈 l 3亿人喜欢的视频平台
+    -3 票圈 l 视频精选
+    -4 票圈 l 祝福
+    -5 票圈 l 福年
+    -6 票圈 l 信仰之路
+    -7 票圈视频
+    -8 票圈短视频
+    -9 老好看视频
+    -10 票圈最惊奇
+    -11 票圈视频+
+"""
+
+platform_list = [
+    "票圈 l 3亿人喜欢的视频平台",
+    "票圈 l 视频精选",
+    "票圈 l 祝福",
+    "票圈 l 福年",
+    "票圈 l 信仰之路",
+    "票圈视频",
+    "票圈短视频",
+    "老好看视频",
+    "票圈最惊奇",
+    "票圈视频+"
+]
+
+

get_info.py

@@ -0,0 +1,33 @@
+import asyncio
+
+from .navigate_and_download import navigate_and_download
+
+
+async def get_info(page, platform):
+    await page.get_by_text("经营工具").click()
+    await asyncio.sleep(5)
+    if platform == "票圈 l 祝福":
+        await navigate_and_download(page, "产品功能实验", platform)
+
+        await navigate_and_download(page, "广告实验", platform)
+    else:
+        await navigate_and_download(page, "产品功能实验", platform)
+
+        await navigate_and_download(page, "产品实验", platform)
+
+        await navigate_and_download(page, "广告实验", platform)
+
+        if platform == "票圈视频+":
+            await navigate_and_download(page, " 算法实验", platform)
+        else:
+            await navigate_and_download(page, "算法实验", platform)
+
+        if platform == "票圈短视频":
+            await navigate_and_download(page, " 产品/广告实验", platform)
+        else:
+            await navigate_and_download(page, "产品/广告实验", platform)
+
+        if platform == "票圈 l 3亿人喜欢的视频平台":
+            await navigate_and_download(page, "广告类型策略实验层", platform)
+        else:
+            await navigate_and_download(page, "广告类型策略实验", platform)

mysql.py

@@ -0,0 +1,54 @@
+import pymysql
+
+
+class Mysql(object):
+
+    def __init__(self):
+        self.connection = pymysql.connect(
+            host="rm-bp1159bu17li9hi94.mysql.rds.aliyuncs.com",  # 数据库IP地址，内网地址
+            port=3306,  # 端口号
+            user="crawler",  # mysql用户名
+            passwd="crawler123456@",  # mysql用户登录密码
+            db="piaoquan-crawler",  # 数据库名
+            charset="utf8mb4"  # 如果数据库里面的文本是utf8编码的，charset指定是utf8
+        )
+
+    def insert_line(self, data):
+        if self.index(data):
+            try:
+                with self.connection.cursor() as cursor:
+                    sql = "INSERT INTO we_analysis_results (mini_program, task_name, subject, date, control_group, test_group1, test_group2, test_group3, test_group4, test_group5, test_group6, test_group7, test_group8, test_group9, test_group10, test_group11, test_group12, test_group13, test_group14, test_group15, test_group16, test_group17, test_group18, test_group19, test_group20, test_group21, test_group22, test_group23, test_group24) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s);"
+                    cursor.execute(sql, data)
+                self.connection.commit()
+            finally:
+                self.connection.close()
+        else:
+            print("数据已经存在")
+
+    def index(self, data):
+        mini_program = data[0]
+        task_name = data[1]
+        subject = data[2]
+        date = data[3]
+        sql = (
+            f'SELECT 1 FROM we_analysis_results where mini_program = "{mini_program}" and task_name = "{task_name}" and subject = "{subject}" and date = "{date}"')
+        print(sql)
+        cursor = self.connection.cursor()
+        cursor.execute(sql)
+        data = cursor.fetchall()
+        if data:
+            return False
+        else:
+            return True
+
+
+
+
+
+
+
+
+
+
+
+

navigate_and_download.py

@@ -0,0 +1,158 @@
+import asyncio
+
+
+async def navigate_and_download(page, experiment_name, platform):
+    """
+    导航到指定的页面并触发下载。
+    :param page: playwright的页面对象。
+    :param experiment_name: 要导航到的实验名称。
+    :param platform: 小程序名称。
+    """
+    if platform == "票圈视频+" and experiment_name in ['广告实验', '产品/广告实验']:
+        if experiment_name == "产品/广告实验":
+            await page.get_by_text("实验工具").click()
+            await page.get_by_text("实验数据").click()
+            await page.locator(f'text=/^{experiment_name}$/').click()
+
+            # 小程序访问人数
+            await asyncio.sleep(10)
+            # await page.get_by_text("小程序访问人数").nth(1).click()
+            await page.locator("dt").filter(has_text="小程序访问人数").first.click()
+            await page.get_by_role("list").get_by_text("小程序访问人数", exact=True).click()
+            await asyncio.sleep(10)
+            await page.get_by_role("img", name="下载").click()
+            await asyncio.sleep(10)
+
+            # 小程序分享人数
+            # await page.get_by_text("小程序访问人数").nth(1).click()
+            await page.locator("dt").filter(has_text="小程序访问人数").first.click()
+            await page.get_by_role("list").get_by_text("小程序分享人数", exact=True).click()
+            await asyncio.sleep(10)
+            await page.get_by_role("img", name="下载").click()
+            await asyncio.sleep(10)
+
+            # 小程序分享次数
+            await page.locator("dt").filter(has_text="小程序分享人数").click()
+            await page.get_by_role("list").get_by_text("小程序分享次数", exact=True).click()
+            await asyncio.sleep(10)
+            await page.get_by_role("img", name="下载").click()
+            await asyncio.sleep(10)
+
+            # 广告曝光次数
+            await page.locator("dt").filter(has_text="小程序分享次数").click()
+            await page.get_by_role("list").get_by_text("广告曝光次数", exact=True).click()
+            await asyncio.sleep(10)
+            await page.get_by_role("img", name="下载").click()
+            await asyncio.sleep(10)
+
+            await page.locator("dt").filter(has_text="广告曝光次数").click()
+            await page.get_by_role("list").get_by_text("广告曝光人数", exact=True).click()
+            await asyncio.sleep(10)
+            await page.get_by_role("img", name="下载").click()
+            await asyncio.sleep(10)
+
+            # 广告 eCPM
+            await page.locator("dt").filter(has_text="广告曝光人数").click()
+            await page.get_by_role("list").get_by_text("广告 eCPM", exact=True).click()
+            await asyncio.sleep(10)
+            await page.get_by_role("img", name="下载").click()
+            await asyncio.sleep(10)
+
+        elif experiment_name == "广告实验":
+            await page.get_by_text("实验工具").click()
+            await page.get_by_text("实验数据").click()
+            await page.locator(f'text=/^{experiment_name}$/').click()
+
+            # 小程序访问人数
+            await asyncio.sleep(10)
+            # await page.get_by_text("小程序访问人数").nth(1).click()
+            await page.locator("dt").filter(has_text="小程序访问人数").first.click()
+            await page.get_by_role("list").get_by_text("小程序访问人数", exact=True).click()
+            await asyncio.sleep(10)
+            await page.get_by_role("img", name="下载").click()
+            await asyncio.sleep(10)
+
+
+            # 广告曝光次数
+            await page.locator("dt").filter(has_text="小程序访问人数").click()
+            await page.get_by_role("list").get_by_text("广告曝光次数", exact=True).click()
+            await asyncio.sleep(10)
+            await page.get_by_role("img", name="下载").click()
+            await asyncio.sleep(10)
+
+            await page.locator("dt").filter(has_text="广告曝光次数").click()
+            await page.get_by_role("list").get_by_text("广告曝光人数", exact=True).click()
+            await asyncio.sleep(10)
+            await page.get_by_role("img", name="下载").click()
+            await asyncio.sleep(10)
+
+            # 广告 eCPM
+            await page.locator("dt").filter(has_text="广告曝光人数").click()
+            await page.get_by_role("list").get_by_text("广告 eCPM", exact=True).click()
+            await asyncio.sleep(10)
+            await page.get_by_role("img", name="下载").click()
+            await asyncio.sleep(10)
+
+            # 广告收入
+            await page.locator("dt").filter(has_text="广告 eCPM").click()
+            await page.get_by_role("list").get_by_text("广告收入", exact=True).click()
+            await asyncio.sleep(10)
+            await page.get_by_role("img", name="下载").click()
+            await asyncio.sleep(10)
+
+    else:
+        await page.get_by_text("实验工具").click()
+        await page.get_by_text("实验数据").click()
+        await page.locator(f'text=/^{experiment_name}$/').click()
+
+        # 小程序访问人数
+        await asyncio.sleep(10)
+        # await page.get_by_text("小程序访问人数").nth(1).click()
+        await page.locator("dt").filter(has_text="小程序访问人数").first.click()
+        await page.get_by_role("list").get_by_text("小程序访问人数", exact=True).click()
+        await asyncio.sleep(10)
+        await page.get_by_role("img", name="下载").click()
+        await asyncio.sleep(10)
+
+        # 小程序分享人数
+        # await page.get_by_text("小程序访问人数").nth(1).click()
+        await page.locator("dt").filter(has_text="小程序访问人数").first.click()
+        await page.get_by_role("list").get_by_text("小程序分享人数", exact=True).click()
+        await asyncio.sleep(10)
+        await page.get_by_role("img", name="下载").click()
+        await asyncio.sleep(10)
+
+        # 小程序分享次数
+        await page.locator("dt").filter(has_text="小程序分享人数").click()
+        await page.get_by_role("list").get_by_text("小程序分享次数", exact=True).click()
+        await asyncio.sleep(10)
+        await page.get_by_role("img", name="下载").click()
+        await asyncio.sleep(10)
+
+        # 广告曝光次数
+        await page.locator("dt").filter(has_text="小程序分享次数").click()
+        await page.get_by_role("list").get_by_text("广告曝光次数", exact=True).click()
+        await asyncio.sleep(10)
+        await page.get_by_role("img", name="下载").click()
+        await asyncio.sleep(10)
+
+        await page.locator("dt").filter(has_text="广告曝光次数").click()
+        await page.get_by_role("list").get_by_text("广告曝光人数", exact=True).click()
+        await asyncio.sleep(10)
+        await page.get_by_role("img", name="下载").click()
+        await asyncio.sleep(10)
+
+        # 广告 eCPM
+        await page.locator("dt").filter(has_text="广告曝光人数").click()
+        await page.get_by_role("list").get_by_text("广告 eCPM", exact=True).click()
+        await asyncio.sleep(10)
+        await page.get_by_role("img", name="下载").click()
+        await asyncio.sleep(10)
+
+        # 广告收入
+        await page.locator("dt").filter(has_text="广告 eCPM").click()
+        await page.get_by_role("list").get_by_text("广告收入", exact=True).click()
+        await asyncio.sleep(10)
+        await page.get_by_role("img", name="下载").click()
+        await asyncio.sleep(10)
+