huanghai/aiData
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

'commit'

Browse Source
This commit is contained in:
HuangHai
2026-01-14 07:37:15 +08:00
parent 1bdfde98c6
commit 77f5e80857
10 changed files with 110 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
Apps/AiTeJiYiChong/Kit.py
View File

@@ -160,11 +160,14 @@ def click_image_template(d, template_path, timeout=5.0, threshold=0.8):
return False return False
def setup_logger(name, log_file=None): def setup_logger(name, log_file=None, clear_old_log=False):
""" """
配置日志，支持同时输出到控制台和文件。 配置日志，支持同时输出到控制台和文件。
使用供应商代号作为父级 Logger，所有子 Logger 继承其 Handler， 使用供应商代号作为父级 Logger，所有子 Logger 继承其 Handler，
并通过 propagate=False 避免与根 Logger 重复。 并通过 propagate=False 避免与根 Logger 重复。
:param name: Logger 名称
:param log_file: 指定日志文件路径，如果不指定则使用默认路径
:param clear_old_log: 是否在启动时清空旧日志文件
""" """
# 1. 获取供应商代号 (如 AiTeJiYiChong) # 1. 获取供应商代号 (如 AiTeJiYiChong)
supplier_code = os.path.basename(os.path.dirname(os.path.abspath(__file__))) supplier_code = os.path.basename(os.path.dirname(os.path.abspath(__file__)))
@@ -174,15 +177,26 @@ def setup_logger(name, log_file=None):
parent_logger.setLevel(logging.INFO) parent_logger.setLevel(logging.INFO)
parent_logger.propagate = False # 禁止向上传递给 root logger，防止重复 parent_logger.propagate = False # 禁止向上传递给 root logger，防止重复
if not parent_logger.handlers: if log_file is None:
if log_file is None: # 获取项目根目录 (aiData)
# 获取项目根目录 (aiData) root_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
root_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) log_dir = os.path.join(root_dir, "Logs")
log_dir = os.path.join(root_dir, "Logs") if not os.path.exists(log_dir):
if not os.path.exists(log_dir): os.makedirs(log_dir)
os.makedirs(log_dir) log_file = os.path.join(log_dir, f"{supplier_code}.log")
log_file = os.path.join(log_dir, f"{supplier_code}.log")
# 如果需要清空旧日志且文件存在
if clear_old_log and os.path.exists(log_file):
try:
# 关闭现有的 handler 以便删除文件
for handler in parent_logger.handlers[:]:
handler.close()
parent_logger.removeHandler(handler)
os.remove(log_file)
except Exception as e:
print(f"无法清空旧日志文件 {log_file}: {e}")
if not parent_logger.handlers:
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
# 控制台 Handler # 控制台 Handler

7
Apps/TeLaiDian/Crawler.py
View File

@@ -31,12 +31,9 @@ class TeLaiDianCrawler(BaseCrawler):
async def start(self): async def start(self):
""" """
实现 BaseCrawler 的启动入口 实现 BaseCrawler 的启动入口
注意：在 T4 等全流程脚本中，小程序由 Opener 提前打开，
此处直接开始爬取逻辑。
""" """
success = await self.open_app()
if not success:
logger.error("打开特来电小程序失败。")
return
d = u2.connect() d = u2.connect()
await self.crawl_list_logic(d) await self.crawl_list_logic(d)

36
Apps/TeLaiDian/Kit.py
View File

@@ -193,13 +193,16 @@ def clear_temp_dir(save_dir=None):
except Exception as e: except Exception as e:
logger.error(f"Error deleting file {file_path}: {e}") logger.error(f"Error deleting file {file_path}: {e}")
def setup_logger(name, log_file=None): def setup_logger(name, log_file=None, clear_old_log=False):
""" """
配置日志，支持同时输出到控制台和文件。 配置日志，支持同时输出到控制台和文件。
使用供应商代号作为父级 Logger，所有子 Logger 继承其 Handler， 使用供应商代号作为父级 Logger，所有子 Logger 继承其 Handler，
并通过 propagate=False 避免与根 Logger 重复。 并通过 propagate=False 避免与根 Logger 重复。
:param name: Logger 名称
:param log_file: 指定日志文件路径，如果不指定则使用默认路径
:param clear_old_log: 是否在启动时清空旧日志文件
""" """
# 1. 获取供应商代号 (如 TeLaiDian) # 1. 获取供应商代号 (如 TelaiDian)
supplier_code = os.path.basename(os.path.dirname(os.path.abspath(__file__))) supplier_code = os.path.basename(os.path.dirname(os.path.abspath(__file__)))
# 2. 获取父级 Logger 并配置 # 2. 获取父级 Logger 并配置
@@ -207,15 +210,26 @@ def setup_logger(name, log_file=None):
parent_logger.setLevel(logging.INFO) parent_logger.setLevel(logging.INFO)
parent_logger.propagate = False # 禁止向上传递给 root logger，防止重复 parent_logger.propagate = False # 禁止向上传递给 root logger，防止重复
if not parent_logger.handlers: if log_file is None:
if log_file is None: # 获取项目根目录 (aiData)
# 获取项目根目录 (aiData) root_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
root_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) log_dir = os.path.join(root_dir, "Logs")
log_dir = os.path.join(root_dir, "Logs") if not os.path.exists(log_dir):
if not os.path.exists(log_dir): os.makedirs(log_dir)
os.makedirs(log_dir) log_file = os.path.join(log_dir, f"{supplier_code}.log")
log_file = os.path.join(log_dir, f"{supplier_code}.log")
# 如果需要清空旧日志且文件存在
if clear_old_log and os.path.exists(log_file):
try:
# 关闭现有的 handler 以便删除文件
for handler in parent_logger.handlers[:]:
handler.close()
parent_logger.removeHandler(handler)
os.remove(log_file)
except Exception as e:
print(f"无法清空旧日志文件 {log_file}: {e}")
if not parent_logger.handlers:
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
# 控制台 Handler # 控制台 Handler
@@ -228,7 +242,7 @@ def setup_logger(name, log_file=None):
fh.setFormatter(formatter) fh.setFormatter(formatter)
parent_logger.addHandler(fh) parent_logger.addHandler(fh)
# 3. 返回子 Logger (如 TeLaiDian.T4_TeLaiDian) # 3. 返回子 Logger (如 TelaiDian.T4_TelaiDian)
if name == supplier_code: if name == supplier_code:
return parent_logger return parent_logger
return logging.getLogger(f"{supplier_code}.{name}") return logging.getLogger(f"{supplier_code}.{name}")

32
Apps/XinDianTu/Kit.py
View File

@@ -246,11 +246,14 @@ def detect_rabbit_ad_close(image_path, debug_dir=None):
return [best[3], best[4]] return [best[3], best[4]]
def setup_logger(name, log_file=None): def setup_logger(name, log_file=None, clear_old_log=False):
""" """
配置日志，支持同时输出到控制台和文件。 配置日志，支持同时输出到控制台和文件。
使用供应商代号作为父级 Logger，所有子 Logger 继承其 Handler， 使用供应商代号作为父级 Logger，所有子 Logger 继承其 Handler，
并通过 propagate=False 避免与根 Logger 重复。 并通过 propagate=False 避免与根 Logger 重复。
:param name: Logger 名称
:param log_file: 指定日志文件路径，如果不指定则使用默认路径
:param clear_old_log: 是否在启动时清空旧日志文件
""" """
# 1. 获取供应商代号 (如 XinDianTu) # 1. 获取供应商代号 (如 XinDianTu)
supplier_code = os.path.basename(os.path.dirname(os.path.abspath(__file__))) supplier_code = os.path.basename(os.path.dirname(os.path.abspath(__file__)))
@@ -260,15 +263,26 @@ def setup_logger(name, log_file=None):
parent_logger.setLevel(logging.INFO) parent_logger.setLevel(logging.INFO)
parent_logger.propagate = False # 禁止向上传递给 root logger，防止重复 parent_logger.propagate = False # 禁止向上传递给 root logger，防止重复
if not parent_logger.handlers: if log_file is None:
if log_file is None: # 获取项目根目录 (aiData)
# 获取项目根目录 (aiData) root_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
root_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) log_dir = os.path.join(root_dir, "Logs")
log_dir = os.path.join(root_dir, "Logs") if not os.path.exists(log_dir):
if not os.path.exists(log_dir): os.makedirs(log_dir)
os.makedirs(log_dir) log_file = os.path.join(log_dir, f"{supplier_code}.log")
log_file = os.path.join(log_dir, f"{supplier_code}.log")
# 如果需要清空旧日志且文件存在
if clear_old_log and os.path.exists(log_file):
try:
# 关闭现有的 handler 以便删除文件
for handler in parent_logger.handlers[:]:
handler.close()
parent_logger.removeHandler(handler)
os.remove(log_file)
except Exception as e:
print(f"无法清空旧日志文件 {log_file}: {e}")
if not parent_logger.handlers:
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
# 控制台 Handler # 控制台 Handler

32
Apps/YeLiTe/Kit.py
View File

@@ -270,11 +270,14 @@ def draw_rectangles(image_path, points, output_path=None):
logger.error(f"绘制矩形框失败: {e}") logger.error(f"绘制矩形框失败: {e}")
return image_path return image_path
def setup_logger(name, log_file=None): def setup_logger(name, log_file=None, clear_old_log=False):
""" """
配置日志，支持同时输出到控制台和文件。 配置日志，支持同时输出到控制台和文件。
使用供应商代号作为父级 Logger，所有子 Logger 继承其 Handler， 使用供应商代号作为父级 Logger，所有子 Logger 继承其 Handler，
并通过 propagate=False 避免与根 Logger 重复。 并通过 propagate=False 避免与根 Logger 重复。
:param name: Logger 名称
:param log_file: 指定日志文件路径，如果不指定则使用默认路径
:param clear_old_log: 是否在启动时清空旧日志文件
""" """
# 1. 获取供应商代号 (如 YeLiTe) # 1. 获取供应商代号 (如 YeLiTe)
supplier_code = os.path.basename(os.path.dirname(os.path.abspath(__file__))) supplier_code = os.path.basename(os.path.dirname(os.path.abspath(__file__)))
@@ -284,15 +287,26 @@ def setup_logger(name, log_file=None):
parent_logger.setLevel(logging.INFO) parent_logger.setLevel(logging.INFO)
parent_logger.propagate = False # 禁止向上传递给 root logger，防止重复 parent_logger.propagate = False # 禁止向上传递给 root logger，防止重复
if not parent_logger.handlers: if log_file is None:
if log_file is None: # 获取项目根目录 (aiData)
# 获取项目根目录 (aiData) root_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
root_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) log_dir = os.path.join(root_dir, "Logs")
log_dir = os.path.join(root_dir, "Logs") if not os.path.exists(log_dir):
if not os.path.exists(log_dir): os.makedirs(log_dir)
os.makedirs(log_dir) log_file = os.path.join(log_dir, f"{supplier_code}.log")
log_file = os.path.join(log_dir, f"{supplier_code}.log")
# 如果需要清空旧日志且文件存在
if clear_old_log and os.path.exists(log_file):
try:
# 关闭现有的 handler 以便删除文件
for handler in parent_logger.handlers[:]:
handler.close()
parent_logger.removeHandler(handler)
os.remove(log_file)
except Exception as e:
print(f"无法清空旧日志文件 {log_file}: {e}")
if not parent_logger.handlers:
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
# 控制台 Handler # 控制台 Handler

1
Python运行环境.txt Normal file
View File

@@ -0,0 +1 @@
D:\anaconda3\envs\py310\python.exe

13
T1_XinDianTu.py
View File

@@ -9,21 +9,16 @@ project_root = os.path.dirname(os.path.abspath(__file__))
if project_root not in sys.path: if project_root not in sys.path:
sys.path.append(project_root) sys.path.append(project_root)
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("T1_XinDianTu")
try: try:
from Apps.XinDianTu.Service import XinDianTuService from Apps.XinDianTu.Service import XinDianTuService
from Apps.XinDianTu import Opener, Crawler, Kit from Apps.XinDianTu import Opener, Crawler, Kit
# 配置日志
logger = Kit.setup_logger("T1_XinDianTu", clear_old_log=True)
except KeyboardInterrupt: except KeyboardInterrupt:
logger.info("\n🛑 用户在初始化阶段手动停止了程序。") print("\n🛑 用户在初始化阶段手动停止了程序。")
sys.exit(0) sys.exit(0)
except Exception as e: except Exception as e:
logger.error(f"❌ 初始化导入失败: {e}") print(f"❌ 初始化导入失败: {e}")
sys.exit(1) sys.exit(1)
async def run_process(): async def run_process():

13
T2_AiTeJiYiChong.py
View File

@@ -9,21 +9,16 @@ project_root = os.path.dirname(os.path.abspath(__file__))
if project_root not in sys.path: if project_root not in sys.path:
sys.path.append(project_root) sys.path.append(project_root)
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("T2_AiTeJiYiChong")
try: try:
from Apps.AiTeJiYiChong.Service import AiTeJiYiChongService from Apps.AiTeJiYiChong.Service import AiTeJiYiChongService
from Apps.AiTeJiYiChong import Opener, Crawler, Kit from Apps.AiTeJiYiChong import Opener, Crawler, Kit
# 配置日志
logger = Kit.setup_logger("T2_AiTeJiYiChong", clear_old_log=True)
except KeyboardInterrupt: except KeyboardInterrupt:
logger.info("\n🛑 用户在初始化阶段手动停止了程序。") print("\n🛑 用户在初始化阶段手动停止了程序。")
sys.exit(0) sys.exit(0)
except Exception as e: except Exception as e:
logger.error(f"❌ 初始化导入失败: {e}") print(f"❌ 初始化导入失败: {e}")
sys.exit(1) sys.exit(1)
async def run_process(): async def run_process():

13
T3_YeLiTe.py
View File

@@ -9,21 +9,16 @@ project_root = os.path.dirname(os.path.abspath(__file__))
if project_root not in sys.path: if project_root not in sys.path:
sys.path.append(project_root) sys.path.append(project_root)
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("T3_YeLiTe")
try: try:
from Apps.YeLiTe.Service import YiLaiTeService from Apps.YeLiTe.Service import YiLaiTeService
from Apps.YeLiTe import Opener, Crawler, Kit from Apps.YeLiTe import Opener, Crawler, Kit
# 配置日志
logger = Kit.setup_logger("T3_YeLiTe", clear_old_log=True)
except KeyboardInterrupt: except KeyboardInterrupt:
logger.info("\n🛑 用户在初始化阶段手动停止了程序。") print("\n🛑 用户在初始化阶段手动停止了程序。")
sys.exit(0) sys.exit(0)
except Exception as e: except Exception as e:
logger.error(f"❌ 初始化导入失败: {e}") print(f"❌ 初始化导入失败: {e}")
sys.exit(1) sys.exit(1)
async def run_process(): async def run_process():

2
T4_TeLaiDian.py
View File

@@ -13,7 +13,7 @@ try:
from Apps.TeLaiDian.Service import TeLaiDianService from Apps.TeLaiDian.Service import TeLaiDianService
from Apps.TeLaiDian import Opener, Crawler, Kit from Apps.TeLaiDian import Opener, Crawler, Kit
# 初始化日志文件记录 # 初始化日志文件记录
logger = Kit.setup_logger("T4_TeLaiDian") logger = Kit.setup_logger("T4_TeLaiDian", clear_old_log=True)
except KeyboardInterrupt: except KeyboardInterrupt:
print("\n🛑 用户在初始化阶段手动停止了程序。") print("\n🛑 用户在初始化阶段手动停止了程序。")
sys.exit(0) sys.exit(0)