-

index.py

@@ -1,6 +1,8 @@
-from modules.load_config import load_config
+from modules.load_config import load_config, dprint
 from modules.send_sign_in_threading import send_sign_in_threading
 
+# ❗敏感信息输出开关, 例如用户名和手机号还有部分接口响应内容, 部署时一定要关掉
+dprint.DEBUG = False
 
 
 def handler(event, context):

modules/check_sign_in.py

@@ -1,4 +1,5 @@
 import requests
+from modules.load_config import dprint
 from datetime import datetime, timedelta
 
 
@@ -17,15 +18,16 @@ def check_sign(check_sign_token, sign_type):
     headers = {"Authorization": f"{check_sign_token}"}
     response = requests.post(check_sign_url, json=data, headers=headers)
     response_data = response.json()
+
     # ❗开发环境输出, 打印打卡记录的响应内容
-    # print(response_data)
+    # dprint(response_data)
 
-    # 检查是否有过签到
+    # 检查是否有过打卡记录
     for record in response_data["data"]:
         if record["type"] == sign_type:
-            # print("打过卡了")
+            # 打过卡
             return True
-    # print("没打过卡")
+    # 没打过卡
     return False
 
 
@@ -55,10 +57,10 @@ def get_current_day():
 
 
 # 开发环境测试环境解除注释
-# print("\n❗注意：以下输出是check_sign模块的开发环境信息，👇")
+# dprint("\n❗注意：以下输出是check_sign模块的开发环境信息，👇")
 # day_start_time, day_end_time = get_current_day()
-# print("打卡信息查询日期范围:\n", day_start_time, "→", day_end_time)
+# dprint("打卡信息查询日期范围:\n", day_start_time, "→", day_end_time)
 # check_sign(
-#     check_sign_token="eyJhbGciOiJIUzUxMiJ9.eyJpc3MiOiJtb2d1ZGluZy11c2VyIiwic3ViIjoie1wibG9naW5UeXBlXCI6XCJ3ZWJcIixcInVzZXJJZFwiOjEwNjY0ODAyN30iLCJhdWQiOiJtb2d1ZGluZyIsImV4cCI6MjA0NTI4Mzc0MiwibmJmIjoxNzI5NjYzNjQyLCJpYXQiOjE3Mjk2NjQ1NDJ9.pPnKypQaNub0l96aqCA-hkv0Hg2igosq8yOdgvONVjyHuS-3Y8Ul0Agf1Ru4PEEMRsHzGAVESQT9uY3DMQ_c1w",
+#     check_sign_token="xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
 #     sign_type="START",
 # )

modules/crypto_aes.py

@@ -1,6 +1,5 @@
 import logging
 from hashlib import md5
-from datetime import datetime, timedelta
 from packages.aes_pkcs5.algorithms.aes_ecb_pkcs5_padding import AESECBPKCS5Padding
 
 logging.basicConfig(
@@ -81,29 +80,3 @@ def aes_decrypt(
     except Exception as e:
         logger.error(f"解密失败: {e}")
         raise ValueError(f"解密失败: {str(e)}")
-
-
-def get_current_month_info() -> dict:
-    """获取当前月份的开始和结束时间。
-
-    该方法计算当前月份的开始日期和结束日期, 并将它们返回为字典,
-    字典中包含这两项的字符串表示。
-
-    :return: 包含当前月份开始和结束时间的字典。
-    :rtype: dict
-    """
-    now = datetime.now()
-
-    start_of_month = datetime(now.year, now.month, 1)
-
-    if now.month == 12:
-        next_month_start = datetime(now.year + 1, 1, 1)
-    else:
-        next_month_start = datetime(now.year, now.month + 1, 1)
-
-    end_of_month = next_month_start - timedelta(days=1)
-
-    start_time_str = start_of_month.strftime("%Y-%m-%d %H:%M:%S")
-    end_time_str = end_of_month.strftime("%Y-%m-%d 00:00:00Z")
-
-    return {"startTime": start_time_str, "endTime": end_time_str}

modules/get_login_token.py

@@ -1,9 +1,8 @@
-import json, time, json, time, uuid, random, base64, struct, requests
+import json, time, requests, uuid, random, base64, struct, requests, cv2
 from modules.crypto_aes import aes_encrypt, aes_decrypt
+from modules.load_config import dprint
 import numpy as np
-import cv2
 from typing import Optional
-import requests
 
 
 request_headers = {
@@ -43,8 +42,9 @@ def get_token(user):
         "登陆失败, 看到这条提示代表官方接口又更新了, 请在github留issue。",
     )
     user_info = json.loads(aes_decrypt(rsp.get("data", "")))
+
     # ❗开发环境输出, 打印登录响应内容, 从此查看是否获取到token
-    # print(user_info)
+    # dprint(user_info)
 
     return user_info
 

modules/get_plan_id.py

@@ -1,5 +1,4 @@
 import json, requests
-
 from modules.map_headers import header_user_agent
 
 

modules/load_config.py

@@ -9,7 +9,10 @@
 # 环境变量/本地配置加载器(json格式)
 def load_config():
     full_config_path = os.path.join(config_path)
-    print(f"\n>>>>>>>>>> 打卡任务启动! <<<<<<<<<<\n")
+    print(f"\n>>>>>>>>>> 打卡任务启动! <<<<<<<<<<\n\n")
+    dprint(
+        "❗当前模式会输出隐私信息, 如您处于线上环境请为index.py文件内容设置debug=False"
+    )
     config_data = {}
     config_loaded = False
     if os.path.exists(full_config_path):
@@ -18,7 +21,7 @@ def load_config():
                 config_data = json.load(file)
                 config_loaded = True
                 print(f">>> 配置文件载入成功: {full_config_path}\n")
-                # print("载入的配置文件信息: ", config_data)
+                # dprint("载入的配置文件信息: ", config_data)
         except Exception as e:
             print(f"XXX 配置文件加载失败, 请检查文件是否存在: {full_config_path}\n")
     if not config_loaded:
@@ -28,7 +31,7 @@ def load_config():
                 config_data = json.loads(env_value)
                 config_loaded = True
                 print(f">>> 环境变量加载成功: {config_sys}\n")
-                # print("载入的环境变量配置信息: ", config_data)
+                # dprint("载入的环境变量配置信息: ", config_data)
             except Exception as e:
                 print(
                     f"XXX 环境变量加载失败, 请检查是否设置了环境变量键名为{config_sys}!\n"
@@ -38,3 +41,12 @@ def load_config():
             f"XXX 未找到配置信息，请检查是否设置了环境变量键名为{config_sys}, 或是本地文件路径{config_path}!\n"
         )
     return config_data
+
+
+# 控制台输出, 此处的开关（布尔值）由主程序定义，否则默认不输出
+def dprint(*args, **kwargs):
+    if dprint.DEBUG:
+        print(*args, **kwargs)
+
+
+dprint.DEBUG = False

modules/msg_pusher.py

@@ -1,4 +1,5 @@
 import requests
+from modules.load_config import dprint
 
 
 # 推送消息到PushPlus
@@ -14,11 +15,14 @@ def push_to_pushplus(users, title, content, token):
     )
     resp = requests.post(url)
     result = resp.json()
+
     if result["code"] == 200:
         for user in users:
-            print(f'√ [{user["remark"]}/{user["phone"]}]消息推送成功！')
+            dprint(f'[{user["remark"]}/{user["phone"]}]', end="")
+            print(f"√ 用户消息推送成功！", end="")
+            dprint(f"响应内容: {resp.text}")
     else:
         for user in users:
-            print(
-                f'× [{user["remark"]}/{user["phone"]}]推送消息提醒失败！原因: {result.get("msg", "未知错误")}'
-            )
+            dprint(f'[{user["remark"]}/{user["phone"]}]', end="")
+            print(f"X 推送消息提醒失败！", end="")
+            dprint(f"响应内容: {resp.text}")

modules/send_sign_in.py

@@ -5,6 +5,7 @@
 from modules.check_sign_in import check_sign
 from modules.save_user_info import save_user_info
 from datetime import datetime, timezone, timedelta
+from modules.load_config import dprint
 
 
 # 操作地时区设置
@@ -17,6 +18,8 @@
 sign_type_mapping = {"START": "上班", "END": "下班"}
 # 收集所有线程签到结果
 global_sign_results = []
+# 覆盖打卡，设置为True会导致每次运行都不检查签到历史从而重复签到，用于测试环境调试
+RepeatCheckIn = True
 
 
 # 处理单个用户的签到流程, 包括获取登录令牌、签到计划 ID、检查是否已签到以及执行签到操作。
@@ -30,8 +33,8 @@ def send_sign_in(user_info, timeout):
 
         user_info_prefix = f"[{remark}/{phone}]"
 
-        # ❗开发环境输出, 打印登录响应内容
-        # print(f"{user_info_prefix} 登录响应: {userInfo}\n")
+        # ❗开发环境输出, 包含敏感内容, 打印登录响应内容
+        # dprint(f"{user_info_prefix} 登录响应: {userInfo}\n")
 
         # 随机化经纬度
         if user_info.get("randomLocation", False):
@@ -54,12 +57,12 @@ def send_sign_in(user_info, timeout):
         # 使用上海时间判断是上班还是下班
         signType = "START" if is_start else "END"
         # 检查是否已经打过卡，避免重复打卡, 不需要检查的话可以注释掉整个if块
-        if check_sign(token, signType):
-            print(
-                f"{user_info_prefix}已打过{sign_type_mapping[signType]}卡, 任务跳过\n"
-            )
+        if check_sign(token, signType) and not RepeatCheckIn:
+            dprint(f"{user_info_prefix}")
+            print(f"已打过{sign_type_mapping[signType]}卡, 任务跳过。\n")
             return
-        print(f"{user_info_prefix}准备{signType}打卡\n")
+        dprint(f"{user_info_prefix}", end="")
+        print(f"准备{sign_type_mapping[signType]}打卡\n")
         # 发送请求
         try:
             signResp, msg, full_response = save_user_info(
@@ -75,11 +78,11 @@ def send_sign_in(user_info, timeout):
             )
         except Exception as e:
             error_message = f"{user_info_prefix}打卡失败, 错误原因: {str(e)}"
-            print(error_message)
+            dprint(error_message)
             global_sign_results.append(error_message)
             return
         # 打印完整的打卡响应内容
-        print(f"{user_info_prefix} 打卡响应: {full_response}\n")
+        dprint(f"{user_info_prefix}打卡响应: {full_response}\n")
         # 构建推送消息
         pushSignType = "上班" if signType == "START" else "下班"
         pushSignIsOK = "成功！" if signResp else "失败！"

modules/send_sign_in_threading.py

@@ -1,5 +1,6 @@
 import threading
 from modules.msg_pusher import push_to_pushplus
+from modules.load_config import dprint
 from modules.send_sign_in import send_sign_in, global_sign_results, gmt_time
 
 
@@ -35,4 +36,5 @@ def send_sign_in_threading(users):
         try:
             push_to_pushplus(users_with_same_pushkey, title, content, pushKey)
         except Exception as e:
-            print(f"推送失败, 错误原因: {str(e)}")
+            print(f"推送失败, 错误原因请在主程序设置debug=True后重新运行程序查看。")
+            dprint(f"{str(e)}")